[multiview] Update to v0.3.0

plugins/multiview/__init__.py

@@ -9,6 +9,7 @@
 import logging
 import os
 import socket
+import threading
 
 logger = logging.getLogger(__name__)
 
@@ -19,7 +20,7 @@
 
 PLUGIN_DB_KEY = "multiview"
 DEFAULT_SERVER_PORT = 9292
-DEFAULT_SERVER_HOST = "127.0.0.1"
+DEFAULT_SERVER_HOST = "0.0.0.0"
 
 
 def _load_submodule(name: str):
@@ -76,15 +77,29 @@ class Plugin:
         {
             "id": "install_pyav_amd64",
             "label": "Install / Update PyAV (amd64 / x86_64)",
-            "description": "Download and install the PyAV media dependency for x86_64 hosts. Required before streaming. Needs internet access.",
+            "description": (
+                "Download and install the PyAV media dependency for x86_64 hosts. "
+                "Required before streaming. Needs internet access. Running this "
+                "once is treated as consent for the plugin to automatically "
+                "reinstall PyAV for you later if it's ever found missing or "
+                "outdated (e.g. after a plugin update resets the vendored copy) "
+                "-- you shouldn't need to click this again after the first time."
+            ),
             "button_label": "Install PyAV (amd64)",
             "button_variant": "filled",
             "button_color": "blue",
         },
         {
             "id": "install_pyav_arm64",
             "label": "Install / Update PyAV (arm64 / aarch64)",
-            "description": "Download and install the PyAV media dependency for aarch64 hosts. Required before streaming. Needs internet access.",
+            "description": (
+                "Download and install the PyAV media dependency for aarch64 hosts. "
+                "Required before streaming. Needs internet access. Running this "
+                "once is treated as consent for the plugin to automatically "
+                "reinstall PyAV for you later if it's ever found missing or "
+                "outdated (e.g. after a plugin update resets the vendored copy) "
+                "-- you shouldn't need to click this again after the first time."
+            ),
             "button_label": "Install PyAV (arm64)",
             "button_variant": "filled",
             "button_color": "blue",
@@ -94,11 +109,18 @@ class Plugin:
     # Lifecycle (init)
 
     def __init__(self):
+        threading.Thread(target=self._auto_repair_pyav, daemon=True).start()
         try:
             self._autostart()
         except Exception as e:
             logger.warning(f"Multiview server auto-start skipped: {e}")
 
+    def _auto_repair_pyav(self):
+        try:
+            self._deps().maybe_auto_install()
+        except Exception as e:  # noqa: BLE001
+            logger.warning(f"Multiview PyAV auto-repair skipped: {e}")
+
     def _autostart(self):
         existing = _server().get_server()
         if existing and existing.is_running():
@@ -164,8 +186,9 @@ def _generate_m3u(self) -> dict:
         lines = ["#EXTM3U"]
         for n in range(1, mv_count + 1):
             name = settings.get(f"multiview_{n}_name", f"Multiview {n}") or f"Multiview {n}"
-            stream_url = f"http://localhost:{DEFAULT_SERVER_PORT}/stream/{n}"
-            lines.append(f'#EXTINF:-1 tvg-id="multiview_{n}" tvg-name="{name}",{name}')
+            safe_name = name.replace('"', "'")  # quotes break EXTINF attribute parsing
+            stream_url = f"http://127.0.0.1:{DEFAULT_SERVER_PORT}/stream/{n}"
+            lines.append(f'#EXTINF:-1 tvg-id="multiview_{n}" tvg-name="{safe_name}",{safe_name}')
             lines.append(stream_url)
 
         m3u_content = "\n".join(lines) + "\n"
@@ -195,7 +218,7 @@ def _generate_m3u(self) -> dict:
                 },
             )
             verb = "created" if created else "updated"
-            self._refresh_m3u_then_epg(account.id, source_id)
+            self._refresh_epg_then_m3u(account.id, source_id)
             return {
                 "status": "success",
                 "message": f"M3U written to {m3u_path} | M3U account {verb} in Dispatcharr",
@@ -207,30 +230,31 @@ def _generate_m3u(self) -> dict:
                 "message": f"M3U written to {m3u_path} (could not create M3U account: {e})",
             }
 
