317 lines
11 KiB
Python
317 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
中继器核心服务模块
|
||
实现持续运行的中继器服务，处理与平台的通信
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
import threading
|
||
import asyncio
|
||
import websockets
|
||
from datetime import datetime
|
||
from queue import Queue, Empty
|
||
|
||
from config import config
|
||
from internal.protocol.lora import LoRaHandler
|
||
from internal.protocol.websocket import WebSocketClient
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RelayService:
|
||
"""中继器服务类"""
|
||
|
||
def __init__(self):
|
||
"""初始化中继器服务"""
|
||
self.device_id = config._config.get('relay', {}).get('device_id', '1')
|
||
self.name = config._config.get('relay', {}).get('name', '默认中继器')
|
||
self.running = False
|
||
self.lora_handler = None
|
||
self.websocket_client = None
|
||
self.command_queue = Queue()
|
||
self.response_queue = Queue()
|
||
self.loop = None
|
||
|
||
logger.info(f"初始化中继器服务: ID={self.device_id}, 名称={self.name}")
|
||
|
||
def initialize(self):
|
||
"""初始化中继器服务"""
|
||
logger.info("开始初始化中继器服务")
|
||
|
||
# 初始化LoRa协议处理器
|
||
self.lora_handler = LoRaHandler(config)
|
||
self.lora_handler.initialize()
|
||
|
||
logger.info("中继器服务初始化完成")
|
||
|
||
def start(self):
|
||
"""启动中继器服务"""
|
||
logger.info("启动中继器服务")
|
||
self.running = True
|
||
|
||
# 在新线程中运行异步事件循环
|
||
websocket_thread = threading.Thread(target=self._run_websocket_loop, daemon=True)
|
||
websocket_thread.start()
|
||
|
||
# 启动命令处理线程
|
||
command_thread = threading.Thread(target=self._command_loop, daemon=True)
|
||
command_thread.start()
|
||
|
||
# 主循环
|
||
self._main_loop()
|
||
|
||
def stop(self):
|
||
"""停止中继器服务"""
|
||
logger.info("停止中继器服务")
|
||
self.running = False
|
||
|
||
# 断开WebSocket连接
|
||
if self.websocket_client:
|
||
asyncio.run_coroutine_threadsafe(
|
||
self.websocket_client.disconnect(),
|
||
self.loop
|
||
)
|
||
|
||
def _run_websocket_loop(self):
|
||
"""运行WebSocket事件循环"""
|
||
# 创建新的事件循环
|
||
self.loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(self.loop)
|
||
|
||
# 运行事件循环
|
||
self.loop.run_until_complete(self._websocket_main())
|
||
|
||
async def _websocket_main(self):
|
||
"""WebSocket主协程"""
|
||
# 创建WebSocket客户端
|
||
self.websocket_client = WebSocketClient(self)
|
||
|
||
# 持续尝试连接到平台
|
||
while self.running:
|
||
# 连接到平台
|
||
if await self.websocket_client.connect():
|
||
# 监听平台消息
|
||
await self.websocket_client.listen()
|
||
|
||
# 如果连接断开但服务仍在运行，则尝试重新连接
|
||
if self.running and not self.websocket_client.is_connected():
|
||
logger.info("连接已断开，尝试重新连接...")
|
||
await self.websocket_client.reconnect()
|
||
else:
|
||
# 连接失败，等待后重试
|
||
logger.info("连接失败，5秒后重试...")
|
||
await asyncio.sleep(5)
|
||
|
||
# 服务停止时断开连接
|
||
await self.websocket_client.disconnect()
|
||
|
||
def _main_loop(self):
|
||
"""主循环"""
|
||
logger.info("中继器服务主循环开始")
|
||
|
||
try:
|
||
while self.running:
|
||
# 短暂休眠以避免过度占用CPU
|
||
time.sleep(0.1)
|
||
except KeyboardInterrupt:
|
||
logger.info("收到中断信号，正在停止服务...")
|
||
except Exception as e:
|
||
logger.error(f"主循环发生错误: {e}")
|
||
finally:
|
||
self.stop()
|
||
logger.info("中继器服务已停止")
|
||
|
||
def _command_loop(self):
|
||
"""命令处理循环"""
|
||
logger.info("命令处理线程启动")
|
||
|
||
while self.running:
|
||
try:
|
||
# 处理命令队列中的指令
|
||
try:
|
||
command_data = self.command_queue.get(timeout=1)
|
||
self._process_command(command_data)
|
||
self.command_queue.task_done()
|
||
except Empty:
|
||
# 队列为空，继续循环
|
||
continue
|
||
|
||
except Exception as e:
|
||
logger.error(f"命令处理线程发生错误: {e}")
|
||
|
||
logger.info("命令处理线程停止")
|
||
|
||
def _process_command(self, command_data):
|
||
"""处理平台指令"""
|
||
try:
|
||
logger.info(f"处理平台指令: {command_data}")
|
||
|
||
command_type = command_data.get('type')
|
||
command_name = command_data.get('command')
|
||
|
||
if command_type == 'command':
|
||
if command_name == 'control_device':
|
||
self._handle_control_device(command_data)
|
||
elif command_name == 'query_device_status':
|
||
self._handle_query_device_status(command_data)
|
||
elif command_name == 'query_all_device_status':
|
||
self._handle_query_all_device_status(command_data)
|
||
elif command_name == 'heartbeat':
|
||
self._handle_heartbeat_command(command_data)
|
||
else:
|
||
logger.warning(f"未知指令: {command_name}")
|
||
self._send_error_response(command_data, f"未知指令: {command_name}")
|
||
elif command_type == 'heartbeat':
|
||
self._handle_heartbeat_message(command_data)
|
||
elif command_type == 'system':
|
||
# 处理系统消息
|
||
logger.info(f"收到系统消息: {command_name}")
|
||
else:
|
||
logger.warning(f"未知消息类型: {command_type}")
|
||
self._send_error_response(command_data, f"未知消息类型: {command_type}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理指令时发生错误: {e}")
|
||
self._send_error_response(command_data, f"处理指令时发生错误: {str(e)}")
|
||
|
||
def _handle_control_device(self, command_data):
|
||
"""处理设备控制指令"""
|
||
data = command_data.get('data', {})
|
||
device_id = data.get('device_id')
|
||
action = data.get('action')
|
||
|
||
logger.info(f"控制设备: ID={device_id}, 动作={action}")
|
||
|
||
if self.lora_handler and config.simulation_enabled:
|
||
result = self.lora_handler.send_command('control_device', data)
|
||
# 发送响应到平台
|
||
self._send_command_response(command_data, result)
|
||
|
||
def _handle_query_device_status(self, command_data):
|
||
"""处理查询设备状态指令"""
|
||
data = command_data.get('data', {})
|
||
device_id = data.get('device_id')
|
||
|
||
logger.info(f"查询设备状态: ID={device_id}")
|
||
|
||
if self.lora_handler and config.simulation_enabled:
|
||
result = self.lora_handler.send_command('query_device_status', data)
|
||
# 发送响应到平台
|
||
self._send_command_response(command_data, result)
|
||
|
||
def _handle_query_all_device_status(self, command_data):
|
||
"""处理查询所有设备状态指令"""
|
||
logger.info("查询所有设备状态")
|
||
|
||
if self.lora_handler and config.simulation_enabled:
|
||
result = self.lora_handler.send_command('query_all_device_status', {})
|
||
# 发送响应到平台
|
||
self._send_command_response(command_data, result)
|
||
|
||
def _handle_heartbeat_command(self, command_data):
|
||
"""处理心跳指令"""
|
||
logger.info("收到平台心跳指令")
|
||
|
||
# 获取所有设备状态
|
||
devices_status = []
|
||
|
||
# 添加中继器自身状态
|
||
devices_status.append({
|
||
"device_id": self.device_id,
|
||
"device_type": "relay",
|
||
"status": "running"
|
||
})
|
||
|
||
# 如果启用了模拟模式，获取模拟设备状态
|
||
if self.lora_handler and config.simulation_enabled and self.lora_handler.device_manager:
|
||
# 获取所有区域主控设备状态
|
||
controllers = self.lora_handler.device_manager.get_all_controllers()
|
||
for controller in controllers:
|
||
devices_status.append({
|
||
"device_id": controller.device_id,
|
||
"device_type": "area_controller",
|
||
"status": controller.status
|
||
})
|
||
|
||
# 获取该控制器下的所有普通设备状态
|
||
devices = controller.get_all_devices()
|
||
for device in devices:
|
||
devices_status.append({
|
||
"device_id": device.device_id,
|
||
"device_type": device.device_type,
|
||
"status": device.status
|
||
})
|
||
|
||
response = {
|
||
"type": "response",
|
||
"command": "heartbeat",
|
||
"data": {
|
||
"devices": devices_status
|
||
},
|
||
"timestamp": command_data.get("timestamp") # 原封不动地返回平台发送的时间戳
|
||
}
|
||
|
||
self._send_response(response)
|
||
|
||
def _handle_heartbeat_message(self, command_data):
|
||
"""处理心跳消息"""
|
||
logger.info("收到平台心跳消息")
|
||
|
||
# 根据RELAY_API.md，平台发送的心跳消息不需要特殊处理
|
||
# 设备的心跳响应已经在_handle_heartbeat_command中处理
|
||
pass
|
||
|
||
def _send_command_response(self, command_data, result_data):
|
||
"""发送指令响应"""
|
||
response = {
|
||
"type": "response",
|
||
"command": command_data.get('command'),
|
||
"data": result_data,
|
||
"timestamp": datetime.utcnow().isoformat() + 'Z'
|
||
}
|
||
|
||
self._send_response(response)
|
||
|
||
def _send_error_response(self, command_data, error_message):
|
||
"""发送错误响应"""
|
||
response = {
|
||
"type": "response",
|
||
"command": command_data.get('command', 'unknown'),
|
||
"status": "failed",
|
||
"message": error_message,
|
||
"timestamp": datetime.utcnow().isoformat() + 'Z'
|
||
}
|
||
|
||
self._send_response(response)
|
||
|
||
def _send_response(self, response):
|
||
"""发送响应到平台"""
|
||
if self.websocket_client and self.websocket_client.is_connected():
|
||
# 在事件循环中发送响应
|
||
asyncio.run_coroutine_threadsafe(
|
||
self.websocket_client.send_message(response),
|
||
self.loop
|
||
)
|
||
else:
|
||
logger.warning("WebSocket未连接，无法发送响应")
|
||
|
||
def handle_platform_command(self, command_json):
|
||
"""
|
||
处理来自平台的指令（供外部调用）
|
||
|
||
Args:
|
||
command_json (str): JSON格式的指令字符串
|
||
"""
|
||
try:
|
||
command_data = json.loads(command_json)
|
||
self.command_queue.put(command_data)
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"无效的JSON格式: {e}")
|
||
except Exception as e:
|
||
logger.error(f"处理平台指令时发生错误: {e}") |