Files
seismo-relay/tests/test_bw_ascii_report.py
T
serversdown cdfe4ad3c8 feat(import): parse paired BW ASCII reports on /db/import/blastware_file
Blastware's ACH writes a per-event ASCII report (.TXT) alongside each
event binary, containing the rich derived per-channel fields BW
computes (PPV, ZC Freq, Time of Peak, Peak Acceleration, Peak
Displacement, Peak Vector Sum + time, sensor self-check Pass/Fail,
monitor-log timestamps).  None of this lives in the BW binary itself.

When the watcher daemon forwards both files to /db/import/blastware_file
in one multipart POST, we now:

  - Pair binaries with their .TXT partners by filename match
  - Parse the report into a structured BwAsciiReport
  - Land the rich fields in a new top-level `bw_report` block of the
    sidecar JSON
  - Overlay the report's peaks/project_info/timestamp/sample_rate/
    record_time/total_samples/pretrig_samples onto the canonical
    sidecar fields (the report values are device-authoritative; the
    BW-binary STRT-derived values had bugs like reading the 0x46
    record-type marker as rectime)

This unblocks the monthly-summary review workflow — events become
sortable/filterable by peak, location, project, etc. — without
depending on the still-undecoded waveform body codec.
2026-05-08 23:56:43 +00:00

260 lines
9.1 KiB
Python

"""
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)