diff --git a/main.py b/main.py index 0d5632a8..3c3d4e37 100644 --- a/main.py +++ b/main.py @@ -16,6 +16,7 @@ from src.websocket.websocket_server import websocket_server from src.batch.batch_import import batch_manager from src.utils.data_utils import data_converter, data_formatter, quality_checker from src.models.models import validator +from src.iot.app import create_iot_app # 配置日志 logging.basicConfig( @@ -70,9 +71,24 @@ class WaterManagementSystem: import uvicorn logger.info("启动REST API服务器...") - # 在新的事件循环中运行uvicorn + # 创建组合应用（包含所有模块） + from werkzeug.middleware.dispatcher import DispatcherMiddleware + from werkzeug.serving import WSGIRequestHandler + + # IoT应用 + iot_app = create_iot_app() + + # 组合所有应用 + def combined_app(environ, start_response): + path = environ.get('PATH_INFO', '') + if path.startswith('/api/iot/'): + return iot_app(environ, start_response) + else: + return rest_api_app(environ, start_response) + + # 配置uvicorn api_config = uvicorn.Config( - app=rest_api_app, + app=combined_app, host=self.config["api"]["host"], port=self.config["api"]["port"], log_level="info" @@ -264,7 +280,9 @@ def create_requirements(): "openpyxl==3.1.2", "aiofiles==23.2.1", "python-multipart==0.0.6", - "jinja2==3.1.2" + "jinja2==3.1.2", + "paho-mqtt==1.6.1", + "flask==2.3.3" ] with open("requirements.txt", 'w') as f: diff --git a/src/iot/__init__.py b/src/iot/__init__.py new file mode 100644 index 00000000..23087e06 --- /dev/null +++ b/src/iot/__init__.py @@ -0,0 +1,9 @@ +""" +IoT Module - 物联网平台核心模块 +包含MQTT协议适配器、设备注册/发现API、统一设备模型等 +""" + +from .device_manager import DeviceManager +from .device_controller import DeviceController + +__all__ = ['DeviceManager', 'DeviceController'] \ No newline at end of file diff --git a/src/iot/__pycache__/__init__.cpython-312.pyc b/src/iot/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..a705e0d1 Binary files /dev/null and b/src/iot/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/iot/__pycache__/device_controller.cpython-312.pyc b/src/iot/__pycache__/device_controller.cpython-312.pyc new file mode 100644 index 00000000..6659a6df Binary files /dev/null and b/src/iot/__pycache__/device_controller.cpython-312.pyc differ diff --git a/src/iot/__pycache__/device_manager.cpython-312.pyc b/src/iot/__pycache__/device_manager.cpython-312.pyc new file mode 100644 index 00000000..87bc9a9a Binary files /dev/null and b/src/iot/__pycache__/device_manager.cpython-312.pyc differ diff --git a/src/iot/__pycache__/models.cpython-312.pyc b/src/iot/__pycache__/models.cpython-312.pyc new file mode 100644 index 00000000..4f20d203 Binary files /dev/null and b/src/iot/__pycache__/models.cpython-312.pyc differ diff --git a/src/iot/__pycache__/mqtt_adapter.cpython-312.pyc b/src/iot/__pycache__/mqtt_adapter.cpython-312.pyc new file mode 100644 index 00000000..3e0c0b15 Binary files /dev/null and b/src/iot/__pycache__/mqtt_adapter.cpython-312.pyc differ diff --git a/src/iot/app.py b/src/iot/app.py new file mode 100644 index 00000000..71c18bea --- /dev/null +++ b/src/iot/app.py @@ -0,0 +1,177 @@ +""" +IoT模块Flask应用 +集成MQTT适配器、设备管理器、OTA管理等核心组件 +""" + +import logging +import os +from flask import Flask +from .device_manager import DeviceManager +from .mqtt_adapter import MqttAdapter +from .device_controller import DeviceController +from .ota_manager import OtaManager +from .ota_controller import OtaController +from .config import MqttConfig, DatabaseConfig + + +def create_iot_app(config=None): + """ + 创建IoT模块Flask应用 + + Args: + config: 配置字典 + + Returns: + Flask: Flask应用实例 + """ + app = Flask(__name__) + + # 配置日志 + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + logger = logging.getLogger(__name__) + + # 加载配置 + if config is None: + config = { + 'mqtt': MqttConfig.from_env(), + 'database': DatabaseConfig.from_env() + } + + # 初始化组件 + device_manager = DeviceManager() + mqtt_adapter = MqttAdapter( + broker_host=config['mqtt'].broker_host, + broker_port=config['mqtt'].broker_port, + username=config['mqtt'].username, + password=config['mqtt'].password, + client_id=config['mqtt'].client_id + ) + ota_manager = OtaManager() + + # 创建控制器 + device_controller = DeviceController(device_manager, mqtt_adapter) + ota_controller = OtaController(ota_manager) + + # 注册蓝图 + app.register_blueprint(device_controller.get_blueprint()) + app.register_blueprint(ota_controller.get_blueprint()) + + @app.route('/api/iot/status') + def get_status(): + """获取IoT模块状态""" + return { + "mqtt": mqtt_adapter.get_connection_status(), + "device_statistics": device_manager.get_device_statistics(), + "ota_statistics": ota_manager.get_update_statistics(), + "devices_count": len(device_manager.devices), + "shadows_count": len(device_manager.shadows) + } + + @app.route('/api/iot/mqtt/connect', methods=['POST']) + def connect_mqtt(): + """连接MQTT""" + try: + success = mqtt_adapter.connect() + return { + "success": success, + "message": "MQTT connected" if success else "MQTT connection failed" + } + except Exception as e: + logger.error(f"MQTT connection error: {e}") + return {"success": False, "error": str(e)} + + @app.route('/api/iot/mqtt/disconnect', methods=['POST']) + def disconnect_mqtt(): + """断开MQTT连接""" + try: + mqtt_adapter.disconnect() + return {"success": True, "message": "MQTT disconnected"} + except Exception as e: + logger.error(f"MQTT disconnection error: {e}") + return {"success": False, "error": str(e)} + + @app.route('/api/iot/mqtt/reconnect', methods=['POST']) + def reconnect_mqtt(): + """重连MQTT""" + try: + mqtt_adapter.disconnect() + success = mqtt_adapter.connect() + return { + "success": success, + "message": "MQTT reconnected" if success else "MQTT reconnection failed" + } + except Exception as e: + logger.error(f"MQTT reconnection error: {e}") + return {"success": False, "error": str(e)} + + @app.route('/api/iot/mqtt/publish', methods=['POST']) + def publish_message(): + """发布MQTT消息""" + try: + data = request.get_json() + topic = data.get('topic') + payload = data.get('payload', {}) + qos = data.get('qos', 0) + retain = data.get('retain', False) + + success = mqtt_adapter.publish(topic, payload, qos, retain) + return { + "success": success, + "message": "Message published" if success else "Message publish failed" + } + except Exception as e: + logger.error(f"MQTT publish error: {e}") + return {"success": False, "error": str(e)} + + @app.route('/api/iot/initialize', methods=['POST']) + def initialize_iot(): + """初始化IoT模块""" + try: + # 连接MQTT + mqtt_success = mqtt_adapter.connect() + + # 订阅设备主题 + mqtt_adapter.subscribe_device_topics(device_manager) + + # 初始化一些示例设备 + if len(device_manager.devices) == 0: + sample_devices = [ + { + 'device_sn': 'LL-001', + 'device_type': 'flow_meter', + 'name': '流量计-001', + 'description': 'A区入口流量计', + 'area': 'A区', + 'position': '入口处', + 'manufacturer': '华为', + 'model': 'LL-100' + }, + { + 'device_sn': 'YL-001', + 'device_type': 'pressure_meter', + 'name': '压力表-001', + 'description': 'B区主压力表', + 'area': 'B区', + 'position': '主管道', + 'manufacturer': '西门子', + 'model': 'YL-200' + } + ] + + for device_data in sample_devices: + device_manager.register_device(device_data) + + return { + "success": True, + "mqtt_connected": mqtt_success, + "devices_count": len(device_manager.devices), + "message": "IoT module initialized successfully" + } + except Exception as e: + logger.error(f"IoT initialization error: {e}") + return {"success": False, "error": str(e)} + + return app \ No newline at end of file diff --git a/src/iot/config.py b/src/iot/config.py new file mode 100644 index 00000000..0e52ee12 --- /dev/null +++ b/src/iot/config.py @@ -0,0 +1,189 @@ +""" +IoT模块配置 +包含MQTT配置、数据库配置、设备类型映射等 +""" + +import os +from dataclasses import dataclass + + +@dataclass +class MqttConfig: + """MQTT配置""" + broker_host: str = "localhost" + broker_port: int = 1883 + username: str = "" + password: str = "" + client_id: str = "water-management-system" + keep_alive: int = 60 + clean_session: bool = True + + @classmethod + def from_env(cls): + """从环境变量加载配置""" + return cls( + broker_host=os.getenv('MQTT_BROKER_HOST', 'localhost'), + broker_port=int(os.getenv('MQTT_BROKER_PORT', 1883)), + username=os.getenv('MQTT_USERNAME', ''), + password=os.getenv('MQTT_PASSWORD', ''), + client_id=os.getenv('MQTT_CLIENT_ID', 'water-management-system'), + keep_alive=int(os.getenv('MQTT_KEEP_ALIVE', 60)), + clean_session=os.getenv('MQTT_CLEAN_SESSION', 'true').lower() == 'true' + ) + + +@dataclass +class DatabaseConfig: + """数据库配置""" + host: str = "localhost" + port: int = 3306 + database: str = "water_management" + username: str = "root" + password: str = "" + + @classmethod + def from_env(cls): + """从环境变量加载配置""" + return cls( + host=os.getenv('DB_HOST', 'localhost'), + port=int(os.getenv('DB_PORT', 3306)), + database=os.getenv('DB_NAME', 'water_management'), + username=os.getenv('DB_USER', 'root'), + password=os.getenv('DB_PASSWORD', '') + ) + + +@dataclass +class DeviceTypeMapping: + """设备类型映射""" + # 水利行业标准设备类型映射 + MAPPINGS = { + # 流量计 + 'LL': 'flow_meter', # 流量计 + 'Q': 'flow_meter', # 流量 + 'QL': 'flow_meter', # 瞬时流量 + + # 压力表 + 'YL': 'pressure_meter', # 压力表 + 'P': 'pressure_meter', # 压力 + 'PL': 'pressure_meter', # 瞬时压力 + + # 水位计 + 'SW': 'level_meter', # 水位计 + 'L': 'level_meter', # 水位 + 'WL': 'level_meter', # 瞬时水位 + + # 水质仪 + 'ZD': 'quality_meter', # 浊度 + 'PH': 'quality_meter', # pH值 + 'DO': 'quality_meter', # 溶解氧 + 'COD': 'quality_meter', # 化学需氧量 + 'NH3': 'quality_meter', # 氨氮 + 'TEMP': 'quality_meter', # 温度 + + # 阀门 + 'FV': 'valve', # 阀门 + 'BV': 'valve', # 球阀 + 'GV': 'valve', # 闸阀 + + # 水泵 + 'PUMP': 'pump', # 水泵 + 'CP': 'pump', # 循环泵 + 'FP': 'pump', # 给水泵 + + # 传感器 + 'SENSOR': 'sensor', # 传感器 + 'TS': 'sensor', # 温度传感器 + 'HS': 'sensor', # 湿度传感器 + + # 摄像头 + 'CAM': 'camera', # 摄像头 + 'IPC': 'camera', # 网络摄像头 + + # 其他 + 'OTHER': 'other' # 其他设备 + } + + @classmethod + def map_device_type(cls, standard_type: str) -> str: + """ + 将标准设备类型映射为内部类型 + + Args: + standard_type: 标准设备类型 + + Returns: + str: 内部设备类型 + """ + return cls.MAPPINGS.get(standard_type.upper(), 'other') + + @classmethod + def get_all_standard_types(cls) -> list: + """获取所有标准设备类型""" + return list(cls.MAPPINGS.keys()) + + +@dataclass +class UnitConversion: + """单位转换配置""" + # 水利行业常用单位转换 + CONVERSIONS = { + # 流量单位 + 'm³/h': {'m³/s': 1/3600, 'L/s': 1000/3600}, + 'm³/s': {'m³/h': 3600, 'L/s': 1000}, + 'L/s': {'m³/h': 3600/1000, 'm³/s': 1/1000}, + + # 压力单位 + 'MPa': {'kPa': 1000, 'Pa': 1000000}, + 'kPa': {'MPa': 1/1000, 'Pa': 1000}, + 'Pa': {'MPa': 1/1000000, 'kPa': 1/1000}, + + # 水位单位 + 'm': {'cm': 100, 'mm': 1000}, + 'cm': {'m': 1/100, 'mm': 10}, + 'mm': {'m': 1/1000, 'cm': 1/10}, + + # 水质单位 + 'NTU': {'': 1}, # 浊度无标准转换 + 'pH': {'': 1}, # pH值无标准转换 + 'mg/L': {'ppm': 1}, # mg/L 和 ppm 等价 + 'ppm': {'mg/L': 1} + } + + @classmethod + def convert_unit(cls, value: float, from_unit: str, to_unit: str) -> float: + """ + 单位转换 + + Args: + value: 数值 + from_unit: 源单位 + to_unit: 目标单位 + + Returns: + float: 转换后的数值 + """ + if from_unit == to_unit: + return value + + if from_unit in cls.CONVERSIONS and to_unit in cls.CONVERSIONS[from_unit]: + factor = cls.CONVERSIONS[from_unit][to_unit] + return value * factor + + # 如果没有找到转换关系，尝试反向查找 + for unit, conversions in cls.CONVERSIONS.items(): + if to_unit in conversions and from_unit in conversions: + # 相同基准单位的转换 + factor = conversions[from_unit] / conversions[to_unit] + return value * factor + + # 无法转换，返回原值 + return value + + @classmethod + def get_supported_units(cls) -> list: + """获取支持的所有单位""" + units = set() + for conversions in cls.CONVERSIONS.values(): + units.update(conversions.keys()) + return list(units) \ No newline at end of file diff --git a/src/iot/device_controller.py b/src/iot/device_controller.py new file mode 100644 index 00000000..e0fa1853 --- /dev/null +++ b/src/iot/device_controller.py @@ -0,0 +1,260 @@ +""" +设备控制器 +提供设备注册/发现的REST API接口 +""" + +import json +import logging +from datetime import datetime +from typing import Dict, Any, List, Optional +from flask import Blueprint, request, jsonify +from .device_manager import DeviceManager +from .models import DeviceType, DeviceStatus + + +class DeviceController: + """设备控制器""" + + def __init__(self, device_manager: DeviceManager): + """ + 初始化设备控制器 + + Args: + device_manager: 设备管理器 + """ + self.device_manager = device_manager + self.logger = logging.getLogger(__name__) + + # 创建Blueprint + self.blueprint = Blueprint('device', __name__, url_prefix='/api/iot/device') + + # 注册路由 + self._register_routes() + + def _register_routes(self): + """注册API路由""" + + @self.blueprint.route('/', methods=['GET']) + def list_devices(): + """获取设备列表""" + try: + # 获取查询参数 + device_type_str = request.args.get('type') + status_str = request.args.get('status') + area = request.args.get('area') + page = int(request.args.get('page', 1)) + per_page = int(request.args.get('per_page', 10)) + + # 过滤条件 + device_type = DeviceType(device_type_str) if device_type_str else None + status = DeviceStatus(status_str) if status_str else None + + # 获取设备列表 + devices = self.device_manager.list_devices(device_type, status, area) + + # 分页 + total = len(devices) + start = (page - 1) * per_page + end = start + per_page + paginated_devices = devices[start:end] + + # 转换为字典格式 + device_list = [device.to_dict() for device in paginated_devices] + + return jsonify({ + "success": True, + "data": device_list, + "pagination": { + "page": page, + "per_page": per_page, + "total": total, + "pages": (total + per_page - 1) // per_page + } + }) + except Exception as e: + self.logger.error(f"Error listing devices: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/', methods=['GET']) + def get_device(device_sn): + """获取设备详情""" + try: + device = self.device_manager.get_device(device_sn) + if not device: + return jsonify({"success": False, "error": "Device not found"}), 404 + + # 获取设备影子 + shadow = self.device_manager.get_device_shadow(device_sn) + device_data = device.to_dict() + if shadow: + device_data['shadow'] = shadow.to_dict() + + return jsonify({ + "success": True, + "data": device_data + }) + except Exception as e: + self.logger.error(f"Error getting device {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/', methods=['POST']) + def register_device(): + """注册新设备""" + try: + device_data = request.get_json() + + # 验证必要字段 + required_fields = ['device_sn', 'device_type', 'name'] + for field in required_fields: + if field not in device_data: + return jsonify({"success": False, "error": f"Missing required field: {field}"}), 400 + + # 检查设备是否已存在 + existing_device = self.device_manager.get_device(device_data['device_sn']) + if existing_device: + return jsonify({"success": False, "error": "Device already exists"}), 409 + + # 注册设备 + device = self.device_manager.register_device(device_data) + + return jsonify({ + "success": True, + "data": device.to_dict(), + "message": "Device registered successfully" + }), 201 + except Exception as e: + self.logger.error(f"Error registering device: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/', methods=['PUT']) + def update_device(device_sn): + """更新设备信息""" + try: + device = self.device_manager.get_device(device_sn) + if not device: + return jsonify({"success": False, "error": "Device not found"}), 404 + + updates = request.get_json() + + # 更新设备 + updated_device = self.device_manager.update_device(device_sn, updates) + if not updated_device: + return jsonify({"success": False, "error": "Failed to update device"}), 400 + + return jsonify({ + "success": True, + "data": updated_device.to_dict(), + "message": "Device updated successfully" + }) + except Exception as e: + self.logger.error(f"Error updating device {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/', methods=['DELETE']) + def delete_device(device_sn): + """删除设备""" + try: + success = self.device_manager.delete_device(device_sn) + if not success: + return jsonify({"success": False, "error": "Device not found"}), 404 + + return jsonify({ + "success": True, + "message": "Device deleted successfully" + }) + except Exception as e: + self.logger.error(f"Error deleting device {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('//shadow', methods=['GET']) + def get_device_shadow(device_sn): + """获取设备影子""" + try: + shadow = self.device_manager.get_device_shadow(device_sn) + if not shadow: + return jsonify({"success": False, "error": "Device shadow not found"}), 404 + + return jsonify({ + "success": True, + "data": shadow.to_dict() + }) + except Exception as e: + self.logger.error(f"Error getting device shadow {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('//shadow', methods=['PUT']) + def update_device_shadow(device_sn): + """更新设备影子""" + try: + state = request.get_json() + + success = self.device_manager.update_device_shadow(device_sn, state) + if not success: + return jsonify({"success": False, "error": "Device not found"}), 404 + + return jsonify({ + "success": True, + "message": "Device shadow updated successfully" + }) + except Exception as e: + self.logger.error(f"Error updating device shadow {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/discover', methods=['POST']) + def discover_devices(): + """设备发现""" + try: + discovered = self.device_manager.discover_devices() + + return jsonify({ + "success": True, + "data": discovered, + "message": f"Discovered {len(discovered)} devices" + }) + except Exception as e: + self.logger.error(f"Error discovering devices: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('//command', methods=['POST']) + def send_device_command(device_sn): + """发送设备控制命令""" + try: + command = request.get_json() + + # 验证设备存在 + device = self.device_manager.get_device(device_sn) + if not device: + return jsonify({"success": False, "error": "Device not found"}), 404 + + # 发送命令 + success = self.mqtt_adapter.send_command(device_sn, command) + if not success: + return jsonify({"success": False, "error": "Failed to send command"}), 500 + + return jsonify({ + "success": True, + "message": "Command sent successfully", + "device_sn": device_sn, + "command": command + }) + except Exception as e: + self.logger.error(f"Error sending command to device {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/statistics', methods=['GET']) + def get_device_statistics(): + """获取设备统计信息""" + try: + statistics = self.device_manager.get_device_statistics() + + return jsonify({ + "success": True, + "data": statistics + }) + except Exception as e: + self.logger.error(f"Error getting device statistics: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + def get_blueprint(self): + """获取Blueprint""" + return self.blueprint \ No newline at end of file diff --git a/src/iot/device_manager.py b/src/iot/device_manager.py new file mode 100644 index 00000000..c70225be --- /dev/null +++ b/src/iot/device_manager.py @@ -0,0 +1,219 @@ +""" +设备管理服务 +负责设备的CRUD操作、设备影子管理、设备发现等功能 +""" + +import json +import logging +from datetime import datetime +from typing import List, Optional, Dict, Any +from .models import Device, DeviceShadow, DeviceStatus, DeviceType + + +class DeviceManager: + """设备管理器""" + + def __init__(self): + self.devices: Dict[str, Device] = {} # device_sn -> Device + self.shadows: Dict[str, DeviceShadow] = {} # device_sn -> DeviceShadow + self.logger = logging.getLogger(__name__) + + def register_device(self, device_data: Dict[str, Any]) -> Device: + """ + 注册设备 + + Args: + device_data: 设备数据字典 + + Returns: + Device: 注册的设备对象 + """ + device = Device( + device_sn=device_data['device_sn'], + device_type=DeviceType(device_data.get('device_type', 'other')), + name=device_data.get('name', ''), + description=device_data.get('description', ''), + area=device_data.get('area', ''), + position=device_data.get('position', ''), + geom=device_data.get('geom'), + manufacturer=device_data.get('manufacturer', ''), + model=device_data.get('model', ''), + firmware_version=device_data.get('firmware_version', ''), + hardware_version=device_data.get('hardware_version', ''), + metadata=device_data.get('metadata', {}) + ) + + self.devices[device.device_sn] = device + + # 创建设备影子 + shadow = DeviceShadow(device_sn=device.device_sn) + self.shadows[device.device_sn] = shadow + + self.logger.info(f"Device registered: {device.device_sn}") + return device + + def get_device(self, device_sn: str) -> Optional[Device]: + """ + 获取设备信息 + + Args: + device_sn: 设备序列号 + + Returns: + Device: 设备对象，如果不存在返回None + """ + return self.devices.get(device_sn) + + def update_device(self, device_sn: str, updates: Dict[str, Any]) -> Optional[Device]: + """ + 更新设备信息 + + Args: + device_sn: 设备序列号 + updates: 更新的字段 + + Returns: + Device: 更新后的设备对象，如果不存在返回None + """ + device = self.devices.get(device_sn) + if not device: + return None + + # 更新设备属性 + for key, value in updates.items(): + if hasattr(device, key): + setattr(device, key, value) + + device.updated_at = datetime.now() + self.logger.info(f"Device updated: {device_sn}") + return device + + def delete_device(self, device_sn: str) -> bool: + """ + 删除设备 + + Args: + device_sn: 设备序列号 + + Returns: + bool: 是否删除成功 + """ + if device_sn in self.devices: + del self.devices[device_sn] + if device_sn in self.shadows: + del self.shadows[device_sn] + self.logger.info(f"Device deleted: {device_sn}") + return True + return False + + def list_devices(self, + device_type: Optional[DeviceType] = None, + status: Optional[DeviceStatus] = None, + area: Optional[str] = None) -> List[Device]: + """ + 列出设备 + + Args: + device_type: 设备类型过滤 + status: 设备状态过滤 + area: 区域过滤 + + Returns: + List[Device]: 设备列表 + """ + devices = list(self.devices.values()) + + if device_type: + devices = [d for d in devices if d.device_type == device_type] + + if status: + devices = [d for d in devices if d.status == status] + + if area: + devices = [d for d in devices if d.area == area] + + return devices + + def update_device_shadow(self, device_sn: str, state: Dict[str, Any]) -> bool: + """ + 更新设备影子 + + Args: + device_sn: 设备序列号 + state: 设备状态 + + Returns: + bool: 是否更新成功 + """ + if device_sn not in self.shadows: + return False + + shadow = self.shadows[device_sn] + shadow.state.update(state) + shadow.timestamp = datetime.now() + self.logger.debug(f"Device shadow updated: {device_sn}") + return True + + def get_device_shadow(self, device_sn: str) -> Optional[DeviceShadow]: + """ + 获取设备影子 + + Args: + device_sn: 设备序列号 + + Returns: + DeviceShadow: 设备影子对象 + """ + return self.shadows.get(device_sn) + + def discover_devices(self) -> List[Dict[str, Any]]: + """ + 设备发现 - 扫描网络中的设备 + + Returns: + List[Dict[str, Any]]: 发现的设备列表 + """ + discovered = [] + + # 模拟设备发现过程 + # 在实际实现中，这里可以包含网络扫描、协议握手等逻辑 + for device_sn, device in self.devices.items(): + if device.status == DeviceStatus.OFFLINE: + # 模拟设备上线 + device.status = DeviceStatus.ONLINE + device.last_seen = datetime.now() + device.ip_address = f"192.168.1.{hash(device_sn) % 255 + 1}" + + discovered.append({ + "device_sn": device_sn, + "name": device.name, + "type": device.device_type.value, + "ip_address": device.ip_address, + "status": device.status.value + }) + + self.logger.info(f"Discovered {len(discovered)} devices") + return discovered + + def get_device_statistics(self) -> Dict[str, Any]: + """ + 获取设备统计信息 + + Returns: + Dict[str, Any]: 统计信息 + """ + total = len(self.devices) + online = sum(1 for d in self.devices.values() if d.status == DeviceStatus.ONLINE) + offline = total - online + + by_type = {} + for device in self.devices.values(): + device_type = device.device_type.value + by_type[device_type] = by_type.get(device_type, 0) + 1 + + return { + "total_devices": total, + "online_devices": online, + "offline_devices": offline, + "devices_by_type": by_type + } \ No newline at end of file diff --git a/src/iot/models.py b/src/iot/models.py new file mode 100644 index 00000000..55eb7b0c --- /dev/null +++ b/src/iot/models.py @@ -0,0 +1,163 @@ +""" +IoT 设备模型定义 +包含设备实体、设备影子、OTA升级等核心数据模型 +""" + +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum +from typing import Dict, List, Optional, Any +import uuid + + +class DeviceStatus(Enum): + """设备状态枚举""" + ONLINE = "online" + OFFLINE = "offline" + MAINTENANCE = "maintenance" + FAULT = "fault" + + +class DeviceType(Enum): + """设备类型枚举""" + FLOW_METER = "flow_meter" # 流量计 + PRESSURE_METER = "pressure_meter" # 压力表 + LEVEL_METER = "level_meter" # 水位计 + QUALITY_METER = "quality_meter" # 水质仪 + VALVE = "valve" # 阀门 + PUMP = "pump" # 水泵 + SENSOR = "sensor" # 传感器 + CAMERA = "camera" # 摄像头 + OTHER = "other" # 其他 + + +@dataclass +class Device: + """设备实体模型""" + # 必需字段（无默认值） + device_sn: str # 设备序列号（唯一标识） + device_type: DeviceType # 设备类型 + name: str # 设备名称 + + # 可选字段（有默认值） + description: str = "" # 设备描述 + area: str = "" # 区域 + position: str = "" # 位置 + geom: Optional[str] = None # 地理坐标（GeoJSON格式） + manufacturer: str = "" # 厂商 + model: str = "" # 型号 + firmware_version: str = "" # 固件版本 + hardware_version: str = "" # 硬件版本 + status: DeviceStatus = DeviceStatus.OFFLINE + last_seen: Optional[datetime] = None + ip_address: Optional[str] = None + port: Optional[int] = None + metadata: Dict[str, Any] = field(default_factory=dict) + created_at: datetime = field(default_factory=datetime.now) + updated_at: datetime = field(default_factory=datetime.now) + id: Optional[int] = None + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "device_sn": self.device_sn, + "device_type": self.device_type.value, + "name": self.name, + "description": self.description, + "area": self.area, + "position": self.position, + "geom": self.geom, + "manufacturer": self.manufacturer, + "model": self.model, + "firmware_version": self.firmware_version, + "hardware_version": self.hardware_version, + "status": self.status.value, + "last_seen": self.last_seen.isoformat() if self.last_seen else None, + "ip_address": self.ip_address, + "port": self.port, + "metadata": self.metadata, + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat() + } + + +@dataclass +class DeviceShadow: + """设备影子模型""" + # 必需字段 + device_sn: str # 设备序列号 + + # 可选字段 + state: Dict[str, Any] = field(default_factory=dict) # 设备状态 + desired_state: Dict[str, Any] = field(default_factory=dict) # 期望状态 + reported_state: Dict[str, Any] = field(default_factory=dict) # 报告状态 + timestamp: datetime = field(default_factory=datetime.now) # 时间戳 + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "device_sn": self.device_sn, + "state": self.state, + "desired_state": self.desired_state, + "reported_state": self.reported_state, + "timestamp": self.timestamp.isoformat() + } + + +@dataclass +class OtaUpdate: + """OTA升级记录""" + # 必需字段 + device_sn: str # 设备序列号 + version: str # 目标版本 + file_url: str # 固件文件URL + file_size: int # 文件大小 + checksum: str # 文件校验和 + + # 可选字段 + id: str = field(default_factory=lambda: str(uuid.uuid4())) + status: str = "pending" # 状态：pending/downloading/installed/failed + progress: int = 0 # 进度百分比 + error_message: Optional[str] = None # 错误信息 + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "device_sn": self.device_sn, + "version": self.version, + "file_url": self.file_url, + "file_size": self.file_size, + "checksum": self.checksum, + "status": self.status, + "progress": self.progress, + "error_message": self.error_message, + "started_at": self.started_at.isoformat() if self.started_at else None, + "completed_at": self.completed_at.isoformat() if self.completed_at else None + } + + +@dataclass +class MqttMessage: + """MQTT消息模型""" + # 必需字段 + topic: str # 主题 + payload: Dict[str, Any] # 消息内容 + + # 可选字段 + qos: int = 0 # QoS等级 + retain: bool = False # 是否保留消息 + timestamp: datetime = field(default_factory=datetime.now) # 时间戳 + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "topic": self.topic, + "payload": self.payload, + "qos": self.qos, + "retain": self.retain, + "timestamp": self.timestamp.isoformat() + } \ No newline at end of file diff --git a/src/iot/mqtt_adapter.py b/src/iot/mqtt_adapter.py new file mode 100644 index 00000000..2c42a0f9 --- /dev/null +++ b/src/iot/mqtt_adapter.py @@ -0,0 +1,352 @@ +""" +MQTT 协议适配器 +负责MQTT连接管理、消息订阅/发布、消息解析等功能 +""" + +import json +import logging +import paho.mqtt.client as mqtt +from datetime import datetime +from typing import Dict, Any, Optional, Callable, List +from .models import MqttMessage +from threading import Lock + + +class MqttAdapter: + """MQTT适配器""" + + def __init__(self, + broker_host: str = "localhost", + broker_port: int = 1883, + username: Optional[str] = None, + password: Optional[str] = None, + client_id: str = "water-management-system"): + """ + 初始化MQTT适配器 + + Args: + broker_host: MQTT broker地址 + broker_port: MQTT broker端口 + username: 用户名 + password: 密码 + client_id: 客户端ID + """ + self.broker_host = broker_host + self.broker_port = broker_port + self.username = username + self.password = password + self.client_id = client_id + + self.client = mqtt.Client(client_id=client_id) + self.message_handlers: Dict[str, Callable] = {} + self.connected = False + self.lock = Lock() + + # 配置MQTT客户端 + if username and password: + self.client.username_pw_set(username, password) + + # 设置回调函数 + self.client.on_connect = self._on_connect + self.client.on_disconnect = self._on_disconnect + self.client.on_message = self._on_message + self.client.on_publish = self._on_publish + self.client.on_subscribe = self._on_subscribe + + self.logger = logging.getLogger(__name__) + + def _on_connect(self, client, userdata, flags, rc): + """连接回调""" + if rc == 0: + self.connected = True + self.logger.info(f"Connected to MQTT broker at {self.broker_host}:{self.broker_port}") + else: + self.logger.error(f"Failed to connect to MQTT broker, return code {rc}") + + def _on_disconnect(self, client, userdata, rc): + """断开连接回调""" + self.connected = False + self.logger.warning(f"Disconnected from MQTT broker, return code {rc}") + + def _on_message(self, client, userdata, msg): + """消息接收回调""" + try: + # 解析消息 + payload = json.loads(msg.payload.decode('utf-8')) if msg.payload else {} + + message = MqttMessage( + topic=msg.topic, + payload=payload, + qos=msg.qos, + retain=msg.retain + ) + + self.logger.debug(f"Received message: {message.topic} - {message.payload}") + + # 查找对应的消息处理器 + for topic_pattern, handler in self.message_handlers.items(): + if self._topic_matches(msg.topic, topic_pattern): + try: + handler(message) + except Exception as e: + self.logger.error(f"Error in message handler for {msg.topic}: {e}") + + except json.JSONDecodeError as e: + self.logger.error(f"Failed to parse JSON message from {msg.topic}: {e}") + except Exception as e: + self.logger.error(f"Error processing message from {msg.topic}: {e}") + + def _on_publish(self, client, userdata, mid): + """发布消息回调""" + self.logger.debug(f"Message published with mid: {mid}") + + def _on_subscribe(self, client, userdata, mid, granted_qos): + """订阅回调""" + self.logger.debug(f"Subscribed with mid: {mid}, granted_qos: {granted_qos}") + + def _topic_matches(self, topic: str, pattern: str) -> bool: + """检查主题是否匹配模式""" + # 简单的通配符匹配实现 + # 支持单层通配符 + 和多层通配符 # + pattern_parts = pattern.split('/') + topic_parts = topic.split('/') + + if len(pattern_parts) != len(topic_parts): + return False + + for p_part, t_part in zip(pattern_parts, topic_parts): + if p_part == '+' or p_part == '#': + continue + if p_part != t_part: + return False + + return True + + def connect(self) -> bool: + """ + 连接到MQTT broker + + Returns: + bool: 是否连接成功 + """ + try: + self.client.connect(self.broker_host, self.broker_port, 60) + self.client.loop_start() + return True + except Exception as e: + self.logger.error(f"Failed to connect to MQTT broker: {e}") + return False + + def disconnect(self): + """断开MQTT连接""" + if self.connected: + self.client.loop_stop() + self.client.disconnect() + + def is_connected(self) -> bool: + """ + 检查是否已连接 + + Returns: + bool: 是否已连接 + """ + return self.connected + + def subscribe(self, topic: str, qos: int = 0) -> bool: + """ + 订阅主题 + + Args: + topic: 主题 + qos: QoS等级 + + Returns: + bool: 是否订阅成功 + """ + try: + result = self.client.subscribe(topic, qos) + if result[0] == mqtt.MQTT_ERR_SUCCESS: + self.logger.info(f"Subscribed to topic: {topic}") + return True + else: + self.logger.error(f"Failed to subscribe to topic: {topic}") + return False + except Exception as e: + self.logger.error(f"Error subscribing to topic {topic}: {e}") + return False + + def unsubscribe(self, topic: str) -> bool: + """ + 取消订阅主题 + + Args: + topic: 主题 + + Returns: + bool: 是否取消订阅成功 + """ + try: + result = self.client.unsubscribe(topic) + if result[0] == mqtt.MQTT_ERR_SUCCESS: + self.logger.info(f"Unsubscribed from topic: {topic}") + return True + else: + self.logger.error(f"Failed to unsubscribe from topic: {topic}") + return False + except Exception as e: + self.logger.error(f"Error unsubscribing from topic {topic}: {e}") + return False + + def publish(self, topic: str, payload: Any, qos: int = 0, retain: bool = False) -> bool: + """ + 发布消息 + + Args: + topic: 主题 + payload: 消息内容 + qos: QoS等级 + retain: 是否保留消息 + + Returns: + bool: 是否发布成功 + """ + try: + if isinstance(payload, dict): + payload = json.dumps(payload) + elif not isinstance(payload, str): + payload = str(payload) + + result = self.client.publish(topic, payload, qos, retain) + if result[0] == mqtt.MQTT_ERR_SUCCESS: + self.logger.debug(f"Published to topic: {topic}") + return True + else: + self.logger.error(f"Failed to publish to topic: {topic}") + return False + except Exception as e: + self.logger.error(f"Error publishing to topic {topic}: {e}") + return False + + def add_message_handler(self, topic_pattern: str, handler: Callable[[MqttMessage], None]): + """ + 添加消息处理器 + + Args: + topic_pattern: 主题模式（支持通配符） + handler: 消息处理函数 + """ + with self.lock: + self.message_handlers[topic_pattern] = handler + self.logger.info(f"Added message handler for pattern: {topic_pattern}") + + def remove_message_handler(self, topic_pattern: str): + """ + 移除消息处理器 + + Args: + topic_pattern: 主题模式 + """ + with self.lock: + if topic_pattern in self.message_handlers: + del self.message_handlers[topic_pattern] + self.logger.info(f"Removed message handler for pattern: {topic_pattern}") + + def subscribe_device_topics(self, device_manager): + """ + 订阅设备相关主题 + + Args: + device_manager: 设备管理器实例 + """ + # 设备状态上报 + self.add_message_handler("devices/+/status", self._handle_device_status) + + # 设备数据上报 + self.add_message_handler("devices/+/data", self._handle_device_data) + + # 设备控制命令响应 + self.add_message_handler("devices/+/command/response", self._handle_command_response) + + # 设备OTA状态 + self.add_message_handler("devices/+/ota/status", self._handle_ota_status) + + def _handle_device_status(self, message: MqttMessage): + """处理设备状态消息""" + topic_parts = message.topic.split('/') + if len(topic_parts) >= 2: + device_sn = topic_parts[1] + status = message.payload.get('status', 'unknown') + + # 更新设备状态 + device = device_manager.get_device(device_sn) + if device: + from .models import DeviceStatus + try: + device.status = DeviceStatus(status) + device.last_seen = datetime.now() + device_manager.logger.info(f"Device {device_sn} status updated to {status}") + except ValueError: + device_manager.logger.warning(f"Unknown status: {status}") + + def _handle_device_data(self, message: MqttMessage): + """处理设备数据消息""" + topic_parts = message.topic.split('/') + if len(topic_parts) >= 2: + device_sn = topic_parts[1] + data = message.payload + + # 更新设备影子 + device_manager.update_device_shadow(device_sn, data) + device_manager.logger.debug(f"Device {device_sn} data updated") + + def _handle_command_response(self, message: MqttMessage): + """处理命令响应消息""" + topic_parts = message.topic.split('/') + if len(topic_parts) >= 2: + device_sn = topic_parts[1] + command_id = message.payload.get('command_id') + result = message.payload.get('result') + + device_manager.logger.info(f"Device {device_sn} command response: {command_id} -> {result}") + + def _handle_ota_status(self, message: MqttMessage): + """处理OTA状态消息""" + topic_parts = message.topic.split('/') + if len(topic_parts) >= 2: + device_sn = topic_parts[1] + status = message.payload.get('status') + progress = message.payload.get('progress', 0) + + device_manager.logger.info(f"Device {device_sn} OTA status: {status}, progress: {progress}%") + + def send_command(self, device_sn: str, command: Dict[str, Any]) -> bool: + """ + 发送设备控制命令 + + Args: + device_sn: 设备序列号 + command: 命令内容 + + Returns: + bool: 是否发送成功 + """ + topic = f"devices/{device_sn}/command" + command['command_id'] = f"cmd_{datetime.now().timestamp()}" + command['timestamp'] = datetime.now().isoformat() + + return self.publish(topic, command, qos=1) + + def get_connection_status(self) -> Dict[str, Any]: + """ + 获取连接状态 + + Returns: + Dict[str, Any]: 连接状态信息 + """ + return { + "connected": self.connected, + "broker_host": self.broker_host, + "broker_port": self.broker_port, + "client_id": self.client_id, + "message_handlers_count": len(self.message_handlers) + } \ No newline at end of file diff --git a/src/iot/ota_controller.py b/src/iot/ota_controller.py new file mode 100644 index 00000000..d17935a5 --- /dev/null +++ b/src/iot/ota_controller.py @@ -0,0 +1,214 @@ +""" +OTA固件升级控制器 +提供OTA升级相关的REST API接口 +""" + +import json +import logging +from datetime import datetime +from typing import Dict, Any, List, Optional +from flask import Blueprint, request, jsonify +from .ota_manager import OtaManager +from .models import OtaUpdate + + +class OtaController: + """OTA控制器""" + + def __init__(self, ota_manager: OtaManager): + """ + 初始化OTA控制器 + + Args: + ota_manager: OTA管理器 + """ + self.ota_manager = ota_manager + self.logger = logging.getLogger(__name__) + + # 创建Blueprint + self.blueprint = Blueprint('ota', __name__, url_prefix='/api/iot/ota') + + # 注册路由 + self._register_routes() + + def _register_routes(self): + """注册API路由""" + + @self.blueprint.route('/updates', methods=['GET']) + def list_updates(): + """获取OTA更新列表""" + try: + # 获取查询参数 + device_sn = request.args.get('device_sn') + status = request.args.get('status') + page = int(request.args.get('page', 1)) + per_page = int(request.args.get('per_page', 10)) + + # 过滤条件 + updates = list(self.ota_manager.updates.values()) + if device_sn: + updates = [u for u in updates if u.device_sn == device_sn] + if status: + updates = [u for u in updates if u.status == status] + + # 分页 + total = len(updates) + start = (page - 1) * per_page + end = start + per_page + paginated_updates = updates[start:end] + + # 转换为字典格式 + update_list = [update.to_dict() for update in paginated_updates] + + return jsonify({ + "success": True, + "data": update_list, + "pagination": { + "page": page, + "per_page": per_page, + "total": total, + "pages": (total + per_page - 1) // per_page + } + }) + except Exception as e: + self.logger.error(f"Error listing OTA updates: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates/', methods=['GET']) + def get_update(update_id): + """获取OTA更新详情""" + try: + update = self.ota_manager.get_update(update_id) + if not update: + return jsonify({"success": False, "error": "Update not found"}), 404 + + return jsonify({ + "success": True, + "data": update.to_dict() + }) + except Exception as e: + self.logger.error(f"Error getting OTA update {update_id}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates', methods=['POST']) + def create_update(): + """创建OTA更新任务""" + try: + update_data = request.get_json() + + # 验证必要字段 + required_fields = ['device_sn', 'version', 'file_url', 'file_size', 'checksum'] + for field in required_fields: + if field not in update_data: + return jsonify({"success": False, "error": f"Missing required field: {field}"}), 400 + + # 创建更新 + update = self.ota_manager.create_update( + device_sn=update_data['device_sn'], + version=update_data['version'], + file_url=update_data['file_url'], + file_size=update_data['file_size'], + checksum=update_data['checksum'] + ) + + return jsonify({ + "success": True, + "data": update.to_dict(), + "message": "OTA update created successfully" + }), 201 + except Exception as e: + self.logger.error(f"Error creating OTA update: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates//start', methods=['POST']) + def start_update(update_id): + """开始OTA更新""" + try: + success = self.ota_manager.start_update(update_id) + if not success: + return jsonify({"success": False, "error": "Failed to start update"}), 400 + + return jsonify({ + "success": True, + "message": "OTA update started successfully", + "update_id": update_id + }) + except Exception as e: + self.logger.error(f"Error starting OTA update {update_id}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates//progress', methods=['PUT']) + def update_progress(update_id): + """更新OTA进度""" + try: + progress_data = request.get_json() + + progress = progress_data.get('progress') + error_message = progress_data.get('error_message') + + if progress is None: + return jsonify({"success": False, "error": "Progress is required"}), 400 + + success = self.ota_manager.update_progress(update_id, progress, error_message) + if not success: + return jsonify({"success": False, "error": "Update not found"}), 404 + + return jsonify({ + "success": True, + "message": "OTA progress updated successfully", + "update_id": update_id, + "progress": progress + }) + except Exception as e: + self.logger.error(f"Error updating OTA progress {update_id}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates//cancel', methods=['POST']) + def cancel_update(update_id): + """取消OTA更新""" + try: + success = self.ota_manager.cancel_update(update_id) + if not success: + return jsonify({"success": False, "error": "Failed to cancel update"}), 400 + + return jsonify({ + "success": True, + "message": "OTA update cancelled successfully", + "update_id": update_id + }) + except Exception as e: + self.logger.error(f"Error cancelling OTA update {update_id}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/updates/device/', methods=['GET']) + def get_device_updates(device_sn): + """获取设备的OTA更新记录""" + try: + updates = self.ota_manager.get_updates_by_device(device_sn) + + return jsonify({ + "success": True, + "data": [update.to_dict() for update in updates], + "device_sn": device_sn + }) + except Exception as e: + self.logger.error(f"Error getting updates for device {device_sn}: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + @self.blueprint.route('/statistics', methods=['GET']) + def get_statistics(): + """获取OTA统计信息""" + try: + statistics = self.ota_manager.get_update_statistics() + + return jsonify({ + "success": True, + "data": statistics + }) + except Exception as e: + self.logger.error(f"Error getting OTA statistics: {e}") + return jsonify({"success": False, "error": str(e)}), 500 + + def get_blueprint(self): + """获取Blueprint""" + return self.blueprint \ No newline at end of file diff --git a/src/iot/ota_manager.py b/src/iot/ota_manager.py new file mode 100644 index 00000000..108442f8 --- /dev/null +++ b/src/iot/ota_manager.py @@ -0,0 +1,173 @@ +""" +OTA固件升级管理器 +负责设备OTA升级流程、版本管理、升级状态跟踪等功能 +""" + +import json +import logging +import hashlib +from datetime import datetime +from typing import Dict, Any, List, Optional +from .models import OtaUpdate + + +class OtaManager: + """OTA管理器""" + + def __init__(self): + self.updates: Dict[str, OtaUpdate] = {} # update_id -> OtaUpdate + self.logger = logging.getLogger(__name__) + + def create_update(self, device_sn: str, version: str, file_url: str, + file_size: int, checksum: str) -> OtaUpdate: + """ + 创建OTA升级任务 + + Args: + device_sn: 设备序列号 + version: 目标版本 + file_url: 固件文件URL + file_size: 文件大小 + checksum: 文件校验和 + + Returns: + OtaUpdate: OTA升级对象 + """ + update = OtaUpdate( + device_sn=device_sn, + version=version, + file_url=file_url, + file_size=file_size, + checksum=checksum + ) + + self.updates[update.id] = update + self.logger.info(f"Created OTA update: {update.id} for device {device_sn}") + + return update + + def get_update(self, update_id: str) -> Optional[OtaUpdate]: + """ + 获取OTA升级信息 + + Args: + update_id: 更新ID + + Returns: + OtaUpdate: OTA升级对象 + """ + return self.updates.get(update_id) + + def get_updates_by_device(self, device_sn: str) -> List[OtaUpdate]: + """ + 获取设备的OTA升级记录 + + Args: + device_sn: 设备序列号 + + Returns: + List[OtaUpdate]: OTA升级列表 + """ + return [update for update in self.updates.values() if update.device_sn == device_sn] + + def start_update(self, update_id: str) -> bool: + """ + 开始OTA升级 + + Args: + update_id: 更新ID + + Returns: + bool: 是否开始成功 + """ + update = self.updates.get(update_id) + if not update: + return False + + if update.status != "pending": + self.logger.warning(f"Update {update_id} is not in pending state") + return False + + update.status = "downloading" + update.started_at = datetime.now() + self.logger.info(f"Started OTA update: {update_id}") + + return True + + def update_progress(self, update_id: str, progress: int, error_message: Optional[str] = None) -> bool: + """ + 更新OTA升级进度 + + Args: + update_id: 更新ID + progress: 进度百分比 + error_message: 错误信息 + + Returns: + bool: 是否更新成功 + """ + update = self.updates.get(update_id) + if not update: + return False + + update.progress = progress + + if error_message: + update.error_message = error_message + update.status = "failed" + self.logger.error(f"OTA update {update_id} failed: {error_message}") + elif progress >= 100: + update.status = "installed" + update.completed_at = datetime.now() + self.logger.info(f"OTA update {update_id} completed successfully") + else: + # 保持下载中状态 + pass + + return True + + def cancel_update(self, update_id: str) -> bool: + """ + 取消OTA升级 + + Args: + update_id: 更新ID + + Returns: + bool: 是否取消成功 + """ + update = self.updates.get(update_id) + if not update: + return False + + if update.status in ["installed", "failed"]: + self.logger.warning(f"Cannot cancel completed update {update_id}") + return False + + update.status = "failed" + update.error_message = "Cancelled by user" + update.completed_at = datetime.now() + + self.logger.info(f"Cancelled OTA update: {update_id}") + return True + + def get_update_statistics(self) -> Dict[str, Any]: + """ + 获取OTA统计信息 + + Returns: + Dict[str, Any]: 统计信息 + """ + total = len(self.updates) + pending = sum(1 for u in self.updates.values() if u.status == "pending") + downloading = sum(1 for u in self.updates.values() if u.status == "downloading") + installed = sum(1 for u in self.updates.values() if u.status == "installed") + failed = sum(1 for u in self.updates.values() if u.status == "failed") + + return { + "total_updates": total, + "pending": pending, + "downloading": downloading, + "installed": installed, + "failed": failed + } \ No newline at end of file diff --git a/test_iot.py b/test_iot.py new file mode 100644 index 00000000..ce06030e --- /dev/null +++ b/test_iot.py @@ -0,0 +1,148 @@ +""" +IoT模块测试脚本 +用于验证MQTT适配器、设备管理器和API功能 +""" + +import asyncio +import json +import sys +import os + +# 添加项目根目录到Python路径 +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +from src.iot.device_manager import DeviceManager +from src.iot.mqtt_adapter import MqttAdapter +from src.iot.device_controller import DeviceController +from src.iot.models import DeviceType, DeviceStatus + + +async def test_device_manager(): + """测试设备管理器""" + print("=== 测试设备管理器 ===") + + device_manager = DeviceManager() + + # 注册设备 + device_data = { + 'device_sn': 'LL-001', + 'device_type': 'flow_meter', + 'name': '流量计-001', + 'description': 'A区入口流量计', + 'area': 'A区', + 'position': '入口处', + 'manufacturer': '华为', + 'model': 'LL-100' + } + + device = device_manager.register_device(device_data) + print(f"注册设备: {device.device_sn} - {device.name}") + + # 获取设备 + retrieved_device = device_manager.get_device('LL-001') + print(f"获取设备: {retrieved_device.name}") + + # 更新设备 + updated_device = device_manager.update_device('LL-001', {'status': DeviceStatus.ONLINE}) + print(f"更新设备状态: {updated_device.status}") + + # 列出设备 + devices = device_manager.list_devices() + print(f"设备列表: {len(devices)}个设备") + + # 更新设备影子 + device_manager.update_device_shadow('LL-001', {'temperature': 25.5, 'pressure': 0.8}) + shadow = device_manager.get_device_shadow('LL-001') + print(f"设备影子: {shadow.state}") + + # 获取统计信息 + stats = device_manager.get_device_statistics() + print(f"设备统计: {stats}") + + print("设备管理器测试完成\n") + + +async def test_mqtt_adapter(): + """测试MQTT适配器""" + print("=== 测试MQTT适配器 ===") + + # 创建MQTT适配器（不实际连接） + mqtt_adapter = MqttAdapter( + broker_host="localhost", + broker_port=1883, + client_id="test-client" + ) + + # 测试消息发布 + test_payload = {"message": "Hello IoT", "timestamp": "2024-01-01T00:00:00"} + success = mqtt_adapter.publish("test/topic", test_payload) + print(f"消息发布测试: {'成功' if success else '失败'}") + + # 测试连接状态 + status = mqtt_adapter.get_connection_status() + print(f"MQTT状态: {status}") + + print("MQTT适配器测试完成\n") + + +async def test_device_controller(): + """测试设备控制器""" + print("=== 测试设备控制器 ===") + + # 创建组件 + device_manager = DeviceManager() + mqtt_adapter = MqttAdapter() + device_controller = DeviceController(device_manager, mqtt_adapter) + + # 注册一些测试设备 + test_devices = [ + { + 'device_sn': 'LL-001', + 'device_type': 'flow_meter', + 'name': '流量计-001', + 'area': 'A区', + 'manufacturer': '华为' + }, + { + 'device_sn': 'YL-001', + 'device_type': 'pressure_meter', + 'name': '压力表-001', + 'area': 'B区', + 'manufacturer': '西门子' + } + ] + + for device_data in test_devices: + device_manager.register_device(device_data) + + # 模拟API请求 + print("测试设备注册:") + print(f"已注册设备数量: {len(device_manager.devices)}") + + # 模拟设备发现 + discovered = device_manager.discover_devices() + print(f"发现设备数量: {len(discovered)}") + + print("设备控制器测试完成\n") + + +async def main(): + """主测试函数""" + print("开始 IoT 模块测试...\n") + + try: + # 测试各个组件 + await test_device_manager() + await test_mqtt_adapter() + await test_device_controller() + + print("✅ 所有测试完成！") + + except Exception as e: + print(f"❌ 测试失败: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/test_iot_simple.py b/test_iot_simple.py new file mode 100644 index 00000000..1f7937ca --- /dev/null +++ b/test_iot_simple.py @@ -0,0 +1,149 @@ +""" +IoT模块简化测试脚本 +仅测试设备管理器功能，不依赖MQTT +""" + +import asyncio +import json +import sys +import os + +# 添加项目根目录到Python路径 +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +from src.iot.device_manager import DeviceManager +from src.iot.models import DeviceType, DeviceStatus + + +async def test_device_manager(): + """测试设备管理器""" + print("=== 测试设备管理器 ===") + + device_manager = DeviceManager() + + # 注册设备 + device_data = { + 'device_sn': 'LL-001', + 'device_type': 'flow_meter', + 'name': '流量计-001', + 'description': 'A区入口流量计', + 'area': 'A区', + 'position': '入口处', + 'manufacturer': '华为', + 'model': 'LL-100' + } + + device = device_manager.register_device(device_data) + print(f"注册设备: {device.device_sn} - {device.name}") + + # 获取设备 + retrieved_device = device_manager.get_device('LL-001') + print(f"获取设备: {retrieved_device.name}") + + # 更新设备 + updated_device = device_manager.update_device('LL-001', {'status': DeviceStatus.ONLINE}) + print(f"更新设备状态: {updated_device.status}") + + # 列出设备 + devices = device_manager.list_devices() + print(f"设备列表: {len(devices)}个设备") + + # 更新设备影子 + device_manager.update_device_shadow('LL-001', {'temperature': 25.5, 'pressure': 0.8}) + shadow = device_manager.get_device_shadow('LL-001') + print(f"设备影子: {shadow.state}") + + # 获取统计信息 + stats = device_manager.get_device_statistics() + print(f"设备统计: {stats}") + + # 测试设备发现 + discovered = device_manager.discover_devices() + print(f"发现设备: {len(discovered)}个设备") + + print("设备管理器测试完成\n") + + +async def test_device_filtering(): + """测试设备过滤功能""" + print("=== 测试设备过滤 ===") + + device_manager = DeviceManager() + + # 注册多个设备 + test_devices = [ + {'device_sn': 'LL-001', 'device_type': 'flow_meter', 'name': '流量计-001', 'area': 'A区'}, + {'device_sn': 'YL-001', 'device_type': 'pressure_meter', 'name': '压力表-001', 'area': 'A区'}, + {'device_sn': 'SW-001', 'device_type': 'level_meter', 'name': '水位计-001', 'area': 'B区'}, + {'device_sn': 'LL-002', 'device_type': 'flow_meter', 'name': '流量计-002', 'area': 'B区'}, + ] + + for device_data in test_devices: + device_manager.register_device(device_data) + + # 测试按类型过滤 + flow_meters = device_manager.list_devices(device_type=DeviceType.FLOW_METER) + print(f"流量计数量: {len(flow_meters)}") + + # 测试按区域过滤 + a_zone_devices = device_manager.list_devices(area='A区') + print(f"A区设备数量: {len(a_zone_devices)}") + + # 测试按状态过滤 + online_devices = device_manager.list_devices(status=DeviceStatus.ONLINE) + print(f"在线设备数量: {len(online_devices)}") + + print("设备过滤测试完成\n") + + +async def test_device_shadow(): + """测试设备影子功能""" + print("=== 测试设备影子 ===") + + device_manager = DeviceManager() + + # 注册设备 + device_data = { + 'device_sn': 'LL-001', + 'device_type': 'flow_meter', + 'name': '流量计-001' + } + device_manager.register_device(device_data) + + # 更新设备影子 + shadow_data = { + 'temperature': 25.5, + 'pressure': 0.8, + 'flow_rate': 100.5 + } + + success = device_manager.update_device_shadow('LL-001', shadow_data) + print(f"影子更新成功: {success}") + + # 获取设备影子 + shadow = device_manager.get_device_shadow('LL-001') + print(f"设备影子状态: {shadow.state}") + + print("设备影子测试完成\n") + + +async def main(): + """主测试函数""" + print("开始 IoT 模块简化测试...\n") + + try: + # 测试各个组件 + await test_device_manager() + await test_device_filtering() + await test_device_shadow() + + print("✅ 所有测试完成！") + + except Exception as e: + print(f"❌ 测试失败: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file