#!/usr/bin/env python3
import paho.mqtt.client as mqtt
import json
import random
import time
import datetime

class TemperatureSensor:
    def __init__(self, device_id, broker_host="localhost", broker_port=1883):
        self.device_id = device_id
        self.broker_host = broker_host
        self.broker_port = broker_port
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.topic = f"sensors/temperature/{device_id}"
        self.running = False

    def connect(self):
        try:
            self.client.connect(self.broker_host, self.broker_port, 60)
            print(f"[温度传感器 {self.device_id}] 已连接到 MQTT Broker")
            return True
        except Exception as e:
            print(f"[温度传感器 {self.device_id}] 连接失败: {e}")
            return False

    def generate_temperature_data(self):
        base_temp = 20.0
        variation = random.uniform(-10, 10)
        temperature = round(base_temp + variation, 2)
        # 5% 概率产生异常数据
        if random.random() < 0.05:
            temperature += random.uniform(15, 25)
        return {
            "device_id": self.device_id,
            "sensor_type": "temperature",
            "value": temperature,
            "unit": "celsius",
            "timestamp": datetime.datetime.now().isoformat(),
            "location": {"lat": 39.9042, "lon": 116.4074}
        }

    def run(self):
        self.running = True
        while self.running:
            try:
                data = self.generate_temperature_data()
                self.client.publish(self.topic, json.dumps(data))
                time.sleep(2)
            except Exception as e:
                print(f"[温度传感器 {self.device_id}] 发送失败: {e}")
                time.sleep(5)

    def stop(self):
        self.running = False
        self.client.disconnect()

#!/usr/bin/env python3
import threading
import time
import signal
import sys
from temperature_sensor import TemperatureSensor
from humidity_sensor import HumiditySensor
from light_sensor import LightSensor
from pressure_sensor import PressureSensor
from motion_sensor import MotionSensor
from dust_sensor import DustSensor

devices = []
threads = []

def create_sensors():
    sensors = [
        TemperatureSensor("temp_001"), TemperatureSensor("temp_002"),
        HumiditySensor("humid_001"),   HumiditySensor("humid_002"),
        LightSensor("light_001"),
        PressureSensor("press_001"),
        MotionSensor("motion_001"),
        DustSensor("dust_001"),
    ]
    for sensor in sensors:
        if sensor.connect():
            devices.append(sensor)
    print(f"共启动 {len(devices)} 个传感器设备")

def start_sensors():
    for device in devices:
        t = threading.Thread(target=device.run, daemon=True)
        t.start()
        threads.append(t)
    print("所有传感器已开始发送数据")

def stop_sensors(signum=None, frame=None):
    print("\n正在停止所有传感器...")
    for device in devices:
        device.stop()
    sys.exit(0)

if __name__ == "__main__":
    signal.signal(signal.SIGINT, stop_sensors)
    signal.signal(signal.SIGTERM, stop_sensors)
    create_sensors()
    start_sensors()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        stop_sensors()

def normalize(self, value, sensor_type):
    """各传感器数值归一化到 [0,1]"""
    ranges = {
        "temperature": (0, 50),   "humidity": (0, 100),
        "light": (0, 2000),       "pressure": (950, 1050),
        "motion": (0, 50),        "dust": (0, 500),
    }
    lo, hi = ranges.get(sensor_type, (0, 100))
    return (value - lo) / (hi - lo) if hi != lo else 0.0

def classify(self, value, sensor_type):
    """基于阈值五级分类"""
    thresholds = {
        "temperature": (-10, 10, 40, 80),     # 低温/正常/高温/超高温
        "humidity":    (0, 30, 70, 100),       # 过干/偏干/正常/偏湿/过湿
        "light":       (0, 300, 1000, 2000),   # 暗/较暗/正常/亮/过亮
        "pressure":    (900, 980, 1030, 1100), # 低压/偏低/正常/偏高/高压
        "motion":      (0, 10, 30, 50),        # 静止/少量/正常/频繁/异常
        "dust":        (0, 75, 150, 500),      # 优/良/轻度/中度/重度
    }
    t = thresholds.get(sensor_type)
    if t is None: return "normal"
    if value < t[0]: return "critical_low"
    elif value < t[1]: return "low"
    elif value < t[2]: return "normal"
    elif value < t[3]: return "high"
    else: return "critical_high"

# 核心检测逻辑
class AnomalyDetector:
    def __init__(self):
        self.history = defaultdict(list)   # 滑动窗口
        self.window_size = 20

    def on_message(self, client, userdata, msg):
        data = json.loads(msg.payload.decode())
        key = f"{data['sensor_type']}/{data['device_id']}"
        self.history[key].append(data['normalized'])
        if len(self.history[key]) > self.window_size:
            self.history[key].pop(0)

        alerts = []
        # 1. 阈值检测
        if data['status'] in ("critical_high", "critical_low"):
            alerts.append({"type": "threshold_violation", "severity": "critical", ...})
        elif data['status'] in ("high", "low"):
            alerts.append({"type": "threshold_violation", "severity": "warning", ...})

        # 2. 统计异常
        arr = np.array(self.history[key])
        mean, std = np.mean(arr), np.std(arr)
        if std > 0 and abs(data['normalized'] - mean) > 2 * std:
            alerts.append({"type": "statistical_outlier", "severity": "warning", ...})

        # 3. 趋势异常
        recent = np.array(self.history[key][-5:])
        trend = np.polyfit(range(len(recent)), recent, 1)[0]
        if abs(trend) > 0.1:
            alerts.append({"type": "trend_anomaly", "severity": "warning", ...})

class CloudAggregator:
    def __init__(self):
        self.data_buffer = defaultdict(list)
        self.aggregation_window = 10  # 秒

    def aggregate_and_publish(self):
        for sensor_type, records in self.data_buffer.items():
            values = [r["value"] for r in records]
            aggregated = {
                "sensor_type": sensor_type,
                "count": int(len(records)),
                "device_count": int(len(set(r["device_id"] for r in records))),
                "avg": float(round(np.mean(values), 2)),
                "max": float(round(np.max(values), 2)),
                "min": float(round(np.min(values), 2)),
                "std": float(round(np.std(values), 2)),
                "avg_normalized": float(round(np.mean([r["normalized"] for r in records]), 4)),
                "timestamp": datetime.datetime.now().isoformat(),
                "window_seconds": int(self.aggregation_window)
            }
            self.client.publish(f"cloud/aggregated/{sensor_type}", json.dumps(aggregated))
        self.data_buffer.clear()

self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

# 错误
f"norm={normalized[normalized]:.2f}"
# 正确
f"norm={normalized['normalized']:.2f}"

# numpy int64 -> Python int
"count": int(len(records))
# numpy float64 -> Python float
"avg": float(round(np.mean(values), 2))