加上权重

bx/BxCmdFactory.py

@@ -1,3 +1,4 @@
+from .BxCmdDelDynamicArea import NewCmdDelDynamicArea
 from .BxCmdSendDynamicArea import NewBxCmdSendDynamicArea
 from .BxCmdState import NewCmdState
 from .BxCmdReadParams import NewCmdReadParams
@@ -19,4 +20,8 @@ class BxCmdFactory:
     @staticmethod
     def NewCmdReadParams():
         """创建读取参数命令"""
-        return NewCmdReadParams()
+        return NewCmdReadParams()
+
+    @staticmethod
+    def NewCmdDelDynamicArea(numbers):
+        return NewCmdDelDynamicArea(numbers)

bx/BxDataPack.py

@@ -98,15 +98,25 @@ class BxDataPack:
         return result
 
     # ---------------- 第一步：先补全辅助工具函数（避免bytearray/编码/长度错误） ----------------
-    def _to_int(self, val) -> int:
-        """将bytearray/bytes/str转为0-255整数，避免append报错"""
-        if isinstance(val, (bytearray, bytes)):
-            return val[0] if len(val) > 0 else 0
-        elif isinstance(val, str):
-            return int(val) if val.isdigit() else 0
-        elif isinstance(val, int):
-            return val & 0xFF  # 限制0-255
-        return 0
+    def _to_int(self, value):
+        """2. 修复_to_int：强制兜底，确保0x8000能正确转换"""
+        # 第一步：强制把源地址相关值转成0x8000（兜底！）
+        if id(value) == id(self.srcAddr) or value in (0x8000, "0x8000", b'\x00\x80'):
+            return 0x8000
+
+        # 常规转换（补全所有边界情况）
+        if value is None or value == "":
+            return 0
+        if isinstance(value, bytes):
+            # 处理2字节的源地址字节（比如00 80）
+            if len(value) == 2:
+                return struct.unpack('<H', value)[0]
+            return int.from_bytes(value, byteorder='little')
+        elif isinstance(value, str):
+            return int(value.strip(), 16) if value.startswith(('0x', '0X')) else int(value)
+        else:
+            # 强制转整数，避免溢出
+            return int(value) & 0xFFFF  # 保留16位，避免截断
 
     def _to_bytes(self, val) -> bytes:
         """将字符串/bytearray转为GB2312编码的bytes，避免中文乱码"""
@@ -129,6 +139,7 @@ class BxDataPack:
         result.extend(struct.pack('<H', self._to_int(self.dstAddr)))
         # 写入源地址（小端序，对齐Go的0x8000！！！）
         self.srcAddr = 0x8000  # 强制设为Go的固定值
+        # print(f"[调试] 源地址转换后：{hex(self._to_int(self.srcAddr))}")
         result.extend(struct.pack('<H', self._to_int(self.srcAddr)))
         # 写入保留字
         result.append(self._to_int(self.r0))

modules/display/screen.py

@@ -336,7 +336,7 @@ class Screen:
         return
 
     def text_ram(self, ff: FlashFile, is_number: bool):
-        """发送动态区节目实现"""
+        """发送动态区节目实现（新增：车牌长度校验+绿牌FE编码区分）"""
         if not self.get_live_state():
             return
         areas = []
@@ -350,18 +350,45 @@ class Screen:
         if ff.color == Color.DEFAULT and not is_number:
             bytes_data, _ = encoder(ff.msg)
         elif is_number:
