hold: a first-frame source whose buffer is full is primed; cli: a lateral that scales is hinted the size it scales to

cli/ffrwd/functions.py

@@ -202,9 +202,8 @@ class RuntimeLateral:
     as written. `params` are the values bound per message and `written` the
     ones the call wrote itself, as SQL. `needs` are the script's own
     definitions an instance calls, in the order the script wrote them.
-    `sizes` are the picture sizes the body declares: each video column that
-    is an input's own picture with a ``shape`` naming its width and height,
-    as ``(column, width, height)``.
+    `sizes` are the picture sizes the body makes, where the compile can tell
+    them, as ``(column, width, height)`` (:func:`_declared_sizes`).
     """
 
     function: str
@@ -3694,14 +3693,46 @@ def _runs_per_message(function: _Function) -> bool:
     )
 
 
-def _declared_sizes(function: _Function) -> tuple[tuple[str, int, int], ...]:
-    """The picture size a run-time lateral's body declares for each video
-    column that is an input's own first picture, ``f.video[1] AS video`` over
-    ``input(url, shape => STRUCT(1920 AS width, 1080 AS height, ...)) f``:
-    ``(column, width, height)``. A column made any other way, or a shape
-    whose size is not written as numbers, has none.
+# Filters a lateral's picture may pass through with its size kept, as the
+# compile reads a body for the size its picture leaves at.
+_SIZE_KEEPING = frozenset({"fps", "setpts", "settb", "format", "null", "copy", "fifo"})
+
+
+def _declared_sizes(
+    function: _Function, written: Mapping[str, str]
+) -> tuple[tuple[str, int, int], ...]:
+    """The picture size a run-time lateral's body makes for each video
+    column, where the compile can tell it: ``(column, width, height)``.
+
+    The column is an input's first picture, ``f.video[1]``, whose
+    ``input(..., shape => STRUCT(... AS width, ... AS height, ...))`` gives
+    its size, passed through filters that keep the size (``fps``,
+    ``setpts``, ``format``), or a ``scale`` of it to a width and a height.
+    A value is a number, or a parameter: what the call wrote, else its
+    ``DEFAULT``. A column made any other way has none.
     """
     body = function.body
+    params = {param.name: param for param in function.params}
+
+    def number(node: exp.Expr | None) -> int | None:
+        if isinstance(node, exp.Paren):
+            node = node.this
+        if isinstance(node, exp.Literal) and not node.is_string:
+            value = node.to_py()
+            return value if isinstance(value, int) and value > 0 else None
+        if isinstance(node, exp.Column) and not node.table:
+            name = node.name
+            if name in written:
+                try:
+                    found = float(written[name])
+                except ValueError:
+                    return None
+                return int(found) if found.is_integer() and found > 0 else None
+            param = params.get(name)
+            if param is not None and param.default is not None:
+                return number(param.default)
+        return None
+
     items: list[exp.Expr] = []
     found = body.args.get("from_")
     if isinstance(found, exp.From):
@@ -3721,42 +3752,49 @@ def _declared_sizes(function: _Function) -> tuple[tuple[str, int, int], ...]:
             struct = argument.expression
             if not isinstance(struct, exp.Struct):
                 continue
-            fields: dict[str, object] = {}
-            for entry in struct.expressions:
-                if isinstance(entry, exp.PropertyEQ) and isinstance(entry.expression, exp.Literal):
-                    literal = entry.expression
-                    fields[str(entry.this.name).lower()] = (
-                        None if literal.is_string else literal.to_py()
-                    )
-            width, height = fields.get("width"), fields.get("height")
-            if (
-                isinstance(width, int)
-                and isinstance(height, int)
-                and not isinstance(width, bool)
-                and not isinstance(height, bool)
-                and width > 0
-                and height > 0
-            ):
+            fields = {
+                str(entry.this.name).lower(): entry.expression
+                for entry in struct.expressions
+                if isinstance(entry, exp.PropertyEQ) and isinstance(entry.expression, exp.Expr)
+            }
+            width, height = number(fields.get("width")), number(fields.get("height"))
+            if width is not None and height is not None:
                 shapes[_ident_name(alias.this)] = (width, height)
