diff --git a/cli/ffrwd/functions.py b/cli/ffrwd/functions.py index 371a079..6d67817 100644 --- a/cli/ffrwd/functions.py +++ b/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] = {} diff --git a/cli/tests/test_laterals.py b/cli/tests/test_laterals.py index 73df308..543c0d4 100644 --- a/cli/tests/test_laterals.py +++ b/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 diff --git a/docs/dialect.md b/docs/dialect.md index d3a9b91..7fcbe48 100644 --- a/docs/dialect.md +++ b/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 =`), 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 diff --git a/docs/known_gaps.md b/docs/known_gaps.md index 4b17ecd..dfd888a 100644 --- a/docs/known_gaps.md +++ b/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 diff --git a/sidecar/NODE-CLI.md b/sidecar/NODE-CLI.md index 448e595..1ccf003 100644 --- a/sidecar/NODE-CLI.md +++ b/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 diff --git a/sidecar/ffrwd-wasm/src/hold.rs b/sidecar/ffrwd-wasm/src/hold.rs index fcc388a..5428579 100644 --- a/sidecar/ffrwd-wasm/src/hold.rs +++ b/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::(); + let bytes = source.members.iter().map(|m| m.bytes).sum::(); + 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::(); - let bytes = source.members.iter().map(|m| m.bytes).sum::(); - 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::(); 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(