Merge branch 'master' into ListInput

comfy_api/latest/_input_impl/video_types.py

@@ -1,6 +1,5 @@
 from av.container import InputContainer
 from av.subtitles.stream import SubtitleStream
-from av.video.reformatter import ColorRange
 from fractions import Fraction
 from typing import Optional
 from .._input import AudioInput, VideoInput
@@ -10,7 +9,6 @@ import itertools
 import json
 import numpy as np
 import math
-import os
 import torch
 from .._util import VideoContainer, VideoCodec, VideoComponents
 import logging
@@ -60,57 +58,6 @@ def video_stream_bit_depth(stream) -> int:
     return max(component.bits for component in stream.format.components)
 
 
-def last_decodable_audio_stream(container: InputContainer):
-    """Streams FFmpeg has no decoder for have no codec context, and decoding their
-    packets crashes the process (e.g. APAC spatial-audio track in iPhone)."""
-    stream = next(
-        (s for s in reversed(container.streams.audio) if s.codec_context is not None),
-        None,
-    )
-    if stream is None and len(container.streams.audio):
-        logging.warning("No decodable audio stream found in video; ignoring audio.")
-    return stream
-
-
-def probe_audio_params(container: InputContainer, audio_stream, max_packets: int = 200):
-    """Containers probed only up to a window (mpegts) leave audio codec parameters unset when
-    audio starts beyond it; learn them by decoding ahead. The caller must seek back afterwards.
-    Returns (sample_rate, channels), zeros when the stream never yields a decodable frame."""
-    for i, packet in enumerate(container.demux(audio_stream)):
-        try:
-            frames = packet.decode()
-        except av.error.FFmpegError:
-            frames = ()
-        if frames:
-            return frames[0].sample_rate, frames[0].layout.nb_channels
-        if i >= max_packets:
-            break
-    return 0, 0
-
-
-def write_output_metadata(container: InputContainer, output, metadata: dict | None):
-    """Copy the source container's metadata, then overlay the caller's tags."""
-    for key, value in container.metadata.items():
-        if metadata is None or key not in metadata:
-            output.metadata[key] = value
-    if metadata is not None:
-        for key, value in metadata.items():
-            output.metadata[key] = value if isinstance(value, str) else json.dumps(value)
-
-
-def mp4_output_open_kwargs(path: str | io.BytesIO, format: VideoContainer, codec: VideoCodec) -> dict:
-    if format != VideoContainer.AUTO and format != VideoContainer.MP4:
-        raise ValueError("Only MP4 format is supported for now")
-    if codec != VideoCodec.AUTO and codec != VideoCodec.H264:
-        raise ValueError("Only H264 codec is supported for now")
-    open_kwargs = {"mode": "w", "options": {"movflags": "use_metadata_tags"}}
-    if isinstance(format, VideoContainer) and format != VideoContainer.AUTO:
-        open_kwargs["format"] = format.value
-    elif isinstance(path, io.BytesIO):
-        open_kwargs["format"] = "mp4"  # no file extension to infer the format from
-    return open_kwargs
-
-
 class VideoFromFile(VideoInput):
     """
     Class representing video input from a file.
@@ -245,10 +192,13 @@ class VideoFromFile(VideoInput):
                     return estimated_frames
 
             # 3. Last resort: decode frames and count them (streaming)
-            start_time, duration = self.get_active_trim_window()
+            if self.__start_time < 0:
+                start_time = max(self._get_raw_duration() + self.__start_time, 0)
+            else:
+                start_time = self.__start_time
             frame_count = 1
             start_pts = int(start_time / video_stream.time_base)
-            end_pts = int((start_time + duration) / video_stream.time_base)
+            end_pts = int((start_time + self.__duration) / video_stream.time_base)
             container.seek(start_pts, stream=video_stream)
             frame_iterator = (
                 container.decode(video_stream)
@@ -303,14 +253,17 @@ class VideoFromFile(VideoInput):
 
     def get_components_internal(self, container: InputContainer) -> VideoComponents:
         video_stream = self._get_first_video_stream(container)
-        start_time, duration = self.get_active_trim_window()
+        if self.__start_time < 0:
+            start_time = max(self._get_raw_duration() + self.__start_time, 0)
+        else:
+            start_time = self.__start_time
 
         # Get video frames
         frames = []
         audio_frames = []
         alphas = None
         start_pts = int(start_time / video_stream.time_base)
-        end_pts = int((start_time + duration) / video_stream.time_base)
+        end_pts = int((start_time + self.__duration) / video_stream.time_base)
 
         if start_pts != 0:
             container.seek(start_pts, stream=video_stream)
@@ -328,11 +281,18 @@ class VideoFromFile(VideoInput):
         video_done = False
         audio_done = True
 
-        audio_stream = last_decodable_audio_stream(container)
+        # Use the last decodable audio stream. Streams FFmpeg has no decoder for have no codec context,
+        # and decoding their packets crashes the process. (e.g. APAC spatial-audio track in iPhone)
+        audio_stream = next(
+            (s for s in reversed(container.streams.audio) if s.codec_context is not None),
+            None,
+        )
         if audio_stream is not None:
             streams += [audio_stream]
             resampler = av.audio.resampler.AudioResampler(format='fltp')
             audio_done = False
+        elif len(container.streams.audio):
+            logging.warning("No decodable audio stream found in video; ignoring audio.")
 
         for packet in container.demux(*streams):
             if video_done and audio_done:
@@ -345,7 +305,7 @@ class VideoFromFile(VideoInput):
                     for frame in packet.decode():
                         if frame.pts < start_pts:
                             continue
-                        if duration and frame.pts >= end_pts:
+                        if self.__duration and frame.pts >= end_pts:
                             video_done = True
                             break
 
@@ -412,7 +372,7 @@ class VideoFromFile(VideoInput):
                     map(resampler.resample, packet.decode())
                 )
                 for frame in aframes:
-                    if duration and frame.time > start_time + duration:
+                    if self.__duration and frame.time > start_time + self.__duration:
                         audio_done = True
                         break
 
@@ -434,8 +394,8 @@ class VideoFromFile(VideoInput):
 
         if len(audio_frames) > 0:
             audio_data = np.concatenate(audio_frames, axis=1)  # shape: (channels, total_samples)
-            if duration:
-                audio_data = audio_data[..., :int(duration * audio_stream.sample_rate)]
+            if self.__duration:
+                audio_data = audio_data[..., :int(self.__duration * audio_stream.sample_rate)]
 
             audio_tensor = torch.from_numpy(audio_data).unsqueeze(0)  # shape: (1, channels, total_samples)
             audio = AudioInput({
@@ -481,14 +441,28 @@ class VideoFromFile(VideoInput):
             if not reuse_streams:
                 if bit_depth is None:
                     bit_depth = source_bit_depth
-                return self._save_transcoded(container, path, format=format, codec=codec, metadata=metadata, bit_depth=bit_depth)
+                components = self.get_components_internal(container)
+                video = VideoFromComponents(components)
+                return video.save_to(
+                    path, format=format, codec=codec, metadata=metadata, bit_depth=bit_depth,
+                )
 
             streams = container.streams
 
             open_kwargs = get_open_write_kwargs(path, container_format, format)
             with av.open(path, **open_kwargs) as output_container:
-                # Add metadata before writing any streams
-                write_output_metadata(container, output_container, metadata)
+                # Copy over the original metadata
+                for key, value in container.metadata.items():
+                    if metadata is None or key not in metadata:
+                        output_container.metadata[key] = value
+
+                # Add our new metadata
+                if metadata is not None:
+                    for key, value in metadata.items():
+                        if isinstance(value, str):
+                            output_container.metadata[key] = value
+                        else:
+                            output_container.metadata[key] = json.dumps(value)
 
                 # Add streams to the new container. Streams with no codec context cannot be used as an output template.
                 stream_map = {}
@@ -506,282 +480,6 @@ class VideoFromFile(VideoInput):
                         packet.stream = stream_map[packet.stream]
                         output_container.mux(packet)
 
-    def _save_transcoded(
-        self,
-        container: InputContainer,
-        path: str | io.BytesIO,
-        format: VideoContainer,
-        codec: VideoCodec,
-        metadata: dict | None,
-        bit_depth: int,
-    ):
-        """Re-encode to H.264/AAC one frame at a time; peak memory does not scale with video length."""
-        open_kwargs = mp4_output_open_kwargs(path, format, codec)
-        video_stream = self._get_first_video_stream(container)
-        start_time, duration = self.get_active_trim_window()
-        start_pts = int(start_time / video_stream.time_base)
-        end_pts = int((start_time + duration) / video_stream.time_base) if duration else None
-        stream_end_pts = None
-        if video_stream.duration is not None:
-            stream_end_pts = (video_stream.start_time or 0) + video_stream.duration
-        output_end_pts = end_pts
-        if stream_end_pts is not None and (output_end_pts is None or stream_end_pts < output_end_pts):
-            output_end_pts = stream_end_pts
-        if start_pts != 0:
-            container.seek(start_pts, stream=video_stream)
-
-        audio_stream = last_decodable_audio_stream(container)
-        pix_fmt = "yuv420p10le" if bit_depth >= 10 else "yuv420p"
-        rate = Fraction(video_stream.average_rate) if video_stream.average_rate else Fraction(1)
-
-        resampler = None
-        sample_rate = 0
-        audio_time_base = None
-        duration_cap = None
-        if audio_stream is not None:
-            sample_rate = audio_stream.codec_context.sample_rate
-            channels = audio_stream.codec_context.channels
-            if not sample_rate:
-                sample_rate, channels = probe_audio_params(container, audio_stream)
-                container.seek(start_pts, stream=video_stream)
-                if sample_rate:
-                    audio_stream.codec_context.flush_buffers()
-                else:
-                    logging.warning("Audio stream parameters could not be determined; ignoring audio.")
-                    audio_stream = None
-        if audio_stream is not None:
-            audio_time_base = Fraction(1, sample_rate)
-            layout = {1: "mono", 2: "stereo", 6: "5.1"}.get(channels, "stereo")
-            resampler = av.audio.resampler.AudioResampler(format="fltp", layout=layout, rate=sample_rate)
-            if duration:
-                duration_cap = math.ceil(duration * sample_rate)
-
-        streams = [video_stream] if audio_stream is None else [video_stream, audio_stream]
-        pts_step = max(1, int(round((1 / rate) / video_stream.time_base)))
-        video_done = False
-        audio_done = audio_stream is None
-        video_pts_offset = None
-        last_video_pts = None
-        last_video_end = None
-        # rebased pts -> true display duration: the mp4 muxer pads the last sample with 1/rate otherwise
-        video_frame_durations = {}
-        source_size = None
-        rotation_k = 0
-        rotation_filter = None
-        audio_started = False
-        samples_written = 0
-        pending_audio = []
-        # The output opens lazily on the first kept frame: it decides the geometry (90/270 rotation swaps dims),
-        # and never seeking back keeps webm/mkv leading audio intact.
-        output = None
-        out_video = None
-        out_audio = None
-
-        def audio_frame_from_ndarray(nd_planar):
-            frame = av.AudioFrame.from_ndarray(np.ascontiguousarray(nd_planar), format="fltp", layout=layout)
-            frame.sample_rate = sample_rate
-            return frame
-
-        def drain_audio(final=False):
-            # Audio may cover the pts span of the video written so far, capped by the requested duration
-            nonlocal samples_written, audio_done
-            if last_video_end is None:
-                cap = 0
-            else:
-                cap = math.ceil(last_video_end * video_stream.time_base * sample_rate)
-            if duration_cap is not None:
-                cap = min(cap, duration_cap)
-            while pending_audio and not audio_done:
-                frame = pending_audio[0]
-                if samples_written + frame.samples <= cap:
-                    frame.pts = samples_written
-                    frame.time_base = audio_time_base
-                    output.mux(out_audio.encode(frame))
-                    samples_written += frame.samples
-                    pending_audio.pop(0)
-                    continue
-                if final:
-                    keep = frame.to_ndarray()[..., :cap - samples_written]
-                    if keep.shape[-1] > 0:
-                        tail = audio_frame_from_ndarray(keep)
-                        tail.pts = samples_written
-                        tail.time_base = audio_time_base
-                        output.mux(out_audio.encode(tail))
-                        samples_written += keep.shape[-1]
-                    pending_audio.clear()
-                break
-            if duration_cap is not None and samples_written >= duration_cap:
-                audio_done = True
-            return cap
-
-        try:
-            for packet in container.demux(*streams):
-                if video_done and audio_done:
-                    break
-
-                if packet.stream == video_stream and not video_done:
-                    try:
-                        frames = packet.decode()
-                    except av.error.InvalidDataError:
-                        logging.info("pyav decode error")
-                        continue
-                    for frame in frames:
-                        if frame.pts is not None and frame.pts < start_pts:
-                            continue
-                        if end_pts is not None and frame.pts is not None and frame.pts >= end_pts:
-                            video_done = True
-                            if last_video_pts is not None:
-                                # the source continues past the window: hold the last kept frame to the window end
-                                end_offset = video_pts_offset if video_pts_offset is not None else start_pts
-                                last_video_end = max(last_video_end, end_pts - end_offset)
-                            break
-                        # the source's true display duration of this frame; average_rate is not a
-                        # frame duration (sparse/VFR sources), so it is only the fallback
-                        frame_duration = frame.duration if frame.duration else pts_step
-                        if end_pts is not None and frame.pts is not None:
-                            frame_duration = min(frame_duration, end_pts - frame.pts)
-                        if output is None:
-                            rotation_k = int(round(frame.rotation // 90)) % 4 if frame.rotation else 0
-                            if rotation_k % 2:
-                                out_width, out_height = frame.height, frame.width
-                            else:
-                                out_width, out_height = frame.width, frame.height
-                            if out_width % 2 or out_height % 2:
-                                raise ValueError(f"H.264 output requires even dimensions, got {out_width}x{out_height}")
-                            source_size = (frame.width, frame.height)
-                            output = av.open(path, **open_kwargs)
-                            # Add metadata before writing any streams
-                            write_output_metadata(container, output, metadata)
-                            out_video = output.add_stream("h264", rate=rate)
-                            # no B-frames: reordering makes mp4 sample durations follow decode order,
-                            # so irregular-VFR spans and trim windows land wrong
-                            out_video.codec_context.max_b_frames = 0
-                            out_video.width = out_width
-                            out_video.height = out_height
-                            out_video.pix_fmt = pix_fmt
-                            # source pts pass through (rebased to 0), so variable frame rate survives
-                            out_video.codec_context.time_base = video_stream.time_base
-                            if audio_stream is not None:
-                                out_audio = output.add_stream("aac", rate=sample_rate, layout=layout)
-                        if (frame.width, frame.height) != source_size:
-                            # encoding would silently rescale the new geometry into the old one
-                            raise ValueError(
-                                f"Video resolution changes mid-stream "
-                                f"({source_size[0]}x{source_size[1]} -> {frame.width}x{frame.height}); cannot transcode"
-                            )
-                        if rotation_k:
-                            if rotation_filter is None:
-                                g = av.filter.Graph()
-                                g_src = g.add_buffer(width=frame.width, height=frame.height,
-                                                     format=frame.format.name, time_base=video_stream.time_base)
-                                tail = g_src
-                                for filter_name, filter_args in {1: [("transpose", "cclock")],
-                                                                 2: [("hflip", None), ("vflip", None)],
-                                                                 3: [("transpose", "clock")]}[rotation_k]:
-                                    step = g.add(filter_name, filter_args)
-                                    tail.link_to(step)
-                                    tail = step
-                                g_sink = g.add("buffersink")
-                                tail.link_to(g_sink)
-                                g.configure()
-                                rotation_filter = (g_src, g_sink)
-                            rotation_filter[0].push(frame)
-                            frame = rotation_filter[1].pull()
-                        if frame.color_range == ColorRange.JPEG:
-                            # compress full-range sources (yuvj/MJPEG) to limited range
-                            frame = frame.reformat(format=pix_fmt, src_color_range="JPEG", dst_color_range="MPEG")
-                        else:
-                            frame = frame.reformat(format=pix_fmt)
-                        frame_output_end = None
-                        if frame.pts is not None:
-                            if video_pts_offset is None:
-                                video_pts_offset = frame.pts
-                            frame.pts -= video_pts_offset
-                            if output_end_pts is not None:
-                                frame_output_end = output_end_pts - video_pts_offset
-                                if frame.pts + frame_duration > frame_output_end:
-                                    clamped_pts = frame_output_end - frame_duration
-                                    if clamped_pts >= 0 and (last_video_pts is None or clamped_pts > last_video_pts):
-                                        frame.pts = min(frame.pts, clamped_pts)
-                                    elif frame.pts < frame_output_end:
-                                        frame_duration = frame_output_end - frame.pts
-                                    else:
-                                        continue
-                        if frame.pts is None or (last_video_pts is not None and frame.pts <= last_video_pts):
-                            # broken sources emit missing/backward timestamps mid-stream, which the
-                            # muxer rejects; nudge them forward by one nominal frame interval
-                            frame.pts = 0 if last_video_pts is None else last_video_pts + pts_step
-                            if frame_output_end is not None and frame.pts + frame_duration > frame_output_end:
-                                if frame.pts >= frame_output_end:
-                                    continue
-                                frame_duration = frame_output_end - frame.pts
-                        last_video_pts = frame.pts
-                        last_video_end = frame.pts + frame_duration
-                        video_frame_durations[frame.pts] = frame_duration
-                        # the decoded pict_type would force x264's frame types (intra-only
-                        # sources like MJPEG/ProRes would come out all-keyframe)
-                        frame.pict_type = 0
-                        for out_packet in out_video.encode(frame):
-                            out_packet.duration = video_frame_durations.pop(out_packet.pts, 0)
-                            output.mux(out_packet)
-                        drain_audio()
-
-                elif packet.stream == audio_stream and not audio_done:
-                    for resampled in itertools.chain.from_iterable(map(resampler.resample, packet.decode())):
-                        frame_start = None
-                        if resampled.pts is not None:
-                            # passthrough frames keep the source stream's time base
-                            tb = resampled.time_base if resampled.time_base else audio_time_base
-                            frame_start = float(resampled.pts * tb)
-                            if duration and not audio_started and frame_start >= start_time + duration:
-                                audio_done = True
-                                break
-                        if not audio_started:
-                            if frame_start is None:
-                                frame_start = 0.0
-                            to_skip = max(0, int((start_time - frame_start) * sample_rate))
-                            if to_skip >= resampled.samples:
-                                continue
-                            audio_started = True
-                            if duration and frame_start > start_time:
-                                duration_cap = min(duration_cap, math.ceil((start_time + duration - frame_start) * sample_rate))
-                            if to_skip:
-                                pending_audio.append(audio_frame_from_ndarray(resampled.to_ndarray()[..., to_skip:]))
-                                continue
-                        pending_audio.append(resampled)
-                        if video_done:
-                            # the video window is complete so the cap is final, but containers
-                            # that interleave audio behind video (fragmented mp4) still owe most
-                            # of it: stop only once the demuxed audio covers the cap
-                            cap = drain_audio()
-                            if pending_audio or samples_written >= cap:
-                                drain_audio(final=True)
-                                audio_done = True
-                                break
-
-            if output is None:
-                raise ValueError(f"No decodable video frames found in file '{self.__file}'")
-            if out_audio is not None and not audio_done:
-                drain_audio(final=True)
-            window_fill = last_video_end - last_video_pts if video_done and last_video_pts is not None else 0
-            for out_packet in out_video.encode(None):
-                duration = video_frame_durations.pop(out_packet.pts, 0)
-                if out_packet.pts == last_video_pts:
-                    duration = max(duration, window_fill)
-                out_packet.duration = duration
-                output.mux(out_packet)
-            if out_audio is not None:
-                output.mux(out_audio.encode(None))
-        except BaseException:
-            if output is not None:
-                output.close()
-                if isinstance(path, (str, os.PathLike)) and os.path.exists(path):
-                    os.remove(path)
-            raise
-        else:
-            if output is not None:
-                output.close()
-
     def _get_first_video_stream(self, container: InputContainer):
         if len(container.streams.video):
             return container.streams.video[0]
@@ -829,12 +527,22 @@ class VideoFromComponents(VideoInput):
         bit_depth: int | None = None,
     ):
         """Save the video to a file path or BytesIO buffer."""
-        open_kwargs = mp4_output_open_kwargs(path, format, codec)
+        if format != VideoContainer.AUTO and format != VideoContainer.MP4:
+            raise ValueError("Only MP4 format is supported for now")
+        if codec != VideoCodec.AUTO and codec != VideoCodec.H264:
+            raise ValueError("Only H264 codec is supported for now")
         # None means "use the depth this video was created with" (CreateVideo's choice).
         if bit_depth is None:
             bit_depth = self.__bit_depth
         is_10bit = bit_depth >= 10
-        with av.open(path, **open_kwargs) as output:
+        extra_kwargs = {}
+        if isinstance(format, VideoContainer) and format != VideoContainer.AUTO:
+            extra_kwargs["format"] = format.value
+        elif isinstance(path, io.BytesIO):
+            # BytesIO has no file extension, so av.open can't infer the format.
+            # Default to mp4 since that's the only supported format anyway.
+            extra_kwargs["format"] = "mp4"
+        with av.open(path, mode='w', options={'movflags': 'use_metadata_tags'}, **extra_kwargs) as output:
             # Add metadata before writing any streams
             if metadata is not None:
                 for key, value in metadata.items():

comfy_api/latest/_io.py

@@ -1261,6 +1261,155 @@ class DynamicSlot(ComfyTypeI):
             out_dict[input_type][finalized_id] = value
             out_dict["dynamic_paths"][finalized_id] = finalize_prefix(curr_prefix, curr_prefix[-1])
 
+@comfytype(io_type="COMFY_DYNAMICGROUP_V3")
+class DynamicGroup(ComfyTypeI):
+    """A repeatable group of widget inputs (e.g. lora_name + strength stacked into N rows).
+
+    At execution time the node receives a ``list[dict]`` where each element is a row.
+
+    Example::
+
+        io.DynamicGroup.Input(
+            "loras",
+            template=[
+                io.Combo.Input("lora_name", options=folder_paths.get_filename_list("loras")),
+                io.Float.Input("strength", default=1.0, min=-100, max=100, step=0.01),
+            ],
+            min=0,
+            max=50,
+        )
+        # execute receives: loras: list[dict] = [{"lora_name": "x.safetensors", "strength": 1.0}, ...]
+    """
+
+    Type = list[dict[str, Any]]
+    _MaxRows = 100
+
+    class Input(DynamicInput):
+        def __init__(
+            self,
+            id: str,
+            template: list["Input"],
+            min: int = 0,
+            max: int = 50,
+            display_name: str = None,
+            optional: bool = False,
+            tooltip: str = None,
+            lazy: bool = None,
+            extra_dict=None,
+            group_name: str = "Group",
+        ):
+            super().__init__(id, display_name, optional, tooltip, lazy, extra_dict)
+            assert len(template) > 0, "DynamicGroup template must have at least one field."
+            for t in template:
+                assert isinstance(t, WidgetInput), (
+                    f"DynamicGroup template field '{t.id}' must be a WidgetInput subclass "
+                    f"(Combo, Float, Int, String, Boolean, Color). Got {type(t).__name__}."
+                )
+                assert not isinstance(t, DynamicInput), (
+                    f"DynamicGroup template field '{t.id}' must not be a DynamicInput. "
+                    "Nesting dynamic inputs inside DynamicGroup is not supported."
+                )
+            field_ids = [t.id for t in template]
+            assert len(field_ids) == len(set(field_ids)), (
+                f"DynamicGroup template field ids must be unique within a row. Got: {field_ids}"
+            )
+            # Reject "." in group id and template field ids: slot_id encoding uses "." as a
+            # delimiter (<group_id>.<row>.<field_id>), so any "." in these names would cause
+            # path.split(".") to produce the wrong number of segments during decoding.
+            assert "." not in id, (
+                f"DynamicGroup id must not contain '.'. Got: '{id}'"
+            )
+            for t in template:
+                assert "." not in t.id, (
+                    f"DynamicGroup template field id must not contain '.'. Got: '{t.id}'"
+                )
+            assert min >= 0, "DynamicGroup min must be >= 0."
+            assert max >= 1, "DynamicGroup max must be >= 1."
+            assert max <= DynamicGroup._MaxRows, f"DynamicGroup max must be <= {DynamicGroup._MaxRows}."
+            assert min <= max, "DynamicGroup min must be <= max."
+            self.template = template
+            self.min = min
+            self.max = max
+            self.group_name = group_name
+
+        def get_all(self) -> list["Input"]:
+            return [self] + list(self.template)
+
+        def as_dict(self):
+            return super().as_dict() | prune_dict({
+                "template": create_input_dict_v1(self.template),
+                "min": self.min,
+                "max": self.max,
+                "group_name": self.group_name,
+            })
+
+        def validate(self):
+            for t in self.template:
+                t.validate()
+
+    @staticmethod
+    def _expand_schema_for_dynamic(
+        out_dict: dict[str, Any],
+        live_inputs: dict[str, Any],
+        value: tuple[str, dict[str, Any]],
+        input_type: str,
+        curr_prefix: list[str] | None,
+    ):
+        info = value[1]
+        min_rows: int = info.get("min", 0)
+        max_rows: int = info.get("max", DynamicGroup._MaxRows)
+        template: dict[str, Any] = info.get("template", {})
+
+        # Collect all template field specs across required/optional sections
+        field_specs: list[tuple[str, tuple[str, dict[str, Any]], bool]] = []
+        for field_required_key in ("required", "optional"):
+            section = template.get(field_required_key, {})
+            is_required_field = field_required_key == "required"
+            for field_id, field_value in section.items():
+                field_specs.append((field_id, field_value, is_required_field))
+
+        # Determine how many rows are currently present by scanning live_inputs
+        finalized_prefix = finalize_prefix(curr_prefix)
+        present_rows = 0
+        for live_key in live_inputs:
+            # Keys look like "<prefix>.<row>.<field_id>"
+            if live_key.startswith(finalized_prefix + "."):
+                remainder = live_key[len(finalized_prefix) + 1:]
+                parts = remainder.split(".", 1)
+                if len(parts) >= 1:
+                    try:
+                        row_idx = int(parts[0])
+                        present_rows = max(present_rows, row_idx + 1)
+                    except ValueError:
+                        pass
+
+        if present_rows > max_rows:
+            raise ValueError(
+                f"DynamicGroup input '{finalized_prefix}' received {present_rows} rows but max is {max_rows}."
+            )
+        row_count = max(min_rows, present_rows)
+
+        for row in range(row_count):
+            for field_id, field_value, is_required_field in field_specs:
+                slot_id = f"{finalized_prefix}.{row}.{field_id}"
+                if row < min_rows and is_required_field:
+                    out_dict["required"][slot_id] = field_value
+                else:
+                    out_dict["optional"][slot_id] = field_value
+                # Register into dynamic_paths so build_nested_inputs places value at the right path
+                out_dict["dynamic_paths"][slot_id] = slot_id
+
+        # Track the list root path so build_nested_inputs can convert the index dict to a list
+        out_dict.setdefault("list_paths", set()).add(finalized_prefix)
+
+        # Handle the empty case (0 rows) – emit an empty-list default for the parent.
+        # This must only fire when there are genuinely no rows; otherwise the parent
+        # path would clobber the per-row dict built from the slot ids above.
+        if row_count == 0:
+            out_dict["dynamic_paths"][finalized_prefix] = finalized_prefix
+            out_dict["dynamic_paths_default_value"][finalized_prefix] = DynamicPathsDefaultValue.EMPTY_LIST
+
+
 @comfytype(io_type="IMAGECOMPARE")
 class ImageCompare(ComfyTypeI):
   Type = dict
@@ -1418,6 +1567,8 @@ def setup_dynamic_input_funcs():
     register_dynamic_input_func(DynamicCombo.io_type, DynamicCombo._expand_schema_for_dynamic)
     # DynamicSlot.Input
     register_dynamic_input_func(DynamicSlot.io_type, DynamicSlot._expand_schema_for_dynamic)
+    # DynamicGroup.Input
+    register_dynamic_input_func(DynamicGroup.io_type, DynamicGroup._expand_schema_for_dynamic)
 
 if len(DYNAMIC_INPUT_LOOKUP) == 0:
     setup_dynamic_input_funcs()
@@ -1429,6 +1580,8 @@ class V3Data(TypedDict):
     'Dictionary where the keys are the input ids and the values dictate how to turn the inputs into a nested dictionary.'
     dynamic_paths_default_value: dict[str, Any]
     'Dictionary where the keys are the input ids and the values are a string from DynamicPathsDefaultValue for the inputs if value is None.'
+    list_paths: set[str]
+    'Set of top-level keys whose index-keyed dict values should be converted to a sorted list[dict] after build_nested_inputs runs.'
     create_dynamic_tuple: bool
     'When True, the value of the dynamic input will be in the format (value, path_key).'
 
@@ -1770,6 +1923,7 @@ def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], i
         "optional": {},
         "dynamic_paths": {},
         "dynamic_paths_default_value": {},
+        "list_paths": set(),
     }
     d = d.copy()
     # ignore hidden for parsing
@@ -1785,6 +1939,10 @@ def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], i
     dynamic_paths_default_value = out_dict.pop("dynamic_paths_default_value", None)
     if dynamic_paths_default_value is not None and len(dynamic_paths_default_value) > 0:
         v3_data["dynamic_paths_default_value"] = dynamic_paths_default_value
+    # list_paths: keys whose nested dict should be post-converted to a sorted list[dict]
+    list_paths = out_dict.pop("list_paths", None)
+    if list_paths:
+        v3_data["list_paths"] = list_paths
     return out_dict, hidden, v3_data
 
 def parse_class_inputs(out_dict: dict[str, Any], live_inputs: dict[str, Any], curr_dict: dict[str, Any], curr_prefix: list[str] | None=None) -> None:
@@ -1820,10 +1978,12 @@ def add_to_dict_v1(i: Input, d: dict):
 
 class DynamicPathsDefaultValue:
     EMPTY_DICT = "empty_dict"
+    EMPTY_LIST = "empty_list"
 
 def build_nested_inputs(values: dict[str, Any], v3_data: V3Data):
     paths = v3_data.get("dynamic_paths", None)
     default_value_dict = v3_data.get("dynamic_paths_default_value", {})
+    list_paths: set[str] = v3_data.get("list_paths", set()) or set()
     if paths is None:
         return values
     values = values.copy()
@@ -1846,6 +2006,8 @@ def build_nested_inputs(values: dict[str, Any], v3_data: V3Data):
                     default_option = default_value_dict.get(key, None)
                     if default_option == DynamicPathsDefaultValue.EMPTY_DICT:
                         value = {}
+                    elif default_option == DynamicPathsDefaultValue.EMPTY_LIST:
+                        value = []
                 if create_tuple:
                     value = (value, key)
                 current[p] = value
@@ -1853,6 +2015,34 @@ def build_nested_inputs(values: dict[str, Any], v3_data: V3Data):
                 current = current.setdefault(p, {})
 
     values.update(result)
+
+    # Post-pass: convert index-keyed dicts to sorted lists for io.DynamicGroup fields
+    for list_path in list_paths:
+        parts = list_path.split(".")
+        # Navigate to the parent container, then convert the leaf
+        container = values
+        for part in parts[:-1]:
+            if not isinstance(container, dict) or part not in container:
+                container = None
+                break
+            container = container[part]
+        if container is None:
+            continue
+        leaf_key = parts[-1]
+        leaf = container.get(leaf_key, None)
+        if isinstance(leaf, dict):
+            try:
+                sorted_rows = [leaf[k] for k in sorted(leaf.keys(), key=int)]
+                container[leaf_key] = sorted_rows
+            except (ValueError, TypeError):
+                # Keys are not all integers; leave as-is
+                pass
+        elif isinstance(leaf, list):
+            # Already a list (e.g. the EMPTY_LIST default was applied above)
+            pass
+        elif leaf is None:
+            container[leaf_key] = []
+
     return values
 
 
@@ -2417,7 +2607,9 @@ __all__ = [
     # Dynamic Types
     "MatchType",
     "DynamicCombo",
+    "DynamicSlot",
     "Autogrow",
+    "DynamicGroup",
     # Other classes
     "HiddenHolder",
     "Hidden",

comfy_extras/nodes_lora_stack.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from typing_extensions import override
+
+import comfy.sd
+import comfy.utils
+import folder_paths
+from comfy_api.latest import ComfyExtension, io
+
+
+def _load_lora_file(lora_name: str):
+    lora_path = folder_paths.get_full_path_or_raise("loras", lora_name)
+    return comfy.utils.load_torch_file(lora_path, safe_load=True, return_metadata=True)
+
+
+def _lora_template() -> list[io.Input]:
+    return [
+        io.Combo.Input("lora_name", options=folder_paths.get_filename_list("loras"),
+                       tooltip="The name of the LoRA file to apply."),
+        io.Float.Input("strength", default=1.0, min=-100.0, max=100.0, step=0.01,
+                       tooltip="How strongly to apply this LoRA. 0 = off, negative inverts the effect."),
+    ]
+
+
+class LoadLoraModel(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoadLoraModel",
+            display_name="Load LoRA (Model)",
+            search_aliases=["lora", "load lora", "apply lora", "lora model", "lora stack"],
+            category="model/loaders",
+            description="Apply a stack of LoRAs to a diffusion model. Add one row per LoRA; "
+                        "each row picks a LoRA file and its strength.",
+            inputs=[
+                io.Model.Input("model", tooltip="The diffusion model the LoRAs will be applied to."),
+                io.DynamicGroup.Input(
+                    "loras",
+                    template=_lora_template(),
+                    min=1,
+                    max=50,
+                    tooltip="Each row applies one LoRA to the model.",
+                    group_name="LoRA",
+                ),
+            ],
+            outputs=[io.Model.Output(tooltip="The modified diffusion model.")],
+        )
+
+    @classmethod
+    def execute(cls, model, loras: list[dict]) -> io.NodeOutput:
+        for row in loras:
+            lora_name = row.get("lora_name")
+            strength = row.get("strength", 1.0)
+            if not lora_name or lora_name == "none" or strength == 0:
+                continue
+            lora, metadata = _load_lora_file(lora_name)
+            model, _ = comfy.sd.load_lora_for_models(model, None, lora, strength, 0, lora_metadata=metadata)
+        return io.NodeOutput(model)
+
+
+class LoadLoraTextEncoder(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoadLoraTextEncoder",
+            display_name="Load LoRA (Text Encoder)",
+            search_aliases=["lora", "load lora", "apply lora", "clip lora", "lora stack"],
+            category="model/loaders",
+            description="Apply a stack of LoRAs to a CLIP text encoder. Add one row per LoRA; "
+                        "each row picks a LoRA file and its strength.",
+            inputs=[
+                io.Clip.Input("clip", tooltip="The CLIP text encoder the LoRAs will be applied to."),
+                io.DynamicGroup.Input(
+                    "loras",
+                    template=_lora_template(),
+                    min=1,
+                    max=50,
+                    tooltip="Each row applies one LoRA to the text encoder.",
+                    group_name="LoRA",
+                ),
+            ],
+            outputs=[io.Clip.Output(tooltip="The modified CLIP text encoder.")],
+        )
+
+    @classmethod
+    def execute(cls, clip, loras: list[dict]) -> io.NodeOutput:
+        for row in loras:
+            lora_name = row.get("lora_name")
+            strength = row.get("strength", 1.0)
+            if not lora_name or lora_name == "none" or strength == 0:
+                continue
+            lora, metadata = _load_lora_file(lora_name)
+            _, clip = comfy.sd.load_lora_for_models(None, clip, lora, 0, strength, lora_metadata=metadata)
+        return io.NodeOutput(clip)
+
+
+class LoraStackExtension(ComfyExtension):
+    @override
+    async def get_node_list(self) -> list[type[io.ComfyNode]]:
+        return [
+            LoadLoraModel,
+            LoadLoraTextEncoder,
+        ]
+
+
+async def comfy_entrypoint() -> LoraStackExtension:
+    return LoraStackExtension()

nodes.py

@@ -2503,6 +2503,7 @@ async def init_builtin_extra_nodes():
         "nodes_triposplat.py",
         "nodes_depth_anything_3.py",
         "nodes_seed.py",
+        "nodes_lora_stack.py",
     ]
 
     import_failed = []

tests-unit/comfy_api_test/io_dynamic_group_test.py

@@ -0,0 +1,204 @@
+"""Unit tests for io.DynamicGroup: expansion/reconstruction (0-row and N-row cases)."""
+import sys
+import types
+import pytest
+
+# Stub torch (type-hint only in _io.py; real torch not available in unit-test env)
+if "torch" not in sys.modules:
+    _torch_stub = types.ModuleType("torch")
+    _torch_stub.Tensor = object  # type: ignore[attr-defined]
+    sys.modules["torch"] = _torch_stub
+
+from comfy_api.latest._io import (  # noqa: E402
+    DynamicGroup,
+    Float,
+    Int,
+    String,
+    Boolean,
+    get_finalized_class_inputs,
+    build_nested_inputs,
+    create_input_dict_v1,
+    setup_dynamic_input_funcs,
+)
+
+# Make sure dynamic input funcs are registered (may already be done at import time)
+setup_dynamic_input_funcs()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_class_inputs(group_input: DynamicGroup.Input) -> dict:
+    """Wrap a DynamicGroup.Input into the required/optional dict structure."""
+    return create_input_dict_v1([group_input])
+
+
+def _run(group_input: DynamicGroup.Input, live_values: dict) -> dict:
+    """End-to-end helper: expand schema + reconstruct values.
+
+    Mirrors the production split in execution.py:
+      1. get_finalized_class_inputs  (schema expansion, line 162)
+      2. build_nested_inputs          (value reconstruction, line 281)
+
+    The two steps are separate in production because the engine resolves
+    linked node outputs between them, but in tests we supply values directly.
+    """
+    class_inputs = _make_class_inputs(group_input)
+    _, _, v3_data = get_finalized_class_inputs(class_inputs, live_values)
+    return build_nested_inputs(dict(live_values), v3_data)
+
+
+# ---------------------------------------------------------------------------
+# Schema construction
+# ---------------------------------------------------------------------------
+
+class TestDynamicGroupInputConstruction:
+    def test_basic_construction(self):
+        inp = DynamicGroup.Input(
+            "loras",
+            template=[
+                Float.Input("strength", default=1.0),
+                String.Input("name"),
+            ],
+            min=0,
+            max=10,
+        )
+        assert inp.id == "loras"
+        assert inp.min == 0
+        assert inp.max == 10
+        assert len(inp.template) == 2
+
+    def test_get_all_includes_self_and_template(self):
+        inp = DynamicGroup.Input(
+            "items",
+            template=[Float.Input("value")],
+        )
+        all_inputs = inp.get_all()
+        assert all_inputs[0] is inp
+        assert all_inputs[1].id == "value"
+
+    def test_as_dict_has_template_min_max(self):
+        inp = DynamicGroup.Input(
+            "items",
+            template=[Float.Input("val", default=0.5)],
+            min=1,
+            max=5,
+        )
+        d = inp.as_dict()
+        assert "template" in d
+        assert d["min"] == 1
+        assert d["max"] == 5
+
+    def test_duplicate_field_ids_raises(self):
+        with pytest.raises(AssertionError):
+            DynamicGroup.Input(
+                "bad",
+                template=[Float.Input("x"), Float.Input("x")],
+            )
+
+    def test_empty_template_raises(self):
+        with pytest.raises(AssertionError):
+            DynamicGroup.Input("bad", template=[])
+
+    def test_min_gt_max_raises(self):
+        with pytest.raises(AssertionError):
+            DynamicGroup.Input("bad", template=[Float.Input("x")], min=5, max=3)
+
+    def test_max_exceeds_limit_raises(self):
+        with pytest.raises(AssertionError):
+            DynamicGroup.Input("bad", template=[Float.Input("x")], max=101)
+
+    def test_dynamic_input_in_template_raises(self):
+        with pytest.raises(AssertionError):
+            DynamicGroup.Input(
+                "bad",
+                template=[DynamicGroup.Input("nested", template=[Float.Input("x")])],
+            )
+
+    def test_validate_calls_through(self):
+        inp = DynamicGroup.Input("items", template=[Float.Input("val", min=-1.0, max=1.0)])
+        inp.validate()  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# 0-row case
+# ---------------------------------------------------------------------------
+
+class TestZeroRows:
+    def test_empty_live_inputs_produces_empty_list(self):
+        """With min=0 and no live values, the result should be an empty list."""
+        inp = DynamicGroup.Input("loras", template=[Float.Input("strength", default=1.0)], min=0, max=10)
+        assert _run(inp, {}).get("loras") == []
+
+    def test_min_zero_with_values(self):
+        """min=0 but 2 rows of live data."""
+        inp = DynamicGroup.Input("loras", template=[Float.Input("strength", default=1.0)], min=0, max=10)
+        result = _run(inp, {"loras.0.strength": 0.8, "loras.1.strength": 0.5})
+        assert result["loras"] == [{"strength": 0.8}, {"strength": 0.5}]
+
+
+# ---------------------------------------------------------------------------
+# N-row case
+# ---------------------------------------------------------------------------
+
+class TestNRows:
+    def test_two_rows_two_fields(self):
+        """Two rows with two fields each produce a list[dict]."""
+        inp = DynamicGroup.Input(
+            "loras",
+            template=[String.Input("lora_name"), Float.Input("strength", default=1.0)],
+            min=0, max=50,
+        )
+        result = _run(inp, {
+            "loras.0.lora_name": "model_a.safetensors", "loras.0.strength": 0.9,
+            "loras.1.lora_name": "model_b.safetensors", "loras.1.strength": 0.4,
+        })
+        assert result["loras"] == [
+            {"lora_name": "model_a.safetensors", "strength": 0.9},
+            {"lora_name": "model_b.safetensors", "strength": 0.4},
+        ]
+
+    def test_rows_are_sorted_by_index(self):
+        """Rows must be in ascending index order even if dict iteration is unordered."""
+        inp = DynamicGroup.Input("items", template=[Int.Input("v", default=0)], min=0, max=10)
+        result = _run(inp, {"items.0.v": 10, "items.2.v": 30, "items.1.v": 20})
+        assert [row["v"] for row in result["items"]] == [10, 20, 30]
+
+    def test_min_rows_schema_slots(self):
+        """With min=2 and no live data, 2 slots must appear in the expanded schema."""
+        inp = DynamicGroup.Input("items", template=[Float.Input("val", default=0.0)], min=2, max=5)
+        out, _, _ = get_finalized_class_inputs(_make_class_inputs(inp), {})
+        all_slots = {**out.get("required", {}), **out.get("optional", {})}
+        assert "items.0.val" in all_slots
+        assert "items.1.val" in all_slots
+
+    def test_min_rows_reconstructs_when_no_values(self):
+        """min=2 with NO live values must still yield a 2-element list,
+        not collapse to [] (regression: parent-path clobber)."""
+        inp = DynamicGroup.Input("items", template=[Float.Input("val", default=0.0)], min=2, max=5)
+        result = _run(inp, {})
+        assert len(result["items"]) == 2
+        assert all("val" in row for row in result["items"])
+
+    def test_min_rows_reconstructs_with_partial_values(self):
+        """min=2 with only the first row's value present still yields 2 rows."""
+        inp = DynamicGroup.Input("items", template=[Float.Input("val", default=0.0)], min=2, max=5)
+        result = _run(inp, {"items.0.val": 0.7})
+        assert len(result["items"]) == 2
+        assert result["items"][0]["val"] == 0.7
+        assert result["items"][1]["val"] is None
+
+    def test_list_paths_in_v3_data(self):
+        """list_paths must contain the group id so build_nested_inputs knows to convert."""
+        inp = DynamicGroup.Input("things", template=[Boolean.Input("flag")], min=0, max=5)
+        _, _, v3_data = get_finalized_class_inputs(_make_class_inputs(inp), {})
+        assert "things" in v3_data.get("list_paths", set())
+
+    def test_no_leftover_flat_keys(self):
+        """Flat keys must be consumed; only the reconstructed list remains."""
+        inp = DynamicGroup.Input("rows", template=[Float.Input("x", default=0.0)], min=0, max=5)
+        result = _run(inp, {"rows.0.x": 1.0, "rows.1.x": 2.0})
+        assert "rows.0.x" not in result
+        assert "rows.1.x" not in result
+        assert isinstance(result["rows"], list)

