[multiview] Update to v0.4.3

plugins/multiview/__init__.py

@@ -251,7 +251,7 @@ def _generate_m3u(self) -> dict:
                 },
             )
             verb = "created" if created else "updated"
-            self._refresh_epg_then_m3u(account.id, source_id)
+            self._refresh_m3u_then_epg(account.id, source_id)
             return {
                 "status": "success",
                 "message": f"M3U written to {m3u_path} | M3U account {verb} in Dispatcharr",
@@ -263,25 +263,25 @@ def _generate_m3u(self) -> dict:
                 "message": f"M3U written to {m3u_path} (could not create M3U account: {e})",
             }
 
-    def _refresh_epg_then_m3u(self, account_id, source_id) -> None:
-        """Refresh EPG first, then M3U, once EPG finishes (success OR error).
+    def _refresh_m3u_then_epg(self, account_id, source_id) -> None:
+        """Refresh M3U first, then parse EPG after its channel mappings exist.
 
         Firing both at once collides on Dispatcharr's shared celery DB connection
-        ("the last operation didn't produce records (command status: INSERT 0 N)"),
-        so M3U must wait for EPG to finish. A plain celery chain achieves that but
-        aborts the M3U task if EPG raises -- which silently skipped the M3U refresh
-        (and all its Dispatcharr-side websocket progress) whenever EPG refresh
-        failed. link_error() also fires the M3U task on EPG's failure path, so it
-        always runs once EPG is done, regardless of outcome.
+        ("the last operation didn't produce records (command status: INSERT 0 N)").
+        The EPG parser only reads programmes for EPG rows mapped to channels, and
+        M3U refresh creates those mappings, so the tasks must be serialized in
+        this order. If M3U refresh fails, skip EPG parsing rather than parse stale
+        channel mappings.
         """
         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
-                epg_sig = refresh_epg_data.si(source_id)
-                epg_sig.link_error(refresh_single_m3u_account.si(account_id))
-                chain(epg_sig, refresh_single_m3u_account.si(account_id)).delay()
+                chain(
+                    refresh_single_m3u_account.si(account_id),
+                    refresh_epg_data.si(source_id),
+                ).delay()
             else:
                 refresh_single_m3u_account.delay(account_id)
         except Exception as e:  # noqa: BLE001

plugins/multiview/channel.py

@@ -5,6 +5,7 @@
 the class that uses them, separate from the encoder and orchestration code.
 """
 
+import collections
 import os
 import platform
 import sys
@@ -37,6 +38,7 @@
 RECONNECT_RETRIES = 12    # consecutive failures before giving up (~8 min total)
 AUDIO_RATE = 48000
 AUDIO_LAYOUT = "stereo"
+VIDEO_QUEUE_FRAMES = 120
 
 # Tolerate flaky IPTV (skip corrupt packets, ignore decode errors, generous
 # probe) and bound I/O so a dead child errors and retries instead of hanging.
@@ -148,6 +150,11 @@ def __init__(self, spec):
             threading.Thread(target=self._load_logo, args=(logo,), daemon=True).start()
         self.running = True
         self.vcount = 0          # decoded video frames (for rate diagnostics)
+        # The compositor selects frames by source PTS. Keeping a short queue
+        # decouples that deterministic choice from decoder-thread wake timing.
+        self.vlock = threading.Lock()
+        self.vframes = collections.deque(maxlen=VIDEO_QUEUE_FRAMES)
+        self.display = self.fallback
         # audio buffer (only used when provides_audio)
         self.alock = threading.Lock()
         self.aframes = []        # list of (pts_s: float|None, ndarray(n,2) int16)
@@ -160,6 +167,7 @@ def __init__(self, spec):
         # 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.audio_resyncs = 0
         self._reconnect_requested = False
 
     def _make_fallback(self, logo):
@@ -223,6 +231,9 @@ def run(self):
                 self.last_taken_pts = None
             self.clk_pts = None
             self.clk_wall = None
+            with self.vlock:
+                self.vframes.clear()
+                self.display = self.fallback
             vcount_before = self.vcount
             try:
                 cont = av.open(self.url, options=DECODE_OPTS)
@@ -231,13 +242,10 @@ def run(self):
                 # rate (single-threaded PyAV decode runs ~22-27fps -> slow motion).
                 vs.thread_type = "AUTO"
                 vs.codec_context.thread_count = 3
-                # Sources are 1080p60 but we output 30fps; skip non-reference
-                # (B) frames at decode to cut decode CPU on the box, which
-                # otherwise saturates (3x 1080p60 decode + encode).
-                try:
-                    vs.codec_context.skip_frame = "NONREF"
-                except Exception:
-                    pass
+                log(f"channel {self.name}: video codec={getattr(vs.codec_context, 'name', None)} "
+                    f"size={vs.width}x{vs.height} avg_rate={getattr(vs, 'average_rate', None)} "
+                    f"base_rate={getattr(vs, 'base_rate', None)} "
+                    f"field_order={getattr(vs.codec_context, 'field_order', None)}")
                 # Lower-effort decode for non-featured tiles: skip the deblocking
                 # loop filter. Big decode-CPU saving; the minor blockiness is
                 # hidden by downscaling small tiles. The featured tile keeps full
