181 lines
6.4 KiB
Python
181 lines
6.4 KiB
Python
|
# 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
|