88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
|
#!/usr/bin/env python
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
import time
|
|||
|
|
import logging
|
|||
|
|
from threading import Thread
|
|||
|
|
from unittest.mock import patch
|
|||
|
|
|
|||
|
|
# 假设你的 RelayController 类在名为 EMV 的模块中
|
|||
|
|
from EMV import RelayController # 替换为你的模块名
|
|||
|
|
|
|||
|
|
# 设置日志格式
|
|||
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|||
|
|
|
|||
|
|
def simulate_sensor(controller, sensor_name, active=True, duration=10):
|
|||
|
|
"""
|
|||
|
|
模拟指定传感器在一段时间内处于激活状态。
|
|||
|
|
:param controller: RelayController 实例
|
|||
|
|
:param sensor_name: 要模拟的传感器名（如 controller.SENSOR2）
|
|||
|
|
:param active: 是否激活传感器信号
|
|||
|
|
:param duration: 模拟运行时间（秒）
|
|||
|
|
"""
|
|||
|
|
logging.info(f"🧪 开始模拟传感器：{controller.sensor_name_map.get(sensor_name, sensor_name)}")
|
|||
|
|
original_method = controller.get_all_device_status
|
|||
|
|
|
|||
|
|
def mock_get_all_device_status(command_type='devices'):
|
|||
|
|
if command_type == 'sensors':
|
|||
|
|
all_status = original_method(command_type)
|
|||
|
|
all_status[sensor_name] = active
|
|||
|
|
return all_status
|
|||
|
|
return original_method(command_type)
|
|||
|
|
|
|||
|
|
# 确定目标函数
|
|||
|
|
target_func = None
|
|||
|
|
if sensor_name == controller.SENSOR1:
|
|||
|
|
target_func = controller.handle_sensor1
|
|||
|
|
elif sensor_name == controller.SENSOR2:
|
|||
|
|
target_func = controller.handle_sensor2
|
|||
|
|
else:
|
|||
|
|
raise ValueError("不支持的传感器名称")
|
|||
|
|
|
|||
|
|
# 设置 _running 为 True，确保线程能进入循环
|
|||
|
|
controller._running = True
|
|||
|
|
|
|||
|
|
# 启动线程
|
|||
|
|
sensor_thread = Thread(target=target_func, daemon=True)
|
|||
|
|
sensor_thread.start()
|
|||
|
|
logging.info(f"✅ {controller.sensor_name_map[sensor_name]} 监听线程已启动")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# Patch get_all_device_status 方法
|
|||
|
|
with patch.object(controller, 'get_all_device_status', mock_get_all_device_status):
|
|||
|
|
logging.info(f"🟢 模拟 {controller.sensor_name_map[sensor_name]} 有信号输入，持续 {duration} 秒")
|
|||
|
|
time.sleep(duration)
|
|||
|
|
except Exception as e:
|
|||
|
|
logging.error(f"🔴 模拟过程中发生错误: {e}")
|
|||
|
|
finally:
|
|||
|
|
# 停止控制器
|
|||
|
|
controller._running = False
|
|||
|
|
logging.info("🛑 停止控制器主循环")
|
|||
|
|
# 等待线程退出
|
|||
|
|
sensor_thread.join(timeout=2)
|
|||
|
|
if sensor_thread.is_alive():
|
|||
|
|
logging.warning("⚠️ 传感器线程未能及时退出")
|
|||
|
|
else:
|
|||
|
|
logging.info("✅ 传感器线程已安全退出")
|
|||
|
|
|
|||
|
|
if __name__ == '__main__':
|
|||
|
|
# 创建控制器实例
|
|||
|
|
relay_controller = RelayController()
|
|||
|
|
|
|||
|
|
# 打印当前配置的传感器名称映射（方便调试）
|
|||
|
|
logging.info("🔧 当前传感器配置:")
|
|||
|
|
for key, val in relay_controller.sensor_name_map.items():
|
|||
|
|
logging.info(f" {key}: {val}")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# 模拟 SENSOR2 有信号输入，运行 10 秒
|
|||
|
|
logging.info("🧪 开始模拟传感器2")#修改这里
|
|||
|
|
simulate_sensor(relay_controller, relay_controller.SENSOR1, active=True, duration=10)#
|
|||
|
|
|
|||
|
|
# 可选：模拟 SENSOR1
|
|||
|
|
# logging.info("🧪 开始模拟传感器1")
|
|||
|
|
# simulate_sensor(relay_controller, relay_controller.SENSOR1, active=True, duration=10)
|
|||
|
|
|
|||
|
|
except KeyboardInterrupt:
|
|||
|
|
logging.info("🛑 用户中断模拟")
|
|||
|
|
finally:
|
|||
|
|
logging.info("🏁 模拟结束")
|