数据概述

        记录了英国5个家庭，每个家庭至少几个月的数据，1号家庭有超过4年的数据，另一方面，它同时有低频6s采样的数据和高频16khz的采样数据（高频只有电表的）

        简介:        

                该数据集记录了五栋房屋的电力需求。在每所房屋中，我们每六秒记录一次全屋主电源需求以及每六秒对单个电器的电力需求。在五个房屋中的三个（房屋 1、2 和 5）中，我们还记录了 16 kHz 的全屋电压和电流。

        

方法1 OpenDataLab 推荐 .dat

https://opendatalab.com/OpenDataLab/UK-DALE

两个tar的区别是采样频率不同,下面频率低

数据如下:

# 数据解释
这是 UK-DALE数据集的disaggregated（分解）版本，采用了 .dat格式 存储

channel_X.dat - 设备X的功耗数据
channel_X_button_press.dat - 设备X的按钮事件数据  //下面h5版本貌似没这个数据

# 数据对应关系
```
channel_1.dat → 设备1的功耗时间序列
channel_2.dat → 设备2的功耗时间序列  
channel_4_button_press.dat → 设备4的按钮事件
channel_5_button_press.dat → 设备5的按钮事件
```

方法2 kaggle h5

https://www.kaggle.com/datasets/abdelmdz/uk-dale

数据内容:

# 训练阶段
```
# 有标签的监督学习
输入: 总电表功耗 (主电表 instance=54)
标签: 各设备功耗 (子电表 instance=2,3,4...53)

训练数据 = {
    'total_power': [总功耗时间序列],
    'appliance_1': [冰箱功耗时间序列],
    'appliance_2': [洗衣机功耗时间序列],
    'appliance_3': [电视功耗时间序列],
    ...
}
```

# 推理阶段
```
# 只需要总电表数据
输入: 新房子的总电表功耗
输出: 预测的各设备功耗分解

model.predict(total_power) → {
    'fridge': [预测冰箱功耗],
    'washer': [预测洗衣机功耗], 
    'tv': [预测电视功耗],
    ...
}
```

电表包含的设备

• 总设备数量: 66台设备 | 电表数量: 53个电表 | 最多设备: 照明系统(16个)

数据时间

数据特征列

只有功率一列:

Columns: [('power', 'apparent')]
   ('power', 'apparent')
0                  768.0
1                  770.0
2                  777.0
3                  772.0
4                  770.0

数据集对比

上面也能看出来 他只有功率,缺乏无功功率等多种特征,下面给了开源数据集对比

中文版:

h5的读取 训练 推理 脚本

"""
NILM（非侵入式负载监测）示例，基于 NILMTK。
数据集：UK-DALE（公开数据集），请先转换为 HDF5。

依赖：
  pip install nilmtk
  （可选）下载 UK-DALE：https://data.ukedc.rl.ac.uk/browse/edc/UKDALE
"""

from pathlib import Path
import h5py
import duckdb
import pandas as pd

from nilmtk import DataSet
from nilmtk.disaggregate import CO


def safe_explore_h5(filename):
    print(f"探索文件: {filename}\n")

    with h5py.File(filename, 'r') as f:
        print("🗂️ 顶层结构:")
        for key in f.keys():
            item = f[key]
            if isinstance(item, h5py.Group):
                print(f"  📁 {key}/ (Group)")
            else:
                print(f"  📄 {key} (Dataset: {item.shape})")

        # 专门查看building1
        if 'building1' in f:
            building = f['building1']
            print(f"\n🏠 Building1 结构:")

            for key in building.keys():
                print(f"  📁 {key}/")

                if key == 'elec' and isinstance(building[key], h5py.Group):
                    elec = building[key]
                    print(f"    电表数量: {len([k for k in elec.keys() if k.startswith('meter')])}")

                    # 只看前几个meter作为示例
                    meter_keys = [k for k in elec.keys() if k.startswith('meter')][:3]
                    for meter_key in meter_keys:
                        print(f"    📊 {meter_key}:")
                        meter = elec[meter_key]
                        for sub_key in meter.keys():
                            sub_item = meter[sub_key]
                            if isinstance(sub_item, h5py.Dataset):
                                print(f"      📈 {sub_key}: {sub_item.shape}")
                            else:
                                print(f"      📁 {sub_key}/")
def show_h5_head(h5_path: Path) -> None:
    # 使用 duckdb 展示 HDF5 的前五行。
    with pd.HDFStore(h5_path) as store:
        keys = store.keys()
        print(f'keys in HDF5: {keys}')
        if not keys:
            raise RuntimeError("HDF5 中没有可用的 key。")
        key = keys[0]
        df = store.select(key, start=0, stop=5)

    con = duckdb.connect()
    con.register("h5data", df)
    print(f"HDF5 key: {key}")
    print(f"Columns: {list(df.columns)}")
    print(con.execute("SELECT * FROM h5data").fetchdf())


