diff --git a/minimateplus/binary_annotate.py b/minimateplus/binary_annotate.py new file mode 100644 index 0000000..25e164d --- /dev/null +++ b/minimateplus/binary_annotate.py @@ -0,0 +1,75 @@ +"""Structural annotation of a Series-3 Blastware waveform binary. + +Pure, no I/O: takes the raw file bytes and returns a flat, gap-free tiling of +labelled :class:`Span` regions for a hex viewer to paint. Every byte is +covered — anything the decoder can't account for becomes an ``unknown`` span, +so undecoded regions (e.g. a stored spectral/FFT block, if one exists) stand +out instead of hiding. + +File layout (see ``blastware_file.py``): ``[header][21B STRT][body][26B footer]``. +The body is the record chain walked by :func:`waveform_codec.walk_records`. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import List + +from .waveform_codec import walk_records + +_STRT_LEN = 21 +_FOOTER_LEN = 26 + + +@dataclass +class Span: + start: int # inclusive byte offset + end: int # exclusive byte offset + label: str # human-readable description + kind: str # 'header' | 'strt' | 'sample' | 'footer' | 'unknown' + + +def _tile(known: List[Span], total: int) -> List[Span]: + """Sort *known* spans and fill every gap with an ``unknown`` span, so the + result is a contiguous, non-overlapping tiling of ``[0, total)``. Overlaps + are resolved by clamping to the running position (first writer wins).""" + out: List[Span] = [] + pos = 0 + for s in sorted(known, key=lambda x: (x.start, x.end)): + if s.end <= pos: + continue # fully behind — dropped overlap + start = max(s.start, pos) + if start > pos: + out.append(Span(pos, start, "unknown", "unknown")) + out.append(s if start == s.start else Span(start, s.end, s.label, s.kind)) + pos = s.end + if pos < total: + out.append(Span(pos, total, "unknown", "unknown")) + return out + + +def annotate_blastware_binary(raw: bytes) -> List[Span]: + """Annotate a Series-3 waveform binary into a gap-free list of spans.""" + total = len(raw) + strt_pos = raw.find(b"STRT") + if strt_pos < 0: + return [Span(0, total, "unrecognized — no STRT record", "unknown")] + + known: List[Span] = [] + if strt_pos > 0: + known.append(Span(0, strt_pos, "File header", "header")) + known.append(Span(strt_pos, strt_pos + _STRT_LEN, "STRT record", "strt")) + + body_start = strt_pos + _STRT_LEN + footer_start = total - _FOOTER_LEN + if footer_start >= body_start: + known.append(Span(footer_start, total, "File footer", "footer")) + else: + footer_start = total # file too short for a footer + + body = raw[body_start:footer_start] + for rec in walk_records(body): + hi, lo = rec["mode"] + label = f"{rec['channel']} record (seg {rec['segment_index']}, mode {hi:02x} {lo:02x})" + known.append(Span(body_start + rec["offset"], body_start + rec["end"], label, "sample")) + + return _tile(known, total) diff --git a/tests/test_binary_annotate.py b/tests/test_binary_annotate.py new file mode 100644 index 0000000..3748041 --- /dev/null +++ b/tests/test_binary_annotate.py @@ -0,0 +1,50 @@ +"""Structural annotation of a Series-3 Blastware binary (for the seismo_lab +Binary Inspector). The annotator maps byte ranges to labelled spans; anything +the decoder can't account for is a first-class ``unknown`` span, so the whole +file is tiled and the gaps (candidate FFT/spectral data) are visible. +""" +from pathlib import Path + +from minimateplus.binary_annotate import annotate_blastware_binary, Span + +# A known-good full-3-channel Series-3 waveform binary (the V70 cracking fixture). +FIXTURE = Path(__file__).parent / "fixtures" / "5-11-26" / "M529LL1L.V70" + + +def _raw() -> bytes: + return FIXTURE.read_bytes() + + +def test_spans_tile_the_whole_file(): + raw = _raw() + spans = annotate_blastware_binary(raw) + assert spans, "expected at least one span" + assert spans[0].start == 0 + assert spans[-1].end == len(raw) + for a, b in zip(spans, spans[1:]): + assert a.end == b.start, f"gap/overlap between {a!r} and {b!r}" + for s in spans: + assert s.start < s.end, f"empty/negative span {s!r}" + + +def test_strt_record_is_located(): + raw = _raw() + spans = annotate_blastware_binary(raw) + strt = [s for s in spans if s.kind == "strt"] + assert strt, "expected a STRT region" + assert raw[strt[0].start : strt[0].start + 4] == b"STRT" + + +def test_geo_sample_records_annotated(): + raw = _raw() + spans = annotate_blastware_binary(raw) + chans = {s.label.split()[0] for s in spans if s.kind == "sample"} + # V70 is a full three-geo-channel event. + assert {"Tran", "Vert", "Long"} <= chans, f"expected geo records, got {chans}" + + +def test_footer_is_last(): + raw = _raw() + spans = annotate_blastware_binary(raw) + assert spans[-1].kind == "footer" + assert spans[-1].end - spans[-1].start == 26