christian/status-indicators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ALARMS key stops every auto-fix too, and says so

Browse Source 
Muted, nothing fixes itself at all: no Spotify exception (while_muted is gone) and no any-hour
READY fix; holds still fix and the Sunday stream-audio failover still switches. The key reads
ALARMS OFF / AUTO-FIX OFF / TIL 12AM, and its note says both. The settings page's switch says the same.

The mute's end is worked out at the press and kept: a watchdog that was not running at 23:59 no
longer leaves the alarms muted until the next schedule change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
christian
2026-10-19 14:46:20 -04:00
co-authored by Claude Opus 5.5
parent 5d918d60af
commit b41b8a49d2
8 changed files with 121 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
status_panel.py
+78 -20
View File
@@ -814,6 +814,13 @@ def in_windows(windows, now):
return any(window_holds(w, day, hm, DAYS[dt.weekday() - 1]) for w in windows)
def clock12(t):
"""12AM, 6:30PM: a time on a key, the way people say it."""
dt = datetime.fromtimestamp(t)
h = dt.hour % 12 or 12
return ("%d" % h if dt.minute == 0 else "%d:%02d" % (h, dt.minute)) + ("AM" if dt.hour < 12 else "PM")
def windows_phrase(windows):
"""Windows in words, for log headers and notes: "on Sundays 06:30-13:30"."""
parts = []
@@ -886,32 +893,71 @@ class State:
class ServiceMode:
"""On inside a scheduled window; a button press overrides until the next window edge."""
"""On inside a scheduled window; a button press overrides until the next window edge. The
edge is worked out at the press and kept, so the override ends there even when the watchdog
was not running at that minute (a restart over 23:59 left the alarms muted for good)."""
def __init__(self, windows, state):
self.windows = windows
self.state = state if isinstance(state, State) else State(state)
saved = self.state.get("service") or {}
self.override, self.last_sched = saved.get("override"), saved.get("last_sched")
self.edge = saved.get("edge")
def scheduled(self, now):
return in_windows(self.windows, now)
def next_change(self, now):
"""The first minute after `now` whose schedule differs from now's, or None (within 8 days)."""
sched, t = self.scheduled(now), int(now // 60) * 60 + 60
while t < now + 8 * 86400:
if self.scheduled(t) != sched:
return t
t += 60
return None
def on(self, now):
sched = self.scheduled(now)
if sched != self.last_sched:
self.last_sched, self.override = sched, None
if self.override is not None and self.edge is None:
self.edge = self.next_change(now) # an override saved before edges were kept
self._save()
if sched != self.last_sched or (self.override is not None and self.edge and now >= self.edge):
self.last_sched, self.override, self.edge = sched, None, None
self._save()
return self.override if self.override is not None else sched
def back_at(self, now):
"""When a manual override (ALARMS pressed) ends by itself, as a timestamp, or None. The
override holds until the schedule next changes; from then on the schedule rules, so the
state flips there, or at the change after (at NA: cleared 23:59, alarms back 00:00).
Looked up once a minute."""
if self.override is None:
return None
minute = int(now // 60)
cached = getattr(self, "_back", None)
if cached and cached[0] == (minute, self.override):
return cached[1]
sched, t, found = self.scheduled(now), minute * 60 + 60, None
while t < now + 8 * 86400:
s = self.scheduled(t)
if s != sched:
sched = s
if s != self.override:
found = t
break
t += 60
self._back = ((minute, self.override), found)
return found
def set(self, value, now):
sched = self.scheduled(now)
self.last_sched = sched
self.override = None if value == sched else value
self.edge = self.next_change(now) if self.override is not None else None
self._save()
def _save(self):
self.state.set("service", {"override": self.override, "last_sched": self.last_sched})
self.state.set("service", {"override": self.override, "last_sched": self.last_sched, "edge": self.edge})
# --- Engine ------------------------------------------------------------------
@@ -984,7 +1030,7 @@ class Engine:
self.auto_watch = {} # "BUTTON FAULT" auto-fix went after -> when (to log when it clears)
self.auto_paused = False
self.auto_on = False # inside the window (for the "auto-fix on" line)
self.auto_muted = False # ...with ALARMS muted, so only the "while_muted" fixes run
self.auto_muted = False # ALARMS muted last round (nothing fixes itself then)
self.auto_last = 0.0
self.deck_ips = {} # HyperDeck connection -> last address Companion reported
self.now = 0.0
@@ -1498,18 +1544,27 @@ class Engine:
if not a or now - self.auto_last < 5 or self.stopping:
return []
self.auto_last = now
# ALARMS muted (maintenance) stops auto-fix, except the harmless fixes in "while_muted"
# (playing the Spotify playlist): those keep running inside the window.
# The ALARMS key is the on-the-fly switch: muted, nothing is fixed by itself at all (holds
# still work). It comes back by itself with the alarms (2026-10-19: the user's choice).
muted = not self.service.on(now)
hours = in_windows(a.get("windows", []), now)
# A fix marked "anytime" (writing the timecode into Planning Center: nothing in the room
# notices) runs by itself at any hour, not only in the auto-fix hours, muted or not.
# A fix marked "anytime" (writing the timecode into Planning Center) runs by itself at any
# hour, not only in the auto-fix hours - but never while muted, like every other.
anytime = {f for f in a.get("fixes", []) if self.cfg["fixes"].get(f, {}).get("anytime")
and not self.cfg["fixes"].get(f, {}).get("never_auto") and not self.cfg["fixes"].get(f, {}).get("disabled")}
keep = (set(a.get("while_muted", [])) | anytime) if muted else None
if not hours or (muted and not keep):
if muted:
if self.autolog and (self.auto_on or (anytime and not self.auto_muted)):
self.autolog.write(now, "PANEL", "auto-fix off (ALARMS muted)")
self.auto_on, self.auto_muted = False, True
self.auto_tries.clear()
self.auto_seen.clear()
return []
if self.auto_muted and self.autolog and not hours and anytime:
self.autolog.write(now, "PANEL", "ALARMS back on: the any-hour fixes (%s) run again" % ", ".join(sorted(anytime)))
self.auto_muted = False
if not hours:
if self.auto_on and self.autolog:
self.autolog.write(now, "PANEL", "auto-fix off (%s)" % ("ALARMS muted" if hours else "end of the window"))
self.autolog.write(now, "PANEL", "auto-fix off (end of the window)")
self.auto_on = False
for ident in [i for i in self.auto_tries if not (i[0] == "fix" and i[1] in anytime)]:
del self.auto_tries[ident] # the window's fixes start fresh next time
@@ -1517,11 +1572,10 @@ class Engine:
if not anytime:
return []
else:
if (not self.auto_on or self.auto_muted != muted) and self.autolog: # opens today's file, so it can be watched from the start
if not self.auto_on and self.autolog: # opens today's file, so it can be watched from the start
ends = [w["end"] for w in a.get("windows", []) if in_windows([w], now)]
self.autolog.write(now, "PANEL", "auto-fix on%s%s" % (" until " + ends[0] if ends else "",
" (ALARMS muted: only %s)" % ", ".join(sorted(keep)) if muted else ""))
self.auto_on, self.auto_muted = True, muted
self.autolog.write(now, "PANEL", "auto-fix on%s" % (" until " + ends[0] if ends else ""))
self.auto_on = True
net_down = any(self.trackers.get("net:" + k) and self.trackers["net:" + k].confirmed
and self.trackers["net:" + k].result.level == ERR for k in ("link", "gateway"))
if net_down or (self.burst and (self.burst["cleared"] is None
@@ -1560,8 +1614,6 @@ class Engine:
for ident, kind, what, chk in self.fix_plan(t["id"], snap, now, a.get("after_seconds", 20), True, levels):
if not ((kind == "fix" and what in gentle) or (kind == "restart" and restarts)):
continue # not gentle: stays a hold
if keep is not None and not (kind == "fix" and what in keep):
continue # ALARMS muted: only the "while_muted" fixes
self.auto_seen[ident] = now
tries, last = self.auto_tries.get(ident, (0, 0.0))
if tries >= a.get("max_tries", 2):
@@ -1926,8 +1978,14 @@ def render(engine, service_on):
out[tid] = colored(tile["label"] + "\n" + note["text"], note["level"])
continue
if kind == "service":
manual = engine.service.override is not None
text = tile["label"] + "\n" + ("ON" if service_on else "OFF") + ("\nmanual" if manual else "")
# One key, two switches: ALARMS muted also stops every auto-fix. Says both, and when
# they come back by themselves after a press.
word = "ON" if service_on else "OFF"
fix_word = "OFF" if (not service_on or (engine.cfg.get("auto_fix") or {}).get("disabled")) else "ON"
text = "%s %s\nAUTO-FIX %s" % (tile["label"], word, fix_word)
back = engine.service.back_at(engine.now)
if back:
text += "\nTIL " + clock12(back)
out[tid] = colored(text, "service" if service_on else "off")
continue
if kind == "master":