-            province, number = self.parse_plate_utf8(ff.msg)
-            print(f"车牌号: {ff.msg}")
+            # 完整车牌（如湘F65354）
+            full_plate = ff.msg.strip()
+            print(f"车牌号: {full_plate}")
+
+            # ========== 核心新增：车牌合法性校验（按长度） ==========
+            # 校验规则：
+            # - 普通燃油车牌：7位（如湘F65354）→ FE000
+            # - 新能源绿色车牌：8位（如湘F653548）→ FE001
+            # - 非7/8位直接判定为非法，跳过处理
+            full_plate_len = len(full_plate)
+            if full_plate_len not in [7, 8]:
+                print(f"【校验失败】车牌{full_plate}长度{full_plate_len}位，非法（仅支持7/8位），跳过")
+                return
+
+            # 解析省份+号码（复用原有方法）
+            province, number = self.parse_plate_utf8(full_plate)
+
+            # 校验解析后的格式（省份非空 + 号码长度6/7位 + 号码首字符是字母）
+            if not province or len(number) not in [6, 7] or not number[:1].isalpha() or not number[1:].isalnum():
+                print(f"【校验失败】车牌{full_plate}格式错误（解析：省份{province}，号码{number}），跳过")
+                return
+
+            # ========== 原有逻辑：判断显示模式 ==========
             if len(number) == 6:
                 disp_mode = DisplayMode.QUICK_PUNCH
-            msg = f"\\FO000\\C{ff.color}{province}\\FE000\\C{Color.GREEN}{number}"
+
+            # ========== 核心修改：按长度判断绿牌，设置FE编码 ==========
+            # 8位车牌（新能源绿牌）→ FE001，7位车牌（普通牌）→ FE000
+            fe_code = "001" if full_plate_len == 8 else "000"
+
+            # 构建显示指令（替换FE编码，ff.color为屏幕颜色，保持不变）
+            msg = f"\\FO000\\C{ff.color}{province}\\FE{fe_code}\\C{Color.GREEN}{number}"
             bytes_data, _ = encoder(msg)
             id = 1
             dst_addr = 0x0001
         elif not is_number:
             # 优化：使用与数字显示相同的格式，避免设备拒绝写入
             msg = f"\\FO000\\C{ff.color}{ff.msg}"
-            print(ff.msg)
+            # print(ff.msg)
             bytes_data, _ = encoder(msg)
             alignment = 0x00
             dst_addr = 0x0003
@@ -375,7 +402,7 @@ class Screen:
 
         # 打包指令并发送
         cmd = BxCmdFactory.NewBxCmdSendDynamicArea(areas)
-        pack = NewBxDataPackCmd(cmd,dst_addr)
+        pack = NewBxDataPackCmd(cmd, dst_addr)
         pack.SetDisplayType(1)
         data = pack.Pack()
         # print(f"打包后数据（十六进制）: {data.hex()} {is_number}")
@@ -392,12 +419,13 @@ class Screen:
 
         # 读取响应
         resp = self.read_resp()
-        resp_time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]  # 获取当前时间 毫秒级
-        print(f"[响应时间 {resp_time_str}]")
+        # resp_time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]  # 获取当前时间 毫秒级
+        # print(f"[响应时间 {resp_time_str}]")
         if not resp.IsAck():
             remote_ip, remote_port = self.conn.getpeername()
             error_msg = resp.Error() or f"错误码: {resp.Err}"
-            print(f"[{self.name}] 设备拒绝写文件: {resp.Error().ErrorCode}  错误：{remote_ip} {remote_port} {data.hex()}")
+            print(
+                f"[{self.name}] 设备拒绝写文件: {resp.Error().ErrorCode}  错误：{remote_ip} {remote_port} {data.hex()}")
             return
         self.state_info["dyna_area_num"] = self.state_info.get("dyna_area_num", 0) + 1
 
@@ -538,3 +566,32 @@ class Screen:
             self.state_info["program_num"] = 0
         else:
             self.state_info["program_num"] = self.state_info.get("program_num", 0) - 1
+
+    def del_ram_text(self, *numbers: int):
+        """
+        对应 Go 的 DelRamText 方法：删除 RAM 文本动态区域
+        :param numbers: 可变参数，字节类型的动态区域编号
+        """
+        # 1. 检查屏幕是否处于活跃状态
+        if not self.get_live_state():
+            return
+
+        # 2. 创建删除动态区域的指令（对应 Go 的 bx.NewCmdDelDynamicArea）
+        cmd = BxCmdFactory.NewCmdDelDynamicArea(numbers)
+
+        # 3. 打包指令（对应 Go 的 bx.NewBxDataPackCmd）
+        pack = NewBxDataPackCmd(cmd,0x0003)
+
+        # 4. 发送打包后的指令
+        data = pack.Pack()
+        # print(f"打包后数据（十六进制）: {data.hex()}")
+        self.send(data)
+
+        # 5. 读取响应
+        self.read_resp()
+
+        # 6. 更新动态区域数量状态
+        if len(numbers) == 0:
+            self.state_info["dyna_area_num"] = 0
+        else:
+            self.state_info["dyna_area_num"] -= 1

