92 lines
3.8 KiB
Python
92 lines
3.8 KiB
Python
"""End-to-end Thor IDFH histogram report PDF rendering."""
|
|
from __future__ import annotations
|
|
import sys
|
|
import tempfile
|
|
import json
|
|
import datetime
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
from sfm.waveform_store import WaveformStore
|
|
from sfm import report_pdf
|
|
|
|
|
|
class FakeDb:
|
|
def __init__(self, event):
|
|
self.event = event
|
|
|
|
def get_event(self, _id):
|
|
return self.event
|
|
|
|
|
|
def main():
|
|
# Use the multi-interval IDFH (81 + trigger row)
|
|
idfh = REPO / "tests/fixtures/THORDATA_example/THORDATA_example/UPMC Presby/UM13981/UM13981_20220805075441.IDFH"
|
|
txt = idfh.parent / "TXT" / f"{idfh.name}.txt"
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = WaveformStore(Path(td))
|
|
ev, rec = store.save_imported_idf(
|
|
idfh.read_bytes(),
|
|
idfh,
|
|
idf_report_text=txt.read_text(errors="replace"),
|
|
)
|
|
print(f"save_imported_idf: h5={rec['hdf5_filename']}, sidecar={rec['sidecar_filename']}")
|
|
|
|
sc_path = Path(td) / "UM13981" / f"{idfh.name}.sfm.json"
|
|
sc = json.loads(sc_path.read_text())
|
|
bw = sc.get("bw_report", {})
|
|
hist = bw.get("histogram", {})
|
|
print(f" bw_report.histogram.start: {hist.get('start')}")
|
|
print(f" bw_report.histogram.stop: {hist.get('stop')}")
|
|
print(f" bw_report.histogram.n_intervals: {hist.get('n_intervals')}")
|
|
print(f" bw_report.histogram.interval_size: {hist.get('interval_size')}")
|
|
print(f" bw_report.histogram.interval_size_s: {hist.get('interval_size_s')}")
|
|
print(f" bw_report.peaks.tran.ppv_ips: {bw.get('peaks', {}).get('tran', {}).get('ppv_ips')}")
|
|
|
|
ts = ev.timestamp
|
|
ts_iso = None
|
|
if ts is not None:
|
|
try:
|
|
ts_iso = datetime.datetime(ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second).isoformat()
|
|
except Exception:
|
|
pass
|
|
fake_row = {
|
|
"serial": "UM13981",
|
|
"blastware_filename": rec["filename"],
|
|
"record_type": "Histogram",
|
|
"timestamp": ts_iso,
|
|
"sample_rate": ev.sample_rate,
|
|
"project": ev.project_info.project if ev.project_info else None,
|
|
"client": ev.project_info.client if ev.project_info else None,
|
|
"operator": ev.project_info.operator if ev.project_info else None,
|
|
"sensor_location": ev.project_info.sensor_location if ev.project_info else None,
|
|
"created_at": None,
|
|
}
|
|
rd = report_pdf.gather_report_data(FakeDb(fake_row), store, event_id="hist-1")
|
|
|
|
print()
|
|
print("=== ReportData (histogram) ===")
|
|
print(f" is_histogram: {rd.is_histogram}")
|
|
print(f" histogram_start: {rd.histogram_start_str}")
|
|
print(f" histogram_stop: {rd.histogram_stop_str}")
|
|
print(f" histogram_n_intervals: {rd.histogram_n_intervals}")
|
|
print(f" histogram_interval_size:{rd.histogram_interval_size}")
|
|
print(f" histogram_interval_times[:3]: {rd.histogram_interval_times[:3]}")
|
|
print(f" histogram_interval_times[-2:]: {rd.histogram_interval_times[-2:]}")
|
|
print(f" channel_stats: {len(rd.channel_stats)} rows")
|
|
for cs in rd.channel_stats:
|
|
print(f" {cs['name']}: PPV={cs['ppv_ips']} ZC={cs['zc_freq_hz']} peak_date={cs['peak_date']} peak_time={cs['peak_time']}")
|
|
|
|
pdf_bytes = report_pdf.render_event_report_pdf(rd)
|
|
out_path = REPO / "analysis_idf" / "thor_report_idfh.pdf"
|
|
out_path.write_bytes(pdf_bytes)
|
|
print()
|
|
print(f" PDF written: {out_path} ({len(pdf_bytes)} bytes)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|