# Backend 集成指南

本文档说明如何在 Backend 服务中集成 JRSoft Subway WebSocket 协议。

## 概述

Backend 作为业务逻辑层，通过 WebSocket 连接到 Gateway 发送命令并接收响应。Backend 使用 FastAPI 框架，支持异步操作。

## 集成步骤

### 1. 安装依赖

```bash
pip install websockets aiohttp pydantic
```

### 2. WebSocket 客户端实现

```python
import asyncio
import json
from typing import Dict, Optional, Callable
import websockets
from datetime import datetime

class GatewayClient:
    def __init__(self, gateway_url: str, client_id: str):
        self.gateway_url = gateway_url
        self.client_id = client_id
        self.websocket: Optional[websockets.WebSocketClientProtocol] = None
        self.pending_requests: Dict[str, asyncio.Future] = {}
        self.message_handlers: Dict[str, Callable] = {}
        
    async def connect(self):
        """连接到 Gateway"""
        self.websocket = await websockets.connect(self.gateway_url)
        
        # 发送注册消息
        await self.register()
        
        # 启动消息接收循环
        asyncio.create_task(self.receive_messages())
        
        # 启动心跳
        asyncio.create_task(self.heartbeat_loop())
        
    async def register(self):
        """注册客户端"""
        register_message = {
            "type": "REGISTER",
            "clientId": self.client_id,
            "clientType": "BACKEND",
            "clientInfo": {
                "name": "JRSoft Backend Service",
                "version": "1.0.0",
                "capabilities": ["command", "query", "monitor"]
            },
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "version": "1.0"
        }
        await self.websocket.send(json.dumps(register_message))
        
    async def heartbeat_loop(self):
        """心跳循环"""
        while True:
            try:
                if self.websocket and not self.websocket.closed:
                    heartbeat = {
                        "type": "HEARTBEAT",
                        "clientId": self.client_id,
                        "timestamp": datetime.utcnow().isoformat() + "Z",
                        "version": "1.0"
                    }
                    await self.websocket.send(json.dumps(heartbeat))
                await asyncio.sleep(30)  # 30秒心跳间隔
            except Exception as e:
                print(f"Heartbeat error: {e}")
                await self.reconnect()
                
    async def receive_messages(self):
        """接收消息循环"""
        try:
            async for message in self.websocket:
                data = json.loads(message)
                await self.handle_message(data)
        except websockets.exceptions.ConnectionClosed:
            await self.reconnect()
            
    async def handle_message(self, message: dict):
        """处理接收到的消息"""
        message_type = message.get("type")
        
        if message_type == "Ucommand_response":
            # 处理命令响应
            request_ref = message.get("requestRef")
            if request_ref in self.pending_requests:
                future = self.pending_requests.pop(request_ref)
                future.set_result(message)
                
        elif message_type == "Uprogress_update":
            # 处理进度更新
            handler = self.message_handlers.get("Uprogress_update")
            if handler:
                await handler(message)
                
        elif message_type == "Uerror":
            # 处理错误消息
            request_ref = message.get("requestRef")
            if request_ref in self.pending_requests:
                future = self.pending_requests.pop(request_ref)
                future.set_exception(Exception(message.get("message")))
```

### 3. 命令发送接口