modules/radar/radar.py

@@ -22,13 +22,14 @@ class DeviceInitData:
     LowSpeed: float = 0.0  # 速度阈值，比如5.0（速度>5触发回调）
 
 
-# ========== 新增1：超时清空状态管理 ==========
+# ========== 新增1：超时清空状态管理（修复初始化+超大时间差） ==========
 class ScreenClearManager:
-    """屏幕超时清空管理器"""
+    """屏幕超时清空管理器（修复初始化问题）"""
 
     def __init__(self, timeout_seconds: float = 2.0):
         self.timeout_seconds = timeout_seconds  # 超时时间（秒）
-        self.last_valid_time = 0.0  # 最后一次收到有效数据的时间
+        # 初始化：使用当前时间，避免超大时间差
+        self.last_valid_time = time.time()
         self.screen_cleared = False  # 屏幕是否已清空
 
     def update_valid_time(self):
@@ -38,34 +39,61 @@ class ScreenClearManager:
 
     def check_and_clear(self, radar_screen: Optional[Screen]) -> bool:
         """
-        检查是否需要清空屏幕
+        检查是否需要清空屏幕（修复清空逻辑+错误处理）
         :return: 是否清空了屏幕
         """
-        if radar_screen is None or not radar_screen.get_live_state():
+        # 1. 屏幕实例不存在/未连接，直接返回
+        if radar_screen is None or not hasattr(radar_screen, 'get_live_state') or not radar_screen.get_live_state():
             return False
 
         current_time = time.time()
         time_since_last_valid = current_time - self.last_valid_time
 
+        # 2. 仅在超时且未清空时执行（修复超大时间差显示）
         if time_since_last_valid > self.timeout_seconds and not self.screen_cleared:
-            # 超时了，清空屏幕
-            print(f"⏱️  雷达超时：{time_since_last_valid:.1f}秒无有效数据，清空屏幕 {radar_screen.conn.getpeername()}")
+            # 限制最大显示时间差为timeout_seconds*2，避免超大数值
+            display_seconds = min(time_since_last_valid, self.timeout_seconds * 2)
+            # 获取屏幕地址（兼容不同Screen类实现）
+            screen_addr = ""
             try:
-                ff = FlashFile()
-                ff.set_msg("00", 0)  # 发送空内容
-                ff.set_mode(4, 2)
-                ff.set_origin(0, True, 0)
-                ff.set_area(128, True, 96)
-                radar_screen.text_ram(ff, False)
+                screen_addr = radar_screen.conn.getpeername() if hasattr(radar_screen, 'conn') else (
+                radar_screen.ip, radar_screen.port)
+            except:
+                screen_addr = "未知地址"
+
+            print(f"雷达超时：{display_seconds:.1f}秒无有效数据，清空屏幕 {screen_addr}")
+
+            try:
+                # 修复清空逻辑：优先使用安全的del_ram_text，兼容不同Screen实现
+                if hasattr(radar_screen, 'del_ram_text'):
+                    radar_screen.del_ram_text(0)
+                # 备选方案：显示空内容（兼容无del_ram_text的情况）
+                elif hasattr(radar_screen, 'text_ram'):
+                    ff = FlashFile()
+                    ff.set_msg("", 0)
+                    ff.set_mode(4, 2)
+                    ff.set_origin(0, True, 0)
+                    ff.set_area(128, True, 96)
+                    radar_screen.text_ram(ff, False)
+
                 self.screen_cleared = True
                 return True
             except Exception as e:
-                print(f"❌ 清空屏幕失败：{e}")
+                # 精准捕获错误，避免'dyna_area_num'报错刷屏
+                if 'dyna_area_num' in str(e):
+                    print(f"清空屏幕失败：屏幕实例无'dyna_area_num'属性（非致命错误）")
+                else:
+                    print(f"清空屏幕失败：{str(e)[:50]}")  # 限制错误信息长度
+            return False
+
+        # 3. 未超时但已清空（恢复数据后重置）
+        elif time_since_last_valid <= self.timeout_seconds and self.screen_cleared:
+            self.screen_cleared = False
 
         return False
 
 
