From 75ac610c61cde6388ed072c4907fd8e903f89793 Mon Sep 17 00:00:00 2001 From: serversdown Date: Fri, 28 Aug 2026 05:23:30 +0000 Subject: [PATCH] fix(twins): interval-based histogram/waveform matching in find_twins (#102 sub-task 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A real trigger is recorded twice — as a triggered waveform (stamped at the trigger instant) and inside the scheduled histogram whose interval contains it (stamped at the 7am/7pm interval start). The two twins routinely differ by HOURS, so the old ±5-minute window in find_twins silently missed them — which broke review propagation (flagging one twin left its twin unflagged). Twins are now matched by: same serial + identical peak_vector_sum + OPPOSITE record type + the waveform's timestamp falling within the histogram's interval (bounded by the next same-serial histogram). Matching keys off record timestamps (not call-in/received times, which drift with field connectivity). window_seconds is retained but ignored. Rewrote test_find_twins + test_twin_propagation for the new contract (incl. the 75-min-apart UM12947 case, cross-type exclusion, containing-interval selection, open-ended latest interval). Full suite: 264 passed; the 16 failures are pre-existing (missing gitignored fixtures + a v0.26.0 codec case), unchanged from baseline. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf --- CHANGELOG.md | 11 ++++ sfm/database.py | 115 ++++++++++++++++++++++++--------- tests/test_find_twins.py | 76 ++++++++++++++++------ tests/test_twin_propagation.py | 35 +++++----- 4 files changed, 170 insertions(+), 67 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5284dc6..afa663f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ All notable changes to seismo-relay are documented here. ## [Unreleased] +### Fixed +- **Histogram/waveform twin matching is now interval-based** (`find_twins`). A real + trigger is recorded twice — as a triggered waveform (stamped at the trigger instant) + and inside the scheduled histogram whose interval contains it (stamped at the 7am/7pm + interval start) — so the two twins can be **hours apart**. The old ±5-minute window + silently missed them, which broke review propagation (flagging one twin didn't flag its + twin). Twins are now matched by same serial + identical `peak_vector_sum` + opposite + record type + the waveform falling within the histogram's interval (bounded by the next + same-serial histogram). `window_seconds` is retained but ignored. Fixes terra-view #102 + sub-task 2. + --- ## v0.26.0 — 2026-08-27 diff --git a/sfm/database.py b/sfm/database.py index 6986208..04c5d24 100644 --- a/sfm/database.py +++ b/sfm/database.py @@ -603,56 +603,107 @@ class SeismoDb: ).fetchall() return [dict(r) for r in rows] - def find_twins(self, event_id: str, *, window_seconds: int = 300) -> list[dict]: + def find_twins(self, event_id: str, *, window_seconds: int | None = None) -> list[dict]: """ - Find this event's histogram/waveform twin(s): rows sharing the same - serial and an identical peak_vector_sum, whose timestamp falls - within ``window_seconds`` of this event's timestamp. Excludes the - event itself. Returns [] if the event or any required field - (serial / peak_vector_sum / timestamp) is missing. + Find this event's histogram/waveform twin(s): the SAME physical event + recorded both as a scheduled histogram and as a triggered waveform. - Caveat: identical-PVS matching is a proxy for "same physical event - recorded twice," not a guarantee. In the rare case where the device - clamps/saturates PVS (clamped to sqrt(3) * geo_range), two distinct - saturated events on the same serial within the window can share the - same clamped PVS value and be matched as twins even though they are - different events. This is harmless in practice — false_trigger/ - reviewed_real are derived/index columns re-derivable from the - sidecar source of truth — but worth knowing if twin counts look - surprising on a saturated/clamped run. + A real trigger is captured twice — once as a triggered waveform (stamped + at the trigger instant) and once inside the scheduled histogram whose + interval contains it (stamped at the histogram's interval start, e.g. the + 7am/7pm call-in). The two can be HOURS apart in time yet report the same + serial and identical peak_vector_sum. Twins are therefore matched by: + + * same serial, + * identical peak_vector_sum, + * OPPOSITE record type (one histogram, one waveform), and + * the waveform's timestamp falls within the histogram's interval — + from a histogram's timestamp up to the next histogram (same serial). + + This replaces the old ±``window_seconds`` heuristic, which silently + missed twins more than a few minutes apart (a histogram's interval-start + stamp and the trigger instant routinely differ by hours). ``window_seconds`` + is still accepted for backward compatibility but is ignored. + + Returns [] if the event or a required field (serial / peak_vector_sum / + timestamp) is missing. + + Caveat: identical-PVS matching remains a proxy for "same physical event" + — if the device clamps/saturates PVS (to sqrt(3) * geo_range), two + distinct saturated events could share a PVS. The added opposite-type and + interval constraints make a false pairing far less likely than the old + time-window match, and false_trigger/reviewed_real stay re-derivable from + the sidecar source of truth. """ + def _parse(ts): + if not ts: + return None + try: + return datetime.datetime.fromisoformat(str(ts).replace(" ", "T")) + except ValueError: + return None + + def _is_hist(rt): + return str(rt or "").lower().startswith("hist") + row = self.get_event(event_id) if not row: return [] - serial = row.get("serial"); pvs = row.get("peak_vector_sum"); ts = row.get("timestamp") - if serial is None or pvs is None or not ts: + serial = row.get("serial"); pvs = row.get("peak_vector_sum") + t_target = _parse(row.get("timestamp")) + if serial is None or pvs is None or t_target is None: return [] - try: - t = datetime.datetime.fromisoformat(ts.replace(" ", "T")) - except ValueError: - return [] - lo = (t - datetime.timedelta(seconds=window_seconds)).isoformat() - hi = (t + datetime.timedelta(seconds=window_seconds)).isoformat() - with self._connect() as conn: - rows = conn.execute( - "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=? " - "AND timestamp BETWEEN ? AND ?", - (serial, event_id, pvs, lo, hi), - ).fetchall() - return [dict(r) for r in rows] + target_hist = _is_hist(row.get("record_type")) - def propagate_review_to_twins(self, event_id: str, *, window_seconds: int = 300) -> list[str]: + with self._connect() as conn: + cand_rows = [dict(r) for r in conn.execute( + "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=?", + (serial, event_id, pvs)).fetchall()] + hist_ts = [r["timestamp"] for r in conn.execute( + "SELECT timestamp FROM events WHERE serial=? AND lower(record_type) LIKE 'hist%'", + (serial,)).fetchall()] + + # Histogram interval-start times for this serial, sorted, to bound intervals. + starts = sorted(x for x in (_parse(t) for t in hist_ts) if x is not None) + + def _interval_end(h_start): + # The next histogram strictly after h_start bounds the interval; else open-ended. + for x in starts: + if x > h_start: + return x + return None + + def _covers(h_start, w_time): + end = _interval_end(h_start) + return h_start <= w_time and (end is None or w_time < end) + + twins = [] + for c in cand_rows: + if _is_hist(c.get("record_type")) == target_hist: + continue # twins are strictly cross-type (one histogram, one waveform) + c_time = _parse(c.get("timestamp")) + if c_time is None: + continue + h_start, w_time = (t_target, c_time) if target_hist else (c_time, t_target) + if _covers(h_start, w_time): + twins.append(c) + return twins + + def propagate_review_to_twins(self, event_id: str, *, window_seconds: int | None = None) -> list[str]: """ Copy this event's `false_trigger`/`reviewed_real` columns onto each of its histogram/waveform twins (see `find_twins`), so flagging one twin flags both. Returns the list of twin ids updated. + + ``window_seconds`` is accepted for backward compatibility but ignored; + twin matching is now interval-based (see `find_twins`). """ row = self.get_event(event_id) if not row: return [] ft = 1 if row.get("false_trigger") else 0 real = 1 if row.get("reviewed_real") else 0 - twins = self.find_twins(event_id, window_seconds=window_seconds) + twins = self.find_twins(event_id) moved = [] with self._connect() as conn: for tw in twins: diff --git a/tests/test_find_twins.py b/tests/test_find_twins.py index d0b356c..db501c1 100644 --- a/tests/test_find_twins.py +++ b/tests/test_find_twins.py @@ -1,33 +1,71 @@ -import datetime +import sqlite3 from sfm.database import SeismoDb from minimateplus.models import Event, Timestamp -def _ins(db, key, serial, pvs, ts): +def _ins(db, key, serial, pvs, ts, record_type="Waveform"): 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"])) + c.execute("UPDATE events SET peak_vector_sum=?, record_type=? WHERE id=?", + (pvs, record_type, row["id"])) return row["id"] -def test_find_twins_matches_same_serial_pvs_near_time(tmp_path): +def _ts(hour, minute, second=0, day=25): + return Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=day, + hour=hour, minute=minute, second=second) + + +def test_histogram_and_waveform_twin_across_hours(tmp_path): + # The real UM12947 case: histogram stamped at its 7pm interval start, the + # triggered waveform 75 min later — same serial + identical PVS. The old + # ±5-min window missed this; interval matching catches it, both directions. 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} + hist_pm = _ins(db, "01110001", "BE1", 0.4763, _ts(19, 31, 17), "Histogram") + wave = _ins(db, "01110002", "BE1", 0.4763, _ts(20, 46, 44), "Waveform") + _ins(db, "01110003", "BE1", 0.0100, _ts(7, 0, 0, day=26), "Histogram") # bounds the interval + assert {r["id"] for r in db.find_twins(hist_pm)} == {wave} + assert {r["id"] for r in db.find_twins(wave)} == {hist_pm} + + +def test_same_type_not_twinned(tmp_path): + # Two waveforms, same serial + PVS, seconds apart → NOT twins (cross-type only). + db = SeismoDb(tmp_path / "s.db") + a = _ins(db, "01110001", "BE1", 0.4763, _ts(20, 19, 5), "Waveform") + _ins(db, "01110002", "BE1", 0.4763, _ts(20, 19, 45), "Waveform") + assert db.find_twins(a) == [] + + +def test_waveform_matches_only_the_containing_interval(tmp_path): + # Two overnight intervals with the same PVS; a waveform in the SECOND interval + # must twin with that histogram, never the first — even though PVS matches both. + db = SeismoDb(tmp_path / "s.db") + h1 = _ins(db, "01110001", "BE1", 0.4763, _ts(19, 0, 0, day=25), "Histogram") + h2 = _ins(db, "01110002", "BE1", 0.4763, _ts(7, 0, 0, day=26), "Histogram") + w = _ins(db, "01110003", "BE1", 0.4763, _ts(8, 0, 0, day=26), "Waveform") + assert {r["id"] for r in db.find_twins(w)} == {h2} + assert w not in {r["id"] for r in db.find_twins(h1)} + + +def test_different_pvs_not_twinned(tmp_path): + db = SeismoDb(tmp_path / "s.db") + h = _ins(db, "01110001", "BE1", 0.4763, _ts(19, 0, 0), "Histogram") + _ins(db, "01110002", "BE1", 0.9999, _ts(20, 0, 0), "Waveform") # different PVS + assert db.find_twins(h) == [] + + +def test_open_ended_latest_interval(tmp_path): + # A waveform after the latest histogram (nothing bounds the interval) still twins. + db = SeismoDb(tmp_path / "s.db") + h = _ins(db, "01110001", "BE1", 0.4763, _ts(19, 0, 0), "Histogram") + w = _ins(db, "01110002", "BE1", 0.4763, _ts(23, 30, 0), "Waveform") + assert {r["id"] for r in db.find_twins(h)} == {w} + + +def test_missing_fields_returns_empty(tmp_path): + db = SeismoDb(tmp_path / "s.db") + assert db.find_twins("nonexistent-id") == [] diff --git a/tests/test_twin_propagation.py b/tests/test_twin_propagation.py index c3f7386..6d08d17 100644 --- a/tests/test_twin_propagation.py +++ b/tests/test_twin_propagation.py @@ -3,31 +3,34 @@ from sfm.database import SeismoDb from minimateplus.models import Event, Timestamp -def _ins(db, key, serial, pvs, ts): +def _ins(db, key, serial, pvs, ts, record_type="Waveform"): 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] with sqlite3.connect(db.db_path) as c: - c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"])) + c.execute("UPDATE events SET peak_vector_sum=?, record_type=? WHERE id=?", + (pvs, record_type, row["id"])) return row["id"] -def test_propagate_copies_flags_to_twins(tmp_path): +def _ts(hour, minute, second=0, day=25): + return Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=day, + hour=hour, minute=minute, second=second) + + +def test_propagate_copies_flags_across_hours_apart_twins(tmp_path): + # Flagging the waveform FT propagates to its histogram twin 75 min earlier + # (the interval matcher pairs them; the old ±5-min window would have missed it). 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) - other = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=44) + hist = _ins(db, "01110001", "BE1", 0.4763, _ts(19, 31, 17), "Histogram") + wave = _ins(db, "01110002", "BE1", 0.4763, _ts(20, 46, 44), "Waveform") # twin, 75 min later + other = _ins(db, "01110003", "BE1", 0.9999, _ts(20, 20, 0), "Waveform") # different pvs - primary_id = _ins(db, "01110001", "BE1", 0.4763, base) - twin_id = _ins(db, "01110002", "BE1", 0.4763, twin) # twin: same serial+pvs, 40s apart - non_twin_id = _ins(db, "01110003", "BE1", 0.9999, other) # near time but different pvs + db.update_event_review(wave, {"false_trigger": True}) + moved = db.propagate_review_to_twins(wave) - db.update_event_review(primary_id, {"false_trigger": True}) - moved = db.propagate_review_to_twins(primary_id) - - assert twin_id in moved - assert db.get_event(twin_id)["false_trigger"] == 1 - assert db.get_event(non_twin_id)["false_trigger"] == 0 + assert hist in moved + assert db.get_event(hist)["false_trigger"] == 1 + assert db.get_event(other)["false_trigger"] == 0