""" Micromate (Series IV / Thor) native data models. These are the right-shaped dataclasses for Thor data — Thor measures the microphone in dB(L) directly, so this model carries ``mic_pspl_dbl`` rather than the pseudo-``psi`` shoehorn that ``minimateplus.PeakValues`` uses for Series III BW data. The ingest pipeline today goes: .IDFW.txt → parse_idf_report() → dict dict → IdfEvent.from_report() → IdfEvent (typed) IdfEvent → IdfEvent.to_minimateplus_event() → shape DB / sidecar machinery expects The ``to_minimateplus_event()`` bridge is a temporary boundary — when we crack the binary IDF codec and have richer per-event data to store, the DB schema will grow Series-IV-specific columns and the bridge will shrink or disappear. """ from __future__ import annotations import datetime from dataclasses import dataclass, field from typing import Any, Dict, Optional, Tuple # ── IdfReport ───────────────────────────────────────────────────────────────── @dataclass class IdfReport: """Typed wrapper around the dict returned by ``parse_idf_report``. All fields optional — Thor's exporter is permissive and some IDF .txt files (especially histograms) omit fields that waveform sidecars include. Use ``.raw`` for any field this dataclass hasn't surfaced yet (the parser keeps every recognised key in the raw dict). """ # Identity / kind serial_number: Optional[str] = None event_type: Optional[str] = None # "Full Waveform" | "Full Histogram" event_datetime: Optional[datetime.datetime] = None filename: Optional[str] = None # echoed by Thor's exporter # Sampling / timing sample_rate: Optional[int] = None # samples/sec record_time_sec: Optional[float] = None pre_trigger_sec: Optional[float] = None # Geophone peaks (in/s) tran_ppv: Optional[float] = None vert_ppv: Optional[float] = None long_ppv: Optional[float] = None peak_vector_sum: Optional[float] = None # Microphone — Thor's native unit is dB(L), NOT psi. mic_pspl_dbl: Optional[float] = None # Zero-crossing frequencies (Hz) tran_zc_freq: Optional[float] = None vert_zc_freq: Optional[float] = None long_zc_freq: Optional[float] = None mic_zc_freq: Optional[float] = None # Per-channel time of peak (sec, since event start) tran_time_of_peak: Optional[float] = None vert_time_of_peak: Optional[float] = None long_time_of_peak: Optional[float] = None mic_time_of_peak: Optional[float] = None # Derived per-channel motion tran_peak_acceleration: Optional[float] = None # g vert_peak_acceleration: Optional[float] = None long_peak_acceleration: Optional[float] = None tran_peak_displacement: Optional[float] = None # in vert_peak_displacement: Optional[float] = None long_peak_displacement: Optional[float] = None # Operator-supplied strings (Thor's TitleString1..4 → semantic slots) project: Optional[str] = None # TitleString1 client: Optional[str] = None # TitleString2 operator: Optional[str] = None # TitleString3 notes: Optional[str] = None # TitleString4 / PostEventNote setup: Optional[str] = None # setup file name # Sensor self-check results tran_test_passed: Optional[bool] = None vert_test_passed: Optional[bool] = None long_test_passed: Optional[bool] = None mic_test_passed: Optional[bool] = None # Device-fixed metadata firmware_version: Optional[str] = None calibration_text: Optional[str] = None battery_volts: Optional[float] = None # Original parser dict — preserves every recognised key (including # raw unit-suffixed strings) for forward-compatible field access. raw: Dict[str, Any] = field(default_factory=dict, repr=False) @classmethod def from_dict(cls, d: Dict[str, Any]) -> "IdfReport": """Build an IdfReport from the dict returned by ``parse_idf_report``.""" ed = d.get("event_datetime") if isinstance(ed, str): try: ed = datetime.datetime.fromisoformat(ed) except ValueError: ed = None return cls( serial_number = d.get("serial_number"), event_type = d.get("event_type"), event_datetime = ed if isinstance(ed, datetime.datetime) else None, filename = d.get("filename"), sample_rate = d.get("sample_rate"), record_time_sec = d.get("record_time_sec"), pre_trigger_sec = d.get("pre_trigger_sec"), tran_ppv = d.get("tran_ppv"), vert_ppv = d.get("vert_ppv"), long_ppv = d.get("long_ppv"), peak_vector_sum = d.get("peak_vector_sum"), mic_pspl_dbl = d.get("mic_ppv"), # parser names it mic_ppv (legacy) tran_zc_freq = d.get("tran_zc_freq"), vert_zc_freq = d.get("vert_zc_freq"), long_zc_freq = d.get("long_zc_freq"), mic_zc_freq = d.get("mic_zc_freq"), tran_time_of_peak = d.get("tran_time_of_peak"), vert_time_of_peak = d.get("vert_time_of_peak"), long_time_of_peak = d.get("long_time_of_peak"), mic_time_of_peak = d.get("mic_time_of_peak"), tran_peak_acceleration = d.get("tran_peak_acceleration"), vert_peak_acceleration = d.get("vert_peak_acceleration"), long_peak_acceleration = d.get("long_peak_acceleration"), tran_peak_displacement = d.get("tran_peak_displacement"), vert_peak_displacement = d.get("vert_peak_displacement"), long_peak_displacement = d.get("long_peak_displacement"), project = d.get("project"), client = d.get("client"), operator = d.get("operator"), notes = d.get("notes"), setup = d.get("setup"), tran_test_passed = d.get("tran_test_passed"), vert_test_passed = d.get("vert_test_passed"), long_test_passed = d.get("long_test_passed"), mic_test_passed = d.get("mic_test_passed"), firmware_version = d.get("version"), calibration_text = d.get("calibration_text"), battery_volts = d.get("battery_volts"), raw = d, ) # ── IdfPeaks / IdfProjectInfo / IdfSensorCheck (narrow grouping types) ─────── @dataclass class IdfPeaks: """Geophone + mic peak values for one Thor event. Native Thor units.""" transverse_ips: Optional[float] = None # in/s vertical_ips: Optional[float] = None # in/s longitudinal_ips: Optional[float] = None # in/s peak_vector_sum_ips: Optional[float] = None # in/s mic_pspl_dbl: Optional[float] = None # dB(L) @dataclass class IdfProjectInfo: """Operator-supplied strings from Thor's TitleString1..4.""" project: Optional[str] = None client: Optional[str] = None operator: Optional[str] = None notes: Optional[str] = None setup: Optional[str] = None @dataclass class IdfSensorCheck: """Per-channel pass/fail from Thor's self-test.""" tran: Optional[bool] = None vert: Optional[bool] = None long: Optional[bool] = None mic: Optional[bool] = None # ── IdfEvent ───────────────────────────────────────────────────────────────── @dataclass class IdfEvent: """A single Thor / Micromate Series IV event. Built from a parsed .IDFW.txt or .IDFH.txt sidecar via ``IdfEvent.from_report()``. The filename is the authoritative source for serial + timestamp + kind; the .txt provides device-authoritative peak values, frequencies, project strings, sensor self-check, firmware, calibration. """ # Identity serial: str timestamp: datetime.datetime kind: str # "Waveform" | "Histogram" filename: str # device-native binary filename, e.g. "UM11719_20231219163444.IDFW" # Sampling / timing sample_rate: Optional[int] = None record_time_sec: Optional[float] = None pre_trigger_sec: Optional[float] = None # Peaks peaks: IdfPeaks = field(default_factory=IdfPeaks) # Per-channel frequencies (Hz) tran_zc_freq: Optional[float] = None vert_zc_freq: Optional[float] = None long_zc_freq: Optional[float] = None mic_zc_freq: Optional[float] = None # Project strings project_info: IdfProjectInfo = field(default_factory=IdfProjectInfo) # Sensor self-check sensor_check: IdfSensorCheck = field(default_factory=IdfSensorCheck) # Device-fixed firmware_version: Optional[str] = None calibration_text: Optional[str] = None battery_volts: Optional[float] = None # The full parsed report — preserves anything not surfaced as a typed field report: IdfReport = field(default_factory=IdfReport) @classmethod def from_report( cls, report: Any, filename: str, ) -> "IdfEvent": """Build an IdfEvent from a parsed report (dict or IdfReport) and the device-native binary filename. The filename is authoritative for serial + timestamp + kind: Thor's filenames are literal ``_.`` and the device's own clock is the canonical event timestamp. If the report carries an ``event_datetime`` that differs from what's in the filename, the report wins (it has finer-grained device-reported time-of-trigger semantics). """ from .idf_ascii_report import parse_event_filename # Normalise input to IdfReport if isinstance(report, IdfReport): rep = report elif isinstance(report, dict): rep = IdfReport.from_dict(report) else: raise TypeError( f"report must be IdfReport or dict; got {type(report).__name__}" ) # Filename → (serial, timestamp, kind). Required — fall back to # report-supplied values only if filename parsing fails. parsed = parse_event_filename(filename) if parsed is not None: fn_serial, fn_ts, fn_kind = parsed kind = "Histogram" if fn_kind == "IDFH" else "Waveform" else: fn_serial = rep.serial_number or "UNKNOWN" fn_ts = rep.event_datetime or datetime.datetime(1970, 1, 1) kind = "Waveform" if (rep.event_type or "").lower().startswith("full waveform") else "Histogram" # Prefer report's event_datetime (device-authoritative) over the filename. ts = rep.event_datetime or fn_ts serial = rep.serial_number or fn_serial return cls( serial=serial, timestamp=ts, kind=kind, filename=filename, sample_rate=rep.sample_rate, record_time_sec=rep.record_time_sec, pre_trigger_sec=rep.pre_trigger_sec, peaks=IdfPeaks( transverse_ips = rep.tran_ppv, vertical_ips = rep.vert_ppv, longitudinal_ips = rep.long_ppv, peak_vector_sum_ips = rep.peak_vector_sum, mic_pspl_dbl = rep.mic_pspl_dbl, ), tran_zc_freq=rep.tran_zc_freq, vert_zc_freq=rep.vert_zc_freq, long_zc_freq=rep.long_zc_freq, mic_zc_freq=rep.mic_zc_freq, project_info=IdfProjectInfo( project=rep.project, client=rep.client, operator=rep.operator, notes=rep.notes, setup=rep.setup, ), sensor_check=IdfSensorCheck( tran=rep.tran_test_passed, vert=rep.vert_test_passed, long=rep.long_test_passed, mic=rep.mic_test_passed, ), firmware_version=rep.firmware_version, calibration_text=rep.calibration_text, battery_volts=rep.battery_volts, report=rep, ) # ── Bridge to minimateplus shape (for the existing DB / sidecar paths) ── def to_minimateplus_event(self, waveform_key: bytes) -> Any: """Project this Thor event into the shape ``minimateplus.Event`` carries, so it can flow through the existing ``SeismoDb.insert_events()`` and ``event_to_sidecar_dict()`` machinery without those code paths needing to know about Thor. Caveats of the bridge: - ``mic_ppv`` on the produced Event carries Thor's dB(L) value verbatim — the UI distinguishes via the ``device_family`` column (Phase 1). Don't run the BW psi→dBL converter on Series IV rows. - Many Thor-specific fields (Peak Acceleration / Displacement, sensor self-check, calibration) don't have a slot in ``Event``. The full IdfReport is preserved on the ``.sfm.json`` sidecar under ``extensions.idf_report`` via ``save_imported_idf`` — that's the source of truth for them. """ from minimateplus.models import ( Event, PeakValues, ProjectInfo, Timestamp, ) ts_obj = Timestamp( raw=bytes(9), flag=0, year=self.timestamp.year, unknown_byte=0, month=self.timestamp.month, day=self.timestamp.day, hour=self.timestamp.hour, minute=self.timestamp.minute, second=self.timestamp.second, ) pv = PeakValues( tran=self.peaks.transverse_ips, vert=self.peaks.vertical_ips, long=self.peaks.longitudinal_ips, micl=self.peaks.mic_pspl_dbl, # dB(L) — see caveat above peak_vector_sum=self.peaks.peak_vector_sum_ips, ) pi = ProjectInfo( setup_name=self.project_info.setup, project=self.project_info.project, client=self.project_info.client, operator=self.project_info.operator, sensor_location=None, # Thor folds location into project string notes=self.project_info.notes, ) ev = Event( index=0, timestamp=ts_obj, sample_rate=self.sample_rate, peak_values=pv, project_info=pi, record_type=self.kind, rectime_seconds=self.record_time_sec, ) ev._waveform_key = waveform_key return ev