serversdownandClaude Opus 4.8
c0cf6547d9
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
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
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"
|