seismo-relay/tests/test_histogram_codec.py

"""
test_histogram_codec.py — regression locks for the histogram body codec.

The codec is verified byte-exact against BW's ASCII export across the
in-repo histogram fixture bundle.  Each test cross-checks decoded
binary fields against the corresponding .TXT row.

Run:
    python -m pytest tests/test_histogram_codec.py -q
"""

from __future__ import annotations

import os
import re
import sys
from pathlib import Path

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from minimateplus.blastware_file import _WAVEFORM_HEADER_SIZE
from minimateplus.histogram_codec import (
    _BLOCK_SIZE,
    decode_histogram_body,
    decode_histogram_body_full,
    geo_count_to_ins,
    half_period_to_hz,
    walk_body,
)
from minimateplus.waveform_codec import mic_count_to_db


_FIXTURE_DIR = Path(__file__).resolve().parent.parent / "example-events" / "histogram"


def _extract_body(path: Path) -> bytes:
    """Locate the body of a BW event file — bytes between the STRT
    record and the 26-byte footer."""
    raw = path.read_bytes()
    body_start = _WAVEFORM_HEADER_SIZE + 21
    pos = body_start
    footer_pos = -1
    while True:
        pos = raw.find(b"\x0e\x08", pos)
        if pos < 0 or pos + 26 > len(raw):
            break
        yr = (raw[pos + 4] << 8) | raw[pos + 5]
        if 2015 <= yr <= 2050:
            footer_pos = pos
            break
        pos += 1
    if footer_pos < 0:
        footer_pos = len(raw) - 26
    return raw[body_start:footer_pos]


def _parse_txt_rows(path: Path) -> list[tuple[str, list]]:
    """Parse a histogram .TXT into ``[(time_str, [10 col values]), …]``.

    Special tokens:
      - ``">100"`` (the BW-display sentinel for freq > 100 Hz) → ``None``
      - non-numeric → ``None``
    """
    text = path.read_text()
    lines = text.splitlines()
    hdr = None
    for i, line in enumerate(lines):
        if re.match(r"^Tran\s+", line.strip()):
            hdr = i + 3      # skip 2-row header + units row
            break
    if hdr is None:
        return []
    rows: list[tuple[str, list]] = []
    for line in lines[hdr:]:
        parts = line.split("\t")
        if len(parts) != 11:
            continue
        vals: list = []
        for p in parts[1:]:
            s = p.strip()
            if s.startswith(">"):
                vals.append(None)        # ">100 Hz" sentinel
                continue
            try:
                vals.append(float(s))
            except ValueError:
                vals.append(None)
        rows.append((parts[0].strip(), vals))
    return rows


# ── Block-walker plumbing ────────────────────────────────────────────────────


@pytest.mark.parametrize("fixture", [
    "N844L20G.630H",
    "N844L21H.2R0H",
    "N844L6Z8.ZR0H",
    "N844L6XE.BH0H",
    "N844L23B.ND0H",
])
def test_walk_body_returns_records(fixture: str):
    """Walker yields at least one valid block per fixture."""
    path = _FIXTURE_DIR / fixture
    if not path.exists():
        pytest.skip(f"fixture missing: {path}")
    records = walk_body(_extract_body(path))
    assert len(records) > 100, f"expected hundreds of blocks, got {len(records)}"


def test_walk_body_record_count_matches_txt_intervals():
    """Block count should match the .TXT interval count (off-by-one
    at the tail is acceptable — last interval may be truncated at
    recording stop)."""
    bin_path = _FIXTURE_DIR / "N844L20G.630H"
    txt_path = _FIXTURE_DIR / "N844L20G_630H_ASCII.TXT"
    if not bin_path.exists() or not txt_path.exists():
        pytest.skip("fixture missing")
    records = walk_body(_extract_body(bin_path))
    txt_rows = _parse_txt_rows(txt_path)
    # Allow off-by-one (final block may have been mid-write at stop)
    assert abs(len(records) - len(txt_rows)) <= 1, (
        f"binary {len(records)} blocks vs TXT {len(txt_rows)} intervals"
    )


