feat(ft): optional false_trigger_reason ("offset" etc.) as an FT subtype

A reason records *why* an event is a false trigger. It is optional (plain
FT flags still record no reason) and is a subtype of the FT flag: setting a
reason implies false_trigger=1, and the reason is cleared whenever FT ends
up 0 (confirm-real, clear-FT, set_false_trigger(false)). Twin propagation
carries the reason to the histogram/waveform twin alongside the FT flag.

New nullable `false_trigger_reason TEXT` column (schema + _migrate ADD
COLUMN only — not the Migration-1 rebuild). 7 tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf
This commit is contained in:
2026-09-03 20:53:57 +00:00
co-authored by Claude Opus 4.8
parent 4cf0fda804
commit c0cf6547d9
2 changed files with 125 additions and 7 deletions
+22 -7
View File
@@ -82,6 +82,7 @@ CREATE TABLE IF NOT EXISTS events (
record_type TEXT, -- "single_shot" | "continuous"
false_trigger INTEGER NOT NULL DEFAULT 0, -- 0=no, 1=yes (manual flag)
reviewed_real INTEGER NOT NULL DEFAULT 0, -- 0=no, 1=operator-confirmed real (mutually exclusive with false_trigger)
false_trigger_reason TEXT, -- optional FT cause ("offset", ...); NULL = none. Only meaningful when false_trigger=1.
blastware_filename TEXT, -- event file within waveform store; extension is per-event (AB0T encodes timestamp)
blastware_filesize INTEGER, -- bytes; NULL if no event file saved
a5_pickle_filename TEXT, -- "<filename>.a5.pkl" sidecar
@@ -234,6 +235,7 @@ class SeismoDb:
("shape_offset_pre", "REAL"),
("shape_offset_spread", "REAL"),
("reviewed_real", "INTEGER NOT NULL DEFAULT 0"),
("false_trigger_reason", "TEXT"),
):
if col not in existing_cols:
log.info("_migrate: events ADD COLUMN %s %s", col, ddl)
@@ -713,9 +715,9 @@ class SeismoDb:
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.
Copy this event's `false_trigger`/`reviewed_real`/`false_trigger_reason`
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`).
@@ -725,12 +727,16 @@ class SeismoDb:
return []
ft = 1 if row.get("false_trigger") else 0
real = 1 if row.get("reviewed_real") else 0
# The reason is a subtype of the FT flag — carry it only when the source
# is actually a false trigger, so a confirmed-real twin never keeps one.
reason = row.get("false_trigger_reason") if ft else None
twins = self.find_twins(event_id)
moved = []
with self._connect() as conn:
for tw in twins:
conn.execute("UPDATE events SET false_trigger=?, reviewed_real=? WHERE id=?",
(ft, real, tw["id"]))
conn.execute(
"UPDATE events SET false_trigger=?, reviewed_real=?, false_trigger_reason=? WHERE id=?",
(ft, real, reason, tw["id"]))
moved.append(tw["id"])
return moved
@@ -751,7 +757,7 @@ class SeismoDb:
)
else:
cur = conn.execute(
"UPDATE events SET false_trigger=0 WHERE id=?",
"UPDATE events SET false_trigger=0, false_trigger_reason=NULL WHERE id=?",
(event_id,),
)
return cur.rowcount > 0
@@ -845,7 +851,8 @@ class SeismoDb:
return False
has_ft = "false_trigger" in review
has_real = "reviewed_real" in review
if not has_ft and not has_real:
has_reason = "false_trigger_reason" in review
if not has_ft and not has_real and not has_reason:
# Nothing derived to update; just confirm the row exists.
with self._connect() as conn:
row = conn.execute(
@@ -858,11 +865,19 @@ class SeismoDb:
sets["false_trigger"] = 1 if review.get("false_trigger") else 0
if has_real:
sets["reviewed_real"] = 1 if review.get("reviewed_real") else 0
if has_reason:
reason = review.get("false_trigger_reason") or None
sets["false_trigger_reason"] = reason
if reason: # a reason is a subtype of FT → implies FT
sets["false_trigger"] = 1
# mutual exclusivity: a true in one forces the other column to 0
if sets.get("false_trigger") == 1:
sets["reviewed_real"] = 0
if sets.get("reviewed_real") == 1:
sets["false_trigger"] = 0
# the reason is only meaningful while flagged FT — clear it if FT ends up 0
if sets.get("false_trigger") == 0:
sets["false_trigger_reason"] = None
assign = ", ".join(f"{k}=?" for k in sets)
params = list(sets.values()) + [event_id]
with self._connect() as conn:
+103
View File
@@ -0,0 +1,103 @@
import sqlite3
from sfm.database import SeismoDb
from minimateplus.models import Event, Timestamp
def _ev(db, key="0111aaaa", serial="BE1"):
ev = Event(index=0)
ev._waveform_key = bytes.fromhex(key)
ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0,
month=6, day=25, hour=8, minute=0, second=0)
ev.record_type = "Waveform"
db.insert_events([ev], serial=serial)
return [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]["id"]
def test_flag_offset_reason_implies_ft(tmp_path):
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger_reason": "offset"})
row = db.get_event(eid)
assert row["false_trigger"] == 1 # a reason is a subtype of FT
assert row["false_trigger_reason"] == "offset"
assert row["reviewed_real"] == 0
def test_plain_ft_leaves_reason_null(tmp_path):
# Reason is OPTIONAL — flagging FT without one records no reason.
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger": True})
row = db.get_event(eid)
assert row["false_trigger"] == 1
assert row["false_trigger_reason"] is None
def test_confirm_real_clears_reason(tmp_path):
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger_reason": "offset"})
db.update_event_review(eid, {"reviewed_real": True})
row = db.get_event(eid)
assert row["reviewed_real"] == 1
assert row["false_trigger"] == 0
assert row["false_trigger_reason"] is None
def test_clear_ft_clears_reason(tmp_path):
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger_reason": "offset"})
db.update_event_review(eid, {"false_trigger": False})
row = db.get_event(eid)
assert row["false_trigger"] == 0
assert row["false_trigger_reason"] is None
def test_set_false_trigger_false_clears_reason(tmp_path):
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger_reason": "offset"})
assert db.set_false_trigger(eid, False) is True
row = db.get_event(eid)
assert row["false_trigger"] == 0
assert row["false_trigger_reason"] is None
def test_reason_can_be_cleared_without_clearing_ft(tmp_path):
# Setting reason to None removes the reason but leaves the FT flag intact.
db = SeismoDb(tmp_path / "s.db")
eid = _ev(db)
db.update_event_review(eid, {"false_trigger_reason": "offset"})
db.update_event_review(eid, {"false_trigger_reason": None})
row = db.get_event(eid)
assert row["false_trigger"] == 1
assert row["false_trigger_reason"] is None
def _ts(h, m, d=25):
return Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0,
month=2, day=d, hour=h, minute=m, second=0)
def test_offset_reason_propagates_to_twin(tmp_path):
# Flag a waveform as offset → its histogram twin also becomes FT with reason=offset.
db = SeismoDb(tmp_path / "s.db")
def ins(key, ts, rt):
ev = Event(index=0); ev._waveform_key = bytes.fromhex(key); ev.timestamp = ts
db.insert_events([ev], serial="BE1")
rid = [r for r in db.query_events(serial="BE1") if r["waveform_key"] == key][0]["id"]
with sqlite3.connect(db.db_path) as c:
c.execute("UPDATE events SET peak_vector_sum=0.4763, record_type=? WHERE id=?", (rt, rid))
return rid
hist = ins("01110001", _ts(19, 31), "Histogram") # interval start
wave = ins("01110002", _ts(20, 46), "Waveform") # trigger inside the interval
db.update_event_review(wave, {"false_trigger_reason": "offset"})
db.propagate_review_to_twins(wave)
row = db.get_event(hist)
assert row["false_trigger"] == 1
assert row["false_trigger_reason"] == "offset"