```python
import uuid
from enum import Enum
from typing import Any, Dict, Optional

class Priority(Enum):
    LOW = "low"
    NORMAL = "Unormal"
    HIGH = "Uhigh"
    URGENT = "urgent"

class CommandSender:
    def __init__(self, gateway_client: GatewayClient):
        self.gateway_client = gateway_client
        
    async def send_command(
        self,
        target_client_id: str,
        command: Dict[str, Any],
        priority: Priority = Priority.NORMAL,
        timeout: int = 5000,
        callback: Optional[str] = None
    ) -> Dict[str, Any]:
        """发送命令并等待响应"""
        request_ref = f"cmd-{uuid.uuid4()}"
        
        message = {
            "type": "COMMAND",
            "requestRef": request_ref,
            "targetClientId": target_client_id,
            "command": command,
            "priority": priority.value,
            "timeout": timeout,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "version": "1.0"
        }
        
        if callback:
            message["callback"] = callback
            
        # 创建 Future 用于等待响应
        future = asyncio.Future()
        self.gateway_client.pending_requests[request_ref] = future
        
        # 发送命令
        await self.gateway_client.websocket.send(json.dumps(message))
        
        # 等待响应或超时
        try:
            response = await asyncio.wait_for(future, timeout=timeout/1000)
            return response
        except asyncio.TimeoutError:
            self.gateway_client.pending_requests.pop(request_ref, None)
            raise TimeoutError(f"Command timeout: {request_ref}")
    
    async def send_simple_command(
        self,
        edge_id: str,
        device_id: str,
        command_code: str,
        device_type: str,
        operation_type: str,
        parameters: Dict[str, Any],
        **kwargs
    ) -> Dict[str, Any]:
        """发送 Simple 类型命令"""
        command = {
            "commandType": "SIMPLE",
            "commandCode": command_code,
            "deviceType": device_type,
            "deviceId": device_id,
            "operationType": operation_type,
            "parameters": parameters
        }
        
        target_client_id = f"{edge_id}:{device_id}"
        return await self.send_command(target_client_id, command, **kwargs)
    
    async def send_batch_command(
        self,
        edge_id: str,
        device_id: str,
        command_code: str,
        device_type: str,
        targets: str,
        operation_type: str,
        parameters: Dict[str, Any],
        **kwargs
    ) -> Dict[str, Any]:
        """发送 Batch 类型命令"""
        command = {
            "commandType": "BATCH",
            "commandCode": command_code,
            "deviceType": device_type,
            "deviceId": targets,  # 批量目标
            "operationType": operation_type,
            "parameters": {
                **parameters,
                "targets": targets,
                "Ubatch": True
            }
        }
        
        target_client_id = f"{edge_id}:{device_id}"
        return await self.send_command(target_client_id, command, **kwargs)
```

### 4. FastAPI 集成

```python
from fastapi import FastAPI, HTTPException, WebSocket
from contextlib import asynccontextmanager
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 全局 Gateway 客户端
gateway_client: Optional[GatewayClient] = None
command_sender: Optional[CommandSender] = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    global gateway_client, command_sender
    
    # 启动时连接 Gateway
    gateway_client = GatewayClient(
        "ws://localhost:18081",
        "backend-001"
    )
    await gateway_client.connect()
    command_sender = CommandSender(gateway_client)
    logger.info("Connected to Gateway")
    
    yield
    
    # 关闭时断开连接
    if gateway_client.websocket:
        await gateway_client.websocket.close()
    logger.info("Disconnected from Gateway")

app = FastAPI(lifespan=lifespan)

# API 端点示例
@app.post("/api/devices/{edge_id}/{device_id}/control")
async def control_device(
    edge_id: str,
    device_id: str,
    command_code: str,
    parameters: Dict[str, Any]
):
    """控制设备的 API 端点"""
    try:
        response = await command_sender.send_simple_command(
            edge_id=edge_id,
            device_id=device_id,
            command_code=command_code,
            device_type="pillar",
            operation_type="Uwrite",
            parameters=parameters,
            priority=Priority.NORMAL,
            timeout=5000
        )
        
        if response.get("status") == "Ucompleted":
            return {"success": True, "data": response.get("result")}
        else:
            raise HTTPException(status_code=500, detail="Command failed")
            
    except TimeoutError:
        raise HTTPException(status_code=504, detail="Command timeout")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.websocket("/ws/monitor/{request_id}")
async def monitor_progress(websocket: WebSocket, request_id: str):
    """监控命令执行进度的 WebSocket 端点"""
    await websocket.accept()
    
    progress_updates = []
    
    async def progress_handler(message):
        if message.get("requestRef") == request_id:
            progress_updates.append(message)
            await websocket.send_json(message)
    
    # 注册进度处理器
    gateway_client.message_handlers["Uprogress_update"] = progress_handler
    
    try:
        # 保持连接直到客户端断开
        while True:
            await websocket.receive_text()
    except:
        pass
    finally:
        # 清理处理器
        gateway_client.message_handlers.pop("Uprogress_update", None)
```

### 5. 异步任务处理

```python
from typing import List
import asyncio

class BatchCommandExecutor:
    """批量命令执行器"""
    
    def __init__(self, command_sender: CommandSender):
        self.command_sender = command_sender
        
    async def execute_on_multiple_devices(
        self,
        devices: List[Dict[str, str]],
        command_code: str,
        parameters: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """在多个设备上执行命令"""
        tasks = []
        
        for device in devices:
            task = self.command_sender.send_simple_command(
                edge_id=device["edge_id"],
                device_id=device["device_id"],
                command_code=command_code,
                device_type=device["device_type"],
                operation_type="Uwrite",
                parameters=parameters
            )
            tasks.append(task)
        
        # 并发执行所有命令
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 处理结果
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                processed_results.append({
                    "Udevice": devices[i],
                    "success": False,
                    "Uerror": str(result)
                })
            else:
                processed_results.append({
                    "Udevice": devices[i],
                    "success": True,
                    "result": result
                })
        
        return processed_results
```