def test_walk_body_segment_id_increments_every_256_blocks():
    """Segment ID advances 0→1→2→… after every 256 blocks within
    one event."""
    path = _FIXTURE_DIR / "N844L20G.630H"
    if not path.exists():
        pytest.skip("fixture missing")
    records = walk_body(_extract_body(path))
    # Group by segment_id and verify counts make sense
    from collections import Counter
    seg_counts = Counter(r["segment_id"] for r in records)
    # First 3 segments should each have exactly 256 blocks (N844L20G has
    # 791 blocks → 256+256+256+23 → segments 0/1/2/3)
    assert seg_counts[0] == 256
    assert seg_counts[1] == 256
    assert seg_counts[2] == 256
    assert seg_counts[3] == len(records) - 3 * 256


# ── Field-by-field decode verification against .TXT ground truth ─────────────


@pytest.mark.parametrize("fixture", [
    "N844L20G.630H",
    "N844L6Z8.ZR0H",
    "N844L6XE.BH0H",
    "N844L23B.ND0H",
])
def test_decoded_geo_peaks_match_txt(fixture: str):
    """For every block, decoded Tran/Vert/Long peak (count × 0.005)
    matches the corresponding .TXT cell."""
    bin_path = _FIXTURE_DIR / fixture
    txt_path = _FIXTURE_DIR / (fixture.replace(".", "_") + "_ASCII.TXT")
    if not bin_path.exists() or not txt_path.exists():
        pytest.skip("fixture missing")
    records  = walk_body(_extract_body(bin_path))
    txt_rows = _parse_txt_rows(txt_path)
    n = min(len(records), len(txt_rows))
    assert n > 0
    for i in range(n):
        rec = records[i]
        _ts, txt = txt_rows[i]
        # TXT cols 0/2/4 are T/V/L peak in in/s
        for slot, key in (("T", "t_peak"), ("V", "v_peak"), ("L", "l_peak")):
            col = {"T": 0, "V": 2, "L": 4}[slot]
            decoded_ips = geo_count_to_ins(rec[key])
            expected = txt[col]
            assert abs(decoded_ips - expected) < 0.0005, (
                f"{fixture} block {i} {slot}_peak: "
                f"decoded={decoded_ips:.4f} vs txt={expected:.4f}"
            )


@pytest.mark.parametrize("fixture", [
    "N844L6Z8.ZR0H",
    "N844L6XE.BH0H",
])
def test_decoded_geo_freqs_match_txt(fixture: str):
    """Decoded half-period → Hz matches the .TXT freq column for blocks
    where the freq is in-range (not the `>100 Hz` sentinel)."""
    bin_path = _FIXTURE_DIR / fixture
    txt_path = _FIXTURE_DIR / (fixture.replace(".", "_") + "_ASCII.TXT")
    if not bin_path.exists() or not txt_path.exists():
        pytest.skip("fixture missing")
    records  = walk_body(_extract_body(bin_path))
    txt_rows = _parse_txt_rows(txt_path)
    n = min(len(records), len(txt_rows))
    for i in range(n):
        rec = records[i]
        _ts, txt = txt_rows[i]
        for slot, key, col in (("T", "t_halfp", 1), ("V", "v_halfp", 3), ("L", "l_halfp", 5)):
            decoded_hz = half_period_to_hz(rec[key])
            expected = txt[col]
            if expected is None:
                # TXT shows `>100 Hz` — codec should also yield None
                assert decoded_hz is None or decoded_hz > 100, (
                    f"{fixture} block {i} {slot}_freq: codec says "
                    f"{decoded_hz} but TXT says >100"
                )
                continue
            # TXT rounds; allow ±1 Hz
            assert decoded_hz is not None
            assert abs(decoded_hz - expected) < 1.0, (
                f"{fixture} block {i} {slot}_freq: "
                f"decoded={decoded_hz:.2f} Hz vs txt={expected:.2f} Hz"
            )


