88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
|
import struct
|
||
|
|
|
||
|
|
def read_u16_le(data: bytes, offset: int) -> int:
|
||
|
|
if offset + 2 > len(data):
|
||
|
|
return None
|
||
|
|
return struct.unpack_from('<H', data, offset)[0]
|
||
|
|
|
||
|
|
def read_f32_le(data: bytes, offset: int) -> float:
|
||
|
|
if offset + 4 > len(data):
|
||
|
|
return None
|
||
|
|
return struct.unpack_from('<f', data, offset)[0]
|
||
|
|
|
||
|
|
with open("/home/prathiyuman/Prathiyuman/ContecMonitor/fiveparaminte-main/scratch/raw_contec_stream.bin", "rb") as f:
|
||
|
|
stream = f.read()
|
||
|
|
|
||
|
|
print(f"Total stream size: {len(stream)} bytes")
|
||
|
|
|
||
|
|
# Extract packets
|
||
|
|
packets = []
|
||
|
|
consumed = 0
|
||
|
|
while consumed < len(stream) - 4:
|
||
|
|
b0 = stream[consumed]
|
||
|
|
b1 = stream[consumed + 1]
|
||
|
|
b2 = stream[consumed + 2]
|
||
|
|
b3 = stream[consumed + 3]
|
||
|
|
|
||
|
|
if (b2 == 0x04 or b2 == 0x01) and b3 == 0x46:
|
||
|
|
pkt_len = b0 | (b1 << 8)
|
||
|
|
if pkt_len < 5 or pkt_len > 8192:
|
||
|
|
consumed += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
if consumed + pkt_len <= len(stream):
|
||
|
|
packets.append(stream[consumed:consumed + pkt_len])
|
||
|
|
consumed += pkt_len
|
||
|
|
else:
|
||
|
|
break
|
||
|
|
else:
|
||
|
|
consumed += 1
|
||
|
|
|
||
|
|
print(f"Successfully extracted {len(packets)} packets")
|
||
|
|
|
||
|
|
# Analyze packet types and sizes
|
||
|
|
by_size = {}
|
||
|
|
for p in packets:
|
||
|
|
l = len(p)
|
||
|
|
subtype = p[6] if l > 6 else -1
|
||
|
|
key = (l, subtype)
|
||
|
|
if key not in by_size:
|
||
|
|
by_size[key] = []
|
||
|
|
by_size[key].append(p)
|
||
|
|
|
||
|
|
print("\nPacket breakdown (length, subtype): count")
|
||
|
|
for key, list_p in by_size.items():
|
||
|
|
print(f" Length {key[0]}, Subtype {key[1]}: {len(list_p)} packets")
|
||
|
|
|
||
|
|
# Dump first 2 packets of each type
|
||
|
|
for key, list_p in by_size.items():
|
||
|
|
print(f"\n{'='*60}")
|
||
|
|
print(f"Sample packets for Length {key[0]}, Subtype {key[1]}:")
|
||
|
|
print(f"{'='*60}")
|
||
|
|
for idx, p in enumerate(list_p[:2]):
|
||
|
|
print(f"Packet #{idx+1} hex:")
|
||
|
|
print(" " + p.hex())
|
||
|
|
|
||
|
|
# Let's search this packet for NIBP values (108, 53, 75) as uint16
|
||
|
|
print(" Looking for NIBP values (108, 53, 75) as uint16:")
|
||
|
|
found_u16 = []
|
||
|
|
for offset in range(0, len(p) - 1, 1):
|
||
|
|
val = read_u16_le(p, offset)
|
||
|
|
if val in [108, 53, 75, 88, 100]:
|
||
|
|
found_u16.append(f"offset {offset}: {val}")
|
||
|
|
if found_u16:
|
||
|
|
print(" uint16 matches: " + ", ".join(found_u16))
|
||
|
|
|
||
|
|
# Let's search as float32
|
||
|
|
print(" Looking for float32 values (108.0, 53.0, 75.0, 88.0, 100.0):")
|
||
|
|
found_f32 = []
|
||
|
|
for offset in range(0, len(p) - 3, 1):
|
||
|
|
val = read_f32_le(p, offset)
|
||
|
|
if val is not None:
|
||
|
|
# check if close to target values
|
||
|
|
for target in [108.0, 53.0, 75.0, 88.0, 100.0]:
|
||
|
|
if abs(val - target) < 0.1:
|
||
|
|
found_f32.append(f"offset {offset}: {val:.1f}")
|
||
|
|
if found_f32:
|
||
|
|
print(" float32 matches: " + ", ".join(found_f32))
|