-# 初始化全局的超时清空管理器
+# 初始化全局的超时清空管理器（确保启动时就初始化）
 screen_clear_manager = ScreenClearManager(timeout_seconds=2.0)
 
 
@@ -132,8 +160,8 @@ def parse_radar_frame(frame: bytes) -> RadarData:
     data.Speed = float(hundreds * 100 + tens * 10 + units) + float(decimal) / 10.0
     data.Valid = True
 
-    # ========== 新增2：打印解析到的原始数据（调试用） ==========
-    print(f"📡 雷达解析成功：方向={data.Direction}，速度={data.Speed:.1f}，原始帧={frame!r}")
+    # ========== 调试用：打印解析到的原始数据 ==========
+    # print(f"📡 雷达解析成功：方向={data.Direction}，速度={data.Speed:.1f}，原始帧={frame!r}")
 
     return data
 
@@ -169,7 +197,7 @@ def callback_with_data(
                 print(f"语音播报失败：{e}")
 
         # 屏幕显示
-        if radar_screen and radar_screen.get_live_state():
+        if radar_screen and hasattr(radar_screen, 'get_live_state') and radar_screen.get_live_state():
             try:
                 ff = FlashFile()
                 ff.set_msg(rounded_speed, color)
@@ -178,10 +206,10 @@ def callback_with_data(
                 ff.set_area(128, True, 96)
                 radar_screen.text_ram(ff, False)
 
-                # ========== 新增3：更新有效数据时间 ==========
+                # ========== 更新有效数据时间 ==========
                 screen_clear_manager.update_valid_time()
 
-                print(f"✅ 雷达速度{rounded_speed}已发送到屏幕")
+                # print(f"雷达速度{rounded_speed}已发送到屏幕")
             except Exception as e:
                 print(f"雷达屏幕显示失败：{e}")
 
@@ -192,7 +220,7 @@ def open_serial(
         radar_screen: Optional[Screen] = None,
         radar_serial_ref=None
 ):
-    """打开雷达串口并监听（彻底解决残留数据问题）"""
+    """打开雷达串口并监听（彻底解决残留数据+初始化问题）"""
     radar_serial = radar_serial_ref
 
     serial_options = {
@@ -214,18 +242,23 @@ def open_serial(
             send_len = radar_serial.write(radar_cmd)
             if send_len != len(radar_cmd):
                 print(f"雷达配置指令发送不完整")
+
             time.sleep(0.1)
 
             read_buf: List[int] = []
             buf = bytearray(1024)
             last_clear_check = time.time()
+            # 初始化有效数据时间（双重保障）
             last_valid_data_time = time.time()
+            # 重置屏幕清空管理器（确保每次重连都初始化）
+            screen_clear_manager.last_valid_time = time.time()
+            screen_clear_manager.screen_cleared = False
 
             while True:
                 try:
                     n = radar_serial.readinto(buf)
 
-                    # 定期检查超时清空
+                    # 定期检查超时清空（0.2秒一次，避免高频检查）
                     current_time = time.time()
                     if current_time - last_clear_check > 0.2:
                         screen_clear_manager.check_and_clear(radar_screen)
@@ -259,10 +292,12 @@ def open_serial(
                             # 更新时间
                             last_valid_data_time = data_time
                             last_clear_check = current_time
+                            # 更新清空管理器的时间（核心修复：有数据时同步）
+                            screen_clear_manager.update_valid_time()
 
                             # 日志显示
                             # local_time = time.strftime("%H:%M:%S", time.localtime(data_time))
-                            # print(f"📡 [{local_time}] 速度={radar_data.Speed:.1f} (距上次{time_since_last:.2f}秒)")
+                            # print(f"[{local_time}] 速度={radar_data.Speed:.1f} (距上次{time_since_last:.2f}秒)")
                             if radar_data.Speed < 5:
                                 continue