feat: 注册 aiot 路由并更新主程序配置
- main.py:注册 aiot_alarm 和 aiot_edge 路由,保留旧路由兼容 - config.py/alert_service.py/mqtt_service.py:同步更新配置和服务 - 添加 CLAUDE.md 项目说明文档 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
82
CLAUDE.md
Normal file
82
CLAUDE.md
Normal file
@@ -0,0 +1,82 @@
|
|||||||
|
# CLAUDE.md
|
||||||
|
|
||||||
|
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
|
||||||
|
|
||||||
|
## Project Overview
|
||||||
|
|
||||||
|
AI Alert Platform backend service built with FastAPI. Receives alerts from edge AI devices via MQTT and HTTP, stores snapshot images, pushes real-time updates over WebSocket, and exposes REST APIs for a Yudao (芋道) Vue 3 frontend. Optionally triggers async big-model analysis on alert snapshots.
|
||||||
|
|
||||||
|
## Commands
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Install dependencies
|
||||||
|
pip install -r requirements.txt
|
||||||
|
|
||||||
|
# Run development server (starts at http://localhost:8000, auto-reload in debug mode)
|
||||||
|
python -m app.main
|
||||||
|
|
||||||
|
# API docs: http://localhost:8000/docs
|
||||||
|
|
||||||
|
# Production run with Gunicorn (Linux only)
|
||||||
|
gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000
|
||||||
|
```
|
||||||
|
|
||||||
|
Environment setup: copy `.env.example` to `.env` and configure. SQLite is the default database; set `DATABASE_URL` for MySQL in production. Set `MQTT_ENABLED=false` if no MQTT broker is available.
|
||||||
|
|
||||||
|
No test suite exists yet. If adding tests, use pytest.
|
||||||
|
|
||||||
|
**Root `main.py`** is a PyCharm template file, not the real entry point. Always use `python -m app.main`.
|
||||||
|
|
||||||
|
## Architecture
|
||||||
|
|
||||||
|
### Data Flow
|
||||||
|
|
||||||
|
```
|
||||||
|
Edge Devices ──MQTT──→ EMQX Broker ──subscribe──→ MQTTService ──→ AlertService ──→ DB
|
||||||
|
│
|
||||||
|
Edge Devices ──HTTP POST──→ /api/v1/alerts ────────────→│
|
||||||
|
↓
|
||||||
|
NotificationService ──WebSocket──→ Frontend
|
||||||
|
```
|
||||||
|
|
||||||
|
### Dual API Surface
|
||||||
|
|
||||||
|
The app exposes two parallel API layers hitting the same services and database:
|
||||||
|
|
||||||
|
1. **Native API** (`/api/v1/`) — endpoints defined inline in `app/main.py`. Standard REST responses.
|
||||||
|
2. **Yudao-compat API** (`/admin-api/`) — routers in `app/routers/yudao_alert.py` and `yudao_auth.py`. Wraps responses in `{"code": 0, "data": ..., "msg": ""}` format expected by the Yudao Vue frontend. Auth/permissions handled by `app/yudao_compat.py` (stub in dev mode: `DEV_MODE=true` skips token validation, returns mock admin).
|
||||||
|
|
||||||
|
### Services (Global Singletons)
|
||||||
|
|
||||||
|
All services are instantiated as module-level singletons, not injected via FastAPI `Depends()`. Access them via factory functions (`get_alert_service()`, `get_mqtt_service()`, etc.) or import the global directly.
|
||||||
|
|
||||||
|
- **AlertService** (`app/services/alert_service.py`) — CRUD, filtering/pagination, statistics, AI analysis update. Handles both HTTP (`create_alert`) and MQTT (`create_alert_from_mqtt`) creation paths. Alert numbers: `ALT` + timestamp + uuid fragment.
|
||||||
|
- **MQTTService** (`app/services/mqtt_service.py`) — paho-mqtt client subscribing to `edge/alert/#`. Routes messages to registered callbacks for alerts vs heartbeats. Compatible with paho-mqtt 1.x and 2.x APIs.
|
||||||
|
- **DeviceService** (`app/services/device_service.py`) — Tracks edge device status from MQTT heartbeats. In-memory cache + DB persistence. Marks devices offline after 90s without heartbeat.
|
||||||
|
- **NotificationService** (`app/services/notification_service.py`) — WebSocket connection manager + broadcast. Bridges sync MQTT callbacks to async WebSocket sends via `asyncio.run_coroutine_threadsafe()`.
|
||||||
|
- **OSSStorage** (`app/services/oss_storage.py`) — Image storage. Currently local-only (`uploads/` dir); Aliyun OSS stubbed but not implemented.
|
||||||
|
- **AIAnalyzer** (`app/services/ai_analyzer.py`) — Async httpx client for optional big-model analysis. Fire-and-forget via `asyncio.create_task()`.
|
||||||
|
|
||||||
|
### Lifecycle (`app/main.py` lifespan)
|
||||||
|
|
||||||
|
Startup: init DB → set async event loop on NotificationService → register MQTT handlers → start MQTT.
|
||||||
|
Shutdown: stop MQTT.
|
||||||
|
|
||||||
|
### Database
|
||||||
|
|
||||||
|
SQLite at `data/alert_platform.db` (auto-created). Three ORM models in `app/models.py`:
|
||||||
|
|
||||||
|
- **Alert** — main table. Enums: `AlertStatus` (pending/confirmed/ignored/resolved/dispatched), `AlertLevel` (low/medium/high/critical). Indexed on alert_no, camera_id, status, trigger_time.
|
||||||
|
- **EdgeDevice** — device status and heartbeat tracking. Enum: `DeviceStatus` (online/offline/error).
|
||||||
|
- **WorkOrder** — framework exists (model + relationship to Alert) but no API/service layer yet.
|
||||||
|
|
||||||
|
Sessions obtained via `get_session()`. Services manage their own session lifecycle with try/finally.
|
||||||
|
|
||||||
|
### Config
|
||||||
|
|
||||||
|
`app/config.py` — dataclass-based config loaded from `.env` via `os.getenv()`. Sections: `DatabaseConfig`, `OSSConfig`, `AppConfig` (includes `dev_mode`), `AIModelConfig`, `MQTTConfig`. Global `settings` instance created at module load.
|
||||||
|
|
||||||
|
### Key MQTT Topics
|
||||||
|
|
||||||
|
- Alert: `edge/alert/{camera_id}/{roi_id}` — JSON with camera_id, roi_id, alert_type, confidence, etc.
|
||||||
|
- Heartbeat: `edge/alert/heartbeat/{device_id}` — JSON with device_id, status, uptime, counters.
|
||||||
@@ -30,6 +30,7 @@ class AppConfig:
|
|||||||
host: str = "0.0.0.0"
|
host: str = "0.0.0.0"
|
||||||
port: int = 8000
|
port: int = 8000
|
||||||
debug: bool = True
|
debug: bool = True
|
||||||
|
dev_mode: bool = True # 开发模式:跳过认证验证,返回超级管理员权限
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
@@ -82,6 +83,7 @@ def load_settings() -> Settings:
|
|||||||
host=os.getenv("APP_HOST", "0.0.0.0"),
|
host=os.getenv("APP_HOST", "0.0.0.0"),
|
||||||
port=int(os.getenv("APP_PORT", "8000")),
|
port=int(os.getenv("APP_PORT", "8000")),
|
||||||
debug=os.getenv("DEBUG", "true").lower() == "true",
|
debug=os.getenv("DEBUG", "true").lower() == "true",
|
||||||
|
dev_mode=os.getenv("DEV_MODE", "true").lower() == "true",
|
||||||
),
|
),
|
||||||
ai_model=AIModelConfig(
|
ai_model=AIModelConfig(
|
||||||
endpoint=os.getenv("AI_MODEL_ENDPOINT", ""),
|
endpoint=os.getenv("AI_MODEL_ENDPOINT", ""),
|
||||||
|
|||||||
15
app/main.py
15
app/main.py
@@ -26,6 +26,8 @@ from app.services.mqtt_service import get_mqtt_service
|
|||||||
from app.services.notification_service import get_notification_service
|
from app.services.notification_service import get_notification_service
|
||||||
from app.services.device_service import get_device_service
|
from app.services.device_service import get_device_service
|
||||||
from app.utils.logger import logger
|
from app.utils.logger import logger
|
||||||
|
from app.routers import yudao_alert_router, yudao_auth_router, yudao_aiot_alarm_router, yudao_aiot_edge_router
|
||||||
|
from app.yudao_compat import yudao_exception_handler
|
||||||
import json
|
import json
|
||||||
|
|
||||||
|
|
||||||
@@ -100,6 +102,19 @@ app.add_middleware(
|
|||||||
allow_headers=["*"],
|
allow_headers=["*"],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# ==================== 芋道兼容路由 ====================
|
||||||
|
# 提供与芋道主平台一致的接口规范,便于前端统一对接
|
||||||
|
app.include_router(yudao_auth_router)
|
||||||
|
app.include_router(yudao_alert_router)
|
||||||
|
|
||||||
|
# ==================== AIoT 统一路由 ====================
|
||||||
|
# aiot 命名空间下的新路由,与旧路由并存
|
||||||
|
app.include_router(yudao_aiot_alarm_router)
|
||||||
|
app.include_router(yudao_aiot_edge_router)
|
||||||
|
|
||||||
|
# 注册芋道格式异常处理器
|
||||||
|
app.add_exception_handler(HTTPException, yudao_exception_handler)
|
||||||
|
|
||||||
|
|
||||||
def get_alert_svc():
|
def get_alert_svc():
|
||||||
return alert_service
|
return alert_service
|
||||||
|
|||||||
@@ -319,6 +319,68 @@ class AlertService:
|
|||||||
finally:
|
finally:
|
||||||
db.close()
|
db.close()
|
||||||
|
|
||||||
|
def get_camera_alert_summary(
|
||||||
|
self,
|
||||||
|
page: int = 1,
|
||||||
|
page_size: int = 10,
|
||||||
|
) -> dict:
|
||||||
|
"""以摄像头为维度获取告警汇总"""
|
||||||
|
from sqlalchemy import func
|
||||||
|
db = get_session()
|
||||||
|
try:
|
||||||
|
# 按摄像头分组统计总数和最近时间
|
||||||
|
query = db.query(
|
||||||
|
Alert.camera_id,
|
||||||
|
func.count(Alert.id).label("total_count"),
|
||||||
|
func.max(Alert.created_at).label("last_alert_time"),
|
||||||
|
).group_by(Alert.camera_id)
|
||||||
|
|
||||||
|
# 总数
|
||||||
|
total = query.count()
|
||||||
|
|
||||||
|
# 分页
|
||||||
|
results = (
|
||||||
|
query.order_by(func.count(Alert.id).desc())
|
||||||
|
.offset((page - 1) * page_size)
|
||||||
|
.limit(page_size)
|
||||||
|
.all()
|
||||||
|
)
|
||||||
|
|
||||||
|
summary_list = []
|
||||||
|
for row in results:
|
||||||
|
# 获取该摄像头待处理告警数量
|
||||||
|
pending_count = (
|
||||||
|
db.query(Alert)
|
||||||
|
.filter(Alert.camera_id == row.camera_id)
|
||||||
|
.filter(Alert.status == AlertStatus.PENDING)
|
||||||
|
.count()
|
||||||
|
)
|
||||||
|
|
||||||
|
# 获取该摄像头最新的一条告警
|
||||||
|
latest_alert = (
|
||||||
|
db.query(Alert)
|
||||||
|
.filter(Alert.camera_id == row.camera_id)
|
||||||
|
.order_by(Alert.created_at.desc())
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
|
||||||
|
summary_list.append({
|
||||||
|
"cameraId": row.camera_id,
|
||||||
|
"cameraName": row.camera_id, # TODO: 从设备服务获取摄像头名称
|
||||||
|
"totalCount": row.total_count,
|
||||||
|
"pendingCount": pending_count,
|
||||||
|
"lastAlertTime": row.last_alert_time.isoformat() if row.last_alert_time else None,
|
||||||
|
"lastAlertType": latest_alert.alert_type if latest_alert else None,
|
||||||
|
"lastAlertTypeName": latest_alert.alert_type if latest_alert else None, # 前端会映射
|
||||||
|
})
|
||||||
|
|
||||||
|
return {
|
||||||
|
"list": summary_list,
|
||||||
|
"total": total,
|
||||||
|
}
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
# 全局单例
|
# 全局单例
|
||||||
alert_service = AlertService()
|
alert_service = AlertService()
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ class MQTTService:
|
|||||||
self._client: Optional[mqtt.Client] = None
|
self._client: Optional[mqtt.Client] = None
|
||||||
self._connected = False
|
self._connected = False
|
||||||
self._running = False
|
self._running = False
|
||||||
|
self._use_v2_callback = False # paho-mqtt 版本标记
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
|
||||||
# 回调函数
|
# 回调函数
|
||||||
@@ -58,11 +59,23 @@ class MQTTService:
|
|||||||
return
|
return
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self._client = mqtt.Client(
|
# 兼容 paho-mqtt 1.x 和 2.x 版本
|
||||||
client_id=settings.mqtt.client_id,
|
try:
|
||||||
protocol=mqtt.MQTTv5,
|
# paho-mqtt 2.0+ 新 API
|
||||||
callback_api_version=mqtt.CallbackAPIVersion.VERSION2
|
self._client = mqtt.Client(
|
||||||
)
|
client_id=settings.mqtt.client_id,
|
||||||
|
protocol=mqtt.MQTTv5,
|
||||||
|
callback_api_version=mqtt.CallbackAPIVersion.VERSION2
|
||||||
|
)
|
||||||
|
self._use_v2_callback = True
|
||||||
|
except AttributeError:
|
||||||
|
# paho-mqtt 1.x 旧 API
|
||||||
|
self._client = mqtt.Client(
|
||||||
|
client_id=settings.mqtt.client_id,
|
||||||
|
protocol=mqtt.MQTTv5
|
||||||
|
)
|
||||||
|
self._use_v2_callback = False
|
||||||
|
logger.info("使用 paho-mqtt 1.x 兼容模式")
|
||||||
|
|
||||||
# 设置回调
|
# 设置回调
|
||||||
self._client.on_connect = self._on_connect
|
self._client.on_connect = self._on_connect
|
||||||
@@ -107,9 +120,17 @@ class MQTTService:
|
|||||||
self._client.disconnect()
|
self._client.disconnect()
|
||||||
logger.info("MQTT 服务已停止")
|
logger.info("MQTT 服务已停止")
|
||||||
|
|
||||||
def _on_connect(self, client, userdata, flags, reason_code, properties):
|
def _on_connect(self, client, userdata, *args):
|
||||||
"""连接回调"""
|
"""连接回调 (兼容 1.x 和 2.x)"""
|
||||||
if reason_code == 0:
|
# 1.x: (client, userdata, flags, rc)
|
||||||
|
# 2.x: (client, userdata, connect_flags, reason_code, properties)
|
||||||
|
if args:
|
||||||
|
reason_code = args[-2] if len(args) >= 2 else args[-1]
|
||||||
|
rc = reason_code if isinstance(reason_code, int) else getattr(reason_code, 'value', reason_code)
|
||||||
|
else:
|
||||||
|
rc = -1
|
||||||
|
|
||||||
|
if rc == 0:
|
||||||
self._connected = True
|
self._connected = True
|
||||||
logger.info("MQTT 连接成功")
|
logger.info("MQTT 连接成功")
|
||||||
|
|
||||||
@@ -123,10 +144,17 @@ class MQTTService:
|
|||||||
self._connected = False
|
self._connected = False
|
||||||
logger.error(f"MQTT 连接失败: {reason_code}")
|
logger.error(f"MQTT 连接失败: {reason_code}")
|
||||||
|
|
||||||
def _on_disconnect(self, client, userdata, reason_code, properties):
|
def _on_disconnect(self, client, userdata, *args):
|
||||||
"""断开连接回调"""
|
"""断开连接回调 (兼容 1.x 和 2.x)"""
|
||||||
|
# 1.x: (client, userdata, rc)
|
||||||
|
# 2.x: (client, userdata, disconnect_flags, reason_code, properties)
|
||||||
self._connected = False
|
self._connected = False
|
||||||
logger.warning(f"MQTT 连接断开: {reason_code}")
|
if args:
|
||||||
|
reason_code = args[-2] if len(args) >= 2 else args[0]
|
||||||
|
rc = reason_code if isinstance(reason_code, int) else getattr(reason_code, 'value', reason_code)
|
||||||
|
logger.warning(f"MQTT 连接断开: {rc}")
|
||||||
|
else:
|
||||||
|
logger.warning("MQTT 连接断开")
|
||||||
|
|
||||||
def _on_message(self, client, userdata, msg: mqtt.MQTTMessage):
|
def _on_message(self, client, userdata, msg: mqtt.MQTTMessage):
|
||||||
"""消息回调"""
|
"""消息回调"""
|
||||||
|
|||||||
Reference in New Issue
Block a user