@pytest.mark.parametrize("fixture", [
    "N844L6XE.BH0H",
    "N844L23B.ND0H",
    "N844L6Z8.ZR0H",
])
def test_decoded_mic_db_matches_txt(fixture: str):
    """Decoded MicL peak count → dB(L) via mic_count_to_db matches
    the .TXT dB(L) column."""
    bin_path = _FIXTURE_DIR / fixture
    txt_path = _FIXTURE_DIR / (fixture.replace(".", "_") + "_ASCII.TXT")
    if not bin_path.exists() or not txt_path.exists():
        pytest.skip("fixture missing")
    records  = walk_body(_extract_body(bin_path))
    txt_rows = _parse_txt_rows(txt_path)
    n = min(len(records), len(txt_rows))
    for i in range(n):
        rec = records[i]
        _ts, txt = txt_rows[i]
        # TXT col 8 = MicL dB(L)
        decoded_db = mic_count_to_db(rec["m_peak"])
        expected   = txt[8]
        if expected is None:
            continue
        # BW rounds to 1 decimal place for display.  Tolerance 0.1 dB
        # absorbs both rounding modes (truncate vs round-half-even).
        assert abs(decoded_db - expected) < 0.1, (
            f"{fixture} block {i} M_dB: "
            f"decoded={decoded_db:.2f} dB vs txt={expected:.2f} dB"
        )


@pytest.mark.parametrize("fixture", [
    "N844L20G.630H",
    "N844L6Z8.ZR0H",
])
def test_decoded_mic_freq_matches_txt(fixture: str):
    """Decoded MicL half-period → freq matches the .TXT col 9 freq."""
    bin_path = _FIXTURE_DIR / fixture
    txt_path = _FIXTURE_DIR / (fixture.replace(".", "_") + "_ASCII.TXT")
    if not bin_path.exists() or not txt_path.exists():
        pytest.skip("fixture missing")
    records  = walk_body(_extract_body(bin_path))
    txt_rows = _parse_txt_rows(txt_path)
    n = min(len(records), len(txt_rows))
    for i in range(n):
        rec = records[i]
        _ts, txt = txt_rows[i]
        decoded_hz = half_period_to_hz(rec["m_halfp"])
        expected   = txt[9]
        if expected is None:
            assert decoded_hz is None or decoded_hz > 100
            continue
        assert decoded_hz is not None
        assert abs(decoded_hz - expected) < 1.0, (
            f"{fixture} block {i} M_freq: "
            f"decoded={decoded_hz:.2f} Hz vs txt={expected:.2f} Hz"
        )


# ── Public API ───────────────────────────────────────────────────────────────


def test_decode_histogram_body_returns_four_channels():
    """The public API returns the standard 4-channel dict shape."""
    path = _FIXTURE_DIR / "N844L20G.630H"
    if not path.exists():
        pytest.skip("fixture missing")
    decoded = decode_histogram_body(_extract_body(path))
    assert decoded is not None
    assert set(decoded.keys()) == {"Tran", "Vert", "Long", "MicL"}
    # All channels same length (one value per histogram interval)
    n = len(decoded["Tran"])
    assert all(len(decoded[ch]) == n for ch in ("Vert", "Long", "MicL"))
    assert n > 100


def test_decode_histogram_body_returns_none_for_non_histogram():
    """A waveform-mode body (starts with 00 02 00) doesn't decode as
    a histogram body."""
    fake_waveform_body = b"\x00\x02\x00" + b"\x00" * 100
    assert decode_histogram_body(fake_waveform_body) is None


def test_decode_histogram_body_returns_none_for_garbage():
    """Bytes that don't form valid blocks return None."""
    assert decode_histogram_body(b"\xff" * 256) is None


