Spotify: SPOTIFY key watches the Christian EDM 2026 playlist during services; hold, restart key and auto-fix play it

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

status_panel.py

@@ -152,6 +152,11 @@ class Snapshot:
         self.conn_hosts = {}     # Companion connection label -> its "host" setting
         self.prefs = {}          # (preferences domain, key) -> value as "defaults read" prints it
         self.atem = None         # atem_watch.js's latest report
+        self.spotify = None      # {"running", "state", "track", "name", "artist"} on NA Ingest
+        self.spotify_recent = deque(maxlen=3)   # last distinct tracks played
+        self.spotify_tracks = set()             # songs known to be in the service playlist
+        self.spotify_complete = False           # ...and that is all of them
+        self.spotify_tracks_time = 0.0
         self.midi_time = 0.0
         self.mdns = set()        # Bonjour _apple-midi._udp instance names
         self.mdns_time = 0.0
@@ -418,6 +423,31 @@ def check_http(chk, snap, ctx):
     return [("http", Result(OK))]
 
 
+def check_spotify(chk, snap, ctx):
+    """Spotify on NA Ingest must be playing the service playlist during the check's "windows"
+    (Sunday service). Outside them the button just says what it is doing."""
+    sp_ = snap.spotify
+    if sp_ is None:
+        return [("spotify", Result(OK))]                  # not read yet
+    state = "CLOSED" if not sp_.get("running") else (sp_.get("state") or "?").upper()
+    if not in_windows(chk.get("windows", []), ctx["now"]):
+        return [("spotify", Result(OK, info=state))]
+    if not sp_.get("running"):
+        return [("spotify", Result(ERR, "SPOTIFY CLOSED", named=True))]
+    if state != "PLAYING":
+        return [("spotify", Result(ERR, "NOT PLAYING"))]
+    known = snap.spotify_tracks
+    if known:
+        recent = list(snap.spotify_recent)
+        if snap.spotify_complete:
+            wrong = bool(recent) and recent[-1] not in known
+        else:                                           # only part of the playlist is known: 3 strangers in a row
+            wrong = len(recent) >= 3 and all(t not in known for t in recent[-3:])
+        if wrong:
+            return [("spotify", Result(ERR, "WRONG PLAYLIST"))]
+    return [("spotify", Result(OK, info="PLAYING"))]
+
+
 def check_pref(chk, snap, ctx):
     """An app setting on NA Ingest, e.g. ATEM Software Control's "Enable MIDI control" must stay off."""
     value = snap.prefs.get((chk["domain"], chk["key"]))
@@ -507,7 +537,7 @@ def check_remote(chk, snap, ctx):
 CHECKS = {
     "conn": check_conn, "hyperdeck": check_hyperdeck, "dante": check_dante, "shure": check_shure,
     "psm": check_psm, "tascam": check_tascam, "ping": check_ping, "launchd": check_launchd,
-    "process": check_process, "http": check_http, "disk": check_disk, "midi_peer": check_midi_peer, "pref": check_pref,
+    "process": check_process, "http": check_http, "disk": check_disk, "midi_peer": check_midi_peer, "pref": check_pref, "spotify": check_spotify,
     "mdns": check_mdns, "midi_rx": check_midi_rx, "remote": check_remote, "artnet": check_artnet,
 }
 
@@ -1205,9 +1235,10 @@ class Engine:
                     self.autolog.write(now, label, "problem: %s - fixing: %s (try %d of %d)" % (
                         ", ".join(faults) or "?", "; ".join(self.describe(k, w) for _, k, w, _ in plan),
                         try_no, a.get("max_tries", 2)))
+                started = time.time()
                 self.run_plan(t["id"], plan, snap, comp, now, auto=True,
-                              on_done=(lambda outcome, label=label: self.autolog.write(time.time(), label, "-> " + outcome))
-                              if self.autolog else None)
+                              on_done=(lambda outcome, label=label, started=started: self.autolog.write(
+                                  now + (time.time() - started), label, "-> " + outcome)) if self.autolog else None)
                 done.append(t["id"])
         forget = a.get("forget_seconds", 600)
         for ident in list(self.auto_tries):
