261 lines
8.4 KiB
Python
261 lines
8.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
激光管理器模块
|
||
|
|
提供激光控制、校准等功能
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from maix import time, camera
|
||
|
|
import threading
|
||
|
|
import config
|
||
|
|
from logger_manager import logger_manager
|
||
|
|
|
||
|
|
|
||
|
|
class LaserManager:
|
||
|
|
"""激光控制管理器（单例）"""
|
||
|
|
_instance = None
|
||
|
|
|
||
|
|
def __new__(cls):
|
||
|
|
if cls._instance is None:
|
||
|
|
cls._instance = super(LaserManager, cls).__new__(cls)
|
||
|
|
cls._instance._initialized = False
|
||
|
|
return cls._instance
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
if self._initialized:
|
||
|
|
return
|
||
|
|
|
||
|
|
# 私有状态
|
||
|
|
self._calibration_active = False
|
||
|
|
self._calibration_result = None
|
||
|
|
self._calibration_lock = threading.Lock()
|
||
|
|
self._laser_point = None
|
||
|
|
|
||
|
|
self._initialized = True
|
||
|
|
|
||
|
|
# ==================== 状态访问（只读属性）====================
|
||
|
|
|
||
|
|
@property
|
||
|
|
def calibration_active(self):
|
||
|
|
"""是否正在校准"""
|
||
|
|
return self._calibration_active
|
||
|
|
|
||
|
|
@property
|
||
|
|
def laser_point(self):
|
||
|
|
"""当前激光点"""
|
||
|
|
return self._laser_point
|
||
|
|
|
||
|
|
# ==================== 业务方法 ====================
|
||
|
|
|
||
|
|
def load_laser_point(self):
|
||
|
|
"""从配置文件加载激光中心点，失败则使用默认值"""
|
||
|
|
try:
|
||
|
|
if "laser_config.json" in os.listdir("/root"):
|
||
|
|
with open(config.CONFIG_FILE, "r") as f:
|
||
|
|
data = json.load(f)
|
||
|
|
if isinstance(data, list) and len(data) == 2:
|
||
|
|
self._laser_point = (int(data[0]), int(data[1]))
|
||
|
|
logger = logger_manager.logger
|
||
|
|
if logger:
|
||
|
|
logger.debug(f"[INFO] 加载激光点: {self._laser_point}")
|
||
|
|
return self._laser_point
|
||
|
|
else:
|
||
|
|
raise ValueError
|
||
|
|
else:
|
||
|
|
self._laser_point = config.DEFAULT_LASER_POINT
|
||
|
|
except:
|
||
|
|
self._laser_point = config.DEFAULT_LASER_POINT
|
||
|
|
|
||
|
|
return self._laser_point
|
||
|
|
|
||
|
|
def save_laser_point(self, point):
|
||
|
|
"""保存激光中心点到配置文件"""
|
||
|
|
try:
|
||
|
|
with open(config.CONFIG_FILE, "w") as f:
|
||
|
|
json.dump([point[0], point[1]], f)
|
||
|
|
self._laser_point = point
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
logger = logger_manager.logger
|
||
|
|
if logger:
|
||
|
|
logger.error(f"[LASER] 保存激光点失败: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def turn_on_laser(self):
|
||
|
|
"""发送指令开启激光，并读取回包（部分模块支持）"""
|
||
|
|
from hardware import hardware_manager
|
||
|
|
logger = logger_manager.logger
|
||
|
|
|
||
|
|
if hardware_manager.distance_serial is None:
|
||
|
|
if logger:
|
||
|
|
logger.error("[LASER] distance_serial 未初始化")
|
||
|
|
return None
|
||
|
|
|
||
|
|
# 打印调试信息
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 发送开启命令: {config.LASER_ON_CMD.hex()}")
|
||
|
|
|
||
|
|
# 清空接收缓冲区
|
||
|
|
try:
|
||
|
|
hardware_manager.distance_serial.read(-1) # 清空缓冲区
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# 发送命令
|
||
|
|
written = hardware_manager.distance_serial.write(config.LASER_ON_CMD)
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 写入字节数: {written}")
|
||
|
|
|
||
|
|
time.sleep_ms(50) # 增加等待时间，让模块有时间响应
|
||
|
|
|
||
|
|
# 读取回包
|
||
|
|
resp = hardware_manager.distance_serial.read(20)
|
||
|
|
if resp:
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
|
||
|
|
if resp == config.LASER_ON_CMD:
|
||
|
|
if logger:
|
||
|
|
logger.info("✅ 激光开启指令已确认")
|
||
|
|
else:
|
||
|
|
if logger:
|
||
|
|
logger.warning("🔇 无回包（可能正常或模块不支持回包）")
|
||
|
|
return resp
|
||
|
|
|
||
|
|
def turn_off_laser(self):
|
||
|
|
"""发送指令关闭激光"""
|
||
|
|
from hardware import hardware_manager
|
||
|
|
logger = logger_manager.logger
|
||
|
|
|
||
|
|
if hardware_manager.distance_serial is None:
|
||
|
|
if logger:
|
||
|
|
logger.error("[LASER] distance_serial 未初始化")
|
||
|
|
return None
|
||
|
|
|
||
|
|
# 打印调试信息
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 发送关闭命令: {config.LASER_OFF_CMD.hex()}")
|
||
|
|
|
||
|
|
# 清空接收缓冲区
|
||
|
|
try:
|
||
|
|
hardware_manager.distance_serial.read(-1)
|
||
|
|
except:
|
||
|
|
pass
|
||
|
|
|
||
|
|
# 发送命令
|
||
|
|
written = hardware_manager.distance_serial.write(config.LASER_OFF_CMD)
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 写入字节数: {written}")
|
||
|
|
|
||
|
|
time.sleep_ms(50) # 增加等待时间
|
||
|
|
|
||
|
|
# 读取回包
|
||
|
|
resp = hardware_manager.distance_serial.read(20)
|
||
|
|
if resp:
|
||
|
|
if logger:
|
||
|
|
logger.info(f"[LASER] 收到回包 ({len(resp)}字节): {resp.hex()}")
|
||
|
|
else:
|
||
|
|
if logger:
|
||
|
|
logger.warning("🔇 无回包")
|
||
|
|
return resp
|
||
|
|
|
||
|
|
def flash_laser(self, duration_ms=1000):
|
||
|
|
"""闪一下激光（用于射箭反馈）"""
|
||
|
|
try:
|
||
|
|
self.turn_on_laser()
|
||
|
|
time.sleep_ms(duration_ms)
|
||
|
|
self.turn_off_laser()
|
||
|
|
except Exception as e:
|
||
|
|
logger = logger_manager.logger
|
||
|
|
if logger:
|
||
|
|
logger.error(f"闪激光失败: {e}")
|
||
|
|
|
||
|
|
def find_red_laser(self, frame, threshold=150):
|
||
|
|
"""在图像中查找最亮的红色激光点（基于 RGB 阈值）"""
|
||
|
|
w, h = frame.width(), frame.height()
|
||
|
|
img_bytes = frame.to_bytes()
|
||
|
|
max_sum = 0
|
||
|
|
best_pos = None
|
||
|
|
for y in range(0, h, 2):
|
||
|
|
for x in range(0, w, 2):
|
||
|
|
idx = (y * w + x) * 3
|
||
|
|
r, g, b = img_bytes[idx], img_bytes[idx+1], img_bytes[idx+2]
|
||
|
|
if r > threshold and r > g * 2 and r > b * 2:
|
||
|
|
rgb_sum = r + g + b
|
||
|
|
if rgb_sum > max_sum:
|
||
|
|
max_sum = rgb_sum
|
||
|
|
best_pos = (x, y)
|
||
|
|
return best_pos
|
||
|
|
|
||
|
|
def calibrate_laser_position(self):
|
||
|
|
"""执行一次激光校准：拍照 → 找红点 → 保存坐标"""
|
||
|
|
time.sleep_ms(80)
|
||
|
|
cam = camera.Camera(640, 480)
|
||
|
|
frame = cam.read()
|
||
|
|
pos = self.find_red_laser(frame)
|
||
|
|
if pos:
|
||
|
|
self.save_laser_point(pos)
|
||
|
|
return pos
|
||
|
|
return None
|
||
|
|
|
||
|
|
def start_calibration(self):
|
||
|
|
"""开始校准（公共方法）"""
|
||
|
|
with self._calibration_lock:
|
||
|
|
if self._calibration_active:
|
||
|
|
return False
|
||
|
|
self._calibration_active = True
|
||
|
|
self._calibration_result = None
|
||
|
|
return True
|
||
|
|
|
||
|
|
def stop_calibration(self):
|
||
|
|
"""停止校准（公共方法）"""
|
||
|
|
with self._calibration_lock:
|
||
|
|
self._calibration_active = False
|
||
|
|
|
||
|
|
def set_calibration_result(self, result):
|
||
|
|
"""设置校准结果（内部方法）"""
|
||
|
|
with self._calibration_lock:
|
||
|
|
self._calibration_result = result
|
||
|
|
|
||
|
|
def get_calibration_result(self):
|
||
|
|
"""获取并清除校准结果（内部方法）"""
|
||
|
|
with self._calibration_lock:
|
||
|
|
result = self._calibration_result
|
||
|
|
self._calibration_result = None
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
# 创建全局单例实例
|
||
|
|
laser_manager = LaserManager()
|
||
|
|
|
||
|
|
# ==================== 向后兼容的函数接口 ====================
|
||
|
|
|
||
|
|
def load_laser_point():
|
||
|
|
"""加载激光点（向后兼容接口）"""
|
||
|
|
return laser_manager.load_laser_point()
|
||
|
|
|
||
|
|
def save_laser_point(point):
|
||
|
|
"""保存激光点（向后兼容接口）"""
|
||
|
|
return laser_manager.save_laser_point(point)
|
||
|
|
|
||
|
|
def turn_on_laser():
|
||
|
|
"""开启激光（向后兼容接口）"""
|
||
|
|
return laser_manager.turn_on_laser()
|
||
|
|
|
||
|
|
def turn_off_laser():
|
||
|
|
"""关闭激光（向后兼容接口）"""
|
||
|
|
return laser_manager.turn_off_laser()
|
||
|
|
|
||
|
|
def flash_laser(duration_ms=1000):
|
||
|
|
"""闪激光（向后兼容接口）"""
|
||
|
|
return laser_manager.flash_laser(duration_ms)
|
||
|
|
|
||
|
|
def find_red_laser(frame, threshold=150):
|
||
|
|
"""查找红色激光点（向后兼容接口）"""
|
||
|
|
return laser_manager.find_red_laser(frame, threshold)
|
||
|
|
|
||
|
|
def calibrate_laser_position():
|
||
|
|
"""校准激光位置（向后兼容接口）"""
|
||
|
|
return laser_manager.calibrate_laser_position()
|
||
|
|
|