-    def _refresh_m3u_then_epg(self, account_id, source_id) -> None:
-        """Refresh the M3U account, then the EPG source, in sequence.
+    def _refresh_epg_then_m3u(self, account_id, source_id) -> None:
+        """Refresh EPG first, then M3U, in sequence.
 
         Firing both refreshes at once collides on Dispatcharr's shared celery DB
         connection ("the last operation didn't produce records (command status:
-        INSERT 0 N)"). A celery chain serializes them; M3U first so the multiview
-        channels exist before the EPG maps programs onto them.
+        INSERT 0 N)"). A celery chain serializes them; EPG first so program data
+        is current before the M3U account sync runs.
+
+        refresh_single_m3u_account internally calls refresh_m3u_groups(full_refresh=True)
+        which calls process_groups, which hits a Python 3.13 / Django ORM incompatibility
+        (StopIteration raised inside QuerySet.__iter__ generator -> RuntimeError).
+        Calling refresh_m3u_groups directly with full_refresh=False skips process_groups
+        and avoids the crash.
         """
         try:
             from celery import chain
             from apps.m3u.tasks import refresh_single_m3u_account
             if source_id is not None:
                 from apps.epg.tasks import refresh_epg_data
-                chain(refresh_single_m3u_account.si(account_id),
-                      refresh_epg_data.si(source_id)).delay()
+                chain(refresh_epg_data.si(source_id),
+                      refresh_single_m3u_account.si(account_id)).delay()
             else:
                 refresh_single_m3u_account.delay(account_id)
         except Exception as e:  # noqa: BLE001
-            logger.warning(f"Could not trigger M3U/EPG refresh chain: {e}")
-            try:  # best-effort fallback: at least refresh the M3U
-                from apps.m3u.tasks import refresh_single_m3u_account
-                refresh_single_m3u_account.delay(account_id)
-            except Exception:
-                pass
+            logger.warning(f"Could not trigger EPG/M3U refresh chain: {e}")
 
     # start_server
 

plugins/multiview/channel.py

@@ -86,9 +86,13 @@ def _even(v):
     return max(2, (int(v) // 2) * 2)
 
 
-def fit_into_tile(frame, w, h):
-    """Scale a decoded frame into a w x h yuv420p tile preserving aspect ratio,
-    centered on black (letterbox/pillarbox) - matches the old scale+pad behavior."""
+def fit_into_tile(frame, w, h, valign="center", halign="center"):
+    """Scale a decoded frame into a w x h yuv420p tile, preserving aspect
+    ratio (content is never cropped). Letterbox/pillarbox padding is
+    centered by default, or pushed to one edge via valign ("center"/"top"/
+    "bottom") and halign ("center"/"left"/"right") so a tile's content can
+    be made to touch an adjacent tile's content with no gap at the shared
+    edge, leaving any padding on the outer edge instead."""
     sw, sh = frame.width, frame.height
     if sw <= 0 or sh <= 0:
         return black_planes(w, h)
@@ -98,8 +102,18 @@ def fit_into_tile(frame, w, h):
     sf = frame.reformat(width=tw, height=th, format="yuv420p")
     sy, su, sv = yuv_planes_from_frame(sf, tw, th)
     Y, U, V = black_planes(w, h)
-    ox = ((w - tw) // 2) & ~1
-    oy = ((h - th) // 2) & ~1
+    if halign == "left":
+        ox = 0
+    elif halign == "right":
+        ox = (w - tw) & ~1
+    else:
+        ox = ((w - tw) // 2) & ~1
+    if valign == "top":
+        oy = 0
+    elif valign == "bottom":
+        oy = (h - th) & ~1
+    else:
+        oy = ((h - th) // 2) & ~1
     Y[oy:oy + th, ox:ox + tw] = sy
     U[oy // 2:oy // 2 + th // 2, ox // 2:ox // 2 + tw // 2] = su
     V[oy // 2:oy // 2 + th // 2, ox // 2:ox // 2 + tw // 2] = sv
@@ -120,6 +134,8 @@ def __init__(self, spec):
         self.provides_audio = bool(spec.get("audio", False))
         self.lang = spec.get("lang", "und")
         self.featured = bool(spec.get("featured", False))
+        self.valign = spec.get("valign", "center")
+        self.halign = spec.get("halign", "center")
         self.fallback = black_planes(self.w, self.h)
         self.latest = self.fallback
         self.fresh_until = 0.0
@@ -135,6 +151,12 @@ def __init__(self, spec):
         # video PTS clock anchor — updated by run(), read by audio_pts_now()
         self.clk_pts: "float | None" = None
         self.clk_wall: "float | None" = None
+        # PTS of the audio content most recently handed to the caller by take() —
+        # used by audio_feeder() to detect drift against the video clock and
+        # periodically re-sync via _align_to_pts(), without waiting for a full
+        # clk_pts reset. Protected by self.alock (same as aframes/abuffered).
+        self.last_taken_pts: "float | None" = None
+        self._reconnect_requested = False
 
     def _make_fallback(self, logo):
         Y, U, V = black_planes(self.w, self.h)
@@ -175,9 +197,16 @@ def _load_logo(self, logo):
         if self.fresh_until == 0.0:         # no real video yet; update latest too
             self.latest = fb
 
+    def reconnect(self):
+        """Signal run() to drop and reconnect; checked inside the demux loop."""
+        self._reconnect_requested = True
+
     def run(self):
         failures = 0
         while self.running:
+            if self._reconnect_requested:
+                self._reconnect_requested = False
+                failures = 0
             if failures >= RECONNECT_RETRIES:
                 log(f"channel {self.name}: giving up after {RECONNECT_RETRIES} failed retries")
                 break
@@ -187,6 +216,7 @@ def run(self):
             with self.alock:
                 self.aframes.clear()
                 self.abuffered = 0
+                self.last_taken_pts = None
             self.clk_pts = None
             self.clk_wall = None
             vcount_before = self.vcount
@@ -222,7 +252,7 @@ def run(self):
                     res = av.AudioResampler(format="s16", layout=AUDIO_LAYOUT, rate=AUDIO_RATE)
                 try:
                     for packet in cont.demux(*streams):
-                        if not self.running:
+                        if not self.running or self._reconnect_requested:
                             break
                         if packet.dts is None:
                             continue
@@ -239,7 +269,7 @@ def run(self):
                                             time.sleep(gap)
                                         elif gap <= -2.0:
                                             self.clk_pts, self.clk_wall = pts_s, time.monotonic()
-                                self.latest = fit_into_tile(frame, self.w, self.h)
+                                self.latest = fit_into_tile(frame, self.w, self.h, self.valign, self.halign)
                                 self.fresh_until = time.monotonic() + TILE_STALE_SECS
                                 self.vcount += 1
                         elif res is not None and packet.stream.type == "audio":
@@ -319,9 +349,13 @@ def take(self, nsamples: int) -> np.ndarray:
                     self.aframes.pop(0)
                     self.abuffered -= chunk.shape[0]
                     filled += chunk.shape[0]
+                    if pts_s is not None:
+                        self.last_taken_pts = pts_s + chunk.shape[0] / AUDIO_RATE
                 else:
                     out[filled:] = chunk[:need]
                     self.aframes[0] = (pts_s, chunk[need:])
                     self.abuffered -= need
                     filled = nsamples
+                    if pts_s is not None:
+                        self.last_taken_pts = pts_s + need / AUDIO_RATE
         return out

plugins/multiview/compositor_worker.py

@@ -28,6 +28,9 @@
 
 from parameters import fps_fraction, build_encoder_cmd, validate_encoder  # noqa: E402
 
+DRIFT_THRESHOLD = 0.25  # seconds of audio-behind-video before we skip the
+                         # FIFO forward to re-sync (see audio_feeder())
+
 
 # ---------------------------------------------------------------- compositing helpers
 
@@ -42,6 +45,29 @@ def _write_all(fd, data):
     return True
 
 
+def stdin_listener(channels, stop):
+    """Read JSON control commands from stdin (sent by the plugin server)."""
+    for line in sys.stdin:
+        if stop.is_set():
+            break
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            cmd = json.loads(line)
+        except Exception:
+            continue
+        if cmd.get("cmd") == "reconnect_channel":
+            idx = cmd.get("idx")
+            if idx is not None and 0 <= idx < len(channels):
+                log(f"reconnect requested: channel {idx} ({channels[idx].name})")
+                channels[idx].reconnect()
+        elif cmd.get("cmd") == "reconnect_all":
+            log("reconnect all channels requested")
+            for c in channels:
+                c.reconnect()
+
+
 def audio_feeder(track, fd, stop):
     CHUNK = int(AUDIO_RATE * 0.02)  # 960 samples = 20ms per tick
     SILENCE = np.zeros((CHUNK, 2), dtype=np.int16)
@@ -76,6 +102,15 @@ def audio_feeder(track, fd, stop):
 
         was_valid = True
 
+        # Only correct audio-behind-video (silent underruns falling further
+        # back over time); a transient audio-ahead-of-video reading is
+        # self-limiting (FIFO capped by _trim(), audio never paced faster
+        # than real time) and left uncorrected, matching the pre-existing
+        # one-shot snap behavior which is also catch-up-only.
+        last_pts = track.last_taken_pts
+        if last_pts is not None and (pts_now - last_pts) > DRIFT_THRESHOLD:
+            track._align_to_pts(pts_now - 0.10)
+
         target = int((time.monotonic() - start) * AUDIO_RATE)
         need = target - written
         if need > 0:
@@ -98,6 +133,7 @@ def main():
 
     for c in channels:
         threading.Thread(target=c.run, name=f"chan-{c.name}", daemon=True).start()
+    threading.Thread(target=stdin_listener, args=(channels, stop), name="stdin-ctrl", daemon=True).start()
 
     # ffmpeg encodes (libx264, multi-core C) + muxes; we feed it the composited
     # yuv420p canvas on stdin and one PCM track per audio channel on inherited fds.