大数据与家居：智能家居数据分析

    import pandas as pd
    import numpy as np
    from scipy import stats
    
    def preprocess_smart_home_data(raw_data):
    # 处理缺失值
    data = raw_data.interpolate(method='time')
    
    # 处理异常值
    z_scores = np.abs(stats.zscore(data.select_dtypes(include=[np.number])))
    data = data[(z_scores < 3).all(axis=1)]
    
    # 时间序列重采样
    data = data.resample('1T').mean()  # 按分钟重采样
    
    # 特征标准化
    numeric_cols = data.select_dtypes(include=[np.number]).columns
    data[numeric_cols] = (data[numeric_cols] - data[numeric_cols].mean()) / data[numeric_cols].std()
    
    return data
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/WRkqzVuc7a2pSZBDNed6m0gtXGfx.png)

    from sklearn.cluster import KMeans
    from sklearn.preprocessing import StandardScaler
    
    def detect_device_states(power_consumption):
    # 转换为适合聚类的格式
    X = power_consumption.values.reshape(-1, 1)
    X = StandardScaler().fit_transform(X)
    
    # 使用K-means聚类
    kmeans = KMeans(n_clusters=3, random_state=42)
    states = kmeans.fit_predict(X)
    
    # 分析聚类中心确定状态
    centers = kmeans.cluster_centers_.flatten()
    state_labels = ['关闭', '待机', ''] if centers[0] < centers[1] else ['', '待机', '关闭']
    
    return pd.Series(states).map(dict(enumerate(state_labels)))
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/YAaJplgDWRHiB9ZkexLC07yNobUK.png)

    from prefixspan import PrefixSpan
    
    def discover_behavior_patterns(device_events, min_support=10):
    # 转换事件数据为序列格式
    sequences = device_events.groupby('user_id')['device_action'].apply(list).tolist()
    
    # 使用PrefixSpan算法
    ps = PrefixSpan(sequences)
    patterns = ps.frequent(min_support)
    
    # 分析常见模式
    behavior_patterns = []
    for (support, pattern) in patterns:
        if len(pattern) > 1:  # 忽略单一事件模式
            behavior_patterns.append({
                'pattern': pattern,
                'support': support,
                'confidence': support / len(sequences)
            })
    
    return sorted(behavior_patterns, key=lambda x: -x['confidence'])
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/iPOcWqBpG1tUdXNCLaRuIlDvjwgz.png)

 * 边缘节点：树莓派4B或NVIDIA Jetson Nano
 * 服务器：至少16GB内存，多核CPU

    # 创建conda环境

    conda create -n smart_home python=3.8
    conda activate smart_home
    
    # 安装核心包
    pip install numpy pandas scikit-learn tensorflow pymongo influxdb
    
    
    bash

 * 时序数据：InfluxDB
 * 非结构化数据：MongoDB
 * 分析结果：PostgreSQL

    from kafka import KafkaConsumer
    import json
    from influxdb import InfluxDBClient
    
    class RealTimePipeline:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'smart_home',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        self.influx = InfluxDBClient(host='localhost', port=8086)
        self.influx.switch_database('iot_data')
    
    def process_message(self, msg):
        # 基础数据验证
        required_fields = {'device_id', 'timestamp', 'value'}
        if not required_fields.issubset(msg.keys()):
            return False
        
        # 构建InfluxDB数据点
        point = {
            "measurement": msg['device_id'],
            "time": msg['timestamp'],
            "fields": {
                "value": float(msg['value'])
            }
        }
        
        # 写入数据库
        self.influx.write_points([point])
        return True
    
    def run(self):
        for message in self.consumer:
            self.process_message(message.value)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/jZ3ocaQFA0wMKmkdzRH84pXUT97e.png)

    import dash
    from dash import dcc, html
    import plotly.express as px
    import pandas as pd
    from influxdb import DataFrameClient
    
    app = dash.Dash(__name__)
    
    def fetch_energy_data():
    client = DataFrameClient(host='localhost', port=8086)
    client.switch_database('iot_data')
    
    query = """
    SELECT mean("value") as "power_usage"
    FROM "smart_plug"
    WHERE time > now() - 7d
    GROUP BY time(1h), "room"
    """
    
    result = client.query(query)
    return pd.concat(result.values(), keys=result.keys())
    
    app.layout = html.Div([
    html.H1("智能家居能耗分析"),
    dcc.Graph(
        id='energy-usage',
        figure=px.line(fetch_energy_data(), y='power_usage', title='过去一周能耗趋势')
    ),
    dcc.Interval(
        id='interval-component',
        interval=60*1000,  # 1分钟刷新
        n_intervals=0
    )
    ])
    
    if __name__ == '__main__':
    app.run_server(debug=True)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-17/MbhAL1GR9DfWtUwaXH6C4PEO2xyK.png)

 * 使用Kafka作为消息队列，处理高吞吐量设备数据
 * InfluxDB作为时序数据库，优化时间序列查询
 * 实现了基本的数据验证和转换逻辑

 * 基于Dash框架构建交互式可视化界面
 * 从InfluxDB查询并聚合能耗数据
 * 自动刷新机制保持数据实时性

 * 模块化设计便于添加新的数据源和分析功能
 * 使用标准协议和接口确保系统兼容性
 * 资源高效利用的设计适合边缘部署