def show_detailed_features(h5_path):
    dataset = DataSet(h5_path)
    try:
        building = dataset.buildings[1]

        # 查看几个代表性电表
        test_meters = [1, 10, 12, 54]  # 包含主电表和几个子电表
        meter_ids = set()
        for m in building.elec.meters:
            mid = m.instance() if callable(m.instance) else m.instance
            meter_ids.add(mid)

        for meter_id in test_meters:
            if meter_id not in meter_ids:
                print(f"⚠️ Meter {meter_id} 不存在，跳过")
                continue

            print(f"\n{'=' * 50}")
            print(f"Meter {meter_id} 详细特征")
            print(f"{'=' * 50}")

            meter = building.elec[meter_id]

            # 显示电器信息
            if meter.appliances:
                appliances = [f"{app.type}" for app in meter.appliances]
                print(f"🔌 电器: {', '.join(appliances)}")

            if meter.is_site_meter():
                print("🏠 主电表")

            # 获取所有可用的物理量
            available_measurements = meter.available_physical_quantities()
            print(f"📊 可用测量类型: {list(available_measurements)}")

            # 查看每种测量类型的详细结构
            for measurement in available_measurements:
                try:
                    print(f"\n📈 {measurement.upper()} 数据:")

                    # 加载少量数据查看结构
                    data_gen = meter.load(
                        physical_quantity=measurement,
                        sample_period=60,  # 1分钟采样
                        chunksize=1000  # 只加载1000个点
                    )

                    chunk = next(data_gen)
                    if chunk is None or chunk.empty:
                        print(f"   ❌ {measurement}: 无可用数据")
                        continue

                    print(f"   ✅ 数据形状: {chunk.shape}")
                    print(f"   ✅ 列名结构: {chunk.columns.tolist()}")
                    print(f"   ✅ 索引类型: {type(chunk.index[0])}")
                    print(f"   ✅ 时间范围: {chunk.index[0]} 到 {chunk.index[-1]}")

                    # 显示统计信息
                    print("   ✅ 数据统计:")
                    stats = chunk.describe()
                    print(stats)

                    # 显示前几行数据
                    print("   ✅ 前3行数据:")
                    print(chunk.head(3))

                except StopIteration:
                    print(f"   ❌ {measurement}: 无可用数据")
                except Exception as e:
                    print(f"   ❌ {measurement}: 读取失败 - {e}")
    finally:
        dataset.store.close()

def main():
    # 优先使用 data/ 下的转换版 HDF5；若不存在则回退到原始目录中的副本。
    root = Path(__file__).resolve().parents[1]
    h5_path = root / "data" / "ukdale.h5"
    if not h5_path.exists():
        alt_path = root / "data" / "ukdale_raw" / "ukdale.h5"
        if alt_path.exists():
            h5_path = alt_path
        else:
            raise FileNotFoundError(
                "Missing data/ukdale.h5. "
                "Convert UK-DALE to HDF5 before running this example."
            )
    #

    show_h5_head(h5_path)
    print('---------h5 features---------')
    safe_explore_h5(h5_path)
    print('-----')
    show_detailed_features(h5_path)
    print('---------h5 features---------')

    dataset = DataSet(str(h5_path))
    # 打印所有电表信息
    print(dataset.buildings[1].elec.meters,'------meters------')
    try:
        # 选择总电表有数据的时间窗口。
        dataset.set_window(start="2013-03-20", end="2013-03-27")

        # 建筑 1 的总表（汇总）用电数据。
        elec = dataset.buildings[1].elec
        mains = elec.mains()

        # 选择几个目标电器。
        desired = ["fridge freezer", "kettle", "microwave"]
        available = set()
        for app in elec.appliances:
            t = app.type
            if isinstance(t, dict) and "type" in t:
                available.add(t["type"])
            else:
                available.add(str(t))
        target_appliances = [a for a in desired if a in available]
        if not target_appliances:
            raise RuntimeError(
                "No target appliances found in building 1. "
                f"Available examples: {sorted(list(available))[:10]}"
            )
        appliances = elec.select_using_appliances(type=target_appliances)

        # 准备训练数据。
        sample_period = 6
        mains_series = mains.power_series_all_data(sample_period=sample_period)
        train_main = [mains_series.to_frame()] if mains_series is not None else [None]
        train_appliances = []
        for meter in appliances.meters:
            series = meter.power_series_all_data(sample_period=sample_period)
            if series is None or series.empty:
                continue
            app = meter.appliances[0] if meter.appliances else None
            if app is None:
                continue
            t = app.type
            app_name = t["type"] if isinstance(t, dict) and "type" in t else str(t)
            train_appliances.append((app_name, [series.to_frame()]))

        if train_main[0] is not None and not train_main[0].empty and train_appliances:
            # ===== 模型训练（学习电器功率状态）=====
            co = CO(params={})
            co.partial_fit(train_main, train_appliances)

            # ===== 模型推理（把总功率分解成各电器功率）=====
            disagg_list = co.disaggregate_chunk(train_main)
            disagg = disagg_list[0]
            app_names = [m["appliance_name"] for m in co.model]
            disagg.columns = app_names

            # 保存结果为 CSV 便于查看。
            output_path = Path("data/ukdale_disag.csv")
            output_path.parent.mkdir(parents=True, exist_ok=True)
            disagg.to_csv(output_path)

            print(f"Done. Disaggregation saved to {output_path}")
        else:
            raise RuntimeError("Training data is empty for selected appliances.")
    finally:
        # 关闭 HDF5，避免 PyTables 的未关闭告警。
        dataset.store.close()


if __name__ == "__main__":
    main()

输出到ukdale_disag.csv内容:

代码中: ["fridge freezer", "kettle", "microwave"]  即代码只挑选了这三种电器作为训练和推理.

内容解释:

代码使用的算法:  组合优化 (CO)算法

co算法简单解释

CO算法的根本问题：

• 🔸 过度简化：将复杂设备建模为简单开关
• 🔸 忽略时间：不考虑时间序列模式
• 🔸 单一特征：仅使用有功功率
• 🔸 独立假设：忽略设备间相关性