+
+    def size(node: exp.Expr) -> tuple[int, int] | None:
+        if isinstance(node, exp.Dot) and isinstance(node.expression, exp.Anonymous):
+            node = node.expression
+        if isinstance(node, exp.Bracket):
+            read = node.this
+            if isinstance(read, exp.Column) and read.name == "video" and subscript_index(node) == 1:
+                return shapes.get(read.table)
+            return None
+        if not isinstance(node, exp.Anonymous) or not node.expressions:
+            return None
+        name = str(node.this).lower()
+        source, *rest = node.expressions
+        if name in _SIZE_KEEPING:
+            return size(source)
+        if name != "scale":
+            return None
+        named = {kwarg_name(arg): arg.expression for arg in rest if isinstance(arg, exp.Kwarg)}
+        if set(named) - {"w", "h", "width", "height"}:
+            return None
+        positional = [arg for arg in rest if not isinstance(arg, exp.Kwarg)]
+        width = positional[0] if positional else named.get("w", named.get("width"))
+        height = positional[1] if len(positional) > 1 else named.get("h", named.get("height"))
+        scaled = number(width), number(height)
+        return (scaled[0], scaled[1]) if scaled[0] and scaled[1] else None
+
     sizes: list[tuple[str, int, int]] = []
     for column in function.columns or ():
         if column.type != _WASM_STREAM:
             continue
         for selected in body.expressions:
-            named = selected.alias_or_name if isinstance(selected, exp.Alias) else None
-            if named != column.name:
-                continue
-            read = selected.this
-            if not (
-                isinstance(read, exp.Bracket)
-                and isinstance(read.this, exp.Column)
-                and read.this.name == "video"
-                and subscript_index(read) == 1
-            ):
-                continue
-            size = shapes.get(read.this.table)
-            if size is not None:
-                sizes.append((column.name, *size))
+            if isinstance(selected, exp.Alias) and selected.alias_or_name == column.name:
+                made = size(selected.this)
+                if made is not None:
+                    sizes.append((column.name, *made))
     return tuple(sizes)
 
 
@@ -5991,7 +6029,7 @@ def _expand_runtime_lateral(
             columns=function.columns or (),
             written=written,
             needs=self._needs(function),
-            sizes=_declared_sizes(function),
+            sizes=_declared_sizes(function, written),
         )
         line, col = _pos(item)
         columns: dict[str, exp.Expr] = {}

cli/tests/test_laterals.py

@@ -1045,6 +1045,60 @@ def test_a_laterals_declared_picture_size_is_its_streams_hint_in_port_order() ->
     ]
 
 