tests-unit/comfy_api_test/video_types_test.py

@@ -2,12 +2,11 @@ import pytest
 import torch
 import tempfile
 import os
-import sys
 import av
 import io
 from fractions import Fraction
 from comfy_api.input_impl.video_types import VideoFromFile, VideoFromComponents
-from comfy_api.util.video_types import VideoComponents, VideoContainer, VideoCodec
+from comfy_api.util.video_types import VideoComponents
 from comfy_api.input.basic_types import AudioInput
 from av.error import InvalidDataError
 
@@ -238,526 +237,3 @@ def test_duration_consistency(video_components):
     manual_duration = float(components.images.shape[0] / components.frame_rate)
 
     assert duration == pytest.approx(manual_duration)
-
-
-def create_transcode_source(
-    width=64, height=64, frames=30, fps=30, audio_streams=1, undecodable_audio=0, rotation=False,
-    container_format="mov", audio_codec="pcm_s16le",
-):
-    """Create a temp video that save_to must transcode (mpeg4 video, so codec != h264).
-
-    ``undecodable_audio`` trailing PCM streams get their fourcc corrupted so no decoder exists
-    (``codec_context is None``), like the APAC track in iPhone spatial-audio recordings.
-    ``rotation`` patches a 90-degree display matrix into the video track header.
-    """
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format=container_format) as container:
-        video_stream = container.add_stream("mpeg4", rate=fps)
-        video_stream.width = width
-        video_stream.height = height
-        video_stream.pix_fmt = "yuv420p"
-        audio = []
-        for _ in range(audio_streams + undecodable_audio):
-            stream = container.add_stream(audio_codec, rate=44100)
-            stream.sample_rate = 44100
-            audio.append(stream)
-
-        for i in range(frames):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((height, width, 3), (i * 7) % 256, dtype=torch.uint8).numpy(),
-                format="rgb24",
-            )
-            container.mux(video_stream.encode(frame.reformat(format="yuv420p")))
-        # write audio in 1024-sample frames, like real decoders produce, so the
-        # per-frame skip/cap logic in the transcode path actually runs
-        for stream in audio:
-            for offset in range(0, 44100 * frames // fps, 1024):
-                n = min(1024, 44100 * frames // fps - offset)
-                audio_frame = av.AudioFrame.from_ndarray(
-                    torch.zeros(1, n, dtype=torch.int16).numpy(), format="s16", layout="mono"
-                )
-                audio_frame.sample_rate = 44100
-                audio_frame.pts = offset
-                container.mux(stream.encode(audio_frame))
-        for stream in [video_stream, *audio]:
-            container.mux(stream.encode(None))
-
-    data = bytearray(buffer.getvalue())
-    end = len(data)
-    for _ in range(undecodable_audio):
-        end = data.rindex(b"sowt", 0, end)
-        data[end:end + 4] = b"Xpac"
-    if rotation:
-        # the 3x3 display matrix sits 40 bytes into the version-0 tkhd payload; first tkhd
-        # inside moov = video track (search from moov so mdat bytes can't false-match)
-        matrix_offset = data.index(b"tkhd", data.rindex(b"moov")) + 4 + 40
-        values = [0, 1 << 16, 0, -(1 << 16), 0, 0, 0, 0, 1 << 30]
-        data[matrix_offset:matrix_offset + 36] = b"".join(v.to_bytes(4, "big", signed=True) for v in values)
-
-    tmp = tempfile.NamedTemporaryFile(suffix=f".{container_format}", delete=False)
-    tmp.write(bytes(data))
-    tmp.close()
-    return tmp.name
-
-
-def transcode_and_probe(video):
-    buffer = io.BytesIO()
-    video.save_to(buffer, format=VideoContainer.MP4, codec=VideoCodec.H264)
-    buffer.seek(0)
-    with av.open(buffer) as container:
-        video_stream = container.streams.video[0]
-        audio_stream = container.streams.audio[0] if container.streams.audio else None
-        frames = 0
-        first_pts = None
-        for packet in container.demux(video_stream):
-            for frame in packet.decode():
-                if first_pts is None:
-                    first_pts = frame.pts
-                frames += 1
-        return {
-            "codec": video_stream.codec_context.name,
-            "width": video_stream.codec_context.width,
-            "height": video_stream.codec_context.height,
-            "frames": frames,
-            "first_pts": first_pts,
-            "video_seconds": float(video_stream.duration * video_stream.time_base) if video_stream.duration else None,
-            "audio_seconds": float(audio_stream.duration * audio_stream.time_base)
-            if audio_stream and audio_stream.duration else None,
-            "audio_codecs": [s.codec_context.name for s in container.streams.audio],
-        }
-
-
-def test_save_to_transcode_streams_without_buffering_frames():
-    """Transcoding must not decode the whole video into memory first (~2 GiB for this source)"""
-    resource = pytest.importorskip("resource")  # no getrusage on Windows
-    rss_scale = 1 if sys.platform == "darwin" else 1024  # ru_maxrss: bytes on macOS, KiB elsewhere
-    # ru_maxrss is a lifetime peak: a heavier test running earlier would shrink the measured
-    # delta and quietly defang this canary, so keep this source the biggest thing in the suite
-    file_path = create_transcode_source(width=640, height=480, frames=300)
-    try:
-        rss_before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * rss_scale
-        result = transcode_and_probe(VideoFromFile(file_path))
-        rss_delta = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * rss_scale - rss_before
-
-        assert result["codec"] == "h264"
-        assert result["frames"] == 300
-        assert rss_delta < 500 * 2**20, f"transcode buffered frames in RAM (peak grew {rss_delta / 2**20:.0f} MiB)"
-    finally:
-        os.unlink(file_path)
-
-
-def test_save_to_transcode_honors_trim_window():
-    """start_time/duration trim applies to both video and audio on the streaming path"""
-    file_path = create_transcode_source(frames=90)  # 3s @ 30fps
-    try:
-        result = transcode_and_probe(VideoFromFile(file_path, start_time=1, duration=1))
-        assert result["frames"] == pytest.approx(30, abs=2)
-        assert result["first_pts"] == 0  # trimmed output is rebased to start at zero
-        assert result["video_seconds"] == pytest.approx(1.0, abs=0.1)
-        assert result["audio_seconds"] == pytest.approx(1.0, abs=0.1)
-    finally:
-        os.unlink(file_path)
-
-
-def test_save_to_transcode_keeps_audio_of_sparse_video():
-    """Audio that runs ahead of a sparse video track (slideshows, timelapses) must be
-    kept in full — it is only clamped to the video's end, never to the video cursor."""
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        video_stream = container.add_stream("mpeg4", rate=30)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        audio_stream = container.add_stream("aac", rate=48000, layout="stereo")
-        for t in (0, 30, 60):  # 3 frames spread over 60 seconds
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), t * 4, dtype=torch.uint8).numpy(), format="rgb24"
-            ).reformat(format="yuv420p")
-            frame.pts = t * 15360
-            frame.time_base = Fraction(1, 15360)
-            container.mux(video_stream.encode(frame))
-        container.mux(video_stream.encode(None))
-        for offset in range(0, 48000 * 60, 1024):
-            n = min(1024, 48000 * 60 - offset)
-            audio_frame = av.AudioFrame.from_ndarray(
-                torch.zeros(2, n, dtype=torch.float32).numpy(), format="fltp", layout="stereo"
-            )
-            audio_frame.sample_rate = 48000
-            audio_frame.pts = offset
-            audio_frame.time_base = Fraction(1, 48000)
-            container.mux(audio_stream.encode(audio_frame))
-        container.mux(audio_stream.encode(None))
-
-    buffer.seek(0)
-    result = transcode_and_probe(VideoFromFile(buffer))
-    assert result["audio_seconds"] == pytest.approx(60.0, abs=1.0)
-
-
-def test_save_to_transcode_vfr_audio_covers_video_span():
-    """A trim window in the sparse region of a VFR file keeps audio for the true pts span
-    of the kept frames. Deriving the span as frames/average_rate undercuts it badly: the
-    average is dominated by the dense region (and can be plain wrong on MediaRecorder files)."""
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        video_stream = container.add_stream("mpeg4", rate=30)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        audio_stream = container.add_stream("aac", rate=48000, layout="stereo")
-        # 10 frames inside the first second, then one every 1.25 s
-        for i, t in enumerate([x / 10 for x in range(10)] + [1.0, 2.25, 3.5, 4.75]):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), (i * 16) % 256, dtype=torch.uint8).numpy(), format="rgb24"
-            ).reformat(format="yuv420p")
-            frame.pts = int(t * 15360)
-            frame.time_base = Fraction(1, 15360)
-            container.mux(video_stream.encode(frame))
-        container.mux(video_stream.encode(None))
-        for offset in range(0, 48000 * 6, 1024):
-            n = min(1024, 48000 * 6 - offset)
-            audio_frame = av.AudioFrame.from_ndarray(
-                torch.zeros(2, n, dtype=torch.float32).numpy(), format="fltp", layout="stereo"
-            )
-            audio_frame.sample_rate = 48000
-            audio_frame.pts = offset
-            audio_frame.time_base = Fraction(1, 48000)
-            container.mux(audio_stream.encode(audio_frame))
-        container.mux(audio_stream.encode(None))
-
-    buffer.seek(0)
-    result = transcode_and_probe(VideoFromFile(buffer, start_time=1, duration=5))
-    # kept frames: 1.0/2.25/3.5/4.75 s -> rebased span 3.75 s + one nominal interval
-    assert result["frames"] == 4
-    assert result["audio_seconds"] == pytest.approx(4.0, abs=0.45)
-
-
-def test_save_to_transcode_trims_audio_in_stream_time_base_units():
-    """Matroska audio timestamps tick in 1/1000, not 1/sample_rate; trim and audio timing
-    must convert through the frame's time base instead of assuming sample units. AAC audio,
-    because it decodes straight to the encoder's format and hits the resampler passthrough
-    that keeps the source time base on the frames."""
-    file_path = create_transcode_source(frames=90, container_format="matroska", audio_codec="aac")
-    try:
-        result = transcode_and_probe(VideoFromFile(file_path, start_time=1, duration=1))
-        assert result["audio_codecs"] == ["aac"]
-        assert result["video_seconds"] == pytest.approx(1.0, abs=0.1)
-        assert result["audio_seconds"] == pytest.approx(1.0, abs=0.1)
-    finally:
-        os.unlink(file_path)
-
-
-def test_save_to_transcode_learns_unprobed_audio_params():
-    """mpegts is only probed a few seconds deep at open, so an audio stream whose first
-    packet comes later (live captures where audio kicks in late) still has sample_rate 0
-    when the transcode starts; the parameters must be learned from the stream itself."""
-    sample_rate, fps, video_seconds, audio_start = 48000, 30, 13, 12
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mpegts") as container:
-        video_stream = container.add_stream("mpeg4", rate=fps)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        audio_stream = container.add_stream("aac", rate=sample_rate, layout="mono")
-        for i in range(video_seconds * fps):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), (i * 7) % 256, dtype=torch.uint8).numpy(), format="rgb24"
-            )
-            container.mux(video_stream.encode(frame.reformat(format="yuv420p")))
-        for offset in range(0, (video_seconds - audio_start) * sample_rate, 1024):
-            n = min(1024, (video_seconds - audio_start) * sample_rate - offset)
-            audio_frame = av.AudioFrame.from_ndarray(
-                torch.zeros(1, n, dtype=torch.float32).numpy(), format="fltp", layout="mono"
-            )
-            audio_frame.sample_rate = sample_rate
-            audio_frame.pts = audio_start * sample_rate + offset
-            container.mux(audio_stream.encode(audio_frame))
-        for stream in (video_stream, audio_stream):
-            container.mux(stream.encode(None))
-
-    buffer.seek(0)
-    with av.open(buffer) as container:
-        # the scenario requires unprobed parameters; if a future FFmpeg probes deeper,
-        # push audio_start/video_seconds further out to restore it
-        assert container.streams.audio[0].codec_context.sample_rate == 0
-    result = transcode_and_probe(VideoFromFile(buffer))
-    assert result["frames"] == video_seconds * fps
-    assert result["audio_codecs"] == ["aac"]
-    assert result["audio_seconds"] == pytest.approx(1.0, abs=0.1)
-
-    buffer.seek(0)
-    trimmed_before_audio = transcode_and_probe(VideoFromFile(buffer, duration=1))
-    assert trimmed_before_audio["frames"] == fps
-    assert trimmed_before_audio["audio_codecs"] == []
-    assert trimmed_before_audio["audio_seconds"] is None
-
-    buffer.seek(0)
-    trimmed_crossing_audio = transcode_and_probe(VideoFromFile(buffer, start_time=11.5, duration=1))
-    assert trimmed_crossing_audio["frames"] == fps
-    assert trimmed_crossing_audio["audio_codecs"] == ["aac"]
-    assert trimmed_crossing_audio["video_seconds"] == pytest.approx(1.0, abs=0.05)
-    assert trimmed_crossing_audio["audio_seconds"] == pytest.approx(0.5, abs=0.1)
-
-
-def test_save_to_transcode_trimmed_fragmented_mp4_keeps_audio():
-    """Fragmented mp4 (MediaRecorder, DASH/HLS-derived files) delivers audio well behind
-    video, so when the trim window's last video frame arrives the audio demuxed so far
-    does not cover the window yet; the transcode must keep demuxing audio until it does
-    instead of finalizing on the first audio frame it sees afterwards."""
-    sample_rate, fps, seconds = 48000, 30, 6
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4", options={"movflags": "frag_keyframe+empty_moov"}) as container:
-        video_stream = container.add_stream("h264", rate=fps)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        audio_stream = container.add_stream("aac", rate=sample_rate, layout="mono")
-        next_audio_pts = 0
-        for i in range(seconds * fps):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), (i * 7) % 256, dtype=torch.uint8).numpy(), format="rgb24"
-            )
-            container.mux(video_stream.encode(frame.reformat(format="yuv420p")))
-            while next_audio_pts / sample_rate <= i / fps:  # feed audio alongside, like a live pipeline
-                audio_frame = av.AudioFrame.from_ndarray(
-                    torch.zeros(1, 1024, dtype=torch.float32).numpy(), format="fltp", layout="mono"
-                )
-                audio_frame.sample_rate = sample_rate
-                audio_frame.pts = next_audio_pts
-                container.mux(audio_stream.encode(audio_frame))
-                next_audio_pts += 1024
-        for stream in (video_stream, audio_stream):
-            container.mux(stream.encode(None))
-
-    result = transcode_and_probe(VideoFromFile(buffer, start_time=0.5, duration=1.0))
-    assert result["video_seconds"] == pytest.approx(1.0, abs=0.05)
-    assert result["audio_seconds"] == pytest.approx(1.0, abs=0.05)
-
-
-def test_save_to_transcode_sparse_video_keeps_true_duration():
-    """average_rate is not a frame duration: a 3-frame video spanning 60 s averages
-    0.05 fps, and padding the last frame with 1/average_rate used to extend the
-    output — and the audio kept with it — about 20 s past the source span."""
-    sample_rate = 48000
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        video_stream = container.add_stream("mpeg4", rate=30)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        audio_stream = container.add_stream("aac", rate=sample_rate, layout="mono")
-        for i, second in enumerate((0, 30, 60)):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), i * 80, dtype=torch.uint8).numpy(), format="rgb24"
-            ).reformat(format="yuv420p")
-            frame.pts = second * 30
-            frame.time_base = Fraction(1, 30)
-            container.mux(video_stream.encode(frame))
-        for offset in range(0, 90 * sample_rate, 1024):
-            n = min(1024, 90 * sample_rate - offset)
-            audio_frame = av.AudioFrame.from_ndarray(
-                torch.zeros(1, n, dtype=torch.float32).numpy(), format="fltp", layout="mono"
-            )
-            audio_frame.sample_rate = sample_rate
-            audio_frame.pts = offset
-            container.mux(audio_stream.encode(audio_frame))
-        for stream in (video_stream, audio_stream):
-            container.mux(stream.encode(None))
-
-    result = transcode_and_probe(VideoFromFile(buffer))
-    assert result["frames"] == 3
-    # the last frame keeps its true stts duration (1/30 s), not 1/average_rate (~20 s)
-    assert result["video_seconds"] == pytest.approx(60.03, abs=0.05)
-    assert result["audio_seconds"] == pytest.approx(60.03, abs=0.1)
-
-    trimmed = transcode_and_probe(VideoFromFile(buffer, duration=45))
-    assert trimmed["frames"] == 2
-    # a kept frame whose source duration crosses the window end is clamped to it
-    assert trimmed["video_seconds"] == pytest.approx(45.0, abs=0.05)
-    assert trimmed["audio_seconds"] == pytest.approx(45.0, abs=0.1)
-
-
-def test_save_to_transcode_clamps_final_pts_to_declared_stream_duration():
-    """Some iPhone MOVs report a video stream duration that ends before the final
-    decoded frame's nominal duration. A transcode must not turn that trailing
-    timestamp quirk into an extra frame interval compared to the source/remux path."""
-    fps = 30
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        video_stream = container.add_stream("mpeg4", rate=fps)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        for i, pts in enumerate([*range(31), 32]):
-            frame = av.VideoFrame.from_ndarray(
-                torch.full((64, 64, 3), (i * 7) % 256, dtype=torch.uint8).numpy(), format="rgb24"
-            ).reformat(format="yuv420p")
-            frame.pts = pts
-            frame.time_base = Fraction(1, fps)
-            container.mux(video_stream.encode(frame))
-        container.mux(video_stream.encode(None))
-
-    class _StreamProxy:
-        def __init__(self, stream, duration):
-            self._stream = stream
-            self.duration = duration
-
-        def __getattr__(self, name):
-            return getattr(self._stream, name)
-
-    class _StreamsProxy:
-        def __init__(self, video_stream):
-            self.video = [video_stream]
-            self.audio = []
-
-    class _PacketProxy:
-        def __init__(self, packet, stream):
-            self._packet = packet
-            self.stream = stream
-
-        def __getattr__(self, name):
-            return getattr(self._packet, name)
-
-    class _ContainerProxy:
-        def __init__(self, container, stream):
-            self._container = container
-            self._stream = stream
-            self.streams = _StreamsProxy(stream)
-
-        def __getattr__(self, name):
-            return getattr(self._container, name)
-
-        def demux(self, *streams):
-            for packet in self._container.demux(self._stream._stream):
-                yield _PacketProxy(packet, self._stream)
-
-    buffer.seek(0)
-    output = io.BytesIO()
-    with av.open(buffer) as container:
-        real_stream = container.streams.video[0]
-        declared_duration = 32 * int(round((1 / fps) / real_stream.time_base))
-        stream = _StreamProxy(real_stream, declared_duration)
-        VideoFromFile(buffer)._save_transcoded(
-            _ContainerProxy(container, stream), output, VideoContainer.MP4, VideoCodec.H264, None, 8
-        )
-
-    output.seek(0)
-    with av.open(output) as container:
-        video_stream = container.streams.video[0]
-        frames = [f for p in container.demux(video_stream) for f in p.decode()]
-        assert len(frames) == 32
-        assert float(video_stream.duration * video_stream.time_base) == pytest.approx(32 / fps, abs=0.01)
-        assert float(frames[-1].pts * frames[-1].time_base) == pytest.approx(31 / fps, abs=0.01)
-
-
-def test_save_to_transcode_irregular_vfr_keeps_span():
-    """B-frames reorder packets, and mp4 sample durations follow decode order: the dts
-    timeline ends before the pts timeline, so an irregular-VFR source's tail holds fell
-    out of the container (this 20.23 s span used to come out as 15.27 s, and the 10 s
-    trim as 6.03 s). The transcode encodes without B-frames so every sample keeps its
-    true display duration."""
-    durations = [1, 1, 60, 1, 1, 120, 1, 180, 1, 1, 150, 90]  # 1/30 s ticks, span 20.2333 s
-    generator = torch.Generator().manual_seed(7)
-    buffer = io.BytesIO()
-    with av.open(buffer, mode="w", format="mp4") as container:
-        video_stream = container.add_stream("mpeg4", rate=30)
-        video_stream.width = video_stream.height = 64
-        video_stream.pix_fmt = "yuv420p"
-        pts = 0
-        for duration in durations:
-            # textured frames, so an encoder with default settings has B-frames to gain from
-            frame = av.VideoFrame.from_ndarray(
-                torch.randint(0, 255, (64, 64, 3), generator=generator, dtype=torch.uint8).numpy(),
-                format="rgb24",
-            ).reformat(format="yuv420p")
-            frame.pts = pts
-            frame.time_base = Fraction(1, 30)
-            pts += duration
-            for packet in video_stream.encode(frame):
-                packet.duration = duration  # exact stts in the source
-                container.mux(packet)
-        container.mux(video_stream.encode(None))
-
-    result = transcode_and_probe(VideoFromFile(buffer))
-    assert result["frames"] == len(durations)
-    assert result["video_seconds"] == pytest.approx(sum(durations) / 30, abs=0.05)
-
-    trimmed = transcode_and_probe(VideoFromFile(buffer, duration=10))
-    assert trimmed["frames"] == 8  # frames at 12.167 s+ fall outside the window
-    assert trimmed["video_seconds"] == pytest.approx(10.0, abs=0.05)
-
-
-def test_save_to_transcode_trim_survives_missing_leading_pts():
-    """A trim should survive pts-less kept frames followed by a real-pts frame past the window."""
-    nulled_frames = 0
-
-    class _PacketProxy:
-        def __init__(self, packet):
-            self._packet = packet
-
-        def __getattr__(self, name):
-            return getattr(self._packet, name)
-
-        @property
-        def stream(self):
-            return self._packet.stream
-
-        def decode(self):
-            nonlocal nulled_frames
-            frames = self._packet.decode()
-            for frame in frames:
-                if nulled_frames < 2:
-                    frame.pts = None
-                    nulled_frames += 1
-            return frames
-
-    class _ContainerProxy:
-        def __init__(self, real):
-            self._real = real
-
-        def __getattr__(self, name):
-            return getattr(self._real, name)
-
-        def demux(self, *streams):
-            for packet in self._real.demux(*streams):
-                yield _PacketProxy(packet)
-
-    file_path = create_transcode_source(frames=10, audio_streams=0)
-    try:
-        buffer = io.BytesIO()
-        with av.open(file_path) as container:
-            # 0.05 s window: both pts-less frames are kept (synthesized pts 0 and 512),
-            # and the first real-pts frame (1024 ticks) already lies past end_pts (768)
-            VideoFromFile(file_path, duration=0.05)._save_transcoded(
-                _ContainerProxy(container), buffer, VideoContainer.MP4, VideoCodec.H264, None, 8
-            )
-        assert nulled_frames == 2
-        buffer.seek(0)
-        with av.open(buffer) as container:
-            video_stream = container.streams.video[0]
-            frames = [f for p in container.demux(video_stream) for f in p.decode()]
-            assert len(frames) == 2
-            assert float(video_stream.duration * video_stream.time_base) == pytest.approx(2 / 30, abs=0.01)
-    finally:
-        os.unlink(file_path)
-
-
-def test_save_to_transcode_bakes_rotation():
-    """A 90-degree display-matrix rotation swaps the output dimensions (portrait video)"""
-    file_path = create_transcode_source(width=64, height=32, rotation=True)
-    try:
-        result = transcode_and_probe(VideoFromFile(file_path))
-        assert (result["width"], result["height"]) == (32, 64)
-        assert result["frames"] == 30
-    finally:
-        os.unlink(file_path)
-
-
-def test_save_to_transcode_skips_undecodable_audio():
-    """Streaming transcode keeps the decodable audio track and drops undecodable ones;
-    with no decodable audio at all the output is video-only instead of crashing."""
-    mixed = all_bad = None
-    try:
-        mixed = create_transcode_source(audio_streams=1, undecodable_audio=1)
-        all_bad = create_transcode_source(audio_streams=0, undecodable_audio=2)
-        result = transcode_and_probe(VideoFromFile(mixed))
-        assert result["audio_codecs"] == ["aac"]
-        assert result["audio_seconds"] == pytest.approx(1.0, abs=0.1)
-        assert transcode_and_probe(VideoFromFile(all_bad))["audio_codecs"] == []
-    finally:
-        for path in (mixed, all_bad):
-            if path:
-                os.unlink(path)