430 lines
14 KiB
Python
430 lines
14 KiB
Python
# Device routes
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from typing import Optional
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
from bson import ObjectId
|
|
from pydantic import BaseModel, Field
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timedelta
|
|
import uuid
|
|
from redis import asyncio as aioredis
|
|
import asyncio
|
|
# Database Configuration
|
|
MONGODB_URL = "mongodb://lab:[email protected]:27017/lab"
|
|
|
|
# 更新 FastAPI 实例化
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# 启动时执行
|
|
await connect_to_mongo()
|
|
await init_redis()
|
|
yield
|
|
# 关闭时执行
|
|
await close_mongo_connection()
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
# CORS configuration
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
expose_headers=["*"]
|
|
)
|
|
|
|
# MongoDB setup
|
|
class PyObjectId(ObjectId):
|
|
"""
|
|
自定义ObjectId类，用于在Pydantic模型中处理MongoDB的ObjectId
|
|
"""
|
|
@classmethod
|
|
def __get_validators__(cls):
|
|
yield cls.validate
|
|
|
|
@classmethod
|
|
def validate(cls, v, handler):
|
|
if not ObjectId.is_valid(v):
|
|
raise ValueError("Invalid ObjectId")
|
|
return ObjectId(v)
|
|
|
|
@classmethod
|
|
def __get_pydantic_json_schema__(cls, _schema_cache, **_kwargs):
|
|
return {
|
|
'type': 'string',
|
|
'description': 'ObjectId',
|
|
'pattern': r'^[0-9a-fA-F]{24}$'
|
|
}
|
|
|
|
@classmethod
|
|
def __modify_schema__(cls, field_schema):
|
|
field_schema.update(
|
|
type='string',
|
|
description='ObjectId',
|
|
pattern=r'^[0-9a-fA-F]{24}$'
|
|
)
|
|
|
|
class SensorModel(BaseModel):
|
|
"""传感器模型"""
|
|
index: str
|
|
sensor_name: str
|
|
sensor_type: str
|
|
unit: str
|
|
|
|
class DeviceModel(BaseModel):
|
|
id: Optional[PyObjectId] = Field(alias="_id", default=None)
|
|
device_name: str
|
|
device_type: str
|
|
device_number: int
|
|
serial_numbers: list = Field(default_factory=list)
|
|
sensors: list[SensorModel] = Field(default_factory=list)
|
|
|
|
class Config:
|
|
populate_by_name = True
|
|
arbitrary_types_allowed = True
|
|
json_encoders = {ObjectId: str}
|
|
|
|
def dict(self, *args, **kwargs):
|
|
"""确保返回的字典包含 _id 字段"""
|
|
kwargs["by_alias"] = True
|
|
return super().dict(*args, **kwargs)
|
|
|
|
# Database connection
|
|
class Database:
|
|
"""数据库连接管理类"""
|
|
client: AsyncIOMotorClient = None
|
|
|
|
db = Database()
|
|
|
|
async def get_database():
|
|
"""获取数据库连接"""
|
|
if db.client is None:
|
|
await connect_to_mongo()
|
|
return db.client["lab"]
|
|
|
|
async def connect_to_mongo():
|
|
try:
|
|
db.client = AsyncIOMotorClient(MONGODB_URL)
|
|
# 验证连接
|
|
await db.client.admin.command('ping')
|
|
print("Successfully connected to MongoDB")
|
|
except Exception as e:
|
|
print(f"Could not connect to MongoDB: {e}")
|
|
raise
|
|
|
|
async def close_mongo_connection():
|
|
db.client.close()
|
|
|
|
@app.post("/devices/devices")
|
|
async def create_device(device: DeviceModel):
|
|
"""
|
|
创建新的设备
|
|
|
|
参数:
|
|
device: 设备模型，包含设备信息和传感器列表
|
|
|
|
返回:
|
|
dict: 包含创建成功消息和新设备ID
|
|
"""
|
|
db = await get_database()
|
|
|
|
# 验证传感器数量是否与通道数匹配
|
|
if len(device.sensors) != device.device_number:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"传感器数量({len(device.sensors)})与通道数({device.device_number})不匹配"
|
|
)
|
|
|
|
device_dict = device.model_dump(exclude={"id"})
|
|
result = await db.devices.insert_one(device_dict)
|
|
|
|
return {
|
|
"message": "设备创建成功",
|
|
"id": str(result.inserted_id)
|
|
}
|
|
|
|
@app.get("/devices/devices")
|
|
async def get_devices():
|
|
"""
|
|
获取所有设备列表
|
|
|
|
返回:
|
|
List[dict]: 设备列表，包含所有设备的信息
|
|
"""
|
|
db = await get_database()
|
|
|
|
devices = []
|
|
async for device in db.devices.find():
|
|
# 确保 _id 被正确序列化
|
|
device["_id"] = str(device["_id"])
|
|
devices.append(device)
|
|
|
|
return devices
|
|
|
|
# 添加删除路由
|
|
@app.delete("/devices/devices/{device_id}")
|
|
async def delete_device(device_id: str):
|
|
"""
|
|
删除指定的设备
|
|
|
|
参数:
|
|
device_id: 设备的ObjectId字符串
|
|
|
|
返回:
|
|
dict: 包含删除操作的结果信息
|
|
"""
|
|
try:
|
|
db = await get_database()
|
|
result = await db.devices.delete_one({"_id": ObjectId(device_id)})
|
|
|
|
if result.deleted_count == 0:
|
|
return {"message": "未找到指定设备"}
|
|
|
|
return {"message": "设备删除成功"}
|
|
except Exception as e:
|
|
return {"message": f"删除设备失败: {str(e)}"}
|
|
# 添加生成序列号的函数
|
|
def generate_serial_number(device_type, batch_number, device_id):
|
|
"""生成序列号
|
|
格式: XX00-UUID(10位)-设备ID后6位
|
|
"""
|
|
# 获取前缀（设备类型前两位大写）
|
|
prefix = f"{device_type[:2].upper()}{batch_number:02d}"
|
|
|
|
# 生成UUID并取前10位
|
|
uuid_part = str(uuid.uuid4()).replace('-', '')[:10]
|
|
|
|
# 获取设备ID的后6位作为校验码
|
|
checksum = str(device_id)[-6:]
|
|
|
|
return f"{prefix}-{uuid_part}-{checksum}"
|
|
|
|
# 添加生成序列号的路由
|
|
@app.post("/devices/devices/{device_id}/serials")
|
|
async def generate_device_serials(device_id: str):
|
|
"""为设备生成10个新的序列号"""
|
|
try:
|
|
db = await get_database()
|
|
|
|
# 获取设备信息
|
|
device = await db.devices.find_one({"_id": ObjectId(device_id)})
|
|
if not device:
|
|
return {"message": "设备不存在"}
|
|
|
|
# 获取当前序列号数量作为批次号
|
|
current_batch = len(device.get("serial_numbers", [])) // 10 + 1
|
|
|
|
# 生成10个新序列号
|
|
new_serials = []
|
|
for _ in range(10):
|
|
serial = generate_serial_number(
|
|
device["device_type"],
|
|
current_batch,
|
|
device_id # 传入设备ID
|
|
)
|
|
new_serials.append({
|
|
"serial": serial,
|
|
"status": "available",
|
|
"created_at": datetime.now().isoformat()
|
|
})
|
|
|
|
# 更新设备记录
|
|
result = await db.devices.update_one(
|
|
{"_id": ObjectId(device_id)},
|
|
{"$push": {"serial_numbers": {"$each": new_serials}}}
|
|
)
|
|
|
|
return {
|
|
"message": "序列号生成成功",
|
|
"serials": new_serials
|
|
}
|
|
except Exception as e:
|
|
return {"message": f"生成序列号失败: {str(e)}"}
|
|
|
|
# 添加获取序列号的路由
|
|
@app.get("/devices/devices/{device_id}/serials")
|
|
async def get_device_serials(device_id: str):
|
|
"""获取设备的所有序列号"""
|
|
try:
|
|
db = await get_database()
|
|
device = await db.devices.find_one({"_id": ObjectId(device_id)})
|
|
|
|
if not device:
|
|
return {"message": "设备不存在"}
|
|
|
|
return {
|
|
"device_id": str(device["_id"]),
|
|
"device_name": device["device_name"],
|
|
"serial_numbers": device.get("serial_numbers", [])
|
|
}
|
|
except Exception as e:
|
|
return {"message": f"获取序列号失败: {str(e)}"}
|
|
|
|
# 添加更新设备的路由
|
|
@app.put("/devices/devices/{device_id}")
|
|
async def update_device(device_id: str, device: DeviceModel):
|
|
"""更新设备信息"""
|
|
try:
|
|
db = await get_database()
|
|
|
|
# 首先获取现有设备数据，特别是serial_numbers
|
|
existing_device = await db.devices.find_one({"_id": ObjectId(device_id)})
|
|
if not existing_device:
|
|
raise HTTPException(status_code=404, detail="设备不存在")
|
|
|
|
# 验证传感器数量
|
|
if len(device.sensors) != device.device_number:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"传感器数量({len(device.sensors)})与通道数({device.device_number})不匹配"
|
|
)
|
|
|
|
# 只更新允许修改的字段，保留原有的serial_numbers
|
|
update_data = {
|
|
"device_name": device.device_name,
|
|
"device_type": device.device_type,
|
|
"device_number": device.device_number,
|
|
"sensors": [sensor.dict() for sensor in device.sensors],
|
|
"serial_numbers": existing_device.get("serial_numbers", []) # 保留原有的序列号
|
|
}
|
|
|
|
result = await db.devices.update_one(
|
|
{"_id": ObjectId(device_id)},
|
|
{"$set": update_data}
|
|
)
|
|
|
|
return {"message": "设备更新成功"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
# 添加Redis连接配置
|
|
REDIS_URL = "redis://:Obscura@[email protected]:6379"
|
|
redis_client = None
|
|
|
|
# 用于存储上一次检查时的stream长度
|
|
last_stream_lengths = {}
|
|
|
|
# 在全局变量中添加设备活跃时间记录
|
|
device_last_active = {}
|
|
|
|
# 初始化Redis连接
|
|
async def init_redis():
|
|
global redis_client
|
|
if redis_client is None:
|
|
redis_client = await aioredis.from_url(REDIS_URL, db=200)
|
|
return redis_client
|
|
|
|
@app.get("/devices/devices/online")
|
|
async def get_online_devices():
|
|
"""获取在线设备列表"""
|
|
try:
|
|
redis = await init_redis()
|
|
db = await get_database()
|
|
|
|
keys = [key.decode('utf-8') for key in await redis.keys("experiment:*")]
|
|
online_devices = {}
|
|
current_time = datetime.now()
|
|
|
|
# 用于临时存储序列号的活跃状态
|
|
serial_active_status = {}
|
|
stream_lengths = {} # 添加存储stream长度的字典
|
|
|
|
print("\n=== 检查 Stream 活跃状态 ===")
|
|
# 检查所有streams
|
|
for key in keys:
|
|
try:
|
|
serial_number = key.split(":")[-1]
|
|
current_length = await redis.xlen(key)
|
|
last_length = last_stream_lengths.get(key, 0)
|
|
|
|
# 记录当前stream长度
|
|
stream_lengths[key] = {
|
|
'current': current_length,
|
|
'last': last_length
|
|
}
|
|
|
|
# 如果该序列号还未被标记为活跃，且当前stream有更新
|
|
if not serial_active_status.get(serial_number, False) and current_length != last_length:
|
|
serial_active_status[serial_number] = True
|
|
device_last_active[serial_number] = current_time
|
|
last_stream_lengths[key] = current_length
|
|
|
|
except Exception as e:
|
|
print(f"处理key {key} 时出错: {str(e)}")
|
|
continue
|
|
|
|
# 处理在线设备
|
|
for serial_number, is_active in serial_active_status.items():
|
|
last_active_time = device_last_active.get(serial_number)
|
|
|
|
# 如果设备在过去5分钟内有活动，则认为它在线
|
|
if last_active_time and (current_time - last_active_time) <= timedelta(minutes=5):
|
|
device = await db.devices.find_one({
|
|
"serial_numbers.serial": serial_number
|
|
})
|
|
|
|
if device:
|
|
online_devices[serial_number] = {
|
|
"device_id": str(device["_id"]),
|
|
"device_name": device["device_name"],
|
|
"device_type": device["device_type"],
|
|
"device_number": device["device_number"],
|
|
"serial_number": serial_number,
|
|
"last_active": last_active_time.isoformat(),
|
|
"sensors": device["sensors"],
|
|
"status": "active" if is_active else "idle",
|
|
"stream_info": { # 添加stream信息
|
|
k: v for k, v in stream_lengths.items()
|
|
if k.endswith(serial_number)
|
|
}
|
|
}
|
|
|
|
print("\n=== 在线设备统计 ===")
|
|
print(f"在线设备数量: {len(online_devices)}")
|
|
for device in online_devices.values():
|
|
print(f"\n设备: {device['device_name']} ({device['serial_number']})")
|
|
print(f" - 状态: {device['status']}")
|
|
print(f" - 最后活跃: {device['last_active']}")
|
|
print(" - Stream信息:")
|
|
for stream_key, info in device['stream_info'].items():
|
|
print(f" * {stream_key}: 当前={info['current']}, 上次={info['last']}")
|
|
|
|
response_data = {
|
|
"online_count": len(online_devices),
|
|
"devices": list(online_devices.values())
|
|
}
|
|
return response_data
|
|
|
|
except Exception as e:
|
|
print(f"获取在线设备出错: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"获取在线设备失败: {str(e)}")
|
|
|
|
# 修改清理函数
|
|
async def cleanup_old_records():
|
|
while True:
|
|
current_time = datetime.now()
|
|
# 清理stream长度记录
|
|
keys_to_remove = []
|
|
for key in last_stream_lengths.keys():
|
|
if current_time - last_stream_lengths.get(key, {}).get('last_check', current_time) > timedelta(minutes=10):
|
|
keys_to_remove.append(key)
|
|
|
|
for key in keys_to_remove:
|
|
last_stream_lengths.pop(key, None)
|
|
|
|
# 清理过期的设备活跃记录
|
|
serials_to_remove = []
|
|
for serial, last_active in device_last_active.items():
|
|
if current_time - last_active > timedelta(minutes=10):
|
|
serials_to_remove.append(serial)
|
|
|
|
for serial in serials_to_remove:
|
|
device_last_active.pop(serial, None)
|
|
|
|
await asyncio.sleep(300) # 每5分钟清理一次
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=6001) |