def test_decode_histogram_body_full_preserves_frequency_data():
    """The structured-record API preserves the per-channel half-period
    fields that the flat-channel API drops."""
    path = _FIXTURE_DIR / "N844L20G.630H"
    if not path.exists():
        pytest.skip("fixture missing")
    records = decode_histogram_body_full(_extract_body(path))
    assert records is not None
    r0 = records[0]
    expected_fields = {
        "segment_id", "block_ctr",
        "t_peak", "t_halfp", "v_peak", "v_halfp",
        "l_peak", "l_halfp", "m_peak", "m_halfp",
        "meta_var",
    }
    assert set(r0.keys()) >= expected_fields


# ── Helpers ──────────────────────────────────────────────────────────────────


def test_half_period_to_hz_sentinel():
    """Half-period ≤ 5 returns None (the `>100 Hz` sentinel)."""
    assert half_period_to_hz(5) is None
    assert half_period_to_hz(1) is None
    # halfp=6 gives 512/6 = 85.3 Hz — below the >100 threshold
    assert half_period_to_hz(6) == pytest.approx(85.33, abs=0.01)


def test_geo_count_to_ins_scale():
    """1 count = 0.005 in/s at Normal range."""
    assert geo_count_to_ins(1)  == pytest.approx(0.005)
    assert geo_count_to_ins(10) == pytest.approx(0.050)
    assert geo_count_to_ins(0)  == 0.0


# ── Regression: peak is uint8 byte[N], NOT uint16 LE byte[N:N+2] ────────────
#
# Block taken verbatim from K558LKZU.RE0H (BE9558) interval 12 — a real
# field event where the Tran channel had developed a DC offset and was
# producing sub-Hz drift content the device couldn't characterize.
# The annotation byte at [7] = 0xd2 is non-zero in that case.  The
# legacy codec read [6:8] as uint16 LE, producing T_peak = 53763 →
# 268 in/s — physically impossible and 35× too high for the actual
# 0.015 in/s value (T_lo = 3 alone gives the correct count).
# Verified against the paired BW ASCII export.
_K558_INTERVAL_12_BLOCK = bytes.fromhex(
    "00 00 0c 01 0a 00 03 d2 45 00 02 00 02 00 02 00"
    "02 00 10 00 06 00 00 00 0e 91 2f 00 1e 0a 00 00".replace(" ", "")
)


def test_extension_byte_does_not_inflate_peak():
    """The byte after each peak must NOT contribute to the peak count.

    Still true, but for a different reason than originally recorded: the
    block is uniformly **big-endian**, so T_peak is uint16 BE at [5:7]
    (= 3 here) and the 0xd2 at [7] is the HIGH BYTE of the big-endian
    half-period at [7:9], not an "annotation" field.  Reading the peak as
    uint16 LE at [6:8] gave 53763 → 268 in/s, which is what this test was
    written to prevent.
    """
    body = _K558_INTERVAL_12_BLOCK
    records = decode_histogram_body_full(body)
    assert records is not None
    assert len(records) == 1
    r = records[0]
    assert r["t_peak"] == 3,    f"T_peak should be 3, got {r['t_peak']}"
    assert r["v_peak"] == 2
    assert r["l_peak"] == 2
    assert r["m_peak"] == 16
    # Half-period is uint16 BE — 0xd245 = 53829 samples → 0.0095 Hz, which
    # is exactly the sub-Hz drift BW rendered as "<1.0" for this interval.
    assert r["t_halfp"] == 0xd245
    assert half_period_to_hz(r["t_halfp"]) < 1.0
    assert r["m_halfp"] == 6       # → 85.3 Hz


def test_extension_byte_decoded_to_correct_in_s():
    """End-to-end: the channel-grouped output for the K558 ext block
    should give T = 3 counts = 0.015 in/s, not 53763 counts = 268 in/s."""
    channels = decode_histogram_body(_K558_INTERVAL_12_BLOCK)
    assert channels is not None
    assert channels["Tran"] == [3]
    assert geo_count_to_ins(channels["Tran"][0]) == pytest.approx(0.015)
    assert channels["Vert"] == [2]
    assert channels["Long"] == [2]
    assert channels["MicL"] == [16]


