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)

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