serversdownandClaude Opus 4.8
8d3cdba1b5
Make the sensor self-check a first-class part of the standardized decoded event
so SFM stops decoding it at report time — device-agnostic, per the store's
decoder→standardized-.h5→SFM model.
* Event gains a `sensor_check` field; both decoders attach the traces where
they set raw_samples — series-3 in event_file_io.read_blastware_file
(minimateplus.sensor_check), series-4 in waveform_store's IDF path
(micromate.sensor_check). Covers ingest and backfill (both re-decode).
* event_hdf5 bumps schema_version 1→2 and writes an optional /sensor_check
group (raw counts, int32, per channel present). read_event_hdf5 returns
it; plot_json_from_hdf5 carries it as a top-level key. Old v1 files still
read cleanly (no group → None), so nothing breaks before the backfill.
* gather_report_data reads sensor_check_waveforms from the .h5 and drops the
report-time series-3 decode — the report no longer reaches into a decoder,
and a series-4 event now lights up the same strip automatically.
Stored as raw counts (a shape diagnostic, rendered fit-to-box): the per-series
count scale differs and a physical mic unit is ill-defined, so conversion would
add complexity for no display benefit — easy to add later if a numeric use
appears.
Tests: .h5 roundtrip + backward-compat + plot_json + real series-3 decode
attaches to the Event.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf
1076 lines
46 KiB
Python
1076 lines
46 KiB
Python
"""
|
||
sfm/report_pdf.py — generate Instantel-style Event Report PDFs.
|
||
|
||
Stub layout for v0.20.0 — the exact visual is iterated against actual
|
||
Blastware reference PDFs (uploaded to docs/reference/instantel/).
|
||
Current output captures all the data fields a real BW Event Report
|
||
contains, but the visual hierarchy / spacing is still approximate.
|
||
|
||
Architecture
|
||
────────────
|
||
1. ``gather_report_data(event_id)`` — assembles a flat dict from three
|
||
sources: the SeismoDb events row, the .sfm.json sidecar (bw_report
|
||
block), and the .h5 waveform samples. Returns ``None`` when the
|
||
event doesn't exist or has no waveform data on disk.
|
||
|
||
2. ``render_event_report_pdf(data)`` — takes that dict and produces a
|
||
single-page letter-sized PDF as bytes, using matplotlib's PDF
|
||
backend (vector output, no rasterization, prints cleanly).
|
||
|
||
3. The HTTP endpoint at ``/db/events/{id}/report.pdf`` wires them
|
||
together: fetch event → gather → render → stream bytes back with
|
||
``Content-Type: application/pdf``.
|
||
|
||
What's in the report (every field BW's printout includes):
|
||
|
||
Header (left): Date/Time, Trigger Source, Range, Sample Rate, Notes,
|
||
Project, Client, User Name, Seis. Loc
|
||
Header (right): Serial + firmware, Battery, Calibration, File Name,
|
||
Post Event Notes
|
||
Mic block: PSPL (dBL + psi), ZC Freq, Channel Test result
|
||
Stats table: per-channel PPV / ZC Freq / Time of Peak /
|
||
Peak Acceleration / Peak Displacement / Sensor Check
|
||
Peak Vector Sum
|
||
Waveform plot: 4 channels stacked (MicL/Long/Vert/Tran), shared
|
||
time axis, trigger marker, peak markers
|
||
USBM RI8507/OSMRE compliance chart: STUBBED — separate work item
|
||
|
||
Histogram events: the layout differs (Number of Intervals header
|
||
field, no trigger marker, per-interval bar chart instead of waveform).
|
||
Handled via a record_type branch in ``render_event_report_pdf``.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import logging
|
||
import math
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import matplotlib
|
||
matplotlib.use("Agg") # headless — no display required
|
||
import matplotlib.pyplot as plt
|
||
import numpy as np
|
||
from matplotlib.backends.backend_pdf import PdfPages
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
# Reference pressure for dB(L) conversion: 20 µPa expressed in psi.
|
||
DBL_REF_PSI = 2.9e-9
|
||
|
||
|
||
# ── Data assembly ────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class ReportData:
|
||
"""All fields needed to render an Instantel-style Event Report.
|
||
|
||
Most fields are Optional — BW's printout shows '—' or just omits
|
||
sections when source data is missing. The renderer mirrors that.
|
||
"""
|
||
# Header — left column
|
||
event_datetime_str: Optional[str] = None
|
||
trigger_source: Optional[str] = None
|
||
geo_range_str: Optional[str] = None
|
||
sample_rate_str: Optional[str] = None
|
||
notes: Optional[str] = None
|
||
project: Optional[str] = None
|
||
client: Optional[str] = None
|
||
operator: Optional[str] = None
|
||
sensor_location: Optional[str] = None
|
||
|
||
# Header — right column
|
||
serial: Optional[str] = None
|
||
firmware: Optional[str] = None
|
||
battery_volts: Optional[float] = None
|
||
calibration_date: Optional[str] = None
|
||
calibration_by: Optional[str] = None
|
||
file_name: Optional[str] = None
|
||
post_event_notes: Optional[str] = None
|
||
|
||
# Microphone block
|
||
mic_pspl_dbl: Optional[float] = None
|
||
mic_pspl_psi: Optional[float] = None
|
||
mic_pspl_time_s: Optional[float] = None
|
||
mic_pspl_when_str: Optional[str] = None # histogram absolute date+time, BW-formatted
|
||
mic_zc_freq_hz: Optional[float] = None
|
||
mic_zc_freq_above_range: bool = False
|
||
mic_channel_test_result: Optional[str] = None
|
||
mic_channel_test_freq_hz: Optional[float] = None
|
||
mic_channel_test_amp_mv: Optional[float] = None
|
||
|
||
# Per-channel stats — list of dicts (one per channel)
|
||
# Keys: name, ppv_ips, zc_freq_hz, time_of_peak_s,
|
||
# peak_accel_g, peak_disp_in, sensor_check
|
||
channel_stats: list[dict] = field(default_factory=list)
|
||
|
||
# Peak Vector Sum
|
||
peak_vector_sum_ips: Optional[float] = None
|
||
peak_vector_sum_time_s: Optional[float] = None
|
||
|
||
# Waveform samples — channels[ch] = list of floats in physical units
|
||
# Time axis derived from sample_rate + pretrig_samples
|
||
channels: dict = field(default_factory=dict)
|
||
sample_rate_sps: Optional[int] = None
|
||
pretrig_samples: Optional[int] = None
|
||
t0_ms: Optional[float] = None
|
||
dt_ms: Optional[float] = None
|
||
|
||
# Sensor self-check traces — {ch: [samples]} in raw counts, read from the
|
||
# standardized .h5 (/sensor_check group, schema v2+) where the per-series
|
||
# decoder stored them at ingest. The little diagnostic waveforms BW draws
|
||
# in its "Sensor Check" strip. Empty when absent (pre-v2 .h5, histogram,
|
||
# or 3-channel unit's MicL).
|
||
sensor_check_waveforms: dict = field(default_factory=dict)
|
||
|
||
# Record-type discriminator
|
||
record_type: Optional[str] = None
|
||
is_histogram: bool = False
|
||
|
||
# Histogram-only fields — only populated for record_type starts with 'Hist'
|
||
histogram_start_str: Optional[str] = None # "22:30:38 May 16, 2026"
|
||
histogram_stop_str: Optional[str] = None
|
||
histogram_n_intervals: Optional[float] = None # 4.00
|
||
histogram_interval_size: Optional[str] = None # "1 minute"
|
||
histogram_interval_size_s: Optional[float] = None # 60.0 — numeric seconds, used to derive interval_times
|
||
histogram_interval_times: list[str] = field(default_factory=list) # per-interval timestamps for x-axis
|
||
|
||
# Peak Vector Sum metadata (histograms show absolute date+time)
|
||
peak_vector_sum_when_str: Optional[str] = None
|
||
|
||
# Bookkeeping
|
||
event_id: Optional[str] = None
|
||
server_received_at: Optional[str] = None
|
||
bw_pc_sw_version: Optional[str] = None
|
||
|
||
|
||
def gather_report_data(
|
||
db,
|
||
store,
|
||
event_id: str,
|
||
) -> Optional[ReportData]:
|
||
"""Collect every field needed to render an event report.
|
||
|
||
Returns ``None`` if the event is unknown or has no waveform data
|
||
on disk (no .h5, no .a5.pkl — same condition the waveform.json
|
||
endpoint 404s on).
|
||
"""
|
||
row = db.get_event(event_id)
|
||
if row is None:
|
||
return None
|
||
serial = row.get("serial")
|
||
filename = row.get("blastware_filename")
|
||
if not serial or not filename:
|
||
return None
|
||
|
||
rd = ReportData(
|
||
event_id=event_id,
|
||
serial=serial,
|
||
file_name=filename,
|
||
record_type=row.get("record_type"),
|
||
is_histogram=str(row.get("record_type", "")).lower().startswith("hist"),
|
||
event_datetime_str=row.get("timestamp"),
|
||
sample_rate_sps=row.get("sample_rate"),
|
||
project=row.get("project"),
|
||
client=row.get("client"),
|
||
operator=row.get("operator"),
|
||
sensor_location=row.get("sensor_location"),
|
||
server_received_at=row.get("created_at"),
|
||
)
|
||
|
||
# ── Sidecar bw_report — the rich BW-derived fields ──
|
||
sidecar_path = store.sidecar_path_for(serial, filename)
|
||
if sidecar_path.exists():
|
||
try:
|
||
sc = json.loads(sidecar_path.read_text())
|
||
except Exception as exc:
|
||
log.warning("gather_report_data: sidecar read failed: %s", exc)
|
||
sc = {}
|
||
bw = sc.get("bw_report") or {}
|
||
|
||
# Trigger / range / sample-rate display
|
||
trig = bw.get("trigger") or {}
|
||
rd.trigger_source = (
|
||
f"{trig.get('channel','')}: {trig.get('geo_level_ips')} in/s"
|
||
if trig.get("channel") or trig.get("geo_level_ips") is not None
|
||
else None
|
||
)
|
||
rec = bw.get("recording") or {}
|
||
rd.geo_range_str = (
|
||
f"Geo: {rec.get('geo_range_ips')} in/s"
|
||
if rec.get("geo_range_ips") is not None else None
|
||
)
|
||
rt = rec.get("record_time_s")
|
||
if rt is not None and rd.sample_rate_sps:
|
||
rd.sample_rate_str = f"{rt:.1f} sec At {rd.sample_rate_sps} Sps"
|
||
|
||
# Device block
|
||
dev = bw.get("device") or {}
|
||
rd.battery_volts = dev.get("battery_volts")
|
||
rd.calibration_date = dev.get("calibration_date")
|
||
rd.calibration_by = dev.get("calibration_by")
|
||
rd.firmware = bw.get("version")
|
||
rd.bw_pc_sw_version = bw.get("pc_sw_version")
|
||
|
||
# Microphone block
|
||
mic = bw.get("mic") or {}
|
||
rd.mic_pspl_dbl = mic.get("pspl_dbl")
|
||
if rd.mic_pspl_dbl is not None and rd.mic_pspl_dbl > 0:
|
||
# Inverse of the dBL formula → psi. Mirrors waveform_codec convention.
|
||
rd.mic_pspl_psi = DBL_REF_PSI * (10 ** (rd.mic_pspl_dbl / 20))
|
||
rd.mic_pspl_time_s = mic.get("time_of_peak_s")
|
||
rd.mic_zc_freq_hz = mic.get("zc_freq_hz")
|
||
rd.mic_zc_freq_above_range = bool(mic.get("zc_freq_above_range"))
|
||
sc_mic = (bw.get("sensor_check") or {}).get("mic") or {}
|
||
rd.mic_channel_test_result = sc_mic.get("result")
|
||
rd.mic_channel_test_freq_hz = sc_mic.get("freq_hz")
|
||
rd.mic_channel_test_amp_mv = sc_mic.get("amplitude_mv")
|
||
|
||
# Per-channel stats (Tran / Vert / Long). Per-channel peak
|
||
# date+time for histograms comes from bw_report.histogram.channel_peak_when
|
||
# (populated when the parser captured it; see the bw_ascii_report
|
||
# parser's histogram-fields handler).
|
||
peaks = bw.get("peaks") or {}
|
||
sc_block = bw.get("sensor_check") or {}
|
||
hist_block = bw.get("histogram") or {}
|
||
peak_when = hist_block.get("channel_peak_when") or {}
|
||
for ch_lc, ch_label in (("tran", "Tran"), ("vert", "Vert"), ("long", "Long")):
|
||
ch = peaks.get(ch_lc) or {}
|
||
sc_ch = sc_block.get(ch_lc) or {}
|
||
ch_when_iso = peak_when.get(ch_label)
|
||
peak_date, peak_time = _split_iso_to_date_time(ch_when_iso)
|
||
rd.channel_stats.append({
|
||
"name": ch_label,
|
||
"ppv_ips": ch.get("ppv_ips"),
|
||
"zc_freq_hz": ch.get("zc_freq_hz"),
|
||
"zc_freq_above_range": bool(ch.get("zc_freq_above_range")),
|
||
"time_of_peak_s": ch.get("time_of_peak_s"),
|
||
"peak_accel_g": ch.get("peak_accel_g"),
|
||
"peak_disp_in": ch.get("peak_disp_in"),
|
||
"sensor_check": sc_ch.get("result"),
|
||
"sc_freq_hz": sc_ch.get("freq_hz"),
|
||
"sc_ratio": sc_ch.get("ratio"),
|
||
"peak_date": peak_date,
|
||
"peak_time": peak_time,
|
||
})
|
||
|
||
# MicL peak time (used in the mic block — "PSPL ... on DATE at TIME")
|
||
mic_when_iso = peak_when.get("MicL")
|
||
rd.mic_pspl_when_str = _fmt_iso_to_bw(mic_when_iso) if mic_when_iso else None
|
||
|
||
# Peak Vector Sum
|
||
vs = peaks.get("vector_sum") or {}
|
||
rd.peak_vector_sum_ips = vs.get("ips")
|
||
rd.peak_vector_sum_time_s = vs.get("time_s")
|
||
# PVS absolute date+time (histograms). Same formatting as Mic.
|
||
pvs_when_iso = vs.get("when")
|
||
rd.peak_vector_sum_when_str = _fmt_iso_to_bw(pvs_when_iso) if pvs_when_iso else None
|
||
|
||
# Histogram-specific header fields — keys match the projection in
|
||
# _bw_report_to_dict ("start" / "stop", not "_str" suffixed).
|
||
if rd.is_histogram:
|
||
rd.histogram_start_str = hist_block.get("start") or rd.event_datetime_str
|
||
rd.histogram_stop_str = hist_block.get("stop")
|
||
rd.histogram_n_intervals = hist_block.get("n_intervals")
|
||
rd.histogram_interval_size = hist_block.get("interval_size")
|
||
rd.histogram_interval_size_s = hist_block.get("interval_size_s")
|
||
rd.histogram_interval_times = hist_block.get("interval_times") or []
|
||
|
||
# ── Waveform samples — from the .h5 via the existing helper ──
|
||
from sfm import event_hdf5
|
||
h5_path = store.hdf5_path_for(serial, filename)
|
||
if h5_path.exists():
|
||
try:
|
||
wf = event_hdf5.plot_json_from_hdf5(h5_path, event_id=event_id)
|
||
rd.channels = {
|
||
ch: (chd.get("values") or [])
|
||
for ch, chd in (wf.get("channels") or {}).items()
|
||
}
|
||
ta = wf.get("time_axis") or {}
|
||
rd.sample_rate_sps = rd.sample_rate_sps or ta.get("sample_rate")
|
||
rd.pretrig_samples = ta.get("pretrig_samples")
|
||
rd.t0_ms = ta.get("t0_ms")
|
||
rd.dt_ms = ta.get("dt_ms")
|
||
# Sensor self-check traces — read from the standardized .h5 (schema
|
||
# v2+). Device-agnostic: whichever decoder produced the event
|
||
# stored them at ingest, so SFM reads them here without knowing or
|
||
# caring about the source instrument series. Empty on pre-v2 files
|
||
# (until backfilled) and on 3-channel / histogram events.
|
||
rd.sensor_check_waveforms = wf.get("sensor_check") or {}
|
||
except Exception as exc:
|
||
log.warning("gather_report_data: hdf5 read failed: %s", exc)
|
||
|
||
# ── Histogram aggregation ──
|
||
# Codec emits ~N per-block samples (typically 1/sec); BW reports
|
||
# one bar per configured interval (1 min / 5 min / etc.). When
|
||
# bw_report.histogram.n_intervals is populated (events ingested
|
||
# with the parser extension), group max-per-group to match. Also
|
||
# derives per-interval timestamps for the x-axis. No-op for
|
||
# waveform events or when n_intervals is missing.
|
||
if rd.is_histogram and rd.histogram_n_intervals and rd.histogram_n_intervals >= 1:
|
||
n = int(rd.histogram_n_intervals)
|
||
for ch, vals in list(rd.channels.items()):
|
||
if not vals:
|
||
continue
|
||
per_group = len(vals) // n
|
||
remainder = len(vals) % n
|
||
agg: list = []
|
||
offset = 0
|
||
for i in range(n):
|
||
grp_size = per_group + (1 if i < remainder else 0)
|
||
if grp_size > 0:
|
||
grp = vals[offset:offset + grp_size]
|
||
agg.append(max((abs(v) for v in grp if v is not None), default=0))
|
||
offset += grp_size
|
||
else:
|
||
agg.append(0)
|
||
rd.channels[ch] = agg
|
||
# Derive per-interval HH:MM:SS labels if we have the start time + size
|
||
if rd.histogram_start_str and rd.histogram_interval_size_s and not rd.histogram_interval_times:
|
||
try:
|
||
import datetime as _dt
|
||
start = _dt.datetime.fromisoformat(rd.histogram_start_str)
|
||
rd.histogram_interval_times = [
|
||
(start + _dt.timedelta(seconds=(i + 1) * rd.histogram_interval_size_s)).strftime("%H:%M:%S")
|
||
for i in range(n)
|
||
]
|
||
except Exception:
|
||
pass
|
||
|
||
return rd
|
||
|
||
|
||
# ── PDF rendering ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def render_event_report_pdf(rd: ReportData) -> bytes:
|
||
"""Render an event report dict to a single-page letter PDF.
|
||
|
||
Branches on ``rd.is_histogram`` — waveform and histogram layouts
|
||
differ in their header fields, stats-table rows, and bottom plot.
|
||
Layout modeled on Blastware's Event Report PDFs (samples in
|
||
docs/reference/instantel/).
|
||
"""
|
||
# Letter portrait — 8.5"×11"
|
||
fig = plt.figure(figsize=(8.5, 11), dpi=100)
|
||
fig.patch.set_facecolor("white")
|
||
|
||
if rd.is_histogram:
|
||
_render_histogram_layout(fig, rd)
|
||
else:
|
||
_render_waveform_layout(fig, rd)
|
||
|
||
# Page footer (common to both layouts) — Created date + event id.
|
||
# Pushed to the very page bottom so it doesn't collide with the
|
||
# waveform footer scale / trigger legend lines just above.
|
||
# Convert UTC server_received_at to local for display.
|
||
created_local = _fmt_iso_to_bw(rd.server_received_at) if rd.server_received_at else "—"
|
||
fig.text(
|
||
0.07, 0.005,
|
||
f"Created: {created_local} • seismo-relay",
|
||
fontsize=6, color="#888", ha="left",
|
||
)
|
||
fig.text(
|
||
0.93, 0.005,
|
||
f"Event {rd.event_id[:8] if rd.event_id else '—'}",
|
||
fontsize=6, color="#888", ha="right",
|
||
)
|
||
|
||
buf = io.BytesIO()
|
||
fig.savefig(buf, format="pdf")
|
||
plt.close(fig)
|
||
return buf.getvalue()
|
||
|
||
|
||
def _render_waveform_layout(fig, rd: ReportData) -> None:
|
||
"""Waveform layout: header / mic+USBM / per-channel stats / waveform plot.
|
||
|
||
Stats table includes Time (Rel. to Trig), Peak Accel, Peak Disp.
|
||
Left margin sized to fit the channel labels (MicL/Long/Vert/Tran).
|
||
Extra bottom margin reserves space for x-axis tick labels +
|
||
"Amplitude Geo: X in/s/div Mic: Y psi(L)/div" footer + trigger
|
||
legend without overlap.
|
||
"""
|
||
gs = fig.add_gridspec(
|
||
nrows=4, ncols=1,
|
||
left=0.11, right=0.94, top=0.97, bottom=0.12,
|
||
height_ratios=[1.7, 2.0, 1.8, 5.5],
|
||
hspace=0.35,
|
||
)
|
||
ax_header = fig.add_subplot(gs[0]); ax_header.axis("off")
|
||
_draw_header_waveform(ax_header, rd)
|
||
|
||
ax_mid = fig.add_subplot(gs[1]); ax_mid.axis("off")
|
||
_draw_mic_and_usbm(ax_mid, rd)
|
||
|
||
ax_stats = fig.add_subplot(gs[2]); ax_stats.axis("off")
|
||
_draw_channel_stats_waveform(ax_stats, rd)
|
||
|
||
_draw_compliance_panel(fig, rd)
|
||
_draw_waveform_subplot(fig, gs[3], rd)
|
||
|
||
|
||
# Compliance-chart placement, in figure fractions. Measured directly off a
|
||
# Blastware Event Report PDF (ref-stuff/n844lqhbzt0w_bw_pdf.pdf) so the chart
|
||
# matches BW's size and position: it spans from just under the header down
|
||
# through the stats band, hard against the right page margin. The left edge
|
||
# leaves room for the y-axis tick labels + "Velocity (in/s)" title, which the
|
||
# compacted stats table (see _draw_channel_stats_waveform) is sized to clear.
|
||
_COMPLIANCE_BOX = (0.489, 0.502, 0.951, 0.867) # x0, y0, x1, y1
|
||
|
||
|
||
def _draw_compliance_panel(fig, rd: ReportData) -> None:
|
||
"""Large USBM RI8507 compliance chart in the upper-right, sized and
|
||
positioned to match Blastware's Event Report (see _COMPLIANCE_BOX)."""
|
||
x0, y0, x1, y1 = _COMPLIANCE_BOX
|
||
fig.text((x0 + x1) / 2, y1 + 0.006, "USBM RI8507 And OSMRE", fontsize=9,
|
||
weight="bold", color="#333", ha="center", va="bottom")
|
||
if rd.channels and rd.sample_rate_sps:
|
||
from sfm.compliance import draw_compliance_chart
|
||
ax = fig.add_axes([x0, y0, x1 - x0, y1 - y0])
|
||
draw_compliance_chart(ax, rd.channels, rd.sample_rate_sps)
|
||
else:
|
||
fig.text((x0 + x1) / 2, (y0 + y1) / 2, "(no waveform data)", fontsize=8,
|
||
color="#bbb", ha="center", va="center", style="italic")
|
||
|
||
|
||
def _render_histogram_layout(fig, rd: ReportData) -> None:
|
||
"""Histogram layout: header / mic-only / per-channel stats / bar plot.
|
||
|
||
No USBM compliance chart (it's a waveform-only concept). Stats table
|
||
uses Date + Time-of-peak instead of relative-time + accel + disp.
|
||
Left margin sized to fit the channel labels. Extra bottom margin
|
||
leaves room for the x-axis time labels + footer scale legend
|
||
without overlap.
|
||
"""
|
||
gs = fig.add_gridspec(
|
||
nrows=4, ncols=1,
|
||
left=0.11, right=0.94, top=0.97, bottom=0.12,
|
||
height_ratios=[1.8, 0.9, 1.7, 5.6],
|
||
hspace=0.35,
|
||
)
|
||
ax_header = fig.add_subplot(gs[0]); ax_header.axis("off")
|
||
_draw_header_histogram(ax_header, rd)
|
||
|
||
ax_mic = fig.add_subplot(gs[1]); ax_mic.axis("off")
|
||
_draw_mic_only(ax_mic, rd)
|
||
|
||
ax_stats = fig.add_subplot(gs[2]); ax_stats.axis("off")
|
||
_draw_channel_stats_histogram(ax_stats, rd)
|
||
|
||
_draw_histogram_subplot(fig, gs[3], rd)
|
||
|
||
|
||
def _to_display_local(iso: str):
|
||
"""Parse an ISO timestamp and return a datetime in the system's local
|
||
timezone (set by the TZ env var, default America/New_York via the
|
||
Dockerfile).
|
||
|
||
Behaviour:
|
||
- "...Z" or "...+HH:MM" suffix → tz-aware UTC → converted to local
|
||
- Naïve "YYYY-MM-DDTHH:MM:SS" (no tz) → returned as-is. This
|
||
matches the convention used elsewhere in seismo-relay: BW's
|
||
recorded-at timestamps are naïve and ALREADY in the unit's
|
||
local clock; we don't second-guess them.
|
||
"""
|
||
import datetime as _dt
|
||
dt = _dt.datetime.fromisoformat(iso.replace("Z", "+00:00"))
|
||
if dt.tzinfo is not None:
|
||
# Convert from UTC (or other tz) → local per the TZ env var.
|
||
# astimezone() without arg uses the system timezone.
|
||
dt = dt.astimezone()
|
||
return dt
|
||
|
||
|
||
def _fmt_iso_to_bw(iso: Optional[str]) -> Optional[str]:
|
||
"""Convert an ISO-8601 timestamp to BW's display format
|
||
'22:30:37 May 16, 2026'. UTC inputs (with Z suffix) are
|
||
converted to the system's local timezone first; naïve inputs
|
||
are formatted as-is. Returns input unchanged on parse failure."""
|
||
if not iso or "T" not in iso:
|
||
return iso
|
||
try:
|
||
return _to_display_local(iso).strftime("%H:%M:%S %B %d, %Y").replace(" 0", " ")
|
||
except Exception:
|
||
return iso
|
||
|
||
|
||
def _split_iso_to_date_time(iso: Optional[str]) -> tuple[Optional[str], Optional[str]]:
|
||
"""Split an ISO timestamp into BW-formatted ('May 27 /26', '06:06:14')
|
||
date+time strings. Used for the histogram stats table where the
|
||
Date and Time rows are presented separately. UTC inputs are
|
||
converted to local time first. Returns (None, None) on parse failure."""
|
||
if not iso:
|
||
return (None, None)
|
||
try:
|
||
dt = _to_display_local(iso)
|
||
# BW format: 'May 27 /26' (3-letter month + 2-digit year)
|
||
date_str = dt.strftime("%b %d /%y").replace(" 0", " ")
|
||
time_str = dt.strftime("%H:%M:%S")
|
||
return (date_str, time_str)
|
||
except Exception:
|
||
return (None, None)
|
||
|
||
|
||
def _kv(ax, x, y, label, value, *, label_w=0.18, fontsize=8):
|
||
"""Render a 'Label Value' row at axes-coordinates (x, y)."""
|
||
ax.text(x, y, label, fontsize=fontsize, color="#555", ha="left", va="top",
|
||
transform=ax.transAxes)
|
||
ax.text(x + label_w, y, _fmt(value), fontsize=fontsize, ha="left", va="top",
|
||
transform=ax.transAxes, family="monospace")
|
||
|
||
|
||
def _fmt(v):
|
||
"""Format any field for display — '—' for None, str otherwise."""
|
||
if v is None:
|
||
return "—"
|
||
if isinstance(v, float):
|
||
return f"{v:.4f}".rstrip("0").rstrip(".")
|
||
return str(v)
|
||
|
||
|
||
def _draw_header_waveform(ax, rd: ReportData) -> None:
|
||
"""Two-column metadata header — waveform variant."""
|
||
rows_left = [
|
||
("Date/Time", _fmt_iso_to_bw(rd.event_datetime_str)),
|
||
("Trigger Source", rd.trigger_source),
|
||
("Range", rd.geo_range_str),
|
||
("Sample Rate", rd.sample_rate_str),
|
||
("Notes", rd.notes),
|
||
("Project:", rd.project),
|
||
("Client:", rd.client),
|
||
("User Name:", rd.operator),
|
||
("Seis. Loc:", rd.sensor_location),
|
||
]
|
||
_draw_header_columns(ax, rows_left, rd)
|
||
|
||
|
||
def _draw_header_histogram(ax, rd: ReportData) -> None:
|
||
"""Two-column metadata header — histogram variant.
|
||
|
||
Histograms have Start / Finish / Intervals fields instead of
|
||
Trigger Source (there's no trigger event for a histogram capture).
|
||
"""
|
||
intervals_str = None
|
||
if rd.histogram_n_intervals is not None and rd.histogram_interval_size:
|
||
intervals_str = f"{rd.histogram_n_intervals} At {rd.histogram_interval_size}"
|
||
rows_left = [
|
||
("Start", _fmt_iso_to_bw(rd.histogram_start_str or rd.event_datetime_str)),
|
||
("Finish", _fmt_iso_to_bw(rd.histogram_stop_str)),
|
||
("Intervals", intervals_str),
|
||
("Range", rd.geo_range_str),
|
||
("Sample Rate", (f"{rd.sample_rate_sps} Sps" if rd.sample_rate_sps else None)),
|
||
("Notes", rd.notes),
|
||
("Project:", rd.project),
|
||
("Client:", rd.client),
|
||
("User Name:", rd.operator),
|
||
("Seis. Loc:", rd.sensor_location),
|
||
]
|
||
_draw_header_columns(ax, rows_left, rd)
|
||
|
||
|
||
def _draw_header_columns(ax, rows_left, rd: ReportData) -> None:
|
||
"""Shared 2-column header rendering used by both layouts."""
|
||
rows_right = [
|
||
("Serial Number", f"{rd.serial or '—'}" + (f" {rd.firmware}" if rd.firmware else "")),
|
||
("Battery Level", f"{rd.battery_volts:.1f} Volts" if rd.battery_volts is not None else None),
|
||
("Unit Calibration", (f"{rd.calibration_date}" + (f" by {rd.calibration_by}" if rd.calibration_by else ""))
|
||
if rd.calibration_date else None),
|
||
("File Name", rd.file_name),
|
||
("Post Event Notes", rd.post_event_notes),
|
||
]
|
||
# fontsize 7.5 (BW's header is a touch smaller than our body text) + a
|
||
# tighter right-column value indent so the long serial+firmware line
|
||
# ("BE##### V ##.##-#.## MiniMate Plus") fits without running off the page.
|
||
y = 0.95
|
||
dy = 0.095
|
||
for label, value in rows_left:
|
||
_kv(ax, 0.0, y, label, value, label_w=0.18, fontsize=7.5)
|
||
y -= dy
|
||
y = 0.95
|
||
for label, value in rows_right:
|
||
_kv(ax, 0.55, y, label, value, label_w=0.14, fontsize=7.5)
|
||
y -= dy
|
||
|
||
|
||
def _draw_mic_only(ax, rd: ReportData) -> None:
|
||
"""Mic block (histogram variant — no USBM chart)."""
|
||
ax.text(0.0, 0.95, "Microphone Linear Weighting", fontsize=8, color="#555",
|
||
transform=ax.transAxes, va="top")
|
||
rows = _mic_rows(rd)
|
||
y = 0.70
|
||
for label, value in rows:
|
||
_kv(ax, 0.0, y, label, value, label_w=0.18)
|
||
y -= 0.22
|
||
|
||
|
||
def _draw_mic_and_usbm(ax, rd: ReportData) -> None:
|
||
"""Mic block on the left + USBM compliance chart placeholder on right.
|
||
(Waveform variant — USBM is a velocity-vs-frequency compliance plot
|
||
that doesn't apply to histograms.)"""
|
||
ax.text(0.0, 0.95, "Microphone Linear Weighting", fontsize=8, color="#555",
|
||
transform=ax.transAxes, va="top")
|
||
rows = _mic_rows(rd)
|
||
y = 0.80
|
||
# Tighter label indent + slightly smaller font so the long "Channel Test
|
||
# Passed (Freq = … Amp = … mv)" line clears the enlarged compliance chart's
|
||
# left edge (_COMPLIANCE_BOX) instead of running behind it.
|
||
for label, value in rows:
|
||
_kv(ax, 0.0, y, label, value, label_w=0.13, fontsize=7)
|
||
y -= 0.15
|
||
# The USBM compliance chart is drawn as its own large square panel spanning
|
||
# the mic + stats rows on the right — see _draw_compliance_panel().
|
||
|
||
|
||
def _mic_rows(rd: ReportData) -> list[tuple[str, Optional[str]]]:
|
||
"""Build the mic-section value rows (shared by both layouts).
|
||
|
||
For histograms, BW formats the PSPL line as
|
||
"125.7 dB(L) on May 27, 2026 at 06:19:14"
|
||
(absolute date+time of peak). Waveform events show the relative
|
||
"at 0.012 sec." instead. Both formats covered here based on which
|
||
field is populated.
|
||
"""
|
||
rows: list[tuple[str, Optional[str]]] = []
|
||
if rd.mic_pspl_dbl is not None:
|
||
line = f"{rd.mic_pspl_dbl:.1f} dB(L)"
|
||
if rd.mic_pspl_when_str:
|
||
# Histogram-style: "PSPL 125.7 dB(L) on May 27, 2026 at 06:19:14"
|
||
# mic_pspl_when_str is already "HH:MM:SS Month DD, YYYY";
|
||
# reformat to "on Month DD, YYYY at HH:MM:SS" for BW match.
|
||
parts = rd.mic_pspl_when_str.split(" ", 1)
|
||
if len(parts) == 2:
|
||
line += f" on {parts[1]} at {parts[0]}"
|
||
else:
|
||
line += f" on {rd.mic_pspl_when_str}"
|
||
elif rd.mic_pspl_time_s is not None:
|
||
# Waveform-style: relative-to-trigger seconds.
|
||
line += f" at {rd.mic_pspl_time_s:.3f} sec."
|
||
rows.append(("PSPL", line))
|
||
if rd.mic_zc_freq_hz is not None:
|
||
prefix = ">" if rd.mic_zc_freq_above_range else ""
|
||
rows.append(("ZC Freq", f"{prefix}{rd.mic_zc_freq_hz:.0f} Hz"))
|
||
if rd.mic_channel_test_result:
|
||
line = rd.mic_channel_test_result
|
||
if rd.mic_channel_test_freq_hz is not None and rd.mic_channel_test_amp_mv is not None:
|
||
line += (f" (Freq = {rd.mic_channel_test_freq_hz:.1f} Hz, "
|
||
f"Amp = {rd.mic_channel_test_amp_mv:.0f} mv)")
|
||
rows.append(("Channel Test", line))
|
||
return rows
|
||
|
||
|
||
def _draw_channel_stats_waveform(ax, rd: ReportData) -> None:
|
||
"""Waveform stats table — has Time (Rel. to Trig), Peak Accel, Peak Disp.
|
||
Followed by Peak Vector Sum line."""
|
||
rows_spec = [
|
||
("PPV", "ppv_ips", "in/s"),
|
||
("ZC Freq", "zc_freq_hz", "Hz"),
|
||
("Time (Rel. to Trig)", "time_of_peak_s", "sec"),
|
||
("Peak Acceleration", "peak_accel_g", "g"),
|
||
("Peak Displacement", "peak_disp_in", "in"),
|
||
("Sensor Check", "sensor_check", ""),
|
||
# Sensor-check sub-rows (indented under "Sensor Check", like BW): the
|
||
# geophone ring-down frequency + overswing ratio from the self-check.
|
||
(" Frequency", "sc_freq_hz", "Hz"),
|
||
(" Overswing Ratio", "sc_ratio", ""),
|
||
]
|
||
# Compacted to the left half so the enlarged compliance chart (BW-sized,
|
||
# right against the page margin) has room — see _COMPLIANCE_BOX.
|
||
_draw_stats_table(
|
||
ax, rd, rows_spec,
|
||
bbox_width=0.42, fontsize=7.5,
|
||
col_widths=[0.185, 0.065, 0.065, 0.065, 0.040],
|
||
)
|
||
_draw_pvs_summary(ax, rd, n_data_rows=len(rows_spec))
|
||
|
||
|
||
def _draw_channel_stats_histogram(ax, rd: ReportData) -> None:
|
||
"""Histogram stats table — PPV, ZC Freq, Date, Time of peak, Sensor Check.
|
||
Followed by Peak Vector Sum line."""
|
||
# Date / Time of peak are per-channel timestamps for the interval at peak.
|
||
# bw_report stores time_of_peak_s as relative seconds, but for histograms
|
||
# BW shows them as absolute date+time. We populate from rd.channel_stats
|
||
# if those absolute fields are present; otherwise fall back to relative.
|
||
rows_spec = [
|
||
("PPV", "ppv_ips", "in/s"),
|
||
("ZC Freq", "zc_freq_hz", "Hz"),
|
||
("Date", "peak_date", ""),
|
||
("Time", "peak_time", ""),
|
||
("Sensor Check", "sensor_check", ""),
|
||
]
|
||
_draw_stats_table(ax, rd, rows_spec)
|
||
_draw_pvs_summary(ax, rd, n_data_rows=len(rows_spec), histogram_when=True)
|
||
|
||
|
||
def _draw_pvs_summary(
|
||
ax,
|
||
rd: ReportData,
|
||
*,
|
||
n_data_rows: int,
|
||
histogram_when: bool = False,
|
||
) -> None:
|
||
"""Render the Peak Vector Sum + 'NA: Not Applicable' caption below the
|
||
stats table.
|
||
|
||
Reads ``ax._stats_table_bottom`` (set by ``_draw_stats_table`` when
|
||
it pins the table via an explicit ``bbox``) so the PVS line lands
|
||
just below the table's known bottom edge instead of guessing at the
|
||
geometry.
|
||
|
||
Centered horizontally for visual balance (the previous left-aligned
|
||
x=0 landed under the label column, not the data, which looked off).
|
||
"""
|
||
if rd.peak_vector_sum_ips is None:
|
||
return
|
||
|
||
line = f"Peak Vector Sum {rd.peak_vector_sum_ips:.3f} in/s"
|
||
if histogram_when and rd.peak_vector_sum_when_str:
|
||
# Histogram absolute date+time. when_str is "HH:MM:SS Month DD, YYYY";
|
||
# reformat to "<value> on <date> At <time>" to match BW.
|
||
parts = rd.peak_vector_sum_when_str.split(" ", 1)
|
||
if len(parts) == 2:
|
||
line += f" on {parts[1]} At {parts[0]}"
|
||
else:
|
||
line += f" on {rd.peak_vector_sum_when_str}"
|
||
elif not histogram_when and rd.peak_vector_sum_time_s is not None:
|
||
line += f" At {rd.peak_vector_sum_time_s:.3f} sec."
|
||
|
||
# _draw_stats_table stashes the bbox bottom on the axes so we don't
|
||
# have to guess geometry. Falls back to a conservative default if
|
||
# the bbox approach hasn't run.
|
||
table_bottom_y = getattr(ax, "_stats_table_bottom", -0.10)
|
||
pvs_y = table_bottom_y - 0.04 # small gap below the table border
|
||
|
||
# Centered under the stats table for visual balance — looks intentional
|
||
# rather than offset. When the table is compacted (waveform layout), it
|
||
# occupies only the left portion of the axes, so center on the table's
|
||
# width rather than the full axes (which would push the line under the
|
||
# compliance chart). The original BW-replica had a "NA: Not Applicable"
|
||
# caption below this line; dropped because we use "—" for missing values.
|
||
table_w = getattr(ax, "_stats_table_width", 0.80)
|
||
if table_w < 0.79:
|
||
# Compacted (waveform) layout: left-align under the table, one point
|
||
# smaller, so the line clears the enlarged compliance chart's
|
||
# bottom-left tick labels on the right.
|
||
ax.text(0.0, pvs_y, line, fontsize=8, weight="bold",
|
||
ha="left", va="top", transform=ax.transAxes)
|
||
else:
|
||
ax.text(0.5, pvs_y, line, fontsize=9, weight="bold",
|
||
ha="center", va="top", transform=ax.transAxes)
|
||
|
||
|
||
def _draw_stats_table(
|
||
ax, rd: ReportData, rows_spec: list[tuple[str, str, str]],
|
||
*, bbox_width: float = 0.80, fontsize: float = 8,
|
||
col_widths: Optional[list[float]] = None,
|
||
) -> None:
|
||
"""Render a per-channel stats table (Tran/Vert/Long).
|
||
|
||
rows_spec: list of (label, field_name_in_channel_stats, unit_string)
|
||
|
||
``bbox_width`` / ``col_widths`` / ``fontsize`` let a caller compact the
|
||
table (the waveform layout packs it into the left half to clear the
|
||
compliance chart; the histogram layout keeps the wider defaults).
|
||
"""
|
||
if col_widths is None:
|
||
col_widths = [0.28, 0.14, 0.14, 0.14, 0.10]
|
||
headers = ["", "Tran", "Vert", "Long", ""]
|
||
ch_lookup = {c["name"]: c for c in rd.channel_stats}
|
||
|
||
def _cell(field, ch_name):
|
||
ch_rec = ch_lookup.get(ch_name, {})
|
||
val = ch_rec.get(field)
|
||
if val is None:
|
||
return "—"
|
||
if isinstance(val, float):
|
||
# ZC Freq is integer-formatted in BW; ">100 Hz" sentinel
|
||
# rendered as ">N" (val carries the threshold). Everything
|
||
# else gets 3 decimals.
|
||
if field == "zc_freq_hz":
|
||
prefix = ">" if ch_rec.get("zc_freq_above_range") else ""
|
||
return f"{prefix}{val:.0f}"
|
||
if field in ("sc_freq_hz", "sc_ratio"):
|
||
return f"{val:.1f}" # BW shows 1 decimal (7.5 Hz, 3.6)
|
||
return f"{val:.3f}"
|
||
return str(val)
|
||
|
||
table_data = [headers]
|
||
for label, field_name, unit in rows_spec:
|
||
table_data.append([
|
||
label,
|
||
_cell(field_name, "Tran"),
|
||
_cell(field_name, "Vert"),
|
||
_cell(field_name, "Long"),
|
||
unit,
|
||
])
|
||
# Pin the table's position+size via bbox so we know exactly where
|
||
# the bottom edge lands. Lets _draw_pvs_summary place the PVS line
|
||
# just below the table without guessing at row heights.
|
||
#
|
||
# bbox = [x, y, width, height] in axes coords. Header + data rows
|
||
# at row_h each; horizontal extent matches sum(colWidths).
|
||
n_rows = len(table_data) # header + data rows
|
||
row_h = 0.12 # axes-fraction per row (fits fontsize=8)
|
||
table_height = n_rows * row_h
|
||
table_bottom = 1.0 - table_height
|
||
tbl = ax.table(
|
||
cellText=table_data,
|
||
colWidths=col_widths,
|
||
cellLoc="left", edges="open",
|
||
bbox=[0.0, table_bottom, bbox_width, table_height],
|
||
)
|
||
tbl.auto_set_font_size(False)
|
||
tbl.set_fontsize(fontsize)
|
||
for j in range(5):
|
||
tbl[(0, j)].set_text_props(weight="bold", color="#555")
|
||
# Stash the bottom Y + width so _draw_pvs_summary can position itself.
|
||
ax._stats_table_bottom = table_bottom
|
||
ax._stats_table_width = bbox_width
|
||
|
||
|
||
def _channel_axis_color(ch: str) -> str:
|
||
return {"MicL": "#cc00cc", "Long": "#0066ff", "Vert": "#009933", "Tran": "#cc0000"}.get(ch, "#444")
|
||
|
||
|
||
def _draw_waveform_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
||
"""4-channel stacked waveform plot — Instantel printout order
|
||
(MicL on top, Tran on bottom), shared x-axis in SECONDS, trigger
|
||
triangle markers at t=0, '0.0' baseline label on right of each.
|
||
|
||
When sensor self-check traces are present (rd.sensor_check_waveforms), a
|
||
narrow "Sensor Check" strip of per-channel mini-plots is drawn to the right,
|
||
aligned to the lanes — matching Blastware's Event Report.
|
||
"""
|
||
from matplotlib.ticker import MaxNLocator
|
||
|
||
order = ["MicL", "Long", "Vert", "Tran"]
|
||
has_sc = bool(rd.sensor_check_waveforms)
|
||
if has_sc:
|
||
# main lanes + a narrow sensor-check strip column, flush against the
|
||
# main panel (BW shares the border — no gap), with the "0.0" baseline
|
||
# labels moved to the right of the strip. Proportions match BW's
|
||
# Event Report (main ~0.75 / strip ~0.10 of the panel width).
|
||
inner = gridspec_cell.subgridspec(4, 2, width_ratios=[1.0, 0.13],
|
||
wspace=0.0, hspace=0.0)
|
||
else:
|
||
inner = gridspec_cell.subgridspec(4, 1, hspace=0.0)
|
||
sr = rd.sample_rate_sps or 1024
|
||
# Convert ms-based time axis to seconds for the x-axis
|
||
dt_s = (rd.dt_ms or (1000.0 / sr)) / 1000.0
|
||
t0_s = (rd.t0_ms if rd.t0_ms is not None else 0.0) / 1000.0
|
||
|
||
# Shared geo scale across Long/Vert/Tran (matches the event modal + BW's
|
||
# single amp/div): all three geo lanes use ONE Y scale = the max |sample|
|
||
# across them (padded, floored), so relative amplitudes stay honest instead
|
||
# of each lane auto-zooming to its own peak. Mic keeps its own (psi) scale.
|
||
GEO_FLOOR_INS = 0.05
|
||
_geo_amax = 0.0
|
||
for _gch in ("Long", "Vert", "Tran"):
|
||
for _x in (rd.channels.get(_gch) or []):
|
||
_a = abs(_x)
|
||
if _a > _geo_amax:
|
||
_geo_amax = _a
|
||
geo_shared = max(_geo_amax * 1.10, GEO_FLOOR_INS)
|
||
|
||
main_axes = []
|
||
sc_axes = []
|
||
last_idx = len(order) - 1
|
||
for i, ch in enumerate(order):
|
||
ax = fig.add_subplot(inner[i, 0] if has_sc else inner[i])
|
||
main_axes.append(ax)
|
||
values = rd.channels.get(ch) or []
|
||
times = [t0_s + j * dt_s for j in range(len(values))]
|
||
|
||
if values:
|
||
color = _channel_axis_color(ch)
|
||
ax.plot(times, values, color=color, linewidth=0.5)
|
||
# Geo: one shared symmetric scale (honest relative amplitudes).
|
||
# Mic: symmetric on its own psi scale (different unit).
|
||
if ch != "MicL":
|
||
ax.set_ylim(-geo_shared, geo_shared)
|
||
else:
|
||
amax = max((abs(v) for v in values), default=0.001)
|
||
ax.set_ylim(-amax * 1.10, amax * 1.10)
|
||
|
||
# Channel label on the LEFT (matches BW)
|
||
ax.set_ylabel(ch, fontsize=8, rotation=0, ha="right", va="center",
|
||
color=_channel_axis_color(ch), weight="bold", labelpad=14)
|
||
# "0.0" baseline label on the RIGHT (BW convention). With the sensor-
|
||
# check strip attached, it goes to the right of the STRIP (drawn below);
|
||
# otherwise just outside the main lane.
|
||
if not has_sc:
|
||
ax.text(1.005, 0.5, "0.0", transform=ax.transAxes,
|
||
fontsize=7, color="#555", va="center", ha="left")
|
||
|
||
ax.grid(True, linestyle="--", linewidth=0.3, color="#bbb", alpha=0.6)
|
||
# Vertical dashed trigger line at t=0
|
||
ax.axvline(0.0, color="#cc0000", linestyle="--", linewidth=0.6, alpha=0.7)
|
||
# Zero baseline horizontal
|
||
ax.axhline(0.0, color=_channel_axis_color(ch), linestyle="-",
|
||
linewidth=0.4, alpha=0.5)
|
||
|
||
if i != last_idx:
|
||
ax.set_xticklabels([])
|
||
ax.tick_params(axis="x", length=0)
|
||
else:
|
||
ax.tick_params(axis="x", labelsize=7)
|
||
ax.tick_params(axis="y", labelsize=6)
|
||
# Stacked lanes touch, so the top/bottom y-tick labels of adjacent lanes
|
||
# would overprint at the shared boundary. Prune the extreme ticks so
|
||
# each boundary shows clean interior ticks (0.5 / 0.0 / -0.5) only.
|
||
ax.yaxis.set_major_locator(MaxNLocator(nbins=4, prune="both"))
|
||
|
||
# Sensor self-check mini-plot in the right strip (aligned to this lane).
|
||
if has_sc:
|
||
scx = fig.add_subplot(inner[i, 1])
|
||
sc_axes.append(scx)
|
||
sc_vals = rd.sensor_check_waveforms.get(ch) or []
|
||
if sc_vals:
|
||
_col = _channel_axis_color(ch)
|
||
# Faint zero baseline (BW draws the channel baseline through the
|
||
# strip) — reference for the one-sided geophone ring-downs.
|
||
scx.axhline(0.0, color=_col, linewidth=0.3, alpha=0.4)
|
||
scx.plot(range(len(sc_vals)), sc_vals, color=_col, linewidth=0.5)
|
||
# Fit the trace to the box (BW-style) rather than a symmetric
|
||
# scale: the geo self-checks are one-sided dips, so a symmetric
|
||
# scale would strand them in the bottom half with an empty top.
|
||
_lo, _hi = min(sc_vals), max(sc_vals)
|
||
_pad = 0.10 * ((_hi - _lo) or 1.0)
|
||
scx.set_ylim(_lo - _pad, _hi + _pad)
|
||
scx.set_xticks([]); scx.set_yticks([])
|
||
for _s in scx.spines.values():
|
||
_s.set_linewidth(0.4); _s.set_color("#999")
|
||
# "0.0" baseline label to the RIGHT of the strip (BW convention)
|
||
scx.text(1.10, 0.5, "0.0", transform=scx.transAxes,
|
||
fontsize=7, color="#555", va="center", ha="left")
|
||
|
||
# Trigger triangle marker ▼ above the top channel at t=0
|
||
top_ax = main_axes[0] # MicL
|
||
top_ax.plot([0], [top_ax.get_ylim()[1]], marker="v", color="black",
|
||
markersize=8, clip_on=False, zorder=10)
|
||
|
||
# "Sensor Check" caption under the strip (BW convention)
|
||
if has_sc and sc_axes:
|
||
pos = sc_axes[-1].get_position()
|
||
fig.text((pos.x0 + pos.x1) / 2, pos.y0 - 0.012, "Sensor Check",
|
||
fontsize=7, color="#555", ha="center", va="top")
|
||
|
||
# Compute scale-per-division for the footer (10 divs across the chart)
|
||
# and find peak geo amplitude for the geo amp/div setting.
|
||
total_s = times[-1] - times[0] if values else 0
|
||
div_s = total_s / 10 if total_s > 0 else 0
|
||
# Footer div value reflects the SHARED geo scale (so it's correct for all
|
||
# three lanes, not just whichever one happened to be checked first).
|
||
geo_amp_div = f"{(geo_shared * 2) / 10:.3f}" if _geo_amax > 0 else "—"
|
||
fig.text(
|
||
0.11, 0.030,
|
||
f"Time(Seconds) {div_s:.2f} sec/div Amplitude Geo: {geo_amp_div} in/s/div Mic: 0.001 psi(L)/div",
|
||
fontsize=7, color="#444", ha="left",
|
||
)
|
||
fig.text(
|
||
0.11, 0.018,
|
||
"Trigger = ▶━━━━━ ━━━━━━◀",
|
||
fontsize=7, color="#444", ha="left",
|
||
)
|
||
|
||
|
||
def _nice_geo_step(amax: float) -> float:
|
||
"""Pick a "nice" per-division step for the geo y-axis.
|
||
|
||
Geo LSB is 0.005 in/s — sub-LSB steps like 0.003/div are nonsense.
|
||
Quantize to the BW-style 1-2-5 sequence (0.005, 0.01, 0.025, 0.05,
|
||
…) and return the smallest step where 5 divisions >= amax, so the
|
||
top of the chart lands on a tick.
|
||
"""
|
||
if amax <= 0:
|
||
return 0.005
|
||
for step in (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0):
|
||
if step * 5 >= amax:
|
||
return step
|
||
return 10.0
|
||
|
||
|
||
def _draw_histogram_subplot(fig, gridspec_cell, rd: ReportData) -> None:
|
||
"""4-channel stacked histogram bar chart — per-interval peaks.
|
||
|
||
X-axis labeled with the actual times from rd.histogram_interval_times
|
||
when available; otherwise interval index.
|
||
|
||
The three geo channels share a single y-axis scale (a BW-style nice
|
||
multiple of the 0.005 in/s LSB) so bar heights are directly
|
||
comparable across channels. MicL has its own auto-scale.
|
||
"""
|
||
inner = gridspec_cell.subgridspec(4, 1, hspace=0.0)
|
||
order = ["MicL", "Long", "Vert", "Tran"]
|
||
last_idx = len(order) - 1
|
||
|
||
# X-axis: use absolute time labels if we have them, else interval index
|
||
have_times = bool(rd.histogram_interval_times)
|
||
|
||
# Shared geo scale: max across Tran/Vert/Long, quantized to a nice
|
||
# tick step. Used for ylim + the footer "Amplitude Geo: X in/s/div".
|
||
geo_amax = 0.0
|
||
for gch in ("Tran", "Vert", "Long"):
|
||
gv = rd.channels.get(gch) or []
|
||
if gv:
|
||
geo_amax = max(geo_amax, max(abs(x) for x in gv if x is not None))
|
||
geo_step = _nice_geo_step(geo_amax)
|
||
geo_top = geo_step * 5 # 5 divisions — top tick lands at this value
|
||
|
||
for i, ch in enumerate(order):
|
||
ax = fig.add_subplot(inner[i])
|
||
values = rd.channels.get(ch) or []
|
||
if values:
|
||
# Histograms record per-interval PEAK magnitudes — always
|
||
# non-negative. Codec output occasionally includes signed
|
||
# values when the underlying .h5 was scaled like a waveform;
|
||
# take the absolute value so the bars rise from zero.
|
||
abs_vals = [abs(v) if v is not None else 0 for v in values]
|
||
xs = np.arange(len(abs_vals))
|
||
color = _channel_axis_color(ch)
|
||
ax.bar(xs, abs_vals, color=color, width=0.85, linewidth=0)
|
||
if ch in ("Tran", "Vert", "Long"):
|
||
ax.set_ylim(0, geo_top)
|
||
ax.set_yticks([j * geo_step for j in range(6)])
|
||
else:
|
||
amax = max(abs_vals, default=0)
|
||
if amax > 0:
|
||
ax.set_ylim(0, amax * 1.10)
|
||
ax.set_ylabel(ch, fontsize=8, rotation=0, ha="right", va="center",
|
||
color=_channel_axis_color(ch), weight="bold", labelpad=14)
|
||
ax.text(1.005, 0.02, "0.0", transform=ax.transAxes,
|
||
fontsize=7, color="#555", va="bottom", ha="left")
|
||
ax.grid(True, axis="y", linestyle="--", linewidth=0.3, color="#bbb", alpha=0.6)
|
||
if i != last_idx:
|
||
ax.set_xticklabels([])
|
||
ax.tick_params(axis="x", length=0)
|
||
else:
|
||
if have_times and len(rd.histogram_interval_times) == len(values):
|
||
# Show 2-4 labels evenly spaced
|
||
n = len(values)
|
||
step = max(1, n // 4)
|
||
tick_positions = list(range(0, n, step))
|
||
ax.set_xticks(tick_positions)
|
||
ax.set_xticklabels([rd.histogram_interval_times[t] for t in tick_positions],
|
||
rotation=0, fontsize=6)
|
||
else:
|
||
ax.set_xlabel("Interval", fontsize=8)
|
||
ax.tick_params(axis="x", labelsize=7)
|
||
ax.tick_params(axis="y", labelsize=6)
|
||
|
||
# Footer scale info — histograms use minute/div. Reuses the shared
|
||
# geo_step computed above so the label matches the actual y-axis
|
||
# tick spacing on every subplot.
|
||
interval_str = rd.histogram_interval_size or "—"
|
||
geo_amp_div = f"{geo_step:.3f}"
|
||
fig.text(
|
||
0.11, 0.030,
|
||
f"Time {interval_str} /div Amplitude Geo: {geo_amp_div} in/s/div Mic: 0.001 psi(L)/div",
|
||
fontsize=7, color="#444", ha="left",
|
||
)
|