@@ -1562,6 +1593,49 @@ def poll_conn_hosts(snap, cfg):
             log.warning("could not read connection addresses from Companion: %s", e)
 
 
+SPOTIFY_STATE = ('tell application "Spotify" to return (player state as text) & "|" & (id of current track) '
+                 '& "|" & (name of current track) & "|" & (artist of current track)')
+
+
+def playlist_tracks(playlist_uri):
+    """(set of track uris, complete?) from Spotify's public pages: the embed preview lists up to
+    100 songs, the playlist page says how many there are."""
+    pid = playlist_uri.split(":")[-1]
+    req = lambda url: urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}),
+                                             timeout=10, context=_UNVERIFIED).read().decode("utf-8", "replace")
+    m = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
+                  req("https://open.spotify.com/embed/playlist/" + pid))
+    entity = json.loads(m.group(1))["props"]["pageProps"]["state"]["data"]["entity"]
+    tracks = {t["uri"] for t in entity.get("trackList", []) if t.get("uri")}
+    n = re.search(r'music:song_count" content="(\d+)"', req("https://open.spotify.com/playlist/" + pid))
+    return tracks, bool(n) and len(tracks) >= int(n.group(1))
+
+
+def poll_spotify(snap, cfg):
+    checks = [c for t in cfg["tiles"] for c in t.get("checks", []) if c["type"] == "spotify"]
+    if not checks:
+        return
+    if time.time() - snap.spotify_tracks_time > 6 * 3600:            # the playlist changes weekly
+        try:
+            snap.spotify_tracks, snap.spotify_complete = playlist_tracks(checks[0]["playlist"])
+        except (OSError, ValueError, KeyError, AttributeError, TypeError) as e:
+            log.warning("Spotify playlist list: %s", e)
+        snap.spotify_tracks_time = time.time()
+    # Never "tell" Spotify unless it is running: that would open it.
+    running = subprocess.run(["/usr/bin/pgrep", "-xq", "Spotify"], timeout=5).returncode == 0
+    if not running:
+        snap.spotify = {"running": False}
+        return
+    r = subprocess.run(["/usr/bin/osascript", "-e", SPOTIFY_STATE], capture_output=True, text=True, timeout=10)
+    if r.returncode != 0:
+        snap.spotify = {"running": True, "state": None, "error": r.stderr.strip()[-120:]}
+        return
+    parts = (r.stdout.strip().split("|") + ["", "", "", ""])[:4]
+    snap.spotify = {"running": True, "state": parts[0], "track": parts[1], "name": parts[2], "artist": parts[3]}
+    if parts[1] and (not snap.spotify_recent or snap.spotify_recent[-1] != parts[1]):
+        snap.spotify_recent.append(parts[1])
+
+
 def poll_midi(snap, cfg):
     import coremidi
     snap.midi, snap.midi_time = coremidi.sessions(), time.time()
@@ -2222,6 +2296,7 @@ def start_pollers(cfg, snap, stop, data_dir=None):
         Periodic("variables", 1.0, lambda: poll_variables(c2, snap, fast), stop),
         Periodic("slow-variables", 3.0, lambda: poll_variables(c3, snap, slow), stop),
         Periodic("conn-hosts", 60.0, lambda: poll_conn_hosts(snap, cfg), stop),
+        Periodic("spotify", 3.0, lambda: poll_spotify(snap, cfg), stop),
         Periodic("pings", 3.0, lambda: poll_pings(snap, ping_hosts(cfg, snap), iface), stop),
         Periodic("midi", 0.5, lambda: poll_midi(snap, cfg), stop),
         Periodic("system", 10.0, lambda: poll_system(snap, cfg), stop),