首次提交完整的python工程代码

2026-06-15 20:38:42 +08:00
commit bca485d748
159 changed files with 248076 additions and 0 deletions
--- a/converter/b_to_d.py
+++ b/converter/b_to_d.py
@@ -0,0 +1,181 @@
+# from typing import Optional
+# import struct
+# from converter.base_converter import BaseConverter
+# from utils.logger import logger
+# from utils.checksum import calc_sum
+# from utils.hex_helper import hex_dump
+# from structure.b_frame import B_STAR_TIME_OFFSET, B_STAR_TIME_LEN
+# from structure.d_frame import D_RAW_DATA_LEN
+# from config.settings import FRAME_CONST
+
+
+# class B2DConverter(BaseConverter):
+#     def convert(self, data: bytes) -> Optional[bytes]:
+#         try:
+#             # 1. 提取星上时
+#             star_time = data[B_STAR_TIME_OFFSET:B_STAR_TIME_OFFSET + B_STAR_TIME_LEN]
+#             if len(star_time) != 6:
+#                 return None
+
+#             sec = star_time[:4]
+#             ms = star_time[4:]
+#             ms_4b = ms + b"\x00\x00"
+
+#             logger.info(f"星上时: {sec.hex()} {ms.hex()} → {ms_4b.hex()}")
+
+#             # 2. 拼装原始数据
+#             raw = sec + ms
+#             raw += b"\x00" * (D_RAW_DATA_LEN - len(raw))
+
+#             # 3. 校验域
+#             src = FRAME_CONST["SRC"]
+#             dst = FRAME_CONST["DST"]
+#             it = FRAME_CONST["IF_TYPE"]
+#             wa = FRAME_CONST["WORD_ADDR"]
+#             res2 = b"\x00\x00\x00\x00"
+
+#             check_data = struct.pack("BBBB", src, dst, it, wa) + res2 + raw
+#             cs = calc_sum(check_data)
+
+#             # 4. 组帧
+#             frame = (
+#                     struct.pack(">H", FRAME_CONST["HEAD"])
+#                     + struct.pack(">H", FRAME_CONST["SIM_ID"])
+#                     + struct.pack(">H", 0x0000)
+#                     + struct.pack(">H", FRAME_CONST["DATA_LEN"])
+#                     + check_data
+#                     + struct.pack(">H", cs)
+#                     + struct.pack(">H", FRAME_CONST["TAIL"])
+#             )
+
+#             hex_dump(frame, "发送D帧")
+#             return frame
+
+#         except Exception as e:
+#             logger.error(f"B→D转换失败: {e}")
+#             return None
+        
+#     @staticmethod
+#     def extract_bits(data: bytes, start_bit: int, bit_len: int) -> bytes:
+#         """
+#         从字节流中精确提取 [start_bit, start_bit+bit_len) 比特
+#         大端序、比特紧密对齐
+#         """
+#         bit_str = ''
+#         for byte in data:
+#             bit_str += f'{byte:08b}'
+
+#         end_bit = start_bit + bit_len
+#         selected = bit_str[start_bit:end_bit]
+
+#         # 不足补0
+#         if len(selected) < bit_len:
+#             selected += '0' * (bit_len - len(selected))
+
+#         # 转回 bytes
+#         res = b''
+#         for i in range(0, len(selected), 8):
+#             byte_bits = selected[i:i+8].ljust(8, '0')
+#             res += int(byte_bits, 2).to_bytes(1, byteorder='big')
+
+#         return res
+    
+from typing import Optional
+import struct
+from converter.base_converter import BaseConverter
+from utils.logger import logger
+from utils.checksum import calc_sum
+from utils.hex_helper import hex_dump
+from structure.d_frame import D_RAW_DATA_LEN
+from config.settings import FRAME_CONST
+
+class B2DConverter(BaseConverter):
+    def convert(self, data: bytes) -> Optional[bytes]:
+        try:
+            # ==============================================
+            # 【2026新版B遥测协议】星上时精确偏移
+            # ==============================================
+            START_BIT = 608      # 新版协议：星上时起始比特
+            STAR_TIME_BITS = 48  # 6B = 48b
+
+            # 从比特流中精确提取星上时
+            star_time_bytes = self.extract_bits(data, START_BIT, STAR_TIME_BITS)
+            if len(star_time_bytes) != 6:
+                logger.warning(f"星上时长度异常：{len(star_time_bytes)}B")
+                return None
+
+            # ==============================================
+            # 拆分：4B秒 + 2B毫秒 → 毫秒扩展为4B
+            # ==============================================
+            seconds = star_time_bytes[:4]
+            ms = star_time_bytes[4:]
+            # ✅ 大端序：高位补 0x0000（你要求的正确方式）
+            ms_4b = b"\x00\x00" + ms
+
+            logger.info("✅ 星上时解析成功（新版协议）")
+            logger.info(f"  原始6B：{star_time_bytes.hex()}")
+            logger.info(f"  秒值(4B)：{seconds.hex()}")
+            logger.info(f"  毫秒(2B→4B)：{ms.hex()} → {ms_4b.hex()}")
+
+            # ==============================================
+            # 封装 D 软件帧
+            # ==============================================
+            #raw_data = seconds + ms
+            raw_data = seconds + ms_4b
+            raw_data += b"\x00" * (D_RAW_DATA_LEN - len(raw_data))
+
+            src = FRAME_CONST["SRC"]#源
+            dst = FRAME_CONST["DST"]#宿
+            if_type = FRAME_CONST["IF_TYPE"]#接口类型
+            word_addr = FRAME_CONST["WORD_ADDR"]#字地址
+            reserved2 = b"\x00\x00\x00\x00"#保留
+
+            check_data = (
+                struct.pack("BBBB", src, dst, if_type, word_addr)
+                + reserved2
+                + raw_data
+            )
+            checksum = calc_sum(check_data)
+
+            frame = (
+                struct.pack(">H", FRAME_CONST["HEAD"])
+                + struct.pack(">H", FRAME_CONST["SIM_ID"])
+                + struct.pack(">H", 0x0000)
+                + struct.pack(">H", FRAME_CONST["DATA_LEN"])
+                + check_data
+                + struct.pack(">H", checksum)
+                + struct.pack(">H", FRAME_CONST["TAIL"])
+            )
+
+            hex_dump(frame, "发送D软件完整帧")
+            return frame
+
+        except Exception as e:
+            logger.error(f"B→D转换失败：{e}", exc_info=True)
+            return None
+
+    @staticmethod
+    def extract_bits(data: bytes, start_bit: int, bit_len: int) -> bytes:
+        """
+        大端序、比特紧密对齐
+        从data的 start_bit 位置，提取 bit_len 长度比特 → 返回bytes
+        """
+        # 把所有字节转成连续的01比特字符串
+        bitstream = ""
+        for byte in data:
+            bitstream += f"{byte:08b}"
+
+        # 截取需要的部分
+        selected_bits = bitstream[start_bit : start_bit + bit_len]
+
+        # 不足补0
+        if len(selected_bits) < bit_len:
+            selected_bits = selected_bits.ljust(bit_len, "0")
+
+        # 转回字节
+        result = b""
+        for i in range(0, len(selected_bits), 8):
+            segment = selected_bits[i:i+8].ljust(8, "0")
+            result += int(segment, 2).to_bytes(1, byteorder="big")
+
+        return result