@@ -273,8 +281,15 @@ 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.valign, self.halign)
-                                self.fresh_until = time.monotonic() + TILE_STALE_SECS
+                                            with self.vlock:
+                                                self.vframes.clear()
+                                                self.display = self.fallback
+                                tile = fit_into_tile(frame, self.w, self.h, self.valign, self.halign)
+                                with self.vlock:
+                                    self.latest = tile
+                                    self.fresh_until = time.monotonic() + TILE_STALE_SECS
+                                    if frame.pts is not None:
+                                        self.vframes.append((pts_s, tile))
                                 self.vcount += 1
                         elif res is not None and packet.stream.type == "audio":
                             for frame in packet.decode():
@@ -310,10 +325,21 @@ def run(self):
                 log(f"channel {self.name}: retry {failures}/{RECONNECT_RETRIES} in {delay:.0f}s")
                 time.sleep(delay)
 
-    def current(self):
-        if time.monotonic() < self.fresh_until:
-            return self.latest
-        return self.fallback
+    def current_at(self, wall_time):
+        """Return the newest decoded frame due at *wall_time* by source PTS."""
+        with self.vlock:
+            if time.monotonic() >= self.fresh_until:
+                return self.fallback
+            if self.clk_pts is None or self.clk_wall is None:
+                return self.latest
+            target_pts = self.clk_pts + wall_time - self.clk_wall
+            while self.vframes and self.vframes[0][0] <= target_pts:
+                _, self.display = self.vframes.popleft()
+            return self.display
+
+    def video_queue_depth(self):
+        with self.vlock:
+            return len(self.vframes)
 
     def _trim(self):
         cap = AUDIO_RATE * 2  # ~2s
@@ -340,6 +366,11 @@ def _align_to_pts(self, pts_limit: float):
                 else:
                     break
 
+    def audio_status(self):
+        """Return a consistent snapshot for compositor A/V diagnostics."""
+        with self.alock:
+            return self.last_taken_pts, self.abuffered, self.audio_resyncs
+
     def take(self, nsamples: int) -> np.ndarray:
         """Return exactly nsamples of int16 (nsamples, 2), silence-padded."""
         out = np.zeros((nsamples, 2), np.int16)
@@ -357,7 +388,11 @@ def take(self, nsamples: int) -> np.ndarray:
                         self.last_taken_pts = pts_s + chunk.shape[0] / AUDIO_RATE
                 else:
                     out[filled:] = chunk[:need]
-                    self.aframes[0] = (pts_s, chunk[need:])
+                    # A buffered chunk's PTS always identifies its first retained
+                    # sample. Without this adjustment, each partial read makes
+                    # drift detection measure from the chunk's original start.
+                    next_pts = pts_s + need / AUDIO_RATE if pts_s is not None else None
+                    self.aframes[0] = (next_pts, chunk[need:])
                     self.abuffered -= need
                     filled = nsamples
                     if pts_s is not None:

plugins/multiview/compositor_worker.py

@@ -31,6 +31,7 @@
 
 DRIFT_THRESHOLD = 0.25  # seconds of audio-behind-video before we skip the
                          # FIFO forward to re-sync (see audio_feeder())
+AUDIO_LEAD_SECS = 0.10  # retain this much audio around the video PTS clock
 
 
 # ---------------------------------------------------------------- compositing helpers
@@ -85,6 +86,7 @@ def audio_feeder(track, fd, stop):
             if was_valid:
                 # Clock just went None -- reconnect in progress; reset snap state
                 # so we re-anchor when the new stream establishes its first frame.
+                log(f"channel {track.name}: audio clock reset")
                 snapped = False
                 start = None
                 written = 0
@@ -96,7 +98,8 @@ def audio_feeder(track, fd, stop):
         if not snapped:
             # New clock available (startup or post-reconnect): snap audio buffer
             # to current video PTS and reset wall-clock counters.
-            track._align_to_pts(pts_now - 0.10)
+            track._align_to_pts(pts_now - AUDIO_LEAD_SECS)
+            log(f"channel {track.name}: audio clock anchor video_pts={pts_now:.3f}")
             start = time.monotonic()
             written = 0
             snapped = True
@@ -108,9 +111,13 @@ def audio_feeder(track, fd, stop):
         # 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
+        last_pts, _, _ = track.audio_status()
         if last_pts is not None and (pts_now - last_pts) > DRIFT_THRESHOLD:
-            track._align_to_pts(pts_now - 0.10)
+            delta = pts_now - last_pts
+            track._align_to_pts(pts_now - AUDIO_LEAD_SECS)
+            with track.alock:
+                track.audio_resyncs += 1
+            log(f"channel {track.name}: audio catch-up delta={delta:.3f}s")
 
         target = int((time.monotonic() - start) * AUDIO_RATE)
         need = target - written