# ── Big-endian block layout + terminal block (2026-08-25) ───────────────────
#
# Verified against 1211 production histograms paired with their Blastware
# ASCII exports: 1211/1211 decode exactly (interval count + every
# per-interval peak), and 842,442 per-interval frequency comparisons match
# with zero mismatches.

def _mk_block(t_peak=3, t_halfp=69, v_peak=2, v_halfp=69, l_peak=2, l_halfp=69,
              m_peak=16, m_halfp=6, seg=0, ctr=256, tail=b"\x1e\x0a\x00\x00",
              b22=0x00):
    """Build one synthetic 32-byte histogram block, big-endian throughout."""
    b = bytearray(32)
    b[0] = 0x00
    b[1] = seg
    b[2], b[3] = ctr & 0xFF, (ctr >> 8) & 0xFF        # block_ctr is uint16 LE
    b[4] = 0x0A                                        # marker, uint8
    for off, val in ((5, t_peak), (7, t_halfp), (9, v_peak), (11, v_halfp),
                     (13, l_peak), (15, l_halfp), (17, m_peak), (19, m_halfp)):
        b[off], b[off + 1] = (val >> 8) & 0xFF, val & 0xFF   # uint16 BE
    b[22] = b22
    b[28:32] = tail
    return bytes(b)


def test_geo_peak_is_uint16_be_not_uint8():
    """A peak above the uint8 ceiling (255 counts = 1.275 in/s) must decode.

    Real example: BE18193/T193LQ9K.OE0H's final interval reads 8.270 in/s
    in the BW export = 1654 counts = 0x0676, which needs both bytes.
    Reading only byte[6] gave 0x76 = 118 = 0.590 in/s.
    """
    r = decode_histogram_body_full(_mk_block(t_peak=1654))
    assert r is not None and len(r) == 1
    assert r[0]["t_peak"] == 1654
    assert geo_count_to_ins(r[0]["t_peak"]) == pytest.approx(8.27)


def test_half_period_is_uint16_be():
    """Half-period spans two bytes big-endian; a large value is sub-Hz."""
    r = decode_histogram_body_full(_mk_block(t_halfp=53829))
    assert r[0]["t_halfp"] == 53829
    assert half_period_to_hz(53829) == pytest.approx(512 / 53829)


def test_terminal_block_tail_is_accepted():
    """The LAST block of a histogram stream carries tail `9c 06 00 42`
    instead of `1e 0a 00 00`.  Rejecting it dropped the final interval —
    which is where the event peak often lives.  Observed in 1206 of 1211
    production histograms, always after every standard-tail block."""
    body = _mk_block(ctr=256) + _mk_block(ctr=257, t_peak=1654,
                                          tail=b"\x9c\x06\x00\x42")
    ch = decode_histogram_body(body)
    assert ch is not None
    assert ch["Tran"] == [3, 1654], "terminal block must not be dropped"


def test_terminal_block_exempt_from_byte22_constraint():
    """Terminal blocks carry arbitrary bytes at [22:24]; only standard
    blocks hold 0x00 there.  Requiring it dropped the final interval on
    files such as BE18438/T438LO30.DC0H."""
    body = _mk_block(ctr=256) + _mk_block(ctr=257, tail=b"\x9c\x06\x00\x42",
                                          b22=0x01)
    ch = decode_histogram_body(body)
    assert ch is not None and len(ch["Tran"]) == 2


def test_standard_block_accepts_nonzero_byte22():
    """block[22] is NOT a constant and must not be tested.

    It was documented as always 0x00, but it carries data on loud blocks.
    Rejecting those threw away the interval holding the event peak:
    BE18350/T350L7HR.NL0H block 92 has block[22]=0x26 and a Tran peak of
    0x0563 = 1379 counts = 6.895 in/s — exactly the device-reported PPV —
    while the file as a whole decoded to 0.015 in/s.

    block[0]==0x00, block[4]==0x0A and the 4-byte tail are six bytes of
    constraint, which is what keeps trailer content out.
    """
    ch = decode_histogram_body(_mk_block(t_peak=1379, b22=0x26))
    assert ch is not None
    assert ch["Tran"] == [1379]
    assert geo_count_to_ins(ch["Tran"][0]) == pytest.approx(6.895)


