From 77771cb1aa1b6a779da1fc9cabc14aa544cf97d2 Mon Sep 17 00:00:00 2001 From: Kelly Fox Date: Tue, 23 Jun 2026 16:44:51 -0500 Subject: [PATCH] feat(video): honor URL t=/start= timestamp as a playback start offset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Quick playback (`c64cast MEDIA…`) now starts a URL at its own timestamp with no new flag: `?t=`/`&start=`/`#t=` (accepting 90, 90s, 1m30s, 1h2m3s) is parsed by quickcast._parse_start_offset and sets the scene's start offset. - SceneCfg.start_s (video-only; negative rejected, non-video rejected) threads through VideoScene into AVFileSource. - AVFileSource seeks the container to the keyframe at/just-before start_s (whole -container seek, AV_TIME_BASE microseconds, backward) before demux starts, and the peak-scan container seeks too so normalization covers the played portion. - The demux loop rebases every video frame's PTS by the first decoded frame so playback restarts at ~0 and tracks the from-0 audio/wall clock; the no-seek path is unchanged. Accuracy is keyframe-granular; A/V stay aligned. Applied as a PyAV seek (not a yt-dlp download-section) so it also covers direct media URLs and config-driven [[scenes]] videos, with no temp download. Docs (CLAUDE.md, usage.md, example TOML) + schema regenerated. Tests: timestamp parsers, start_s validation, and a fake-container demux PTS-rebase unit test. --- CLAUDE.md | 4 ++- c64cast.schema.json | 7 +++++ c64cast/config.py | 25 ++++++++++++++++ c64cast/quickcast.py | 52 ++++++++++++++++++++++++++++++++- c64cast/scenes.py | 10 ++++++- c64cast/video.py | 33 +++++++++++++++++++++ config/c64cast.example.toml | 5 ++++ docs/usage.md | 5 +++- tests/test_config.py | 16 ++++++++++ tests/test_quickcast.py | 55 +++++++++++++++++++++++++++++++++++ tests/test_video.py | 58 +++++++++++++++++++++++++++++++++++++ 11 files changed, 266 insertions(+), 4 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index fb16c22..18b12d3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,7 +19,7 @@ scripts/c64cast.sh --doctor --skip-probe **Connection target (`-u/--url`).** A single scheme-aware string selects both the hardware backend and its endpoint (the granular `--backend`/`--tr-*`/`--dma-port` flags were removed in favor of this — `git log` for the rationale). The parser is [c64cast/connect.py](c64cast/connect.py) (`parse_connection_uri` → `ConnectionSpec` → `apply_to_config`); it decomposes into the existing config fields (`[hardware].backend`, `[ultimate64].url`/`dma_port`, `[teensyrom].transport`/`serial_port`/`host`/…), which stay the canonical store a TOML sets directly. `make_backend` is unchanged. Schemes: `u64://HOST` or `http(s)://HOST` (Ultimate — the only HTTP-speaking backend, so http is deterministically Ultimate); `tr://` (TeensyROM+ USB serial, device auto-detected on macOS), `tr:///dev/cu.usbmodemXYZ` or `tr://COM3` (explicit serial device), `tr://HOST[:PORT]` (TeensyROM+ TCP, default port 2112). Rare per-link knobs as `?query` params (`u64://host?dma_port=64`, `tr://host?tcp_port=2113`, `tr:///dev/…?baud=2000000`, `tr://?storage=usb`). `$C64CAST_URL` is the env fallback. On the CLI the `-u` target overrides the config's connection sections (single-system runs only — in ensemble mode connection comes from the per-system TOMLs). -**Quick playback (positional `MEDIA` args).** Passing media files/dirs/globs/URLs as positional arguments (mutually exclusive with `--config`) builds an **in-memory-only** `Config` (no file on disk) with one scene per argument, in order, **no loop** (override with `--loop`). Each argument is mapped to a scene type by extension — video → `video`, `.sid` → `waveform`, image → `slideshow`, `.prg`/`.crt` → `launcher` — and a directory/glob is passed straight through as the scene's `file` spec (so the scene random-picks at setup, e.g. a dir of SIDs plays a random one). A URL becomes a `video`: direct media URLs play as-is (PyAV opens http(s)), and YouTube/other sites are resolved by yt-dlp (the optional `yt` extra) to a single progressive stream. Audio-only files (mp3/wav over a test pattern) are recognized but deferred to a follow-up. The classifier library is [c64cast/quickcast.py](c64cast/quickcast.py) (`build_config`); `cli.main` dispatches to it via `_resolve_configs` when it sees positional args, then runs the result through the normal path (`build_stack` → `_run_playlists` → `teardown_stack`), so behavior matches a config-driven run. +**Quick playback (positional `MEDIA` args).** Passing media files/dirs/globs/URLs as positional arguments (mutually exclusive with `--config`) builds an **in-memory-only** `Config` (no file on disk) with one scene per argument, in order, **no loop** (override with `--loop`). Each argument is mapped to a scene type by extension — video → `video`, `.sid` → `waveform`, image → `slideshow`, `.prg`/`.crt` → `launcher` — and a directory/glob is passed straight through as the scene's `file` spec (so the scene random-picks at setup, e.g. a dir of SIDs plays a random one). A URL becomes a `video`: direct media URLs play as-is (PyAV opens http(s)), and YouTube/other sites are resolved by yt-dlp (the optional `yt` extra) to a single progressive stream. A timestamp on the URL (`?t=`/`&start=`/`#t=`, accepting `90`, `90s`, `1m30s`, `1h2m3s`) is parsed by `_parse_start_offset` and sets the scene's `start_s` (no flag) so playback begins at that offset — see the `SceneCfg.start_s` / `AVFileSource` seek note below. Audio-only files (mp3/wav over a test pattern) are recognized but deferred to a follow-up. The classifier library is [c64cast/quickcast.py](c64cast/quickcast.py) (`build_config`); `cli.main` dispatches to it via `_resolve_configs` when it sees positional args, then runs the result through the normal path (`build_stack` → `_run_playlists` → `teardown_stack`), so behavior matches a config-driven run. ```bash scripts/c64cast.sh -u u64://192.168.2.64 clip.mp4 tune.sid assets/pictures/ @@ -199,6 +199,8 @@ Sample encoding can optionally apply **TPDF dither** (±1 LSB triangular, contro **`AVFileSource`** is for video playback. The demuxer thread reads packets from one container, pushes resampled mono int16 audio straight through to AudioStreamer, and queues decoded video frames keyed by PTS. Consumers call `current_frame(audio_position_s)` which returns the latest frame whose PTS ≤ the clock and drops anything behind. **Drift can't accumulate** because the audio clock IS the reference. +**Start offset (`start_s`).** `AVFileSource(..., start_s=N)` seeks the container to the keyframe at/just-before N (whole-container `seek` in AV_TIME_BASE microseconds, `backward=True`) before the demux thread starts, and the peak-scan container seeks too so normalization covers only the played portion. Because the playback clock starts at 0 (audio samples / wall-clock) while post-seek frame PTS are ~N, the demux loop rebases every video frame's PTS by the **first decoded frame's PTS** (`_pts_offset`) so video restarts at ~0 and tracks the clock — the no-seek path is unchanged (offset ≈ 0). Post-seek audio packets are interleaved near the same byte offset, so A/V stay aligned to sub-GOP precision; accuracy is keyframe-granular (exact-to-the-second start via decode-and-discard is a future refinement). Carried by `SceneCfg.start_s` (video-only; rejected on other types, negative rejected) → `VideoScene` → here. Quick playback fills it from a URL timestamp; a `[[scenes]]` video can set it directly. + **EOF handling**: `current_frame` normally keeps the chosen frame in `_video_buf` so a clock stall doesn't black-frame the display. After demux EOFs (`self._eof = True`) that stall-protection becomes a trap — the buffer stays size-1 forever, `finished` (which checks `_eof and not _video_buf`) never flips, `VideoScene.process_frame` never returns False, and the audio worker pads NEUTRAL indefinitely (visible as a 3-min `writes=4/s bytes=4KiB/s` streak in audio logs). The fix is in `current_frame`: when `_eof` is set AND the consumed index is the last buffered frame, clear the buffer entirely so `finished` can flip on the next check. ### `modes.py` — DisplayMode hierarchy diff --git a/c64cast.schema.json b/c64cast.schema.json index f4d05b2..c3d98ea 100644 --- a/c64cast.schema.json +++ b/c64cast.schema.json @@ -1504,6 +1504,13 @@ ], "description": "Asset spec (comma-separated paths/dirs/globs). Videos for video, .sid for waveform, images for slideshow, .prg/.crt for launcher, .sid for generative when audio_source = sid." }, + "start_s": { + "type": [ + "number", + "null" + ], + "description": "Seconds into the source to begin playback (video only). Quick playback fills this from a URL's t=/start= timestamp; can also be set directly on a [[scenes]] video. Unset/0 = play from the start." + }, "song": { "type": "integer", "description": "SID subtune index to play (0 = the SID's default; 1-based otherwise). For generative scenes, only with audio_source = sid.", diff --git a/c64cast/config.py b/c64cast/config.py index 8461208..69df4aa 100644 --- a/c64cast/config.py +++ b/c64cast/config.py @@ -622,6 +622,20 @@ class SceneCfg: "applies_to": ("video", "waveform", "slideshow", "launcher", "generative"), }, ) + # Start offset for video playback. Quick playback (`c64cast MEDIA…`) fills + # this from a URL's t=/start= timestamp; it can also be set directly on a + # [[scenes]] video. Honored by VideoScene -> AVFileSource (container seek to + # the keyframe at/just-before this time). Video-only; rejected elsewhere. + start_s: float | None = field( + default=None, + metadata={ + "help": "Seconds into the source to begin playback (video only). " + "Quick playback fills this from a URL's t=/start= timestamp; " + "can also be set directly on a [[scenes]] video. " + "Unset/0 = play from the start.", + "applies_to": ("video",), + }, + ) image_duration_s: float = field( default=5.0, metadata={ @@ -2194,6 +2208,8 @@ def _validate_video(s: SceneCfg, cfg: Config) -> DisplayMode: "config; use a [[scenes]] timeout via a different scene type " "if you want a hard cap." ) + if s.start_s is not None and s.start_s < 0: + raise ValueError(f"video: start_s must be >= 0, got {s.start_s!r}") return _display_mode_for_scene(s.display, s, cfg) @@ -2463,6 +2479,14 @@ def validate_scene_cfg(s: SceneCfg, cfg: Config, *, audio_enabled: bool) -> None f"video frame). Supported: {tuple(sorted(_EFFECT_SCENE_TYPES))}." ) + # start_s is a video-only start offset (the only scene whose source has a + # seekable timeline). Reject it elsewhere rather than silently ignoring it. + if s.start_s is not None and s.type != "video": + raise ValueError( + f"start_s is only supported on video scenes, not {s.type!r}. " + "Remove the field (it would be a silent no-op here)." + ) + if s.type == "webcam": mode = _display_mode_for_scene(s.display, s, cfg) elif s.type == "blank": @@ -2621,6 +2645,7 @@ def build_scene( s.file, prepend_alignment_marker=(cfg.audio.source_alignment_marker and cfg.audio.use_reu_pump), color=cfg.color, + start_s=s.start_s or 0.0, ) if s.target_fps is None: fps = _frame_push_default_fps(mode, scene_audio is not None, cfg.ultimate64.system) diff --git a/c64cast/quickcast.py b/c64cast/quickcast.py index 4fb699f..db23048 100644 --- a/c64cast/quickcast.py +++ b/c64cast/quickcast.py @@ -60,6 +60,11 @@ _URL_RE = re.compile(r"^https?://", re.IGNORECASE) _GLOB_CHARS = re.compile(r"[*?\[]") +# Timestamp grammar for URL start offsets: bare seconds ("90", "90.5") or the +# YouTube [Nh][Nm][Ns] form ("90s", "1m30s", "1h2m3s", "1h"). At least one of +# h/m/s must be present for the unit form to match. +_TIMESTR_HMS_RE = re.compile(r"^(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$", re.IGNORECASE) + # Display mode for video/slideshow scenes. mhires is the richest bitmap mode # and suits arbitrary film/photo content; SceneCfg's default (hires_edges) is # tuned for live-webcam Canny edges, not playback. @@ -76,6 +81,45 @@ def _is_url(arg: str) -> bool: return bool(_URL_RE.match(arg)) +def _parse_timestr(s: str) -> float | None: + """Parse a timestamp string to seconds. Accepts bare seconds ("90", + "90.5") and the [Nh][Nm][Ns] form ("90s", "1m30s", "1h2m3s"). Returns + None for anything else (so an unparseable t= is ignored, not fatal).""" + s = s.strip() + if not s: + return None + try: + return float(s) # bare seconds, e.g. "90" or "90.5" + except ValueError: + pass + m = _TIMESTR_HMS_RE.match(s) + if not m or not any(m.groups()): + return None + h, mi, sec = (int(g) if g else 0 for g in m.groups()) + return float(h * 3600 + mi * 60 + sec) + + +def _parse_start_offset(url: str) -> float | None: + """Start offset (seconds) from a media URL's timestamp, or None. + + Honors the `t` and `start` query params (in that order) and the `#t=` + fragment — the forms YouTube and friends use for "start here" links. An + unparseable or absent timestamp yields None (playback from the start).""" + parts = urllib.parse.urlsplit(url) + query = urllib.parse.parse_qs(parts.query) + for key in ("t", "start"): + values = query.get(key) + if values: + offset = _parse_timestr(values[0]) + if offset is not None: + return offset + # `#t=90` / `#t=1m30s` fragment form. + frag = parts.fragment + if frag.lower().startswith("t="): + return _parse_timestr(frag[2:]) + return None + + def _type_for_ext(ext: str) -> str | None: """Scene type for a single extension (case-insensitive). Returns the sentinel ``"audio"`` for audio-only formats (deferred), or ``None`` for an @@ -238,7 +282,13 @@ def classify_url(arg: str, *, display: str | None) -> SceneCfg: f"{arg!r} resolves to audio only, which isn't supported yet " "(test-pattern-over-audio is a planned follow-up)." ) - return _make_scene("video", stream_url, display=display, duration_s=None, name=title) + scene = _make_scene("video", stream_url, display=display, duration_s=None, name=title) + # Honor a start timestamp on the ORIGINAL url (t=/start=/#t=); the resolved + # stream_url won't carry it. The scene seeks to this offset on setup. + start_s = _parse_start_offset(arg) + if start_s: + scene.start_s = start_s + return scene def build_config(args: argparse.Namespace) -> Config: diff --git a/c64cast/scenes.py b/c64cast/scenes.py index dfb6bb8..4ced9eb 100644 --- a/c64cast/scenes.py +++ b/c64cast/scenes.py @@ -836,6 +836,7 @@ def __init__( file: str, prepend_alignment_marker: bool = False, color: ColorCfg | None = None, + start_s: float = 0.0, ): """`file` is a comma-separated `resolve_file_spec` spec (or a single literal path — the spec grammar treats one path as a one-entry @@ -868,6 +869,10 @@ def __init__( self.filepath = candidates[0] self.source: AVFileSource | None = None self._start_time = 0.0 + # Seconds into the file to begin playback (0 = from the start). Passed + # to AVFileSource at setup(), which seeks + rebases PTS. Quick playback + # derives this from a URL's t=/start= timestamp. + self.start_s = max(0.0, start_s) self._last_rendered_img: np.ndarray | None = None # Rolling-window auto_fit state (set up in setup() when [color].auto_fit # is on without force_palette). None = no online fit (pre-scanned or @@ -963,7 +968,10 @@ def setup(self) -> None: will_push_audio = self.audio is not None and not getattr(self.audio, "use_reu_pump", False) try: self.source = AVFileSource( - self.filepath, target_sample_rate=sr, scan_audio_peak=will_push_audio + self.filepath, + target_sample_rate=sr, + scan_audio_peak=will_push_audio, + start_s=self.start_s, ) except PermissionError as e: log.error("video: permission denied opening %s (%s)", self.filepath, e) diff --git a/c64cast/video.py b/c64cast/video.py index b3a8660..8cd25d1 100644 --- a/c64cast/video.py +++ b/c64cast/video.py @@ -273,6 +273,7 @@ def __init__( max_video_buffer: int = 240, source_noise_gate_enabled: bool = False, scan_audio_peak: bool = True, + start_s: float = 0.0, ): if not _ensure_pyav(): raise RuntimeError("PyAV not installed; install with `pip install c64cast[video]`") @@ -280,6 +281,14 @@ def __init__( self.path = path self.target_sr = target_sample_rate self.max_video_buffer = max_video_buffer + # Seconds into the source to begin playback (0 = from the start). When + # set, the container is sought to the keyframe at/just-before this time + # and frame PTS are rebased to ~0 (see _pts_offset + _demux_loop) so the + # from-0 playback clock in VideoScene still lines up. + self.start_s = max(0.0, start_s) + # Captured from the first decoded video frame so every later frame's PTS + # can be rebased to a ~0 origin. None until that first frame arrives. + self._pts_offset: float | None = None self.container = av.open(path) self.v_stream = self.container.streams.video[0] @@ -288,6 +297,15 @@ def __init__( self.video_fps = float(self.v_stream.average_rate) if self.v_stream.average_rate else 30.0 self.video_time_base = float(self.v_stream.time_base or 0) + # Seek before any demux so there's no decoder state to flush. Whole- + # container seek in AV_TIME_BASE units (microseconds); backward=True + # (the default) lands on the keyframe <= target, so playback starts at + # most one GOP early. Audio packets are interleaved near the same byte + # offset, so A/V stay aligned once video PTS are rebased. + if self.start_s > 0: + self.container.seek(int(self.start_s * 1_000_000)) + log.info("av %s: seek to start_s=%.3fs", os.path.basename(self.path), self.start_s) + if self.a_stream is not None: self._resampler = av.AudioResampler( format="s16", layout="mono", rate=target_sample_rate @@ -353,6 +371,11 @@ def _scan_audio_peak(self) -> int: container = av.open(self.path) try: a_stream = container.streams.audio[0] + # Match the played portion: with a start_s seek, normalize over + # [start_s, end] only — both more correct (gain reflects what's + # heard) and cheaper (no decoding the skipped head on long clips). + if self.start_s > 0: + container.seek(int(self.start_s * 1_000_000)) resampler = av.AudioResampler(format="s16", layout="mono", rate=self.target_sr) for packet in container.demux(a_stream): for frame in packet.decode(): @@ -393,6 +416,16 @@ def _demux_loop(self): if frame.pts is not None else 0.0 ) + # Rebase PTS so the first decoded frame sits at ~0. With + # a start_s seek the raw PTS are ~start_s; the playback + # clock (audio samples / wall-clock) starts at 0, so + # without this current_frame() would find no frame <= 0 + # for start_s seconds. Offset is captured from the first + # frame (the keyframe the seek landed on), so the no-seek + # path is unchanged (offset == first PTS, ~0). + if self._pts_offset is None: + self._pts_offset = pts + pts -= self._pts_offset # Backpressure: wait if the buffer is at capacity. # The old behavior (silent-drop oldest frames) was a # safety net under host-DMA mode, where AudioStreamer's diff --git a/config/c64cast.example.toml b/config/c64cast.example.toml index d4e079f..edf6636 100644 --- a/config/c64cast.example.toml +++ b/config/c64cast.example.toml @@ -864,6 +864,11 @@ palette_mode = "grayscale" # # file = "assets/videos, assets/extra/promo.mp4" # file = "assets/videos/your-video.mp4" # display = "hires_edges" +# # start_s — seconds into the file to begin playback (video-only; omit or 0 +# # = from the start). Seeks to the keyframe at/just-before this time. Quick +# # playback (`c64cast MEDIA…`) fills this automatically from a URL's +# # t=/start= timestamp (e.g. https://youtu.be/?t=1m30s). +# start_s = 90.0 # # Video scenes run until the video file ends — `duration_s` is # rejected by the config loader (the playlist advances when the source diff --git a/docs/usage.md b/docs/usage.md index 4099bc4..d77ea56 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -138,6 +138,9 @@ scripts/c64cast.sh -u tr:// clip.mp4 tune.sid # Direct play a YouTube URL (needs the `yt` extra: `uv sync --extra yt`): scripts/c64cast.sh -u u64://192.168.2.64 'https://youtu.be/dQw4w9WgXcQ' + +# A URL timestamp (?t= / &start= / #t=) starts playback at that offset: +scripts/c64cast.sh -u u64://192.168.2.64 'https://youtu.be/dQw4w9WgXcQ?t=1m30s' ``` Each argument is mapped to a scene type: @@ -149,7 +152,7 @@ Each argument is mapped to a scene type: | image (`.jpg`, `.png` …) | `slideshow` | | `.prg` / `.crt` | `launcher` | | directory or glob | inferred from the contents (a single kind); the spec is passed through, so the scene random-picks at setup | -| URL | `video` (direct media plays as-is; YouTube/others resolved by yt-dlp) | +| URL | `video` (direct media plays as-is; YouTube/others resolved by yt-dlp). A `?t=`/`&start=`/`#t=` timestamp (`90`, `90s`, `1m30s`, `1h2m3s`) seeks playback to that offset | Audio is **on by default** — pass `--no-audio` to mute. Flags: `-u/--url`, `-s/--system`, `--display MODE` (default `mhires` for video/slideshow), diff --git a/tests/test_config.py b/tests/test_config.py index a9f679b..ad665c0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -347,6 +347,22 @@ def test_video_scene_without_duration_s_passes(self): s = cfgmod.SceneCfg(type="video", file="video.mp4") cfgmod.validate_scene_cfg(s, self._cfg(), audio_enabled=False) + def test_video_scene_accepts_start_s(self): + s = cfgmod.SceneCfg(type="video", file="video.mp4", start_s=90.0) + cfgmod.validate_scene_cfg(s, self._cfg(), audio_enabled=False) + + def test_video_scene_rejects_negative_start_s(self): + s = cfgmod.SceneCfg(type="video", file="video.mp4", start_s=-1.0) + with self.assertRaisesRegex(ValueError, "start_s must be >= 0"): + cfgmod.validate_scene_cfg(s, self._cfg(), audio_enabled=False) + + def test_start_s_rejected_on_non_video(self): + # start_s is a video-only seek; setting it elsewhere is a no-op the + # loader rejects rather than silently ignores. + s = cfgmod.SceneCfg(type="slideshow", file="pic.jpg", start_s=10.0) + with self.assertRaisesRegex(ValueError, "start_s is only supported on video"): + cfgmod.validate_scene_cfg(s, self._cfg(), audio_enabled=False) + def test_waveform_scene_falls_back_to_default_dir(self): cwd = os.getcwd() with tempfile.TemporaryDirectory() as tmp: diff --git a/tests/test_quickcast.py b/tests/test_quickcast.py index 6ad7e65..fa96164 100644 --- a/tests/test_quickcast.py +++ b/tests/test_quickcast.py @@ -266,12 +266,67 @@ def test_missing_ytdlp_raises_runtime_error(self): self.assertIn("yt-dlp", str(cm.exception)) +class ParseTimestrTest(unittest.TestCase): + def test_bare_seconds(self): + self.assertEqual(quickcast._parse_timestr("90"), 90.0) + self.assertEqual(quickcast._parse_timestr("90.5"), 90.5) + self.assertEqual(quickcast._parse_timestr("0"), 0.0) + + def test_hms_form(self): + self.assertEqual(quickcast._parse_timestr("90s"), 90.0) + self.assertEqual(quickcast._parse_timestr("1m30s"), 90.0) + self.assertEqual(quickcast._parse_timestr("1h2m3s"), 3723.0) + self.assertEqual(quickcast._parse_timestr("1h"), 3600.0) + self.assertEqual(quickcast._parse_timestr("2m"), 120.0) + self.assertEqual(quickcast._parse_timestr("1H30M"), 5400.0) # case-insensitive + + def test_unparseable_returns_none(self): + self.assertIsNone(quickcast._parse_timestr("")) + self.assertIsNone(quickcast._parse_timestr("abc")) + self.assertIsNone(quickcast._parse_timestr("1x2y")) + + +class ParseStartOffsetTest(unittest.TestCase): + def test_t_query_param(self): + self.assertEqual(quickcast._parse_start_offset("https://youtu.be/x?t=90"), 90.0) + self.assertEqual(quickcast._parse_start_offset("https://youtu.be/x?t=1m30s"), 90.0) + + def test_start_query_param(self): + self.assertEqual(quickcast._parse_start_offset("https://yt/watch?v=x&start=45"), 45.0) + + def test_t_preferred_over_start(self): + self.assertEqual(quickcast._parse_start_offset("https://yt/x?t=10&start=45"), 10.0) + + def test_fragment_form(self): + self.assertEqual(quickcast._parse_start_offset("http://host/clip.mp4#t=30"), 30.0) + self.assertEqual(quickcast._parse_start_offset("http://host/clip.mp4#t=1m"), 60.0) + + def test_no_timestamp(self): + self.assertIsNone(quickcast._parse_start_offset("https://youtu.be/x")) + self.assertIsNone(quickcast._parse_start_offset("http://host/clip.mp4?list=foo")) + + def test_garbage_timestamp_ignored(self): + self.assertIsNone(quickcast._parse_start_offset("https://youtu.be/x?t=soon")) + + class ClassifyUrlTest(unittest.TestCase): def test_video_url_becomes_video(self): scene = quickcast.classify_url("http://host/clip.mp4", display=None) self.assertEqual(scene.type, "video") self.assertEqual(scene.file, "http://host/clip.mp4") self.assertEqual(scene.display, "mhires") + self.assertIsNone(scene.start_s) + + def test_url_timestamp_sets_start_s(self): + # Direct media URL passes through resolve_media_url untouched (no + # yt-dlp), so this exercises start_s wiring without a network/dep. + scene = quickcast.classify_url("http://host/clip.mp4?t=1m30s", display=None) + self.assertEqual(scene.type, "video") + self.assertEqual(scene.start_s, 90.0) + + def test_url_fragment_timestamp_sets_start_s(self): + scene = quickcast.classify_url("http://host/clip.mp4#t=45", display=None) + self.assertEqual(scene.start_s, 45.0) def test_audio_url_deferred(self): with self.assertRaises(ValueError) as cm: diff --git a/tests/test_video.py b/tests/test_video.py index 8afdbd9..2d71863 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -143,5 +143,63 @@ def test_partial_consume_at_eof_keeps_unconsumed_frames(self): self.assertFalse(src.finished, "still frames ahead of clock → not finished") +class _FakeFrame: + def __init__(self, pts: int): + self.pts = pts + + def to_ndarray(self, format: str): # noqa: A002 - matches PyAV's kwarg name + return np.zeros((2, 2, 3), dtype=np.uint8) + + +class _FakeStream: + type = "video" + + +class _FakePacket: + def __init__(self, frames: list[_FakeFrame]): + self.stream = _FakeStream() + self._frames = frames + + def decode(self): + return self._frames + + +class _FakeContainer: + def __init__(self, packets: list[_FakePacket]): + self._packets = packets + + def demux(self): + return iter(self._packets) + + +class DemuxRebaseTest(unittest.TestCase): + """`_demux_loop` rebases video PTS by the first decoded frame so a seeked + source (frame PTS ~start_s) still starts at the from-0 playback clock. + Driven with a fake container — no PyAV, no real file.""" + + def _run_demux(self, frame_ptss: list[int]) -> list[float]: + src = AVFileSource.__new__(AVFileSource) + src._closed = False + src._pts_offset = None + src.video_time_base = 1.0 # 1 PTS tick == 1 second + src._video_buf = [] + src._lock = threading.Lock() + src.max_video_buffer = 240 + src._resampler = None + src._audio_push = None + src.path = "fake" + src.container = _FakeContainer([_FakePacket([_FakeFrame(p)]) for p in frame_ptss]) + src._demux_loop() + return [pts for pts, _ in src._video_buf] + + def test_seeked_source_rebases_to_zero(self): + # Frame PTS ~100s (post-seek) must rebase so the buffer starts at 0. + self.assertEqual(self._run_demux([100, 101, 102]), [0.0, 1.0, 2.0]) + + def test_no_seek_unchanged(self): + # First frame already at 0 → offset 0 → buffer unchanged. + self.assertEqual(self._run_demux([0, 1, 2]), [0.0, 1.0, 2.0]) + + if __name__ == "__main__": unittest.main()