seismo-relay/tests/test_bw_ascii_report.py

"""
test_bw_ascii_report.py — parser for Blastware's per-event ASCII export.

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

from __future__ import annotations

import datetime
import os
import sys
from pathlib import Path

import pytest

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

from minimateplus.bw_ascii_report import (
    BwAsciiReport,
    parse_report,
    parse_report_file,
)


FIXTURES = Path(__file__).parent.parent / "decode-re" / "5-8-26"


def _fixture(event_name: str) -> Path:
    """Find the .TXT file inside a fixture event folder."""
    for p in (FIXTURES / event_name).iterdir():
        if p.suffix.lower() == ".txt":
            return p
    raise FileNotFoundError(f"no .TXT in {FIXTURES / event_name}")


# ── Identity / config ───────────────────────────────────────────────────────


def test_event_c_identity_and_config():
    r = parse_report_file(_fixture("event-c"))

    assert r.event_type == "Full Waveform"
    assert r.serial == "BE11529"
    assert r.file_name == "M529LK44.AB0"
    assert r.event_datetime == datetime.datetime(2026, 4, 23, 15, 56, 35)

    assert r.trigger_channel == "Vert"
    assert r.geo_trigger_level_ips == pytest.approx(0.5)
    assert r.pretrig_s == pytest.approx(-0.25)
    assert r.record_time_s == pytest.approx(1.0)
    assert r.record_stop_mode == "Fixed"
    assert r.sample_rate_sps == 1024
    assert r.battery_volts == pytest.approx(6.8)
    assert r.calibration_date == datetime.date(2025, 4, 29)
    assert r.calibration_by == "Instantel"
    assert r.units == "in/s and dB(L)"


def test_event_c_operator_metadata():
    r = parse_report_file(_fixture("event-c"))

    # The "Project: : value" pattern (key has its own trailing colon)
    # is handled by stripping the colon at lookup time.
    assert r.project == "Test4-21-26"
    assert r.client == "Test-Client1"
    assert r.operator == "Brian and claude"
    assert r.sensor_location == "catbed"


def test_event_c_geo_range():
    r = parse_report_file(_fixture("event-c"))
    assert r.geo_range_ips == pytest.approx(10.0)


# ── Per-channel derived stats ───────────────────────────────────────────────


def test_event_c_per_channel_stats():
    r = parse_report_file(_fixture("event-c"))

    tran = r.channels["Tran"]
    assert tran.ppv_ips         == pytest.approx(0.065)
    assert tran.zc_freq_hz      == pytest.approx(47.0)
    assert tran.time_of_peak_s  == pytest.approx(0.007)
    assert tran.peak_accel_g    == pytest.approx(0.066)
    assert tran.peak_disp_in    == pytest.approx(0.001)

    vert = r.channels["Vert"]
    assert vert.ppv_ips         == pytest.approx(0.610)
    assert vert.zc_freq_hz      == pytest.approx(16.0)
    assert vert.time_of_peak_s  == pytest.approx(0.024)
    assert vert.peak_accel_g    == pytest.approx(0.437)
    assert vert.peak_disp_in    == pytest.approx(0.006)

    long_ = r.channels["Long"]
    assert long_.ppv_ips        == pytest.approx(0.070)
    assert long_.zc_freq_hz     == pytest.approx(22.0)
    assert long_.time_of_peak_s == pytest.approx(0.019)
    assert long_.peak_accel_g   == pytest.approx(0.040)
    assert long_.peak_disp_in   == pytest.approx(0.001)


def test_event_c_micl_stats():
    r = parse_report_file(_fixture("event-c"))

    # MicL specific block
    assert r.mic.weighting       == "Linear Weighting"
    assert r.mic.pspl_dbl        == pytest.approx(88.0)
    assert r.mic.zc_freq_hz      == pytest.approx(57.0)
    assert r.mic.time_of_peak_s  == pytest.approx(-0.004)

    # Mirrored onto channels["MicL"] for uniform per-channel access
    micl_ch = r.channels["MicL"]
    assert micl_ch.zc_freq_hz     == pytest.approx(57.0)
    assert micl_ch.time_of_peak_s == pytest.approx(-0.004)


def test_event_c_vector_sum():
    r = parse_report_file(_fixture("event-c"))
    assert r.peak_vector_sum_ips    == pytest.approx(0.612)
    assert r.peak_vector_sum_time_s == pytest.approx(0.024)


# ── Sensor self-check ───────────────────────────────────────────────────────


def test_event_c_sensor_check_geo_channels():
    r = parse_report_file(_fixture("event-c"))

    for ch_name, expected_freq, expected_ratio in [
        ("Tran", 7.4, 3.7),
        ("Vert", 7.6, 3.5),
        ("Long", 7.5, 3.8),
    ]:
        sc = r.sensor_check[ch_name]
        assert sc.test_freq_hz   == pytest.approx(expected_freq), ch_name
        assert sc.test_ratio     == pytest.approx(expected_ratio), ch_name
        assert sc.test_results   == "Passed", ch_name
        # Geo channels don't have an Test Amplitude
        assert sc.test_amplitude_mv is None


def test_event_c_sensor_check_micl():
    r = parse_report_file(_fixture("event-c"))

    sc = r.sensor_check["MicL"]
    assert sc.test_freq_hz       == pytest.approx(20.1)
    assert sc.test_amplitude_mv  == pytest.approx(533.0)
    assert sc.test_results       == "Passed"
    # MicL doesn't have a ratio — it has amplitude instead
    assert sc.test_ratio is None


# ── Monitor log + tooling ───────────────────────────────────────────────────


def test_event_c_monitor_log_and_pc_version():
    r = parse_report_file(_fixture("event-c"))

    assert len(r.monitor_log) == 1
    e = r.monitor_log[0]
    assert e.start_time == datetime.datetime(2026, 4, 23, 15, 46, 16)
    assert e.stop_time  == datetime.datetime(2026, 4, 23, 15, 56, 36)
    assert e.description == "Event recorded."

    assert r.pc_sw_version == "V 10.74"


# ── Sample table ─────────────────────────────────────────────────────────────


def test_event_c_sample_table_parsed_when_requested():
    r = parse_report_file(_fixture("event-c"), parse_samples=True)

    # 1 sec event @ 1024 sps + 0.25 sec pretrig = 1280 samples
    assert r.samples is not None
    assert len(r.samples) == 1280, f"expected 1280 samples, got {len(r.samples)}"

    # First row: "0.000     \t0.005     \t0.005     \t-81.94"
    t, v, l, m = r.samples[0]
    assert t == pytest.approx(0.000)
    assert v == pytest.approx(0.005)
    assert l == pytest.approx(0.005)
    assert m == pytest.approx(-81.94)


def test_event_c_sample_table_skipped_by_default():
    r = parse_report_file(_fixture("event-c"))
    assert r.samples is None


# ── Cross-event smoke ───────────────────────────────────────────────────────


@pytest.mark.parametrize("event_name", ["event-a", "event-b", "event-c", "event-d"])
def test_all_fixtures_parse_without_error(event_name):
    """Every fixture in the bundle must parse cleanly with the same parser."""
    r = parse_report_file(_fixture(event_name))

    # Common invariants: serial, event_datetime, sample rate, all four
    # channels surfaced.
    assert r.serial == "BE11529"
    assert r.event_datetime is not None
    assert r.sample_rate_sps in (1024, 2048, 4096)
    for ch in ("Tran", "Vert", "Long", "MicL"):
        assert ch in r.channels
        assert ch in r.sensor_check

    # PVS should be present and positive on triggered events
    if r.peak_vector_sum_ips is not None:
        assert r.peak_vector_sum_ips >= 0


# ── Edge cases / defensive parsing ──────────────────────────────────────────


def test_parse_empty_input():
    r = parse_report("")
    assert r.serial is None
    assert r.event_datetime is None
    assert all(cs.ppv_ips is None for cs in r.channels.values())


def test_parse_unknown_keys_ignored():
    """Forward-compat: future BW versions may add fields we don't recognise.
    Those should be silently dropped, not raise."""
    text = (
        '"Serial Number : BE99999"\n'
        '"Future Field That Does Not Exist : 42 widgets"\n'
        '"Tran PPV : 0.123 in/s"\n'
    )
    r = parse_report(text)
    assert r.serial == "BE99999"
    assert r.channels["Tran"].ppv_ips == pytest.approx(0.123)


def test_parse_numeric_with_units_strips_unit():
    text = (
        '"Vert PPV : 1.275 in/s"\n'
        '"Vert ZC Freq : 23 Hz"\n'
        '"MicL  Test Amplitude : 569 mv"\n'
    )
    r = parse_report(text)
    assert r.channels["Vert"].ppv_ips    == pytest.approx(1.275)
    assert r.channels["Vert"].zc_freq_hz == pytest.approx(23.0)
    assert r.sensor_check["MicL"].test_amplitude_mv == pytest.approx(569.0)


def test_parse_handles_micl_double_space_in_key():
    """BW writes "MicL  Time of Peak" with TWO spaces; the parser must
    normalise whitespace before key lookup."""
    text = (
        '"MicL  Time of Peak : 0.012 sec"\n'
        '"MicL  ZC Freq : 51 Hz"\n'
    )
    r = parse_report(text)
    assert r.mic.time_of_peak_s == pytest.approx(0.012)
    assert r.mic.zc_freq_hz     == pytest.approx(51.0)


# ── Operator-field label normalisation ──────────────────────────────────────


@pytest.mark.parametrize("label,expected_field,expected_value", [
    # project — both with-colon and bare
    ("Project:", "project", "Test4-21-26"),
    ("Project",  "project", "Test4-21-26"),
    # client
    ("Client:",  "client",  "Acme Inc"),
    ("Client",   "client",  "Acme Inc"),
    # operator (User Name)
    ("User Name:", "operator", "Brian"),
    ("User Name",  "operator", "Brian"),
    # sensor_location — every Seis*/Sensor Location variant we know about
    ("Seis Loc:",       "sensor_location", "Catbed"),
    ("Seis Loc",        "sensor_location", "Catbed"),
    ("Seis. Location",  "sensor_location", "Catbed"),
    ("Seis Location",   "sensor_location", "Catbed"),
    ("Sensor Location", "sensor_location", "Catbed"),
])
def test_operator_label_variants_route_to_correct_field(label, expected_field, expected_value):
    """All known label spellings of the operator-supplied fields route
    to the same dataclass attribute."""
    text = '"{} : {}"\n'.format(label, expected_value)
    r = parse_report(text)
    assert getattr(r, expected_field) == expected_value, (label, expected_field)


def test_label_normalisation_is_case_insensitive():
    """Lowercase / uppercase / mixed-case labels all hit the same slot."""
    for label in ("seis loc", "SEIS LOC", "SeIs LoC"):
        text = '"{} : Catbed"\n'.format(label)
        r = parse_report(text)
        assert r.sensor_location == "Catbed", label


def test_label_normalisation_collapses_extra_whitespace():
    """Internal whitespace runs in labels are collapsed before lookup."""
    text = '"Seis  Loc : Catbed"\n'   # two spaces between "Seis" and "Loc"
    r = parse_report(text)
    assert r.sensor_location == "Catbed"


def test_real_histogram_fixture_populates_sensor_location():
    """End-to-end: the histogram fixture from example-events/histogram/
    uses 'Seis. Location' (with period) and must successfully populate
    sensor_location after the label-normalisation fix."""
    fixture_dir = (
        Path(__file__).parent.parent / "example-events" / "histogram"
    )
    if not fixture_dir.exists():
        pytest.skip("histogram fixtures not present")
    txt = next(fixture_dir.glob("*_ASCII.TXT"), None)
    if txt is None:
        pytest.skip("no histogram TXT in fixture dir")

    r = parse_report_file(txt)
    # The histogram TXTs verified to use "Seis. Location" — should now
    # populate sensor_location instead of being silently dropped.
    assert r.sensor_location is not None
    assert len(r.sensor_location) > 0
    # Sanity: other shared fields still parse correctly
    assert r.serial is not None
    assert r.serial.startswith("BE")
    assert r.geo_range_ips is not None