def test_marker_is_single_byte_not_uint16():
    """block[4] alone is the 0x0A marker.  Treating [4:6] as a uint16 LE
    marker forced block[5] to zero, which capped every geo peak at 255
    counts — block[5] is the peak's high byte."""
    r = decode_histogram_body_full(_mk_block(t_peak=0x0676))
    assert r is not None, "block[5] != 0 must not disqualify the block"
    assert r[0]["t_peak"] == 0x0676


# ── Multi-interval block variant (2026-08-26) ───────────────────────────────
#
# When the histogram interval is SHORTER than one minute the device packs
# several intervals into one block, so that every block still covers exactly
# one minute of data:
#
#     interval  intervals/block  stride
#     1 min     1                32   (the standard big-endian block)
#     15 s      4                92
#     2 s       30               612
#
#     stride = 12 + n_intervals * 20
#
# Block layout:
#     [0]      0x00
#     [1]      segment_id (256 blocks per segment)
#     [2:4]    block_ctr  uint16 LE
#     [4]      0x0a marker
#     [5]      0x00
#     [6 ...]  n x 20-byte interval records, each 8 x uint16 LITTLE-endian:
#                  T_peak, T_halfp, V_peak, V_halfp,
#                  L_peak, L_halfp, M_peak, M_halfp
#              followed by 2 words (first is 0x0000)
#     [-6:]    6-byte block trailer
#
# NOTE the endianness flip: the standard 32-byte block is big-endian, this
# variant is little-endian.
#
# Ground truth: BE9440/K440L3AQ.T70H (15 s intervals, 5,710 of them) decodes
# with 17,130/17,130 geo peak counts, 22,840/22,840 frequencies and
# 5,710/5,710 mic dB(L) values matching its Blastware ASCII export exactly.

from minimateplus.histogram_codec import (        # noqa: E402
    detect_multi_interval_stride,
    walk_multi_interval_blocks,
)


def _mk_multi_block(intervals, seg=0, ctr=256):
    """Build one multi-interval block from a list of 8-tuples."""
    b = bytearray()
    b += bytes([0x00, seg])
    b += int(ctr).to_bytes(2, "little")
    b += bytes([0x0A, 0x00])
    for iv in intervals:
        for v in iv:
            b += int(v).to_bytes(2, "little")
        b += (0).to_bytes(2, "little")
        b += (5).to_bytes(2, "little")
    b += bytes(6)
    assert len(b) == 12 + 20 * len(intervals)
    return bytes(b)


def test_stride_is_twelve_plus_twenty_per_interval():
    for n in (4, 30):
        ivs = [(1, 1, 2, 2, 3, 3, 4, 4)] * n
        body = _mk_multi_block(ivs, ctr=256) + _mk_multi_block(ivs, ctr=257)
        assert detect_multi_interval_stride(body) == 12 + 20 * n


def test_multi_interval_block_decodes_all_four_channels():
    ivs = [(1, 1, 2, 2, 3, 3, 4, 4), (5, 6, 7, 8, 9, 10, 11, 12)]
    body = _mk_multi_block(ivs) + _mk_multi_block(ivs, ctr=257)
    recs = walk_multi_interval_blocks(body)
    assert len(recs) == 4
    r = recs[0]
    assert (r["t_peak"], r["v_peak"], r["l_peak"], r["m_peak"]) == (1, 2, 3, 4)
    assert (r["t_halfp"], r["v_halfp"], r["l_halfp"], r["m_halfp"]) == (1, 2, 3, 4)
    assert recs[1]["t_peak"] == 5 and recs[1]["m_halfp"] == 12