+# foundmeat's feeds: the main feed's picture and sound as they come, a
+# cam's scaled to the size its parameters give, its sound beside it.
+_FEEDS = """CREATE FUNCTION main_feed(launch data_stream, url text)
+RETURNS TABLE(video video_stream, audio audio_stream) AS $$
+  SELECT f.video[1] AS video, f.audio[1] AS audio
+  FROM input(url, format => 'mpegts', shape => STRUCT(1920 AS width, 1080 AS height,
+             60 AS fps, 'h264' AS video_codec, 48000 AS rate, 2 AS channels)) f
+$$ LANGUAGE sql;
+CREATE FUNCTION cam_feed(launch data_stream, url text,
+                         width number DEFAULT 800, height number DEFAULT 450,
+                         rate number DEFAULT 30)
+RETURNS TABLE(video video_stream, audio audio_stream) AS $$
+  SELECT fps(scale(f.video[1], width, height), rate) AS video, f.audio[1] AS audio
+  FROM input(url, format => 'mpegts', shape => STRUCT(width AS width, height AS height,
+             rate AS fps, 'h264' AS video_codec, 48000 AS rate, 2 AS channels)) f
+$$ LANGUAGE sql;"""
+_THREE_FEEDS = """COPY (
+  WITH prog AS (SELECT s.video[1] AS v, s.data[1] AS d FROM input('leaf.nut') s)
+  SELECT compose(prog.v, ARRAY[m.video, c1.video, c2.video],
+                 audio => ARRAY[m.audio, c1.audio, c2.audio])
+  FROM prog, LATERAL main_feed(prog.d) m, LATERAL cam_feed(prog.d) c1,
+       LATERAL cam_feed(prog.d, width => 640, height => 360) c2
+) TO 'out.nut'"""
+
+
+def test_a_lateral_that_scales_is_hinted_the_size_it_scales_to_beside_its_sound() -> None:
+    """Each feed brings its picture and its sound on one connection. The
+    main feed's picture is hinted the size its shape declares; a cam's, the
+    size its body scales to: its parameters' defaults, or what the call
+    wrote. The sound of each is hinted nothing."""
+    modules = _compose_modules({"type": "array", "items": {"type": "integer"}})
+    plan = compile_all(
+        _FEEDS + "\n" + _COMPOSE_AUDIO + "\n" + _THREE_FEEDS,
+        describe=lambda path: modules[path],
+        shape=_compose_audio_shape,
+    ).plan
+    assert plan is not None
+    compose = _compose_node(plan)
+    ports = [lateral.connections[0].port for lateral in plan.laterals]
+    assert compose.args["port"] == ports
+    unknown = {"rate": None}
+    assert json.loads(compose.bound)[1:] == [
+        {
+            "input": "inputs",
+            "streams": [
+                {"rate": None, "width": 1920, "height": 1080},
+                {"rate": None, "width": 800, "height": 450},
+                {"rate": None, "width": 640, "height": 360},
+            ],
+        },
+        {"input": "audio", "streams": [unknown, unknown, unknown]},
+    ]
+
+
 def test_a_port_written_by_hand_stays_one_integer_where_a_list_also_fits() -> None:
     query = """COPY (
   SELECT compose(s.video[1], port => 9100.0) FROM input('leaf.nut') s

docs/dialect.md

@@ -702,12 +702,16 @@ dest    := 'path' | STDOUT | ( value-expression ) | sink(value, ...)
     its clock, a rate clock's own, the rate of the input a `rate-of`
     clock names, or its clock input's over its stride. A feed by port, a
     self-clocked node's output and data have none. A lateral's picture
-    whose body reads an input's own first picture with a declared size
-    (`f.video[1] AS video` over `input(url, shape => STRUCT(1920 AS
-    width, 1080 AS height, ...)) f`) carries that width and height, and
-    the sidecar conforms the feed to them; without one, the feed is
-    conformed to the size of the input its port follows, or the clock
-    input's. Each call carries the
+    carries the width and height its body makes, where the compile can
+    tell them, and the sidecar conforms the feed to them: an input's own
+    first picture whose `shape` gives its size (`f.video[1] AS video`
+    over `input(url, shape => STRUCT(1920 AS width, 1080 AS height,
+    ...)) f`), through `fps`, `setpts` or `format`, or a `scale` of it to
+    a width and a height, each a number or a parameter the call writes or
+    gives a `DEFAULT` (`fps(scale(f.video[1], width, height), rate)`). A
+    message binding another size has its picture conformed to that one.
+    Without a size, the feed is conformed to the size of the input its
+    port follows, or the clock input's. Each call carries the
     list it was asked with (`-bound <name>=<json>`), which the host asks
     the shape with again, so the shape a node runs with is the one the
     query compiled against. A node at a COPY's TO is asked once for its

docs/known_gaps.md

@@ -73,14 +73,14 @@ from.
   of one ends with an upload. Only 8-bit and 10-bit 4:2:0 pictures, and
   8-bit 4:4:4 HEVC on CUDA, stay on the card; 4:2:2, 10-bit 4:4:4 and an
   input whose pixel format the probe does not report run on the CPU.
-- **A lateral's picture size is known only where its body declares it.**
-  A node input fed on a port by a run-time lateral is conformed to the
-  size the lateral's body declares (`f.video[1]` of an `input(...)` whose
-  `shape` gives `width` and `height`). A body that reads a probed file, or
-  scales its picture, declares none, and its feed is conformed to the size
-  of the input the port follows, or the clock input's, which for a small
-  clock is a blur. Declare the `shape` in the body to keep the picture's
-  own size.
+- **A lateral's picture size is known only where its body says it.** A
+  node input fed on a port by a run-time lateral is conformed to the size
+  the lateral's body makes: its input's declared `shape`, or a `scale` to
+  numbers or to parameters with a value the call writes or a `DEFAULT`.
+  A body that reads a probed file, scales to a value each message binds
+  with no `DEFAULT`, or passes its picture through other filters says
+  none, and its feed is conformed to the size of the input the port
+  follows, or the clock input's, which for a small clock is a blur.
 - **The sidecar has no device filters on macOS.** Its bundled
   libavfilter carries none for VideoToolbox, so a filter between two
   modules holding their pictures on the GPU runs in ffmpeg there, and the

sidecar/NODE-CLI.md

@@ -413,7 +413,9 @@ tick that held them.
   while the source is behind and skipping forward when it catches up, and
   an audio input hands the tick's samples re-cut from what arrived, or
   none while the source is behind. A `first-frame` source is primed until
-  `lead` seconds of it are held, or its end, and scheduled `lead` ahead of
+  `lead` seconds of it are held, or its end, or as much of it as the host
+  holds of a source (600 frames or 96 MiB, which is 12 pictures of 1080p
+  rgba), and scheduled `lead` ahead of
   the clock, on the clock's grid; a `shared-clock` source's first frame
   waits for the clock to reach its pts, and one the clock has passed shows
   at once from where the clock is. A port feed arrives in real time, so a

sidecar/ffrwd-wasm/src/hold.rs

@@ -64,6 +64,13 @@ use crate::trace::trace_feed;
 pub const MAX_HELD_FRAMES: usize = 600;
 pub const MAX_HELD_BYTES: usize = 96 << 20;
 
+/// Whether `source` holds as many frames or bytes as it may.
+fn full(source: &Source) -> bool {
+    let held = source.members.iter().map(|m| m.queue.len()).sum::<usize>();
+    let bytes = source.members.iter().map(|m| m.bytes).sum::<usize>();
+    held >= MAX_HELD_FRAMES || bytes >= MAX_HELD_BYTES
+}
+
 /// A step in a source's time, or the clock's, that is backwards or further
 /// forward than this is a discontinuity.
 pub const MAX_STEP_SECONDS: f64 = 1.0;
@@ -573,11 +580,10 @@ impl Group {
         let Some(source) = self.sources.back() else {
             return MAX_HELD_FRAMES;
         };
-        let held = source.members.iter().map(|m| m.queue.len()).sum::<usize>();
-        let bytes = source.members.iter().map(|m| m.bytes).sum::<usize>();
-        if held >= MAX_HELD_FRAMES || bytes >= MAX_HELD_BYTES {
+        if full(source) {
             return 0;
         }
+        let held = source.members.iter().map(|m| m.queue.len()).sum::<usize>();
         let waiting = match (&self.feed, self.now) {
             (Some(feed), Some((now, _))) => now < feed.at,
             (None, Some((now, clock_base))) => {
@@ -618,13 +624,15 @@ impl Group {
     }
 
     /// Whether a `first-frame` source holds `lead` seconds of frames, or
-    /// has ended with some.
+    /// has ended with some, or holds all it may: a buffer full short of
+    /// `lead` (large pictures at a high rate) would otherwise wait for
+    /// frames it has no room to take.
     fn primed(&self, source: &Source) -> bool {
         let lead = &source.members[source.lead];
         let (Some(first), Some(last)) = (lead.queue.front(), lead.queue.back()) else {
             return false;
         };
-        if source.closed {
+        if source.closed || full(source) {
             return true;
         }
         last.pts.saturating_sub(first.pts) >= ticks_of(self.hold.lead, lead.base)
@@ -1558,6 +1566,42 @@ mod tests {
         assert_eq!(mark(&handed[0]), Some(11));
     }
 
+    #[test]
+    fn a_first_frame_port_feed_whose_buffer_fills_short_of_lead_still_comes_up() {
+        let (mut g, lines) = group(
+            hold(Anchor::FirstFrame, 0.3, None, None),
+            vec![video_member(1, MILLIS)],
+            true,
+        );
+        let grid = Grid::exact();
+        // Four pictures fill the buffer, 50 ms of a lead of 300.
+        let picture = Arc::new(vec![7u8; MAX_HELD_BYTES / 4 + 1]);
+        let big = |pts: i64| TickFrame {
+            pts,
+            duration: None,
+            data: Arc::clone(&picture).into(),
+            rows: Vec::new(),
+        };
+        for pts in [0, 16, 33] {
+            g.arrive(0, big(pts));
+        }
+        let handed = g.tick(100, Some(101), THIRTIETHS, &grid, true);
+        assert!(handed[0].feed.is_none(), "room left: still priming");
+        g.arrive(0, big(50));
+        assert_eq!(g.room(), 0, "the buffer is full");
+        let handed = g.tick(101, Some(102), THIRTIETHS, &grid, true);
+        let feed = handed[0]
+            .feed
+            .clone()
+            .expect("a full buffer starts the feed");
+        assert_eq!(feed.start.first_pts, 0);
+        assert!(
+            lines.lock().unwrap()[0].contains("comes up"),
+            "{:?}",
+            lines.lock().unwrap()
+        );
+    }
+
     #[test]
     fn a_shared_clock_feed_waits_for_the_clock_and_skips_what_the_clock_passed() {
         let (mut g, _) = group(