Resi: checks keep quiet while a Resi event starts or stops (weekly schedule + PCO special services)

User 2026-10-26: three services each have a Resi event that starts and stops on its own; the cloud,
decoder and Resi audio checks threw false alarms then. resi_events (weekly, from studio.resi.io) and the
special services' times from PCO; checks marked quiet resi_events show OK (EVENT CHANGE) within 10 min of
an event's start or end. The encoder's ping and web checks still alarm.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

config.json

@@ -2415,7 +2415,8 @@
           "type": "atem",
           "part": "silence",
           "msg": "RESI AUDIO SILENT",
-          "fix": "atem_audio"
+          "fix": "atem_audio",
+          "quiet": "resi_events"
         },
         {
           "fix": "atem_camera",
@@ -2460,7 +2461,8 @@
         {
           "conn": "RESI_Decoder",
           "tag": "DECODER",
-          "type": "conn"
+          "type": "conn",
+          "quiet": "resi_events"
         },
         {
           "host": "10.1.10.203",
@@ -2481,12 +2483,14 @@
         {
           "conn": "resi-studio-cues",
           "type": "conn",
-          "amber": true
+          "amber": true,
+          "quiet": "resi_events"
         },
         {
           "name": "CLOUD",
           "type": "http",
-          "url": "https://central.resi.io/"
+          "url": "https://central.resi.io/",
+          "quiet": "resi_events"
         }
       ],
       "id": "resi",
@@ -2756,5 +2760,46 @@
       "ms": 3000,
       "page": 98
     }
-  ]
+  ],
+  "resi_events": {
+    "weekly": [
+      {
+        "days": [
+          "sun"
+        ],
+        "start": "07:30",
+        "end": "09:15"
+      },
+      {
+        "days": [
+          "sun"
+        ],
+        "start": "09:30",
+        "end": "11:15"
+      },
+      {
+        "days": [
+          "sun"
+        ],
+        "start": "11:30",
+        "end": "13:15"
+      },
+      {
+        "days": [
+          "mon"
+        ],
+        "start": "09:00",
+        "end": "10:00"
+      },
+      {
+        "days": [
+          "wed"
+        ],
+        "start": "16:30",
+        "end": "22:00"
+      }
+    ],
+    "margin_minutes": 10,
+    "pco_specials": true
+  }
 }

readiness.py

@@ -1860,6 +1860,11 @@ def look(cfg, now=None, data_dir=None, sources=None):
                 lines["start"] = eval_start(rc, services, src["start"](services), now)
             except Exception as e:  # noqa: BLE001
                 failed(["start"], e)
+        if ((cfg.get("resi_events") or {}).get("pco_specials")) and not sources:
+            try:
+                LAST["resi_specials"] = resi_specials(rc, now)
+            except Exception as e:  # noqa: BLE001 - only the quiet windows miss the special services
+                log.warning("READY resi specials: %s", e)
         if rc.get("resi_vox"):
             try:
                 try:
@@ -2038,6 +2043,40 @@ def upcoming_services(rc, now, fresh=False):
     return out
 
 