def test_multi_interval_values_are_little_endian():
    """The standard 32-byte block is big-endian; this variant is not.

    A peak of 0x0100 must decode as 256, not 1.
    """
    ivs = [(0x0100, 1, 1, 1, 1, 1, 1, 1), (1, 1, 1, 1, 1, 1, 1, 1)]
    body = _mk_multi_block(ivs) + _mk_multi_block(ivs, ctr=257)
    assert walk_multi_interval_blocks(body)[0]["t_peak"] == 0x0100


def test_decode_histogram_body_falls_back_to_the_variant():
    ivs = [(1, 1, 2, 2, 3, 3, 4, 4)] * 4
    body = _mk_multi_block(ivs) + _mk_multi_block(ivs, ctr=257)
    ch = decode_histogram_body(body)
    assert ch is not None
    assert len(ch["Tran"]) == 8
    assert ch["Long"][0] == 3


def test_standard_blocks_still_take_precedence():
    """A body of standard 32-byte blocks must not be re-read as the variant."""
    std = _mk_block(t_peak=7) + _mk_block(t_peak=9, ctr=257,
                                          tail=b"\x9c\x06\x00\x42")
    ch = decode_histogram_body(std)
    assert ch is not None and ch["Tran"] == [7, 9]


# ── Ground truth for the multi-interval variant ─────────────────────────────
# Fixture is gitignored (like the rest of tests/fixtures); skips when absent.

_MULTI_FIXTURE = os.path.join(
    os.path.dirname(__file__), "fixtures", "histogram-multi-interval",
    "K440L3AQ.T70H",
)


@pytest.mark.skipif(not os.path.exists(_MULTI_FIXTURE),
                    reason="multi-interval fixture not present")
def test_multi_interval_matches_blastware_ascii_exactly():
    """BE9440/K440L3AQ.T70H — 5,710 intervals at 15 s, 4 per 92-byte block.

    Every geo peak, every frequency and every mic dB(L) in the file matches
    the Blastware ASCII export: 17,130 / 22,840 / 5,710 values, zero
    mismatches.  Before this decoder the file produced nothing at all.
    """
    import math
    import re as _re
    from minimateplus import blastware_file as _bwf

    raw = open(_MULTI_FIXTURE, "rb").read()
    bs = _bwf._WAVEFORM_HEADER_SIZE + 21
    pos, fp = bs, -1
    while True:
        pos = raw.find(b"\x0e\x08", pos)
        if pos < 0 or pos + 26 > len(raw):
            break
        if 2015 <= ((raw[pos + 4] << 8) | raw[pos + 5]) <= 2050:
            fp = pos
            break
        pos += 1
    recs = walk_multi_interval_blocks(raw[bs:fp])

    rows = []
    for line in open(_MULTI_FIXTURE + "_ASCII.TXT", errors="replace"):
        p = [x.strip() for x in line.split("\t")]
        if len(p) >= 11 and _re.match(r"^\d{2}:\d{2}:\d{2}$", p[0]):
            rows.append(p)

    assert len(recs) == len(rows) == 5710

    def want_count(x):
        return round(float(x) / 0.005)

    for rec, row in zip(recs, rows):
        assert rec["t_peak"] == want_count(row[1])
        assert rec["v_peak"] == want_count(row[3])
        assert rec["l_peak"] == want_count(row[5])
        # mic dB(L)
        assert abs((81.94 + 20 * math.log10(rec["m_peak"])) - float(row[9])) <= 0.06
        # frequency: half-period <= 5 is BW's ">100 Hz" sentinel
        for hp, cell in ((rec["t_halfp"], row[2]), (rec["v_halfp"], row[4]),
                         (rec["l_halfp"], row[6]), (rec["m_halfp"], row[10])):
            hz = None if hp <= 5 else 512.0 / hp
            if cell.startswith(">"):
                assert hz is None
            elif not cell.startswith("<"):
                assert hz is not None and abs(hz - float(cell)) <= max(0.55, float(cell) * 0.02)