289 lines
10 KiB
Python
289 lines
10 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)
|
|||
|
|
|
|||
|
|
# 连接到平台
|
|||
|
|
if await self.websocket_client.connect():
|
|||
|
|
# 启动心跳任务
|
|||
|
|
heartbeat_task = asyncio.create_task(self._heartbeat_routine())
|
|||
|
|
|
|||
|
|
# 监听平台消息
|
|||
|
|
await self.websocket_client.listen()
|
|||
|
|
|
|||
|
|
# 取消心跳任务
|
|||
|
|
heartbeat_task.cancel()
|
|||
|
|
else:
|
|||
|
|
logger.error("无法连接到平台")
|
|||
|
|
|
|||
|
|
async def _heartbeat_routine(self):
|
|||
|
|
"""心跳协程"""
|
|||
|
|
logger.info("心跳协程启动")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
while self.running:
|
|||
|
|
if self.websocket_client and self.websocket_client.is_connected():
|
|||
|
|
await self.websocket_client.send_heartbeat()
|
|||
|
|
|
|||
|
|
# 每30秒发送一次心跳
|
|||
|
|
await asyncio.sleep(30)
|
|||
|
|
except asyncio.CancelledError:
|
|||
|
|
logger.info("心跳协程被取消")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"心跳协程发生错误: {e}")
|
|||
|
|
|
|||
|
|
logger.info("心跳协程停止")
|
|||
|
|
|
|||
|
|
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)
|
|||
|
|
else:
|
|||
|
|
logger.warning(f"未知指令: {command_name}")
|
|||
|
|
self._send_error_response(command_data, f"未知指令: {command_name}")
|
|||
|
|
elif command_type == 'heartbeat':
|
|||
|
|
self._handle_heartbeat(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(self, command_data):
|
|||
|
|
"""处理心跳消息"""
|
|||
|
|
logger.info("收到平台心跳消息")
|
|||
|
|
|
|||
|
|
# 发送心跳响应
|
|||
|
|
response = {
|
|||
|
|
"type": "response",
|
|||
|
|
"command": "heartbeat",
|
|||
|
|
"status": "success",
|
|||
|
|
"message": "心跳响应",
|
|||
|
|
"timestamp": datetime.utcnow().isoformat() + 'Z'
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
self._send_response(response)
|
|||
|
|
|
|||
|
|
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}")
|