### 6. 错误处理和重试

```python
from typing import TypeVar, Callable
import asyncio

T = TypeVar('T')

async def retry_async(
    func: Callable[..., T],
    max_attempts: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple = (Exception,)
) -> T:
    """异步重试装饰器"""
    attempt = 0
    current_delay = delay
    
    while attempt < max_attempts:
        try:
            return await func()
        except exceptions as e:
            attempt += 1
            if attempt >= max_attempts:
                raise
            
            logger.warning(f"Attempt {attempt} failed: {e}. Retrying in {current_delay}s...")
            await asyncio.sleep(current_delay)
            current_delay *= backoff

# 使用示例
async def send_command_with_retry(target_client_id: str, command: dict):
    return await retry_async(
        lambda: command_sender.send_command(target_client_id, command),
        max_attempts=3,
        delay=1.0,
        exceptions=(TimeoutError, ConnectionError)
    )
```

## 最佳实践

### 1. 连接管理

```python
class ConnectionManager:
    """连接管理器，处理断线重连"""
    
    def __init__(self, gateway_client: GatewayClient):
        self.gateway_client = gateway_client
        self.reconnect_delay = 5
        self.max_reconnect_delay = 60
        
    async def ensure_connected(self):
        """确保连接可用"""
        if not self.gateway_client.websocket or self.gateway_client.websocket.closed:
            await self.reconnect()
            
    async def reconnect(self):
        """重连逻辑"""
        delay = self.reconnect_delay
        
        while True:
            try:
                logger.info(f"Attempting to reconnect...")
                await self.gateway_client.connect()
                logger.info("Reconnected successfully")
                self.reconnect_delay = 5  # 重置延迟
                break
            except Exception as e:
                logger.error(f"Reconnection failed: {e}")
                await asyncio.sleep(delay)
                delay = min(delay * 2, self.max_reconnect_delay)
```

### 2. 性能优化

```python
# 使用连接池
class GatewayConnectionPool:
    def __init__(self, size: int = 5):
        self.connections: List[GatewayClient] = []
        self.size = size
        self.current = 0
        
    async def initialize(self, gateway_url: str, client_id_prefix: str):
        for i in range(self.size):
            client = GatewayClient(
                gateway_url,
                f"{client_id_prefix}-{i}"
            )
            await client.connect()
            self.connections.append(client)
    
    def get_connection(self) -> GatewayClient:
        """轮询获取连接"""
        conn = self.connections[self.current]
        self.current = (self.current + 1) % self.size
        return conn
```

### 3. 监控和日志

```python
import time
from prometheus_client import Counter, Histogram, Gauge

# Prometheus 指标
command_sent_total = Counter('backend_command_sent_total', 'Total commands sent')
command_duration = Histogram('backend_command_duration_seconds', 'Command execution duration')
active_connections = Gauge('backend_active_connections', 'Number of active Gateway connections')

class MonitoredCommandSender(CommandSender):
    async def send_command(self, *args, **kwargs):
        start_time = time.time()
        command_sent_total.inc()
        
        try:
            result = await super().send_command(*args, **kwargs)
            command_duration.observe(time.time() - start_time)
            return result
        except Exception as e:
            command_duration.observe(time.time() - start_time)
            raise
```

## 故障排查

### 常见问题

1. **连接失败**
   ```python
   # 检查 Gateway 是否运行
   # 检查网络连接
   # 验证 WebSocket URL
   ```

2. **命令超时**
   ```python
   # 增加超时时间
   # 检查目标设备是否在线
   # 查看 Gateway 日志
   ```

3. **消息格式错误**
   ```python
   # 使用协议提供的验证函数
   # 检查必填字段
   # 验证数据类型
   ```

### 调试技巧

```python
# 启用详细日志
logging.getLogger("websockets").setLevel(logging.DEBUG)

# 消息追踪
class DebugGatewayClient(GatewayClient):
    async def send(self, message: str):
        logger.debug(f"Sending: {message}")
        await super().send(message)
        
    async def handle_message(self, message: dict):
        logger.debug(f"Received: {message}")
        await super().handle_message(message)
```

## 总结

Backend 集成的关键点：

1. **异步架构** - 充分利用 Python 的异步特性
2. **错误处理** - 完善的错误处理和重试机制
3. **性能优化** - 连接池、批量处理、并发控制
4. **监控日志** - 完整的监控指标和日志记录
5. **容错设计** - 自动重连、优雅降级

遵循这些指南，可以构建一个稳定、高效的 Backend 服务。