138 lines
5.0 KiB
Python
138 lines
5.0 KiB
Python
"""Decode IDFH histogram intervals + verify against sidecar."""
|
||
from __future__ import annotations
|
||
import sys
|
||
import struct
|
||
from pathlib import Path
|
||
|
||
REPO = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(REPO))
|
||
|
||
|
||
SEGMENT_MAGIC = b"\x02\xda\x0a\x00\x00\x00"
|
||
SEGMENT_SIZE = 732 # = 10-byte header + 10 × 72-byte intervals + 2-byte tail
|
||
INTERVAL_SIZE = 72
|
||
CHANNELS = ("Tran", "Vert", "Long", "MicL")
|
||
|
||
|
||
def decode_interval(buf72: bytes) -> dict:
|
||
"""Decode one 72-byte interval into per-channel min/max/halfp."""
|
||
out = {}
|
||
for i, ch in enumerate(CHANNELS):
|
||
block = buf72[i*16 : (i+1)*16]
|
||
mn = struct.unpack_from(">h", block, 0)[0]
|
||
mx = struct.unpack_from(">h", block, 2)[0]
|
||
sb = struct.unpack_from(">h", block, 4)[0]
|
||
halfp = struct.unpack_from(">H", block, 6)[0]
|
||
f10 = struct.unpack_from(">H", block, 10)[0]
|
||
f14 = struct.unpack_from(">H", block, 14)[0]
|
||
peak_count = max(abs(mn), abs(mx))
|
||
out[ch] = {
|
||
"min": mn,
|
||
"max": mx,
|
||
"field4": sb,
|
||
"halfp": halfp,
|
||
"field10": f10,
|
||
"field14": f14,
|
||
"peak": peak_count,
|
||
"freq_hz": (512.0 / halfp) if halfp > 5 else None,
|
||
}
|
||
out["_tail"] = buf72[64:].hex(" ")
|
||
return out
|
||
|
||
|
||
def walk_idfh(buf: bytes) -> list:
|
||
"""Walk all interval records in an IDFH file."""
|
||
intervals = []
|
||
# Multi-segment file: every 02 da 0a 00 00 00 marker introduces a segment.
|
||
# Single-interval file: just one body header at 0xf96 of form ?? ?? 0a 00 00 00.
|
||
# Find them all.
|
||
i = 0
|
||
while True:
|
||
j = buf.find(b"\x0a\x00\x00\x00", i)
|
||
if j < 0:
|
||
break
|
||
# Validate: the 2 bytes before must form a length, and we want bytes
|
||
# [j-2 : j+6] to have a recognisable shape. Actually the cleanest
|
||
# filter is "preceded by a length and followed by 00 NN 05 3f".
|
||
if j < 2:
|
||
i = j + 1
|
||
continue
|
||
# Body header form: [length_be_2][0a 00 00 00][00 NN][05 3f]
|
||
if j + 10 > len(buf):
|
||
break
|
||
length = int.from_bytes(buf[j-2:j], "big")
|
||
# Verify the segment-marker shape: [length_be][0a 00 00 00][00 NN][05 3f]
|
||
if buf[j+4] != 0x00:
|
||
i = j + 1
|
||
continue
|
||
if buf[j+6:j+8] != b"\x05\x3f":
|
||
i = j + 1
|
||
continue
|
||
# Header layout (10 bytes): [length_be 2B][0a 00 00 00 4B][00 NN 2B][05 3f 2B]
|
||
# Followed by N interval records of 72 bytes each, then 2 tail bytes.
|
||
# length value = (N × 72) + 10 (counts bytes from 0x0a... through interval data).
|
||
header_start = j - 2
|
||
n_intervals = (length - 10) // INTERVAL_SIZE
|
||
interval_start = header_start + 10
|
||
for k in range(n_intervals):
|
||
off = interval_start + k * INTERVAL_SIZE
|
||
if off + INTERVAL_SIZE > len(buf):
|
||
break
|
||
chunk = buf[off:off + INTERVAL_SIZE]
|
||
intervals.append({"offset": off, **decode_interval(chunk)})
|
||
i = header_start + length + 2
|
||
return intervals
|
||
|
||
|
||
def main():
|
||
# Test against multi-segment IDFH
|
||
target = REPO / "tests/fixtures/THORDATA_example/THORDATA_example/UPMC Presby/UM13981/UM13981_20220805075441.IDFH"
|
||
sc_path = target.parent / "TXT" / f"{target.name}.txt"
|
||
buf = target.read_bytes()
|
||
intervals = walk_idfh(buf)
|
||
print(f"=== {target.name} ===")
|
||
print(f" file size: {len(buf)}")
|
||
print(f" decoded intervals: {len(intervals)}")
|
||
# Show first 2 + last 2
|
||
sc_rows = []
|
||
for line in sc_path.read_text(errors="replace").splitlines():
|
||
if line.startswith("2022-") or line.startswith("2023-"):
|
||
sc_rows.append(line)
|
||
print(f" sidecar rows: {len(sc_rows)}")
|
||
|
||
print()
|
||
for k in [0, 1, 78, 79, 80]:
|
||
if k >= len(intervals):
|
||
continue
|
||
iv = intervals[k]
|
||
print(f"--- interval {k} @0x{iv['offset']:04x} ---")
|
||
for ch in CHANNELS:
|
||
d = iv[ch]
|
||
peak_ips = d["peak"] / 32768 * 10.0
|
||
print(f" {ch}: peak={d['peak']:5d} ({peak_ips:.4f} in/s) halfp={d['halfp']:5d} freq={d['freq_hz']}")
|
||
# sidecar row
|
||
if k < len(sc_rows):
|
||
print(f" SC: {sc_rows[k]}")
|
||
|
||
# Test single-interval IDFH
|
||
print()
|
||
target2 = REPO / "tests/fixtures/THORDATA_example/THORDATA_example/UPMC Presby/UM11719/UM11719_20231219162648.IDFH"
|
||
sc2 = target2.parent / "TXT" / f"{target2.name}.txt"
|
||
buf2 = target2.read_bytes()
|
||
intervals2 = walk_idfh(buf2)
|
||
print(f"=== {target2.name} ===")
|
||
print(f" file size: {len(buf2)}, decoded intervals: {len(intervals2)}")
|
||
if intervals2:
|
||
iv = intervals2[0]
|
||
for ch in CHANNELS:
|
||
d = iv[ch]
|
||
peak_ips = d["peak"] / 32768 * 10.0
|
||
print(f" {ch}: peak={d['peak']:5d} ({peak_ips:.4f} in/s) halfp={d['halfp']:5d} freq={d['freq_hz']}")
|
||
sc_rows2 = [l for l in sc2.read_text(errors='replace').splitlines() if l.startswith("2023-")]
|
||
if sc_rows2:
|
||
print(f" SC: {sc_rows2[0]}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|