34 lines
1.7 KiB
Python
34 lines
1.7 KiB
Python
import datetime
|
|
from sfm.database import SeismoDb
|
|
from minimateplus.models import Event, Timestamp
|
|
|
|
|
|
def _ins(db, key, serial, pvs, ts):
|
|
ev = Event(index=0)
|
|
ev._waveform_key = bytes.fromhex(key)
|
|
ev.timestamp = ts
|
|
# peak_vector_sum comes from peak_values; simplest: insert then UPDATE pvs directly
|
|
db.insert_events([ev], serial=serial)
|
|
row = [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]
|
|
import sqlite3
|
|
with sqlite3.connect(db.db_path) as c:
|
|
c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"]))
|
|
return row["id"]
|
|
|
|
|
|
def test_find_twins_matches_same_serial_pvs_near_time(tmp_path):
|
|
db = SeismoDb(tmp_path / "s.db")
|
|
base = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=5)
|
|
twin = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=45)
|
|
far = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=21, minute=0, second=0)
|
|
# d needs a timestamp distinct from `twin` (UNIQUE(serial, timestamp) would
|
|
# otherwise collide with b and UPSERT onto its row instead of inserting a
|
|
# new one) while staying near `base` in time.
|
|
near = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=44)
|
|
a = _ins(db, "01110001", "BE1", 0.4763, base)
|
|
b = _ins(db, "01110002", "BE1", 0.4763, twin) # twin: same serial+pvs, 40s apart
|
|
c = _ins(db, "01110003", "BE1", 0.4763, far) # same pvs but >window away
|
|
d = _ins(db, "01110004", "BE1", 0.9999, near) # near time but different pvs
|
|
ids = {r["id"] for r in db.find_twins(a, window_seconds=300)}
|
|
assert ids == {b}
|