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 |