From 5eae031679fcf31fd837c77a598417b0909ea0d7 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Mon, 15 Jun 2026 11:59:00 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E6=95=B0=E6=8D=AE=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=E5=B1=82=E5=8A=9F=E8=83=BD=EF=BC=88REST=20API=20+=20W?= =?UTF-8?q?ebSocket=20+=20=E6=89=B9=E9=87=8F=E5=AF=BC=E5=85=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 实现REST API服务器，支持IoT数据、手动录入和批量导入接口 - 实现WebSocket服务器，支持实时数据推送和连接管理 - 实现批量导入模块，支持CSV、Excel、JSON多种格式 - 实现数据处理工具，包含字段映射和单位转换功能 - 实现数据模型定义和数据验证机制 - 创建主程序入口和配置文件 - 添加详细的使用文档和API说明 --- README.md | 299 ++++++++++++++++++++++ main.py | 305 ++++++++++++++++++++++ requirements.txt | 10 + src/api/rest_api.py | 181 +++++++++++++ src/batch/batch_import.py | 336 ++++++++++++++++++++++++ src/models/models.py | 317 +++++++++++++++++++++++ src/utils/data_utils.py | 409 ++++++++++++++++++++++++++++++ src/websocket/websocket_server.py | 214 ++++++++++++++++ 8 files changed, 2071 insertions(+) create mode 100644 README.md create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 src/api/rest_api.py create mode 100644 src/batch/batch_import.py create mode 100644 src/models/models.py create mode 100644 src/utils/data_utils.py create mode 100644 src/websocket/websocket_server.py diff --git a/README.md b/README.md new file mode 100644 index 00000000..f66111af --- /dev/null +++ b/README.md @@ -0,0 +1,299 @@ +# 水务管理系统 - 数据接入层 + +## 项目概述 + +本项目是水务管理系统中的数据接入层，实现了多源数据接入、实时WebSocket推送和批量数据导入功能。 + +## 功能特性 + +### 1. REST API 数据接入 +- **IoT设备数据接入**：支持实时接收传感器数据 +- **手动数据录入**：支持人工录入数据 +- **批量API导入**：支持通过API批量导入数据 +- **数据查询接口**：提供按数据类型、时间范围等条件的数据查询 +- **统计分析接口**：提供数据统计和分析功能 + +### 2. WebSocket 实时推送 +- **实时数据推送**：传感器数据实时推送到客户端 +- **连接管理**：支持多客户端连接和订阅管理 +- **数据历史**：新连接客户端可以获取历史数据 +- **警报推送**：支持实时警报推送 +- **心跳检测**：支持连接状态监控 + +### 3. 批量数据导入 +- **多格式支持**：支持CSV、Excel、JSON格式导入 +- **数据验证**：内置数据验证和错误处理 +- **字段映射**：支持水利行业标准字段映射 +- **单位转换**：自动进行单位转换和标准化 +- **批量处理**：支持大批量数据处理 + +## 技术栈 + +- **后端框架**：FastAPI +- **WebSocket**：websockets +- **数据处理**：pandas, numpy +- **异步处理**：asyncio +- **数据验证**：pydantic +- **文件处理**：aiofiles + +## 项目结构 + +``` +water-management-system/ +├── src/ +│ ├── api/ # REST API模块 +│ │ └── rest_api.py # 主API服务器 +│ ├── websocket/ # WebSocket模块 +│ │ └── websocket_server.py # WebSocket服务器 +│ ├── batch/ # 批量导入模块 +│ │ └── batch_import.py # 批量导入功能 +│ ├── utils/ # 工具模块 +│ │ └── data_utils.py # 数据处理工具 +│ └── models/ # 数据模型 +│ └── models.py # 数据模型定义 +├── main.py # 主程序入口 +├── requirements.txt # 依赖文件 +├── config.json # 配置文件 +└── README.md # 项目说明 +``` + +## API 接口文档 + +### REST API 端点 + +#### 1. IoT 数据接收 +``` +POST /api/iot/data +Content-Type: application/json + +{ + "device_id": "device_001", + "data_type": "LL", + "value": 25.5, + "timestamp": "2024-01-01T12:00:00", + "location": "A区" +} +``` + +#### 2. 手动数据录入 +``` +POST /api/manual/data +Content-Type: application/json + +{ + "source": "manual", + "data_type": "YL", + "value": 0.8, + "timestamp": "2024-01-01T12:00:00", + "operator": "张三", + "notes": "定期数据录入" +} +``` + +#### 3. 批量导入 +``` +POST /api/batch/import +Content-Type: application/json + +{ + "batch_id": "batch_001", + "data_source": "system_import", + "records": [ + { + "device_id": "device_002", + "data_type": "SW", + "value": 5.2, + "timestamp": "2024-01-01T12:00:00", + "location": "B区" + } + ] +} +``` + +#### 4. 数据查询 +``` +GET /api/data/{data_type}?limit=100&offset=0 +GET /api/data/recent?hours=24&limit=100 +GET /api/stats +``` + +### WebSocket 连接 + +#### 连接端点 +``` +ws://localhost:8765 +``` + +#### 消息格式 + +**发送消息**： +```json +{ + "type": "subscribe", + "subscription": "LL" // 订阅特定类型数据，"all"订阅所有 +} +``` + +**接收消息**： +```json +{ + "type": "sensor_data", + "data_type": "LL", + "device_id": "device_001", + "value": 25.5, + "location": "A区", + "timestamp": "2024-01-01T12:00:00Z" +} +``` + +## 安装和运行 + +### 1. 安装依赖 +```bash +pip install -r requirements.txt +``` + +### 2. 启动系统 +```bash +# 普通模式 +python main.py + +# 演示模式（自动生成数据） +python main.py --demo + +# 指定配置文件 +python main.py --config custom_config.json +``` + +### 3. 初始化项目 +```bash +python main.py --init +``` + +## 配置文件 + +创建 `config.json` 文件： + +```json +{ + "api": { + "host": "0.0.0.0", + "port": 8000 + }, + "websocket": { + "host": "0.0.0.0", + "port": 8765 + }, + "batch": { + "max_file_size_mb": 100, + "supported_formats": ["csv", "excel", "json"] + }, + "demo_mode": true, + "logging": { + "level": "INFO", + "file": "water_management.log" + } +} +``` + +## 数据类型说明 + +支持的水利行业标准数据类型： + +| 数据类型 | 描述 | 单位 | +|---------|------|------| +| LL | 流量 | m³/h | +| YL | 压力 | MPa | +| SW | 水位 | m | +| ZD | 浊度 | NTU | +| PH | pH值 | - | +| WD | 温度 | °C | +| DD | 电导率 | μS/cm | +| YD | 硬度 | mg/L | + +## 开发规范 + +### 代码结构 +- 遵循模块化设计，功能解耦 +- 使用异步编程提高性能 +- 统一的错误处理机制 +- 完整的日志记录 + +### 数据处理 +- 数据验证和清洗 +- 标准化字段映射 +- 单位自动转换 +- 质量评分机制 + +### 安全考虑 +- 输入数据验证 +- 文件大小限制 +- 连接状态监控 +- 错误信息脱敏 + +## 测试和验证 + +### 数据验证 +```python +from src.utils.data_utils import data_converter + +# 验证传感器数据 +validation_result = data_converter.validate_sensor_data({ + "device_id": "device_001", + "data_type": "LL", + "value": 25.5, + "location": "A区" +}) + +print(validation_result) +``` + +### 批量导入测试 +```python +import asyncio +from src.batch.batch_import import batch_manager + +async def test_import(): + result = await batch_manager.import_file( + file_path="data.csv", + batch_id="test_batch", + data_source="test" + ) + print(result) + +asyncio.run(test_import()) +``` + +## 部署建议 + +### 1. 生产环境配置 +- 使用反向代理（Nginx） +- 配置SSL证书 +- 设置防火墙规则 +- 监控系统资源使用 + +### 2. 性能优化 +- 数据库连接池 +- 缓存机制 +- 异步处理优化 +- 连接数限制 + +### 3. 监控和日志 +- 应用性能监控 +- 错误日志收集 +- 性能指标统计 +- 告警机制 + +## 许可证 + +本项目遵循 MIT 许可证。 + +## 贡献指南 + +欢迎提交 Issue 和 Pull Request来贡献代码。 + +## 联系方式 + +如有问题，请通过以下方式联系： +- 邮箱：bot_dev1@xayunmei.com +- 项目地址：http://git.xayunmei.com/bot_ym/water-management-system \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 00000000..0d5632a8 --- /dev/null +++ b/main.py @@ -0,0 +1,305 @@ +""" +水务管理系统主程序 +集成REST API、WebSocket和批量导入功能 +""" +import asyncio +import logging +import signal +import sys +from pathlib import Path +import argparse +from datetime import datetime + +# 导入各个模块 +from src.api.rest_api import app as rest_api_app +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 + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler('water_management.log'), + logging.StreamHandler(sys.stdout) + ] +) +logger = logging.getLogger(__name__) + +class WaterManagementSystem: + """水务管理系统主类""" + + def __init__(self, config_file: str = "config.json"): + self.config_file = config_file + self.running = False + self.tasks = [] + self.config = self.load_config() + + def load_config(self) -> dict: + """加载配置文件""" + config_path = Path(self.config_file) + if config_path.exists(): + import json + with open(config_path, 'r', encoding='utf-8') as f: + return json.load(f) + else: + # 默认配置 + return { + "api": { + "host": "0.0.0.0", + "port": 8000 + }, + "websocket": { + "host": "0.0.0.0", + "port": 8765 + }, + "batch": { + "max_file_size_mb": 100, + "supported_formats": ["csv", "excel", "json"] + }, + "logging": { + "level": "INFO", + "file": "water_management.log" + } + } + + async def start_api_server(self): + """启动API服务器""" + import uvicorn + logger.info("启动REST API服务器...") + + # 在新的事件循环中运行uvicorn + api_config = uvicorn.Config( + app=rest_api_app, + host=self.config["api"]["host"], + port=self.config["api"]["port"], + log_level="info" + ) + api_server = uvicorn.Server(api_config) + + # 在后台任务中运行 + await api_server.serve() + + async def start_websocket_server(self): + """启动WebSocket服务器""" + logger.info("启动WebSocket服务器...") + + # 启动WebSocket服务器 + server = await websocket_server.start_server() + + # 添加服务器关闭处理 + def cleanup(): + logger.info("关闭WebSocket服务器...") + server.close() + asyncio.create_task(server.wait_closed()) + + return server, cleanup + + async def start_data_generator(self): + """启动数据生成器（用于演示）""" + logger.info("启动数据生成器...") + + while self.running: + # 生成模拟数据 + import random + sensor_types = ["LL", "YL", "SW", "ZD"] + sensor_type = random.choice(sensor_types) + + # 根据传感器类型生成合理的数值范围 + if sensor_type == "LL": # 流量 + value = random.uniform(10, 100) + elif sensor_type == "YL": # 压力 + value = random.uniform(0.1, 1.0) + elif sensor_type == "SW": # 水位 + value = random.uniform(0, 10) + else: # ZD 浊度 + value = random.uniform(0, 50) + + sensor_data = { + "data_type": sensor_type, + "device_id": f"device_{random.randint(1, 10)}", + "value": round(value, 2), + "location": random.choice(["A区", "B区", "C区", "D区"]) + } + + # 通过WebSocket发送数据 + await websocket_server.send_sensor_data(sensor_data) + + # 等待5秒 + await asyncio.sleep(5) + + async def handle_batch_import(self, file_path: str, batch_id: str, data_source: str): + """处理批量导入请求""" + try: + logger.info(f"开始批量导入: {file_path}") + + # 验证文件 + validation_result = await batch_manager.validate_file(file_path) + if not validation_result["valid"]: + raise Exception(f"文件验证失败: {validation_result['error']}") + + # 导入文件 + result = await batch_manager.import_file( + file_path=file_path, + batch_id=batch_id, + data_source=data_source, + file_type="auto" + ) + + logger.info(f"批量导入完成: {result}") + return result + + except Exception as e: + logger.error(f"批量导入失败: {str(e)}") + raise + + async def start_system(self): + """启动系统""" + logger.info("启动水务管理系统...") + + # 标记系统为运行状态 + self.running = True + + try: + # 启动WebSocket服务器 + ws_server, ws_cleanup = await self.start_websocket_server() + self.tasks.append(ws_server) + + # 启动数据生成器（如果启用） + if self.config.get("demo_mode", False): + generator_task = asyncio.create_task(self.start_data_generator()) + self.tasks.append(generator_task) + + # 启动API服务器 + await self.start_api_server() + + except Exception as e: + logger.error(f"系统启动失败: {str(e)}") + await self.stop_system() + raise + + async def stop_system(self): + """停止系统""" + logger.info("停止水务管理系统...") + + self.running = False + + # 取消所有任务 + for task in self.tasks: + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + # 清理WebSocket服务器 + if hasattr(websocket_server, 'server') and websocket_server.server: + websocket_server.server.close() + await websocket_server.server.wait_closed() + + logger.info("水务管理系统已停止") + + async def run(self): + """运行系统""" + # 设置信号处理 + def signal_handler(): + logger.info("收到停止信号...") + asyncio.create_task(self.stop_system()) + + for sig in [signal.SIGINT, signal.SIGTERM]: + signal.signal(sig, signal_handler) + + try: + await self.start_system() + + # 保持运行直到收到停止信号 + while self.running: + await asyncio.sleep(1) + + except KeyboardInterrupt: + logger.info("收到键盘中断信号") + except Exception as e: + logger.error(f"系统运行时出错: {str(e)}") + finally: + await self.stop_system() + +def create_sample_config(): + """创建示例配置文件""" + sample_config = { + "api": { + "host": "0.0.0.0", + "port": 8000 + }, + "websocket": { + "host": "0.0.0.0", + "port": 8765 + }, + "batch": { + "max_file_size_mb": 100, + "supported_formats": ["csv", "excel", "json"] + }, + "demo_mode": True, + "logging": { + "level": "INFO", + "file": "water_management.log" + } + } + + import json + with open("config.json", 'w', encoding='utf-8') as f: + json.dump(sample_config, f, indent=2, ensure_ascii=False) + + logger.info("示例配置文件已创建: config.json") + +def create_requirements(): + """创建requirements.txt文件""" + requirements = [ + "fastapi==0.104.1", + "uvicorn[standard]==0.24.0", + "websockets==12.0", + "pandas==2.1.3", + "openpyxl==3.1.2", + "aiofiles==23.2.1", + "python-multipart==0.0.6", + "jinja2==3.1.2" + ] + + with open("requirements.txt", 'w') as f: + f.write('\n'.join(requirements)) + + logger.info("依赖文件已创建: requirements.txt") + +def main(): + """主函数""" + parser = argparse.ArgumentParser(description="水务管理系统") + parser.add_argument("--config", "-c", default="config.json", help="配置文件路径") + parser.add_argument("--init", action="store_true", help="初始化项目（创建配置文件和依赖）") + parser.add_argument("--demo", action="store_true", help="启动演示模式") + + args = parser.parse_args() + + if args.init: + create_sample_config() + create_requirements() + logger.info("项目初始化完成") + return + + # 创建系统实例 + system = WaterManagementSystem(args.config) + + # 如果启用演示模式 + if args.demo: + system.config["demo_mode"] = True + logger.info("启用演示模式") + + # 运行系统 + try: + asyncio.run(system.run()) + except KeyboardInterrupt: + logger.info("程序已退出") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 00000000..d276217d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,10 @@ +fastapi==0.104.1 +uvicorn[standard]==0.24.0 +websockets==12.0 +pandas==2.1.3 +openpyxl==3.1.2 +aiofiles==23.2.1 +python-multipart==0.0.6 +jinja2==3.1.2 +requests==2.31.0 +python-dateutil==2.8.2 \ No newline at end of file diff --git a/src/api/rest_api.py b/src/api/rest_api.py new file mode 100644 index 00000000..d1180989 --- /dev/null +++ b/src/api/rest_api.py @@ -0,0 +1,181 @@ +""" +REST API 数据接入模块 +支持 IoT 设备数据、手动录入和 API 批量导入 +""" +from fastapi import FastAPI, HTTPException, Depends +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel +from typing import List, Optional, Dict, Any +import uvicorn +import asyncio +import json +from datetime import datetime + +# 创建FastAPI应用 +app = FastAPI(title="Water Management System Data API", version="1.0.0") + +# CORS配置 +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# 数据模型 +class IoTData(BaseModel): + device_id: str + data_type: str # "LL", "YL", "SW", "ZD" 等 + value: float + timestamp: datetime + location: str + +class ManualInputData(BaseModel): + source: str + data_type: str + value: float + timestamp: datetime + operator: str + notes: Optional[str] = None + +class BatchImportRequest(BaseModel): + batch_id: str + data_source: str + records: List[Dict[str, Any]] + +# 数据存储（示例，实际应该用数据库） +data_store = [] + +@app.get("/") +async def root(): + """API根路径""" + return {"message": "Water Management System API", "version": "1.0.0"} + +@app.post("/api/iot/data") +async def receive_iot_data(data: IoTData): + """接收IoT设备数据""" + try: + data_dict = data.dict() + data_store.append({ + **data_dict, + "id": len(data_store) + 1, + "type": "iot" + }) + return {"status": "success", "id": len(data_store), "message": "IoT data received"} + except Exception as e: + raise HTTPException(status_code=400, detail=str(e)) + +@app.post("/api/manual/data") +async def receive_manual_data(data: ManualInputData): + """接收手动录入数据""" + try: + data_dict = data.dict() + data_store.append({ + **data_dict, + "id": len(data_store) + 1, + "type": "manual" + }) + return {"status": "success", "id": len(data_store), "message": "Manual data received"} + except Exception as e: + raise HTTPException(status_code=400, detail=str(e)) + +@app.post("/api/batch/import") +async def batch_import(request: BatchImportRequest): + """批量导入数据""" + try: + imported_count = 0 + failed_count = 0 + + for record in request.records: + # 验证记录 + if not all(k in record for k in ['device_id', 'data_type', 'value']): + failed_count += 1 + continue + + # 创建数据对象 + data_record = { + "device_id": record['device_id'], + "data_type": record['data_type'], + "value": float(record['value']), + "timestamp": record.get('timestamp', datetime.now()), + "location": record.get('location', 'unknown'), + "batch_id": request.batch_id, + "source": request.data_source, + "id": len(data_store) + 1, + "type": "batch" + } + + data_store.append(data_record) + imported_count += 1 + + return { + "status": "success", + "imported_count": imported_count, + "failed_count": failed_count, + "message": f"Batch import completed: {imported_count} records imported, {failed_count} failed" + } + except Exception as e: + raise HTTPException(status_code=400, detail=str(e)) + +@app.get("/api/data/{data_type}") +async def get_data_by_type(data_type: str, limit: int = 100, offset: int = 0): + """根据数据类型获取数据""" + filtered_data = [ + item for item in data_store + if item.get('data_type') == data_type + ] + + return { + "data": filtered_data[offset:offset+limit], + "total": len(filtered_data), + "limit": limit, + "offset": offset + } + +@app.get("/api/data/recent") +async def get_recent_data(hours: int = 24, limit: int = 100): + """获取最近的数据""" + from datetime import timedelta + cutoff_time = datetime.now() - timedelta(hours=hours) + + recent_data = [ + item for item in data_store + if item.get('timestamp', datetime.now()) > cutoff_time + ] + + return { + "data": recent_data[-limit:], + "total": len(recent_data), + "hours": hours, + "limit": limit + } + +@app.get("/api/stats") +async def get_statistics(): + """获取数据统计信息""" + stats = { + "total_records": len(data_store), + "by_type": {}, + "by_device": {}, + "by_hour": {} + } + + for item in data_store: + data_type = item.get('data_type', 'unknown') + device_id = item.get('device_id', 'unknown') + hour = item.get('timestamp', datetime.now()).strftime('%Y-%m-%d %H:00:00') + + stats['by_type'][data_type] = stats['by_type'].get(data_type, 0) + 1 + stats['by_device'][device_id] = stats['by_device'].get(device_id, 0) + 1 + stats['by_hour'][hour] = stats['by_hour'].get(hour, 0) + 1 + + return stats + +@app.get("/health") +async def health_check(): + """健康检查""" + return {"status": "healthy", "timestamp": datetime.now()} + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file diff --git a/src/batch/batch_import.py b/src/batch/batch_import.py new file mode 100644 index 00000000..7fec50d3 --- /dev/null +++ b/src/batch/batch_import.py @@ -0,0 +1,336 @@ +""" +批量数据导入模块 +支持CSV、Excel、JSON等多种格式的批量数据导入 +""" +import pandas as pd +import json +import csv +import asyncio +import aiofiles +from typing import List, Dict, Any, Optional, Union +from pathlib import Path +from datetime import datetime +import logging +from concurrent.futures import ThreadPoolExecutor + +# 配置日志 +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +class BatchImportError(Exception): + """批量导入异常""" + pass + +class DataValidator: + """数据验证器""" + + # 水利行业标准字段映射 + STANDARD_FIELDS = { + "LL": "流量", + "YL": "压力", + "SW": "水位", + "ZD": "浊度", + "PH": "pH值", + "WD": "温度", + "DD": "电导率", + "YD": "硬度" + } + + # 单位映射 + UNIT_MAP = { + "LL": "m³/h", + "YL": "MPa", + "SW": "m", + "ZD": "NTU", + "PH": "", + "WD": "°C", + "DD": "μS/cm", + "YD": "mg/L" + } + + @classmethod + def validate_data_type(cls, data_type: str) -> bool: + """验证数据类型是否有效""" + return data_type in cls.STANDARD_FIELDS + + @classmethod + def get_field_description(cls, data_type: str) -> str: + """获取字段描述""" + return cls.STANDARD_FIELDS.get(data_type, "未知类型") + + @classmethod + def get_unit(cls, data_type: str) -> str: + """获取单位""" + return cls.UNIT_MAP.get(data_type, "") + + @classmethod + def validate_record(cls, record: Dict[str, Any]) -> Dict[str, Any]: + """验证单条记录""" + errors = [] + validated_record = {} + + # 必需字段检查 + required_fields = ["device_id", "data_type", "value"] + for field in required_fields: + if field not in record: + errors.append(f"缺少必需字段: {field}") + else: + validated_record[field] = record[field] + + # 数据类型验证 + if "data_type" in validated_record: + if not cls.validate_data_type(validated_record["data_type"]): + errors.append(f"无效的数据类型: {validated_record['data_type']}") + + # 数值验证 + if "value" in validated_record: + try: + validated_record["value"] = float(validated_record["value"]) + except (ValueError, TypeError): + errors.append(f"无效的数值: {validated_record['value']}") + + # 时间戳处理 + if "timestamp" in record: + try: + if isinstance(record["timestamp"], str): + validated_record["timestamp"] = datetime.fromisoformat(record["timestamp"]) + else: + validated_record["timestamp"] = record["timestamp"] + except (ValueError, TypeError): + # 如果时间戳无效，使用当前时间 + validated_record["timestamp"] = datetime.now() + else: + validated_record["timestamp"] = datetime.now() + + # 地点字段处理 + validated_record["location"] = record.get("location", "未知") + + return { + "validated": len(errors) == 0, + "record": validated_record, + "errors": errors + } + +class BatchImporter: + """批量导入器""" + + def __init__(self): + self.validator = DataValidator() + + async def import_csv(self, file_path: str, batch_id: str, data_source: str) -> Dict[str, Any]: + """导入CSV文件""" + try: + # 使用线程池执行文件读取 + loop = asyncio.get_event_loop() + with ThreadPoolExecutor() as executor: + df = await loop.run_in_executor( + executor, + lambda: pd.read_csv(file_path, encoding='utf-8') + ) + + return await self._process_dataframe(df, batch_id, data_source) + + except Exception as e: + raise BatchImportError(f"CSV文件导入失败: {str(e)}") + + async def import_excel(self, file_path: str, batch_id: str, data_source: str, sheet_name: str = 0) -> Dict[str, Any]: + """导入Excel文件""" + try: + # 使用线程池执行文件读取 + loop = asyncio.get_event_loop() + with ThreadPoolExecutor() as executor: + df = await loop.run_in_executor( + executor, + lambda: pd.read_excel(file_path, sheet_name=sheet_name) + ) + + return await self._process_dataframe(df, batch_id, data_source) + + except Exception as e: + raise BatchImportError(f"Excel文件导入失败: {str(e)}") + + async def import_json(self, file_path: str, batch_id: str, data_source: str) -> Dict[str, Any]: + """导入JSON文件""" + try: + async with aiofiles.open(file_path, 'r', encoding='utf-8') as f: + content = await f.read() + + data = json.loads(content) + + # 处理不同的JSON格式 + if isinstance(data, list): + return await self._process_records(data, batch_id, data_source) + elif isinstance(data, dict): + if "records" in data: + return await self._process_records(data["records"], batch_id, data_source) + else: + return await self._process_records([data], batch_id, data_source) + else: + raise BatchImportError("不支持的JSON格式") + + except Exception as e: + raise BatchImportError(f"JSON文件导入失败: {str(e)}") + + async def _process_dataframe(self, df: pd.DataFrame, batch_id: str, data_source: str) -> Dict[str, Any]: + """处理DataFrame数据""" + # 转换为字典列表 + records = df.to_dict('records') + return await self._process_records(records, batch_id, data_source) + + async def _process_records(self, records: List[Dict[str, Any]], batch_id: str, data_source: str) -> Dict[str, Any]: + """处理记录列表""" + import_count = 0 + error_count = 0 + errors = [] + imported_records = [] + + for i, record in enumerate(records): + validation_result = self.validator.validate_record(record) + + if validation_result["validated"]: + # 添加批次信息 + import_record = { + **validation_result["record"], + "batch_id": batch_id, + "data_source": data_source, + "import_time": datetime.now(), + "type": "batch" + } + + imported_records.append(import_record) + import_count += 1 + else: + error_count += 1 + error_msg = f"记录 {i+1}: {', '.join(validation_result['errors'])}" + errors.append(error_msg) + logger.warning(error_msg) + + # 保存导入的记录到文件（实际项目中应该保存到数据库） + await self._save_imported_records(imported_records) + + return { + "status": "completed" if error_count == 0 else "completed_with_errors", + "imported_count": import_count, + "error_count": error_count, + "total_count": len(records), + "success_rate": import_count / len(records) if records else 0, + "batch_id": batch_id, + "data_source": data_source, + "errors": errors[:10], # 只返回前10个错误 + "imported_records": imported_records[:5] # 返回前5条记录作为示例 + } + + async def _save_imported_records(self, records: List[Dict[str, Any]]): + """保存导入的记录""" + # 这里可以将记录保存到数据库或文件 + # 为了示例，我们只保存到日志 + for record in records: + logger.info(f"导入记录: {record}") + + async def get_import_summary(self, batch_id: str) -> Dict[str, Any]: + """获取导入摘要""" + # 这里应该从数据库查询批次信息 + # 为了示例，返回一个空摘要 + return { + "batch_id": batch_id, + "status": "not_found", + "message": "批次信息未找到（示例实现）" + } + +class BatchImportManager: + """批量导入管理器""" + + def __init__(self): + self.importer = BatchImporter() + + async def import_file(self, file_path: str, batch_id: str, data_source: str, + file_type: str = "auto", **kwargs) -> Dict[str, Any]: + """导入文件""" + file_path_obj = Path(file_path) + + if not file_path_obj.exists(): + raise BatchImportError(f"文件不存在: {file_path}") + + # 自动检测文件类型 + if file_type == "auto": + if file_path_obj.suffix.lower() == '.csv': + file_type = "csv" + elif file_path_obj.suffix.lower() in ['.xlsx', '.xls']: + file_type = "excel" + elif file_path_obj.suffix.lower() == '.json': + file_type = "json" + else: + raise BatchImportError(f"不支持的文件类型: {file_path_obj.suffix}") + + logger.info(f"开始导入{file_type}文件: {file_path}") + + if file_type == "csv": + result = await self.importer.import_csv(file_path, batch_id, data_source) + elif file_type == "excel": + sheet_name = kwargs.get("sheet_name", 0) + result = await self.importer.import_excel(file_path, batch_id, data_source, sheet_name) + elif file_type == "json": + result = await self.importer.import_json(file_path, batch_id, data_source) + else: + raise BatchImportError(f"不支持的文件类型: {file_type}") + + logger.info(f"文件导入完成: {result}") + return result + + async def validate_file(self, file_path: str) -> Dict[str, Any]: + """验证文件格式""" + file_path_obj = Path(file_path) + + if not file_path_obj.exists(): + return {"valid": False, "error": "文件不存在"} + + file_size = file_path_obj.stat().st_size + if file_size > 100 * 1024 * 1024: # 100MB限制 + return {"valid": False, "error": "文件过大，最大支持100MB"} + + # 尝试读取文件前几行进行验证 + try: + with open(file_path, 'r', encoding='utf-8') as f: + first_line = f.readline() + if not first_line: + return {"valid": False, "error": "文件为空"} + except Exception as e: + return {"valid": False, "error": f"无法读取文件: {str(e)}"} + + return {"valid": True, "size": file_size, "format": file_path_obj.suffix.lower()} + +# 全局导入管理器实例 +batch_manager = BatchImportManager() + +# 示例用法 +async def example_usage(): + """示例用法""" + import os + + # 创建示例CSV文件 + sample_data = [ + {"device_id": "device_001", "data_type": "LL", "value": 25.5, "location": "A区"}, + {"device_id": "device_002", "data_type": "YL", "value": 0.8, "location": "B区"}, + {"device_id": "device_003", "data_type": "SW", "value": 5.2, "location": "C区"} + ] + + sample_file = "/tmp/sample_data.csv" + with open(sample_file, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=["device_id", "data_type", "value", "location"]) + writer.writeheader() + writer.writerows(sample_data) + + # 导入文件 + try: + result = await batch_manager.import_file( + file_path=sample_file, + batch_id="batch_" + datetime.now().strftime("%Y%m%d_%H%M%S"), + data_source="manual_test", + file_type="csv" + ) + print("导入结果:", result) + except Exception as e: + print(f"导入失败: {str(e)}") + +if __name__ == "__main__": + asyncio.run(example_usage()) \ No newline at end of file diff --git a/src/models/models.py b/src/models/models.py new file mode 100644 index 00000000..0126cca4 --- /dev/null +++ b/src/models/models.py @@ -0,0 +1,317 @@ +""" +数据模型定义 +定义水务管理系统的各种数据结构 +""" +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Any +from datetime import datetime +from enum import Enum + +class DataType(Enum): + """数据类型枚举""" + LL = "LL" # 流量 + YL = "YL" # 压力 + SW = "SW" # 水位 + ZD = "ZD" # 浊度 + PH = "PH" # pH值 + WD = "WD" # 温度 + DD = "DD" # 电导率 + YD = "YD" # 硬度 + +class AlertLevel(Enum): + """警报级别枚举""" + INFO = "info" + WARNING = "warning" + ERROR = "error" + CRITICAL = "critical" + +@dataclass +class Device: + """设备模型""" + id: str + name: str + device_type: str + location: str + description: Optional[str] = None + install_date: Optional[datetime] = None + status: str = "active" # active, inactive, maintenance + metadata: Dict[str, Any] = field(default_factory=dict) + +@dataclass +class SensorData: + """传感器数据模型""" + id: str + device_id: str + data_type: DataType + value: float + unit: str + timestamp: datetime + location: str + quality_score: float = 1.0 + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "device_id": self.device_id, + "data_type": self.data_type.value, + "value": self.value, + "unit": self.unit, + "timestamp": self.timestamp.isoformat(), + "location": self.location, + "quality_score": self.quality_score, + "metadata": self.metadata + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'SensorData': + """从字典创建对象""" + return cls( + id=data["id"], + device_id=data["device_id"], + data_type=DataType(data["data_type"]), + value=float(data["value"]), + unit=data.get("unit", ""), + timestamp=datetime.fromisoformat(data["timestamp"]), + location=data.get("location", ""), + quality_score=float(data.get("quality_score", 1.0)), + metadata=data.get("metadata", {}) + ) + +@dataclass +class Alert: + """警报模型""" + id: str + device_id: str + alert_type: str + level: AlertLevel + message: str + timestamp: datetime + resolved: bool = False + resolved_by: Optional[str] = None + resolved_at: Optional[datetime] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "device_id": self.device_id, + "alert_type": self.alert_type, + "level": self.level.value, + "message": self.message, + "timestamp": self.timestamp.isoformat(), + "resolved": self.resolved, + "resolved_by": self.resolved_by, + "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None, + "metadata": self.metadata + } + +@dataclass +class BatchImport: + """批量导入记录模型""" + id: str + batch_id: str + data_source: str + total_records: int + successful_records: int + failed_records: int + status: str # pending, processing, completed, failed + file_name: Optional[str] = None + import_time: Optional[datetime] = None + completed_time: Optional[datetime] = None + error_messages: List[str] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "batch_id": self.batch_id, + "data_source": self.data_source, + "total_records": self.total_records, + "successful_records": self.successful_records, + "failed_records": self.failed_records, + "status": self.status, + "file_name": self.file_name, + "import_time": self.import_time.isoformat() if self.import_time else None, + "completed_time": self.completed_time.isoformat() if self.completed_time else None, + "error_messages": self.error_messages, + "metadata": self.metadata + } + +@dataclass +class APIRequest: + """API请求模型""" + id: str + method: str + endpoint: str + params: Dict[str, Any] + headers: Dict[str, Any] + body: Optional[Any] = None + timestamp: datetime = field(default_factory=datetime.now) + response_code: Optional[int] = None + response_time_ms: Optional[float] = None + response_body: Optional[Any] = None + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "method": self.method, + "endpoint": self.endpoint, + "params": self.params, + "headers": self.headers, + "body": self.body, + "timestamp": self.timestamp.isoformat(), + "response_code": self.response_code, + "response_time_ms": self.response_time_ms, + "response_body": self.response_body + } + +@dataclass +class WebSocketConnection: + """WebSocket连接模型""" + id: str + client_ip: str + connected_at: datetime + disconnected_at: Optional[datetime] = None + subscriptions: List[str] = field(default_factory=list) + message_count: int = 0 + last_message_at: Optional[datetime] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "id": self.id, + "client_ip": self.client_ip, + "connected_at": self.connected_at.isoformat(), + "disconnected_at": self.disconnected_at.isoformat() if self.disconnected_at else None, + "subscriptions": self.subscriptions, + "message_count": self.message_count, + "last_message_at": self.last_message_at.isoformat() if self.last_message_at else None, + "metadata": self.metadata + } + +@dataclass +class SystemStats: + """系统统计模型""" + timestamp: datetime + total_records: int + total_devices: int + active_connections: int + api_requests_count: int + alerts_count: int + data_quality_score: float + memory_usage_mb: float + cpu_usage_percent: float + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return { + "timestamp": self.timestamp.isoformat(), + "total_records": self.total_records, + "total_devices": self.total_devices, + "active_connections": self.active_connections, + "api_requests_count": self.api_requests_count, + "alerts_count": self.alerts_count, + "data_quality_score": self.data_quality_score, + "memory_usage_mb": self.memory_usage_mb, + "cpu_usage_percent": self.cpu_usage_percent + } + +class DataValidator: + """数据验证器""" + + @staticmethod + def validate_sensor_data(data: Dict[str, Any]) -> List[str]: + """验证传感器数据""" + errors = [] + + # 必需字段检查 + required_fields = ["device_id", "data_type", "value", "location"] + for field in required_fields: + if field not in data: + errors.append(f"缺少必需字段: {field}") + + # 数据类型验证 + if "data_type" in data: + try: + DataType(data["data_type"]) + except ValueError: + errors.append(f"无效的数据类型: {data['data_type']}") + + # 数值验证 + if "value" in data: + try: + value = float(data["value"]) + # 根据数据类型进行数值范围检查 + data_type = data.get("data_type") + if data_type == "LL" and value < 0: + errors.append("流量不能为负数") + elif data_type == "YL" and value < 0: + errors.append("压力不能为负数") + elif data_type == "SW" and value < 0: + errors.append("水位不能为负数") + except (ValueError, TypeError): + errors.append(f"无效的数值: {data['value']}") + + # 时间戳验证 + if "timestamp" in data: + try: + if isinstance(data["timestamp"], str): + datetime.fromisoformat(data["timestamp"]) + except (ValueError, TypeError): + errors.append(f"无效的时间戳格式: {data['timestamp']}") + + return errors + + @staticmethod + def validate_device_data(data: Dict[str, Any]) -> List[str]: + """验证设备数据""" + errors = [] + + # 必需字段检查 + required_fields = ["id", "name", "device_type", "location"] + for field in required_fields: + if field not in data: + errors.append(f"缺少必需字段: {field}") + + # 设备ID格式验证 + if "id" in data: + device_id = data["id"] + if not isinstance(device_id, str) or not device_id.strip(): + errors.append("设备ID不能为空") + elif len(device_id) > 50: + errors.append("设备ID长度不能超过50个字符") + + # 状态验证 + if "status" in data and data["status"] not in ["active", "inactive", "maintenance"]: + errors.append("设备状态必须是: active, inactive, maintenance") + + return errors + + @staticmethod + def validate_alert_data(data: Dict[str, Any]) -> List[str]: + """验证警报数据""" + errors = [] + + # 必需字段检查 + required_fields = ["device_id", "alert_type", "level", "message"] + for field in required_fields: + if field not in data: + errors.append(f"缺少必需字段: {field}") + + # 警报级别验证 + if "level" in data: + try: + AlertLevel(data["level"]) + except ValueError: + errors.append(f"无效的警报级别: {data['level']}") + + return errors + +# 全局验证器实例 +validator = DataValidator() \ No newline at end of file diff --git a/src/utils/data_utils.py b/src/utils/data_utils.py new file mode 100644 index 00000000..4a336d38 --- /dev/null +++ b/src/utils/data_utils.py @@ -0,0 +1,409 @@ +""" +数据处理工具模块 +提供数据验证、转换、格式化等工具函数 +""" +import json +import csv +import pandas as pd +from typing import Dict, List, Any, Optional, Union +from datetime import datetime, timedelta +import hashlib +import logging + +# 配置日志 +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +class DataConverter: + """数据转换器""" + + # 水利行业标准字段映射 + FIELD_MAPPING = { + "流量": "LL", + "压力": "YL", + "水位": "SW", + "浊度": "ZD", + "pH值": "PH", + "温度": "WD", + "电导率": "DD", + "硬度": "YD", + # 支持常见的中文字段名 + "流量计": "LL", + "压力表": "YL", + "水位计": "SW", + "浊度仪": "ZD", + "pH计": "PH", + "温度计": "WD", + "电导率仪": "DD", + "硬度计": "YD" + } + + # 单位转换 + UNIT_CONVERSIONS = { + # 流量单位转换 (m³/h) + "m³/h": 1.0, + "L/s": 3.6, # L/s = m³/h / 1000 * 3600 + "m³/d": 1/24, # m³/d = m³/h / 24 + "L/min": 1/60, # L/min = m³/h / 1000 * 60 + + # 压力单位转换 (MPa) + "MPa": 1.0, + "kPa": 0.001, # kPa = MPa / 1000 + "bar": 0.1, # bar = MPa * 10 + "kgf/cm²": 0.0980665, # kgf/cm² = MPa / 0.0980665 + + # 水位单位转换 (m) + "m": 1.0, + "cm": 0.01, # cm = m / 100 + "mm": 0.001, # mm = m / 1000 + + # 浊度单位转换 (NTU) + "NTU": 1.0, + "FNU": 1.0, # FNU ≈ NTU + + # pH值单位转换 + "pH": 1.0, + + # 温度单位转换 (°C) + "°C": 1.0, + "K": 1.0, # 相对差值 + "°F": lambda x: (x - 32) / 1.8, # °F to °C + + # 电导率单位转换 (μS/cm) + "μS/cm": 1.0, + "mS/cm": 1000, # mS/cm = μS/cm * 1000 + "S/m": 10000 # S/m = μS/cm * 100 + } + + @classmethod + def normalize_field_name(cls, field_name: str) -> str: + """标准化字段名""" + if not field_name: + return "" + + field_name = field_name.strip().upper() + + # 如果已经是标准格式，直接返回 + if field_name in cls.FIELD_MAPPING.values(): + return field_name + + # 查映射表 + if field_name in cls.FIELD_MAPPING: + return cls.FIELD_MAPPING[field_name] + + # 英文映射 + english_mapping = { + "flow": "LL", + "pressure": "YL", + "level": "SW", + "turbidity": "ZD", + "ph": "PH", + "temperature": "WD", + "conductivity": "DD", + "hardness": "YD" + } + + if field_name.lower() in english_mapping: + return english_mapping[field_name.lower()] + + return field_name + + @classmethod + def convert_unit(cls, value: float, from_unit: str, to_unit: str) -> float: + """单位转换""" + if from_unit == to_unit: + return value + + if from_unit not in cls.UNIT_CONVERSIONS: + raise ValueError(f"不支持的单位: {from_unit}") + + if to_unit not in cls.UNIT_CONVERSIONS: + raise ValueError(f"不支持的目标单位: {to_unit}") + + from_conv = cls.UNIT_CONVERSIONS[from_unit] + to_conv = cls.UNIT_CONVERSIONS[to_unit] + + if callable(from_conv): + value = from_conv(value) + + if callable(to_conv): + return value / to_conv + else: + return value * (to_conv / from_conv) + + @classmethod + def validate_sensor_data(cls, data: Dict[str, Any]) -> Dict[str, Any]: + """验证传感器数据""" + errors = [] + validated_data = {} + + # 必需字段验证 + required_fields = ["device_id", "data_type", "value"] + for field in required_fields: + if field not in data: + errors.append(f"缺少必需字段: {field}") + else: + validated_data[field] = data[field] + + # 数据类型验证和标准化 + if "data_type" in validated_data: + original_type = validated_data["data_type"] + validated_data["data_type"] = cls.normalize_field_name(original_type) + + if validated_data["data_type"] != original_type: + logger.info(f"字段名标准化: {original_type} -> {validated_data['data_type']}") + + # 数值验证 + if "value" in validated_data: + try: + validated_data["value"] = float(validated_data["value"]) + # 检查数值范围 + data_type = validated_data.get("data_type", "") + if data_type == "LL" and validated_data["value"] < 0: + errors.append("流量不能为负数") + elif data_type == "YL" and validated_data["value"] < 0: + errors.append("压力不能为负数") + elif data_type == "SW" and validated_data["value"] < 0: + errors.append("水位不能为负数") + except (ValueError, TypeError): + errors.append(f"无效的数值: {validated_data['value']}") + + # 地点验证 + if "location" not in validated_data or not validated_data["location"]: + validated_data["location"] = "未知" + + # 时间戳处理 + if "timestamp" in data: + try: + if isinstance(data["timestamp"], str): + validated_data["timestamp"] = datetime.fromisoformat(data["timestamp"]) + else: + validated_data["timestamp"] = data["timestamp"] + except (ValueError, TypeError): + validated_data["timestamp"] = datetime.now() + else: + validated_data["timestamp"] = datetime.now() + + return { + "valid": len(errors) == 0, + "data": validated_data, + "errors": errors + } + +class DataFormatter: + """数据格式化器""" + + @staticmethod + def format_sensor_data(data: Dict[str, Any], format_type: str = "json") -> str: + """格式化传感器数据""" + if format_type == "json": + return json.dumps(data, ensure_ascii=False, indent=2) + elif format_type == "csv": + # CSV格式只包含关键字段 + csv_fields = ["device_id", "data_type", "value", "location", "timestamp"] + csv_data = {k: data.get(k, "") for k in csv_fields} + import io + output = io.StringIO() + writer = csv.DictWriter(output, fieldnames=csv_fields) + writer.writeheader() + writer.writerow(csv_data) + return output.getvalue() + else: + raise ValueError(f"不支持的格式类型: {format_type}") + + @staticmethod + def format_statistics(stats: Dict[str, Any], format_type: str = "text") -> str: + """格式化统计数据""" + if format_type == "json": + return json.dumps(stats, ensure_ascii=False, indent=2) + elif format_type == "text": + lines = ["数据统计报告", "=" * 20] + lines.append(f"总记录数: {stats.get('total_records', 0)}") + + if "by_type" in stats: + lines.append("\n按数据类型统计:") + for data_type, count in stats["by_type"].items(): + lines.append(f" {data_type}: {count} 条") + + if "by_device" in stats: + lines.append("\n按设备统计:") + for device_id, count in list(stats["by_device"].items())[:10]: # 只显示前10个 + lines.append(f" {device_id}: {count} 条") + + return "\n".join(lines) + else: + raise ValueError(f"不支持的格式类型: {format_type}") + +class DataHasher: + """数据哈希工具""" + + @staticmethod + def calculate_data_hash(data: Dict[str, Any]) -> str: + """计算数据哈希值""" + # 将数据转换为字符串 + data_str = json.dumps(data, sort_keys=True, ensure_ascii=False) + + # 计算MD5哈希 + hash_md5 = hashlib.md5(data_str.encode()) + return hash_md5.hexdigest() + + @staticmethod + def generate_data_id(device_id: str, data_type: str, timestamp: datetime) -> str: + """生成数据ID""" + # 使用设备ID、数据类型和时间戳生成唯一ID + time_str = timestamp.strftime("%Y%m%d_%H%M%S") + hash_input = f"{device_id}_{data_type}_{time_str}" + hash_md5 = hashlib.md5(hash_input.encode()) + return f"{data_type}_{device_id}_{hash_md5.hexdigest()[:8]}" + +class DataQualityChecker: + """数据质量检查器""" + + @staticmethod + def check_data_quality(records: List[Dict[str, Any]]) -> Dict[str, Any]: + """检查数据质量""" + quality_report = { + "total_records": len(records), + "valid_records": 0, + "invalid_records": 0, + "quality_score": 0, + "issues": [], + "statistics": {} + } + + if not records: + quality_report["quality_score"] = 0 + return quality_report + + valid_records = [] + + for record in records: + issues = [] + + # 检查必需字段 + required_fields = ["device_id", "data_type", "value"] + for field in required_fields: + if field not in record or not record[field]: + issues.append(f"缺少必需字段: {field}") + + # 检查数据类型 + if "data_type" in record and record["data_type"]: + valid_types = ["LL", "YL", "SW", "ZD", "PH", "WD", "DD", "YD"] + if record["data_type"] not in valid_types: + issues.append(f"无效的数据类型: {record['data_type']}") + + # 检查数值范围 + if "value" in record and record["value"]: + try: + value = float(record["value"]) + data_type = record.get("data_type", "") + + if data_type == "LL" and value < 0: + issues.append("流量不能为负数") + elif data_type == "YL" and value < 0: + issues.append("压力不能为负数") + elif data_type == "SW" and value < 0: + issues.append("水位不能为负数") + + # 检查异常值 + if data_type == "LL" and value > 10000: + issues.append("流量值异常大") + elif data_type == "YL" and value > 10: + issues.append("压力值异常大") + + except (ValueError, TypeError): + issues.append("无效的数值格式") + + if not issues: + valid_records.append(record) + quality_report["valid_records"] += 1 + else: + quality_report["invalid_records"] += 1 + quality_report["issues"].extend(issues) + + # 计算质量分数 + quality_report["quality_score"] = quality_report["valid_records"] / len(records) + + # 统计信息 + if records: + quality_report["statistics"] = { + "completeness": quality_report["valid_records"] / len(records), + "uniqueness": len(set(r.get("device_id", "") for r in valid_records)) / len(valid_records) if valid_records else 0, + "timeliness": quality_report.calculate_timeliness(records) + } + + return quality_report + + @staticmethod + def calculate_timeliness(records: List[Dict[str, Any]]) -> float: + """计算数据及时性（24小时内的数据比例）""" + if not records: + return 0 + + now = datetime.now() + recent_count = 0 + + for record in records: + timestamp = record.get("timestamp") + if timestamp: + try: + if isinstance(timestamp, str): + timestamp = datetime.fromisoformat(timestamp) + + time_diff = now - timestamp + if time_diff <= timedelta(hours=24): + recent_count += 1 + except: + pass + + return recent_count / len(records) + +class DataExporter: + """数据导出工具""" + + @staticmethod + def export_to_csv(records: List[Dict[str, Any]], file_path: str) -> bool: + """导出为CSV文件""" + try: + if not records: + return False + + # 获取所有字段 + all_fields = set() + for record in records: + all_fields.update(record.keys()) + + # 排序字段 + field_order = ["device_id", "data_type", "value", "location", "timestamp"] + for field in all_fields: + if field not in field_order: + field_order.append(field) + + with open(file_path, 'w', newline='', encoding='utf-8') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=field_order) + writer.writeheader() + writer.writerows(records) + + return True + + except Exception as e: + logger.error(f"导出CSV失败: {str(e)}") + return False + + @staticmethod + def export_to_json(records: List[Dict[str, Any]], file_path: str) -> bool: + """导出为JSON文件""" + try: + with open(file_path, 'w', encoding='utf-8') as jsonfile: + json.dump(records, jsonfile, ensure_ascii=False, indent=2, default=str) + return True + except Exception as e: + logger.error(f"导出JSON失败: {str(e)}") + return False + +# 全局工具实例 +data_converter = DataConverter() +data_formatter = DataFormatter() +data_hasher = DataHasher() +quality_checker = DataQualityChecker() +data_exporter = DataExporter() \ No newline at end of file diff --git a/src/websocket/websocket_server.py b/src/websocket/websocket_server.py new file mode 100644 index 00000000..36d8d274 --- /dev/null +++ b/src/websocket/websocket_server.py @@ -0,0 +1,214 @@ +""" +WebSocket 实时数据推送服务器 +支持实时数据推送、连接管理和数据广播 +""" +import asyncio +import json +import websockets +from datetime import datetime +from typing import Set, Dict, Any +import logging + +# 配置日志 +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +class WebSocketServer: + """WebSocket服务器类""" + + def __init__(self, host: str = "0.0.0.0", port: int = 8765): + self.host = host + self.port = port + self.clients: Set[websockets.WebSocketServerProtocol] = set() + self.data_history: list = [] # 存储最近的数据用于新连接 + + async def register_client(self, websocket: websockets.WebSocketServerProtocol): + """注册新客户端""" + self.clients.add(websocket) + client_ip = websocket.remote_address[0] + logger.info(f"新客户端连接: {client_ip}") + + # 发送历史数据给新连接的客户端 + if self.data_history: + await websocket.send(json.dumps({ + "type": "history", + "data": self.data_history[-50:] # 发送最近50条数据 + })) + + # 发送欢迎消息 + await websocket.send(json.dumps({ + "type": "welcome", + "message": "已连接到水务管理系统实时数据服务器", + "timestamp": datetime.now().isoformat() + })) + + async def unregister_client(self, websocket: websockets.WebSocketServerProtocol): + """注销客户端""" + if websocket in self.clients: + self.clients.remove(websocket) + client_ip = websocket.remote_address[0] + logger.info(f"客户端断开连接: {client_ip}") + + async def broadcast_data(self, data: Dict[str, Any]): + """广播数据到所有连接的客户端""" + if not self.clients: + return + + # 添加时间戳 + data["timestamp"] = datetime.now().isoformat() + + # 保存历史数据 + self.data_history.append(data) + if len(self.data_history) > 1000: # 只保留最近1000条记录 + self.data_history.pop(0) + + # 广播数据 + message = json.dumps(data) + disconnected_clients = [] + + for client in self.clients: + try: + await client.send(message) + except websockets.exceptions.ConnectionClosed: + disconnected_clients.append(client) + + # 清理已断开的连接 + for client in disconnected_clients: + await self.unregister_client(client) + + async def handle_client_message(self, websocket: websockets.WebSocketServerProtocol, message: str): + """处理客户端消息""" + try: + data = json.loads(message) + + if data.get("type") == "subscribe": + # 处理订阅请求 + subscription_type = data.get("subscription", "all") + response = { + "type": "subscription_ack", + "subscription": subscription_type, + "message": f"已订阅 {subscription_type} 类型数据" + } + await websocket.send(json.dumps(response)) + logger.info(f"客户端订阅了 {subscription_type} 类型数据") + + elif data.get("type") == "ping": + # 响应心跳检测 + response = { + "type": "pong", + "timestamp": datetime.now().isoformat() + } + await websocket.send(json.dumps(response)) + + else: + logger.warning(f"未知的消息类型: {data.get('type', 'unknown')}") + + except json.JSONDecodeError: + logger.error("无效的JSON消息") + except Exception as e: + logger.error(f"处理客户端消息时出错: {str(e)}") + + async def client_handler(self, websocket: websockets.WebSocketServerProtocol, path: str): + """处理客户端连接""" + await self.register_client(websocket) + + try: + async for message in websocket: + await self.handle_client_message(websocket, message) + except websockets.exceptions.ConnectionClosed: + pass + finally: + await self.unregister_client(websocket) + + async def start_server(self): + """启动WebSocket服务器""" + logger.info(f"启动WebSocket服务器: {self.host}:{self.port}") + + # 创建并启动服务器 + self.server = await websockets.serve( + self.client_handler, + self.host, + self.port + ) + + logger.info("WebSocket服务器已启动") + return self.server + + async def send_sensor_data(self, sensor_data: Dict[str, Any]): + """发送传感器数据""" + data = { + "type": "sensor_data", + "data_type": sensor_data.get("data_type"), + "device_id": sensor_data.get("device_id"), + "value": sensor_data.get("value"), + "location": sensor_data.get("location"), + "timestamp": datetime.now().isoformat() + } + await self.broadcast_data(data) + + async def send_alert(self, alert_data: Dict[str, Any]): + """发送警报信息""" + data = { + "type": "alert", + "level": alert_data.get("level", "warning"), + "message": alert_data.get("message"), + "device_id": alert_data.get("device_id"), + "timestamp": datetime.now().isoformat() + } + await self.broadcast_data(data) + +# 全局WebSocket服务器实例 +websocket_server = WebSocketServer() + +# 示例数据生成器 +async def data_generator(): + """模拟数据生成器""" + import random + + while True: + await asyncio.sleep(5) # 每5秒发送一次数据 + + # 模拟不同的传感器数据 + sensor_types = ["LL", "YL", "SW", "ZD"] + sensor_type = random.choice(sensor_types) + + # 根据传感器类型生成合理的数值范围 + if sensor_type == "LL": # 流量 + value = random.uniform(10, 100) + elif sensor_type == "YL": # 压力 + value = random.uniform(0.1, 1.0) + elif sensor_type == "SW": # 水位 + value = random.uniform(0, 10) + else: # ZD 浊度 + value = random.uniform(0, 50) + + sensor_data = { + "data_type": sensor_type, + "device_id": f"device_{random.randint(1, 10)}", + "value": round(value, 2), + "location": random.choice(["A区", "B区", "C区", "D区"]) + } + + await websocket_server.send_sensor_data(sensor_data) + +# 启动服务器和生成器 +async def main(): + """主函数""" + # 启动WebSocket服务器 + server = await websocket_server.start_server() + + # 启动数据生成器 + generator_task = asyncio.create_task(data_generator()) + + # 保持服务器运行 + try: + await asyncio.Future() # 永远等待 + except KeyboardInterrupt: + logger.info("收到中断信号，正在关闭服务器...") + server.close() + await server.wait_closed() + generator_task.cancel() + await generator_task + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file