70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
from __future__ import annotations
|
|
from pathlib import Path
|
|
|
|
from sfm.database import SeismoDb
|
|
from minimateplus.models import Event, Timestamp, PeakValues
|
|
|
|
|
|
def _event(waveform_key="0111abcd"):
|
|
ev = Event(index=0)
|
|
ev._waveform_key = bytes.fromhex(waveform_key)
|
|
ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0,
|
|
month=6, day=25, hour=8, minute=50, second=0)
|
|
ev.record_type = "Waveform"
|
|
ev.peak_values = PeakValues(tran=0.075, vert=0.220, long=0.045,
|
|
peak_vector_sum=0.231, micl=0.01)
|
|
return ev
|
|
|
|
|
|
def test_insert_stores_shape_from_record(tmp_path: Path):
|
|
db = SeismoDb(tmp_path / "s.db")
|
|
ev = _event()
|
|
rec = {ev._waveform_key.hex(): {
|
|
"filename": "F.CE0W", "filesize": 10,
|
|
"shape_crest_factor": 34.0, "shape_near_peak_count": 3,
|
|
"shape_sample_count": 1024, "shape_axis": "Long"}}
|
|
db.insert_events([ev], serial="BE1", waveform_records=rec)
|
|
row = db.query_events(serial="BE1")[0]
|
|
assert row["shape_crest_factor"] == 34.0
|
|
assert row["shape_near_peak_count"] == 3
|
|
assert row["shape_axis"] == "Long"
|
|
|
|
|
|
def test_upsert_refreshes_shape_from_record(tmp_path: Path):
|
|
db = SeismoDb(tmp_path / "s.db")
|
|
ev = _event()
|
|
rec1 = {ev._waveform_key.hex(): {
|
|
"shape_crest_factor": 10.0, "shape_near_peak_count": 1,
|
|
"shape_sample_count": 512, "shape_axis": "Tran"}}
|
|
db.insert_events([ev], serial="BE1", waveform_records=rec1)
|
|
|
|
ev2 = _event()
|
|
rec2 = {ev2._waveform_key.hex(): {
|
|
"shape_crest_factor": 22.5, "shape_near_peak_count": 7,
|
|
"shape_sample_count": 2048, "shape_axis": "Vert"}}
|
|
db.insert_events([ev2], serial="BE1", waveform_records=rec2)
|
|
|
|
row = db.query_events(serial="BE1")[0]
|
|
assert row["shape_crest_factor"] == 22.5
|
|
assert row["shape_near_peak_count"] == 7
|
|
assert row["shape_sample_count"] == 2048
|
|
assert row["shape_axis"] == "Vert"
|
|
|
|
|
|
def test_upsert_preserves_shape_when_record_missing(tmp_path: Path):
|
|
db = SeismoDb(tmp_path / "s.db")
|
|
ev = _event()
|
|
rec1 = {ev._waveform_key.hex(): {
|
|
"shape_crest_factor": 10.0, "shape_near_peak_count": 1,
|
|
"shape_sample_count": 512, "shape_axis": "Tran"}}
|
|
db.insert_events([ev], serial="BE1", waveform_records=rec1)
|
|
|
|
ev2 = _event() # re-import with no waveform_records (e.g. sample missing)
|
|
db.insert_events([ev2], serial="BE1")
|
|
|
|
row = db.query_events(serial="BE1")[0]
|
|
assert row["shape_crest_factor"] == 10.0
|
|
assert row["shape_near_peak_count"] == 1
|
|
assert row["shape_sample_count"] == 512
|
|
assert row["shape_axis"] == "Tran"
|