+RESI_SPECIALS = {}
+
+
+def resi_specials(rc, now):
+    """(start, end) of the special services' service times (First Tuesday, Flourish...) in the next 8 days, from
+    PCO's plan times - Resi streams those too, and the status page keeps quiet while they start and stop."""
+    if RESI_SPECIALS.get("key") == int(now // 1800):
+        return RESI_SPECIALS["list"]
+    auth, home = pco_login(rc)
+    types, _ = pco_get(auth, "/service_types?per_page=100", pages=3)
+    by_id = {x["id"]: x for x in types}
+    parent_of = lambda x: ((((x or {}).get("relationships") or {}).get("parent") or {}).get("data") or {}).get("id")
+    name_of = lambda x: ((x or {}).get("attributes") or {}).get("name") or ""
+    want = re.compile(rc.get("start_types", START_TYPES), re.I)
+    skip = re.compile(rc.get("start_skip_types", START_SKIP), re.I)
+    out = []
+    for x in types:
+        if x["id"] == home or parent_of(x) != parent_of(by_id.get(home)) or not want.search(name_of(x)) or skip.search(name_of(x)):
+            continue
+        plans, _ = pco_get(auth, "/service_types/%s/plans?filter=future&order=sort_date&per_page=3" % x["id"], pages=1)
+        for p in plans:
+            t = parse_iso((p.get("attributes") or {}).get("sort_date"))
+            if t is None or t > now + 8 * 86400:
+                continue
+            times, _ = pco_get(auth, "/service_types/%s/plans/%s/plan_times" % (x["id"], p["id"]), pages=1)
+            for pt in times:
+                a = pt.get("attributes") or {}
+                s, e = parse_iso(a.get("starts_at")), parse_iso(a.get("ends_at"))
+                if a.get("time_type") == "service" and s and e:
+                    out.append((s, e))
+    RESI_SPECIALS.update(key=int(now // 1800), list=out)
+    return out
+
+
 def start_params(rc, services, mode):
     projects = rc.get("projects", "")
     return {"mode": mode, "projects": projects, "site_folder": rc.get("site_folder", ""),

schema.py

@@ -101,6 +101,9 @@ field("service_windows", "list", [{"days": list(sp.DAYS), "start": "00:00", "end
 field("service_windows[].days", "days", None, "", "Schedules", required=True)
 field("service_windows[].start", "hhmm", None, "", "Schedules", required=True)
 field("service_windows[].end", "hhmm", None, "", "Schedules", required=True)
+field("resi_events", "dict", None, "Resi's scheduled events (studio.resi.io), so the checks marked quiet \"resi_events\" keep quiet "
+      "while an event starts or stops: {\"weekly\": [{\"days\": [\"sun\"], \"start\": \"07:30\", \"end\": \"09:15\"}, ...], "
+      "\"margin_minutes\": 10, \"pco_specials\": true (special services' times from Planning Center)}.", "Site")
 field("debounce_seconds", "int", 3, "A fault must last this long before it shows (one bad reading is not an "
       "alarm).", "Schedules", advanced=True, lo=0, hi=60)
 field("flap", "dict", {}, "Repeated faults show amber 'xN'.", "Schedules", advanced=True)
@@ -293,6 +296,8 @@ field("tiles[].checks[].fix", "str", None, "What holding the key runs for this c
 field("tiles[].checks[].tag", "str", None, "Names the device when one key covers several.", "Keys")
 field("tiles[].checks[].service_only", "bool", False, "Legacy: only during a service (has no effect while alarms "
       "are armed all day).", "Keys", advanced=True)
+field("tiles[].checks[].quiet", "enum", None, "Keep quiet while a Resi event starts or stops (resi_events).", "Keys", advanced=True,
+      options=["resi_events"])
 field("tiles[].checks[].debounce_seconds", "int", None, "Wait before this check's fault shows.", "Keys", advanced=True,
       lo=0, hi=60)
 field("tiles[].checks[].amber", "bool", False, "A Companion connection whose errors show amber, not red (a cloud helper "
@@ -529,7 +534,7 @@ CHECKS = {
     "ready_manual": {"fields": ["item", "msg"], "subs": ["done"], "identity": ["item"], "module": None,
                   "question": "Is the weekend ready?", "words": "A line of the weekly checklist a person taps done."},
 }
-COMMON_CHECK_FIELDS = ["type", "fix", "tag", "service_only", "debounce_seconds", "disabled"]
+COMMON_CHECK_FIELDS = ["type", "fix", "tag", "service_only", "debounce_seconds", "disabled", "quiet"]
 FIX_KINDS = sp.FIX_KINDS
 FIX_KIND_FIELDS = {
     "ssh": ["host", "run", "timeout", "reconnect", "conn"], "local": ["run", "timeout", "reconnect", "conn"],

status_panel.py

@@ -1421,7 +1421,8 @@ class Engine:
     def evaluate(self, snap, now, service_on):
         ctx = {"now": now, "service": service_on, "any_recording": False, "dante_all": {},
                "ready_due": in_windows(self.ready_cfg.get("due", []), now), "ready_done": self.ready_done(now),
-               "bypassed": self.bypass_names()[0], "bypassed_fixes": self.bypass_fixes(), "autofix_windows": (self.cfg.get("auto_fix") or {}).get("windows", [])}
+               "bypassed": self.bypass_names()[0], "bypassed_fixes": self.bypass_fixes(),
+               "resi_quiet": resi_quiet(self.cfg, now), "autofix_windows": (self.cfg.get("auto_fix") or {}).get("windows", [])}
         for t in all_tiles(self.cfg):
             for chk in t.get("checks", []):
                 if chk["type"] == "dante":
@@ -1439,6 +1440,8 @@ class Engine:
                     results = CHECKS[chk["type"]](chk, snap, ctx)
                     if isinstance(chk.get("fix"), str) and chk["fix"] in ctx["bypassed_fixes"]:
                         results = [(sub, Result(OK)) for sub, _ in results]     # switched off by its bypass key
+                    if chk.get("quiet") == "resi_events" and ctx.get("resi_quiet"):
+                        results = [(sub, Result(OK, info="EVENT CHANGE")) for sub, _ in results]
                 except Exception as e:  # a bad check must not take the panel down
                     log.exception("check %s/%s failed", t["id"], chk["type"])
                     results = [("error", Result(WARN, "CHECK BUG"))]
@@ -3497,6 +3500,38 @@ class AtemWatcher:
 ATEM_WATCHER = None
 
 
+def resi_event_edges(cfg, now):
+    """Start and end times (unix) of the Resi events around `now`: the weekly ones in config "resi_events" and
+    the special services PCO lists (readiness.LAST["resi_specials"], fetched with the READY page)."""
+    re_ = cfg.get("resi_events") or {}
+    edges = []
+    d0 = datetime.fromtimestamp(now)
+    for back in (1, 0, -1):
+        day = datetime(d0.year, d0.month, d0.day) - timedelta(days=back)
+        for w in re_.get("weekly") or []:
+            if day.strftime("%a").lower()[:3] not in w.get("days", []):
+                continue
+            for k in ("start", "end"):
+                h, m = (int(x) for x in w[k].split(":"))
+                edges.append(time.mktime((day + timedelta(hours=h, minutes=m)).timetuple()))
+    if re_.get("pco_specials", True):
+        try:
+            import readiness
+            edges += [t for ev in readiness.LAST.get("resi_specials") or [] for t in ev]
+        except Exception:  # noqa: BLE001 - no READY page here
+            pass
+    return edges
+
+
+def resi_quiet(cfg, now):
+    """True within "margin_minutes" (10) of a Resi event starting or stopping: the cloud, the decoder and the
+    Resi audio change over then, and those checks keep quiet (2026-10-26: false alarms at every event change)."""
+    if not cfg.get("resi_events"):
+        return False
+    margin = (cfg["resi_events"].get("margin_minutes", 10)) * 60
+    return any(abs(now - t) <= margin for t in resi_event_edges(cfg, now))
+
+
 def check_atem(chk, snap, ctx):
     """chk["part"] "audio" or "camera": the switcher must match the restore file. "silence": a metered input
     (Resi) has had no audio at all for a while; only while a HyperDeck records (a service is on): between services

tests/test_status_panel.py

@@ -359,6 +359,24 @@ class TestChecks(unittest.TestCase):
         tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, self.now, 4)
         self.assertNotEqual(level(tiles["resi"]), "err")
 
+    def test_resi_quiet_while_an_event_starts_or_stops(self):
+        """2026-10-26: the Resi cloud / decoder checks threw false alarms at every Resi event change."""
+        sunday = datetime(2026, 10, 25, 9, 0).timestamp()
+        self.assertFalse(sp.resi_quiet(CONFIG, sunday))                                  # between 7:30-9:15's margins
+        self.assertTrue(sp.resi_quiet(CONFIG, sunday + 14 * 60))                          # 9:14, the 9:15 end
+        self.assertTrue(sp.resi_quiet(CONFIG, sunday + 38 * 60))                          # 9:38, the 9:30 start
+        tuesday = datetime(2026, 11, 3, 18, 35).timestamp()
+        self.assertFalse(sp.resi_quiet(CONFIG, tuesday))
+        import readiness
+        readiness.LAST["resi_specials"] = [(datetime(2026, 11, 3, 18, 30).timestamp(), datetime(2026, 11, 3, 21, 0).timestamp())]
+        try:
+            self.assertTrue(sp.resi_quiet(CONFIG, tuesday))                               # First Tuesday from PCO
+        finally:
+            readiness.LAST.pop("resi_specials", None)
+        self.snap.conns["resi-studio-cues"] = {"id": "r", "enabled": True, "category": "error", "message": "boom"}
+        tiles, _ = run_steps(sp.Engine(CONFIG), self.snap, sunday + 38 * 60, 4)
+        self.assertEqual(tiles["resi"]["x"], "RESI\nEVENT CHANGE")
+
     def test_resi_audio_silent(self):
         """An input stuck at the meter floor (2026-10-25: the Resi AJA passed no Dante audio and nothing said so):
         red while a deck records (a service is on); between services the Resi send is quiet on purpose."""