@@ -230,11 +237,13 @@ def pump_out():
     log_at = start + 30.0
     prev_t = start
     prev_counts = [0] * len(channels)
+    prev_audio_resyncs = [0] * len(audio_chs)
     log(f"started: {len(channels)} tiles, {len(audio_chs)} audio, {out_w}x{out_h}@{cfg['fps']}")
     try:
         while not stop.is_set():
+            frame_time = start + n / fps_f
             for t in channels:
-                Yt, Ut, Vt, ox, oy, tw, th = t.current()
+                Yt, Ut, Vt, ox, oy, tw, th = t.current_at(frame_time)
                 x, y, w, h = t.x, t.y, t.w, t.h
                 Yc[y:y + h, x:x + w] = bg_Y[y:y + h, x:x + w]
                 Uc[y // 2:(y + h) // 2, x // 2:(x + w) // 2] = bg_U[y // 2:(y + h) // 2, x // 2:(x + w) // 2]
@@ -249,12 +258,22 @@ def pump_out():
             now = time.monotonic()
             if now >= log_at:   # heartbeat: per-channel decode fps (CPU health)
                 dt = now - prev_t
-                rates = " ".join(f"{c.name[:7]}={(c.vcount - prev_counts[i]) / dt:.0f}fps"
+                rates = " ".join(f"{c.name[:7]}={(c.vcount - prev_counts[i]) / dt:.0f}fps/q{c.video_queue_depth()}"
                                  for i, c in enumerate(channels))
+                audio = []
+                for i, c in enumerate(audio_chs):
+                    last_pts, buffered, resyncs = c.audio_status()
+                    video_pts = c.audio_pts_now()
+                    delta = video_pts - last_pts if video_pts is not None and last_pts is not None else None
+                    delta_text = f"{delta:+.3f}s" if delta is not None else "n/a"
+                    audio.append(f"{c.name[:7]}=d{delta_text}/q{buffered / AUDIO_RATE:.2f}s/"
+                                 f"r{resyncs - prev_audio_resyncs[i]}")
                 import resource as _res
                 rss_mb = _res.getrusage(_res.RUSAGE_SELF).ru_maxrss // 1024
-                log(f"out {n / (now - start):.1f}fps; decode {rates}; rss={rss_mb}MB")
+                log(f"out {n / (now - start):.1f}fps; decode {rates}; "
+                    f"audio {' '.join(audio) or 'none'}; rss={rss_mb}MB")
                 prev_counts = [c.vcount for c in channels]
+                prev_audio_resyncs = [c.audio_status()[2] for c in audio_chs]
                 prev_t = now
                 log_at = now + 30.0
             delay = (start + n / fps_f) - now

plugins/multiview/dash/static/index.html

@@ -10,7 +10,7 @@
     <link rel="icon" type="image/png" href="./logo.png" />
     <link rel="manifest" href="./manifest.json" />
     <title>Multiview</title>
-    <script type="module" crossorigin src="./assets/index-C4ShuQoB.js"></script>
+    <script type="module" crossorigin src="./assets/index-CQrSzl7U.js"></script>
     <link rel="stylesheet" crossorigin href="./assets/index-BHGMNXvw.css">
   </head>
   <body>

plugins/multiview/epg.py

@@ -47,6 +47,17 @@ def _fmt_xmltv_time(dt) -> str:
     return utc.strftime("%Y%m%d%H%M%S +0000")
 
 
+def _active_tvg_ids(order: list) -> set:
+    """Return the XMLTV channel IDs currently generated by this plugin."""
+    return {f"mv-{layout_id}" for layout_id in order}
+
+
+def _remove_stale_epg_data(source, active_tvg_ids: set) -> int:
+    """Remove rows for layouts no longer emitted by the plugin-owned source."""
+    deleted_count, _ = source.epgs.exclude(tvg_id__in=active_tvg_ids).delete()
+    return deleted_count
+
+
 def _emit_custom_props(cp: dict, lines: list) -> None:
     """Append XMLTV inner elements derived from a ProgramData.custom_properties dict."""
     if not cp:
@@ -266,4 +277,7 @@ def generate_epg(settings: dict, plugin_dir: str) -> "int | None":
             "is_active": True,
         },
     )
+    deleted_count = _remove_stale_epg_data(source, _active_tvg_ids(order))
+    if deleted_count:
+        logger.info("Removed %s stale EPG records from Dispatcharr Multiview source %s", deleted_count, source.id)
     return source.id

plugins/multiview/plugin.json

@@ -4,7 +4,7 @@
   "description": "Tile multiple Dispatcharr channel streams into multi-view outputs using FFmpeg",
   "author": "sethwv",
 
-  "version": "0.4.2",
+  "version": "0.4.3",
   "min_dispatcharr_version": "v0.27.0",
 
   "discord_thread": "https://discord.com/channels/1340492560220684331/1509200002407465001",