53 lines
2.3 KiB
Python
53 lines
2.3 KiB
Python
"""End-to-end ingest test: feed an IDFW + .txt to save_imported_idf in a tmp store."""
|
|
from __future__ import annotations
|
|
import sys
|
|
from pathlib import Path
|
|
import tempfile
|
|
import shutil
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO))
|
|
|
|
from sfm.waveform_store import WaveformStore
|
|
|
|
|
|
def main():
|
|
idfw = REPO / "tests/fixtures/THORDATA_example/THORDATA_example/UPMC Presby/UM11719/UM11719_20231219162723.IDFW"
|
|
txt = idfw.parent / "TXT" / f"{idfw.name}.txt"
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = WaveformStore(Path(td))
|
|
ev, rec = store.save_imported_idf(
|
|
idfw.read_bytes(),
|
|
idfw,
|
|
serial_hint=None,
|
|
idf_report_text=txt.read_text(errors="replace"),
|
|
)
|
|
print("=== Save result ===")
|
|
print(f" serial: {rec['serial']}")
|
|
print(f" filename: {rec['filename']}")
|
|
print(f" filesize: {rec['filesize']}")
|
|
print(f" h5: {rec['hdf5_filename']}")
|
|
print(f" sidecar: {rec['sidecar_filename']}")
|
|
print()
|
|
print("=== Event ===")
|
|
print(f" serial: {ev.serial if hasattr(ev,'serial') else '(n/a)'}")
|
|
print(f" timestamp: {ev.timestamp}")
|
|
print(f" sample_rate: {ev.sample_rate}")
|
|
print(f" record_type: {ev.record_type}")
|
|
print(f" rectime_sec: {ev.rectime_seconds}")
|
|
print(f" raw_samples: Tran={len(ev.raw_samples.get('Tran', [])) if ev.raw_samples else 0}, Vert={len(ev.raw_samples.get('Vert', [])) if ev.raw_samples else 0}, Long={len(ev.raw_samples.get('Long', [])) if ev.raw_samples else 0}, MicL={len(ev.raw_samples.get('MicL', [])) if ev.raw_samples else 0}")
|
|
if ev.peak_values:
|
|
print(f" peaks (txt): Tran={ev.peak_values.tran} Vert={ev.peak_values.vert} Long={ev.peak_values.long}")
|
|
print()
|
|
|
|
# Verify the h5 file actually got written
|
|
h5path = Path(td) / "UM11719" / f"{idfw.name}.h5"
|
|
print(f" h5 exists: {h5path.exists()} size={h5path.stat().st_size if h5path.exists() else 0}")
|
|
sidecar = Path(td) / "UM11719" / f"{idfw.name}.sfm.json"
|
|
print(f" sidecar exists:{sidecar.exists()} size={sidecar.stat().st_size if sidecar.exists() else 0}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|