diff --git a/docs/decisions/agent-readable-processor-catalog.md b/docs/decisions/agent-readable-processor-catalog.md index 6bb531e32..2ec042c3e 100644 --- a/docs/decisions/agent-readable-processor-catalog.md +++ b/docs/decisions/agent-readable-processor-catalog.md @@ -58,9 +58,14 @@ the other way, code to document, and attaches to nothing on a link. processor adds no dependency; three local config enums gain the derive. - The descriptor carries a schema document where it carried a type-name string. - Every Python processor that took keyword configuration changes shape: the engine-tree - fixtures migrate with the change; example and extension-wheel processors lag as - consumers do, and the extension wheels' required parameters become required keys in a - config class rather than defaults. + fixtures migrate with the change, and the extension wheels' required parameters become + required keys in a config class rather than defaults. ~~Example and extension-wheel + processors lag as consumers do~~ — amended 2026-09-11 by the owner's ruling that the + four extension-wheel processors migrate in the same PR as the engine half, as the + deliberate canary §Consumers reserves for in-flight work: `packages/` is the only + consumer tree with a CI lane, so migrating it is what proves the new construction path + on real processors and keeps that lane green. The fourteen example processors lag as + consumers do, unchanged. - Reconfiguration takes the config object; `configure(self, **config)` goes with the keyword form. - A helper process importing a decorated class must register nothing, so the wheel has diff --git a/docs/decisions/importable-python-library.md b/docs/decisions/importable-python-library.md index 69d09267d..d019f2c0b 100644 --- a/docs/decisions/importable-python-library.md +++ b/docs/decisions/importable-python-library.md @@ -198,11 +198,16 @@ process boundary). least one input port defaults; one declaring none must say what it is. A source has nothing to react to, so the default would hand the author a processor that silently never runs — the one case where the convenient default is a trap. -- **Configuration is constructor keyword arguments.** `rt.add(Blur, config={"radius": 3})` - constructs `Blur(radius=3)`, so a processor's settings are ordinary Python parameters with - ordinary defaults and there is no configuration object to learn. It travels as JSON on the graph - node rather than captured in a closure, because one class added twice must yield two - independently configured instances — and because that keeps it visible in `graph`. +> ~~**Configuration is constructor keyword arguments.** `rt.add(Blur, config={"radius": 3})` +> constructs `Blur(radius=3)`, so a processor's settings are ordinary Python parameters with +> ordinary defaults and there is no configuration object to learn.~~ — Superseded 2026-09-11 by +> `agent-readable-processor-catalog.md`: a processor's config is one class, named by the +> annotation on its `__init__`'s `config` parameter, and the helper constructs that class from +> the mapping. Nothing recorded a keyword signature anywhere, so an agent could only learn a key +> by adding the node and reading the failure; a class has annotations and defaults a schema is +> derived from. The rest of the bullet stands: configuration travels as JSON on the graph node +> rather than captured in a closure, because one class added twice must yield two independently +> configured instances — and because that keeps it visible in `graph`. - **Python ports declare no schema.** The wire is self-describing and consuming is a cast at read time, so a port carries a name, a description and (on inputs) a delivery profile. Adding a schema hint here would build on the per-read matching being deleted. diff --git a/docs/plan/changes/agent-readable-processor-catalog.md b/docs/plan/changes/agent-readable-processor-catalog.md index 1dc7966a9..dfaaa30fa 100644 --- a/docs/plan/changes/agent-readable-processor-catalog.md +++ b/docs/plan/changes/agent-readable-processor-catalog.md @@ -150,11 +150,16 @@ from this renders anything new on a port. is constructed into the class's config class; the `processor` decorator's doc names the `config` rule; `stubtest` and pyright gate both as today. - **The six engine-tree fixtures migrate** to a config class in the change; the string - fixture in `test_live_graph_mutation.py` with them. The fourteen example processors and - the four extension-wheel processors lag as §Consumers states + fixture in `test_live_graph_mutation.py` with them. The fourteen example processors + ~~and the four extension-wheel processors~~ lag as §Consumers states (`docs/plan/ARCHITECTURE.md:327-436`: consumers are never in a migration's scope; a converted consumer's breakage is filed as tracked backlog at that consumer), with the - backlog issues filed at ship naming each file. + backlog issues filed at ship naming each file. — Amended 2026-09-11 by the owner's + ruling that the four extension-wheel processors migrate in the same PR as the engine + half (#2222), as the deliberate canary §Consumers reserves at `:430-433` for in-flight + work: `packages/` is the only consumer tree with a CI lane, so migrating it is what + proves the new construction path on real processors rather than on fixtures alone, and + it keeps that lane green. Only the fourteen example processors owe backlog at ship. ## ADDED: §Processor model — declaration registers diff --git a/packages/streamlib-moq/pyproject.toml b/packages/streamlib-moq/pyproject.toml index 9a92a75a2..a2422bdad 100644 --- a/packages/streamlib-moq/pyproject.toml +++ b/packages/streamlib-moq/pyproject.toml @@ -29,13 +29,20 @@ classifiers = [ # The engine is a binary dependency, never a source one: this wheel links no # streamlib crate and speaks no engine internals. # -# 0.18.52 is the earliest release carrying every half this wheel needs. The -# capability-extension mechanism `extension.py:load` is declared against arrived -# in 0.18.48; `read_from_inbound_link_with_timestamp`, which every bag is read -# through, arrived in 0.18.49; and `encode_bag_to_msgpack_bytes`, which every -# data track object is built with, arrived in 0.18.52. Each resolved by `git -# describe --contains` on the commit that introduced it into `_engine.pyi`, not -# from a changelog. +# What this wheel needs arrived across several releases: the +# capability-extension mechanism `extension.py:load` is declared against in +# 0.18.48; `read_from_inbound_link_with_timestamp`, which every bag is read +# through, in 0.18.49; `encode_bag_to_msgpack_bytes`, which every data track +# object is built with, in 0.18.52. Each resolved by `git describe --contains` +# on the commit that introduced it into `_engine.pyi`, not from a changelog. +# +# The binding half is newer than all of them and cannot be named. These +# processors take a config class, which only an engine whose helper constructs +# one can host, and that engine ships in the release *after* this floor was last +# touched — a version the ceiling rule below forbids naming. So 0.20.0 is the +# highest floor that is legal here, not the floor that is true: an install +# against exactly 0.20.0 resolves and then fails in the helper child. Bump this +# to the release carrying config-class hosting once it exists. # # The ceiling is the last *published* engine version, not the one in the tree. # Two lanes install this wheel and they resolve differently: the PR lane @@ -43,7 +50,7 @@ classifiers = [ # lane installs from the published index, which is rebuilt only after the # extension wheels are built. So a floor naming the engine version released in # the same run would fail the release smoke test. -dependencies = ["streamlib>=0.18.52"] +dependencies = ["streamlib>=0.20.0"] # What pip records at install and the engine reads back through # `importlib.metadata` when a process takes an engine role. diff --git a/packages/streamlib-moq/python/streamlib_moq/__init__.py b/packages/streamlib-moq/python/streamlib_moq/__init__.py index d3adea2f6..b4803ec6b 100644 --- a/packages/streamlib-moq/python/streamlib_moq/__init__.py +++ b/packages/streamlib-moq/python/streamlib_moq/__init__.py @@ -10,6 +10,13 @@ """ from .processors import MoqBroadcastPublisher as MoqBroadcastPublisher +from .processors import MoqBroadcastPublisherConfig as MoqBroadcastPublisherConfig from .processors import MoqBroadcastSubscriber as MoqBroadcastSubscriber +from .processors import MoqBroadcastSubscriberConfig as MoqBroadcastSubscriberConfig -__all__ = ["MoqBroadcastPublisher", "MoqBroadcastSubscriber"] +__all__ = [ + "MoqBroadcastPublisher", + "MoqBroadcastPublisherConfig", + "MoqBroadcastSubscriber", + "MoqBroadcastSubscriberConfig", +] diff --git a/packages/streamlib-moq/python/streamlib_moq/processors.py b/packages/streamlib-moq/python/streamlib_moq/processors.py index f724ab3f0..a349b0f26 100644 --- a/packages/streamlib-moq/python/streamlib_moq/processors.py +++ b/packages/streamlib-moq/python/streamlib_moq/processors.py @@ -16,7 +16,7 @@ import threading from collections.abc import Mapping, Sequence from dataclasses import dataclass -from typing import Any, Literal, Protocol +from typing import Annotated, Any, Literal, Protocol from streamlib import ( EncodedAudioPacket, @@ -528,6 +528,27 @@ def _optional_track_names(track_names: Any) -> "list[str] | None": return names +@dataclass +class MoqBroadcastPublisherConfig: + """What a `MoqBroadcastPublisher` is configured with.""" + + relay_url: Annotated[str, "The MoQ relay to publish through."] + broadcast: Annotated[ + "str | None", "The namespace to publish under; minted when absent." + ] = None + container_format: Annotated[ + ContainerFormat, "How each track is packaged on the wire." + ] = "cmaf" + delivery_deadline_ms: Annotated[ + "int | None", + "How old a bag may be, by its own monotonic stamp, and still be published.", + ] = None + track_names: Annotated[ + "Sequence[str] | None", + "Names the tracks positionally in wiring order; `streamlib_bag` only.", + ] = None + + @processor( description=( "Publishes encoded video, encoded audio and data bags to a MoQ " @@ -542,9 +563,10 @@ class MoqBroadcastPublisher: for a bag with no `bitstream` key, data. A data track carries any bag at all, nested whole inside an object beside the publisher's own `sequence_index` and the bag's stamp, under `streamlib_bag` only. Its - settings are ordinary constructor parameters — `relay_url`, `broadcast`, + settings are a `MoqBroadcastPublisherConfig` — `relay_url`, `broadcast`, `container_format` and `track_names` — which is what - `rt.add(MoqBroadcastPublisher, config={"relay_url": ...})` passes. + `rt.add(MoqBroadcastPublisher, config={"relay_url": ...})` is constructed + into. `track_names`, under `streamlib_bag`, names the tracks positionally in wiring order — the order `runtime.connect` ran — so a subscriber in @@ -593,21 +615,16 @@ class MoqBroadcastPublisher: throughput of about 40 Mbit/s at a 100 ms round trip to the relay. """ - def __init__( - self, - relay_url: str, - broadcast: "str | None" = None, - container_format: ContainerFormat = "cmaf", - delivery_deadline_ms: "int | None" = None, - track_names: "Sequence[str] | None" = None, - ) -> None: - self._relay_url = _required_relay_url(relay_url, "MoqBroadcastPublisher") - self._broadcast = broadcast + def __init__(self, config: MoqBroadcastPublisherConfig) -> None: + self._relay_url = _required_relay_url(config.relay_url, "MoqBroadcastPublisher") + self._broadcast = config.broadcast self._container_format = _required_container_format( - container_format, "MoqBroadcastPublisher" + config.container_format, "MoqBroadcastPublisher" + ) + self._delivery_deadline_ms = _optional_delivery_deadline_ms( + config.delivery_deadline_ms ) - self._delivery_deadline_ms = _optional_delivery_deadline_ms(delivery_deadline_ms) - self._track_names = _optional_track_names(track_names) + self._track_names = _optional_track_names(config.track_names) self._session: "_native.MoqBroadcastPublishingSession | None" = None self._kind_by_inbound_link: "dict[str, str]" = {} self._next_data_sequence_index_by_inbound_link: "dict[str, int]" = {} @@ -804,6 +821,26 @@ def _color_axes_of(frame: EncodedVideoFrame) -> "dict[str, str] | None": return stated or None +@dataclass +class MoqBroadcastSubscriberConfig: + """What a `MoqBroadcastSubscriber` is configured with.""" + + relay_url: Annotated[str, "The MoQ relay to subscribe through."] + broadcast: Annotated[str, "The namespace to subscribe to."] + video_track: Annotated[ + "str | None", "The track feeding `encoded_video`; unnamed means no video." + ] = None + audio_track: Annotated[ + "str | None", "The track feeding `encoded_audio`; unnamed means no audio." + ] = None + container_format: Annotated[ + ContainerFormat, "How each track is packaged on the wire." + ] = "cmaf" + data_track: Annotated[ + "str | None", "The track feeding `data_bags`; `streamlib_bag` only." + ] = None + + @processor( execution="manual", description=( @@ -835,43 +872,46 @@ class MoqBroadcastSubscriber: would do with the same bytes. """ - def __init__( - self, - relay_url: str, - broadcast: str, - video_track: "str | None" = None, - audio_track: "str | None" = None, - container_format: ContainerFormat = "cmaf", - data_track: "str | None" = None, - ) -> None: - self._relay_url = _required_relay_url(relay_url, "MoqBroadcastSubscriber") - if not isinstance(broadcast, str) or not broadcast: + def __init__(self, config: MoqBroadcastSubscriberConfig) -> None: + self._relay_url = _required_relay_url( + config.relay_url, "MoqBroadcastSubscriber" + ) + if not isinstance(config.broadcast, str) or not config.broadcast: raise ValueError( "MoqBroadcastSubscriber: `broadcast` is required and names the " - f"namespace to subscribe to; got {broadcast!r}" + f"namespace to subscribe to; got {config.broadcast!r}" ) - if video_track is None and audio_track is None and data_track is None: + if ( + config.video_track is None + and config.audio_track is None + and config.data_track is None + ): raise ValueError( "MoqBroadcastSubscriber: name at least one of `video_track`, " "`audio_track` and `data_track`; a subscriber naming none would " "subscribe to nothing and produce nothing." ) _refuse_track_names_no_broadcast_can_serve( - (("video_track", video_track), ("audio_track", audio_track), ("data_track", data_track)) + ( + ("video_track", config.video_track), + ("audio_track", config.audio_track), + ("data_track", config.data_track), + ) ) self._container_format = _required_container_format( - container_format, "MoqBroadcastSubscriber" + config.container_format, "MoqBroadcastSubscriber" ) - if data_track is not None and self._container_format == "cmaf": + if config.data_track is not None and self._container_format == "cmaf": raise ValueError( f"MoqBroadcastSubscriber: `data_track` names a data track " - f"({data_track!r}), and the `cmaf` container has no packaging for " - f"one; a data track rides `container_format=\"streamlib_bag\"` only." + f"({config.data_track!r}), and the `cmaf` container has no packaging " + "for one; a data track rides " + '`container_format="streamlib_bag"` only.' ) - self._broadcast = broadcast - self._video_track = video_track - self._audio_track = audio_track - self._data_track = data_track + self._broadcast = config.broadcast + self._video_track = config.video_track + self._audio_track = config.audio_track + self._data_track = config.data_track self._stop = threading.Event() self._reader: "threading.Thread | None" = None self._reported_an_oversized_bag = False diff --git a/packages/streamlib-moq/tests/test_data_track_round_trip.py b/packages/streamlib-moq/tests/test_data_track_round_trip.py index 669be7fb3..1f0dea6cd 100644 --- a/packages/streamlib-moq/tests/test_data_track_round_trip.py +++ b/packages/streamlib-moq/tests/test_data_track_round_trip.py @@ -43,7 +43,13 @@ import pytest from streamlib import ProcessorLinkDataAccess, decode_msgpack_bytes_to_python_object -from streamlib_moq import MoqBroadcastPublisher, MoqBroadcastSubscriber, _native +from streamlib_moq import ( + MoqBroadcastPublisher, + MoqBroadcastPublisherConfig, + MoqBroadcastSubscriber, + MoqBroadcastSubscriberConfig, + _native, +) from streamlib_moq.processors import DATA_BAGS_OUTPUT_PORT, TRACKS_INPUT_PORT A_RELAY = "https://relay.invalid/a-token" @@ -200,14 +206,18 @@ def data_track_round_trip( ) publishing_session = _ThePublishingSessionKeepingWhatItWasHanded() - publisher = MoqBroadcastPublisher(relay_url=A_RELAY, container_format="streamlib_bag") + publisher = MoqBroadcastPublisher( + MoqBroadcastPublisherConfig( + relay_url=A_RELAY, container_format="streamlib_bag" + ) + ) publisher._session = publishing_session # type: ignore[assignment] - subscriber = MoqBroadcastSubscriber( + subscriber = MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, container_format="streamlib_bag", data_track=THE_DATA_TRACK_NAME, - ) + )) yield DataTrackRoundTripUnderTest( publisher, diff --git a/packages/streamlib-moq/tests/test_processors.py b/packages/streamlib-moq/tests/test_processors.py index fb152ee02..4b47dbd4e 100644 --- a/packages/streamlib-moq/tests/test_processors.py +++ b/packages/streamlib-moq/tests/test_processors.py @@ -28,7 +28,13 @@ output, processor, ) -from streamlib_moq import MoqBroadcastPublisher, MoqBroadcastSubscriber, _native +from streamlib_moq import ( + MoqBroadcastPublisher, + MoqBroadcastPublisherConfig, + MoqBroadcastSubscriber, + MoqBroadcastSubscriberConfig, + _native, +) from streamlib_moq import processors as processors_module from streamlib_moq.processors import ( BAGS_BETWEEN_PROGRESS_REPORTS, @@ -186,12 +192,17 @@ def test_both_container_formats_are_addable(runtime, container_format): def test_a_container_format_this_wheel_does_not_write_is_refused_by_name(): with pytest.raises(ValueError, match="container_format"): - MoqBroadcastPublisher(relay_url=A_RELAY, container_format="mpegts") # type: ignore[arg-type] + MoqBroadcastPublisher( + MoqBroadcastPublisherConfig( + relay_url=A_RELAY, + container_format="mpegts", # type: ignore[arg-type] + ) + ) def test_a_publisher_without_a_relay_is_refused_by_name(): with pytest.raises(ValueError, match="relay_url"): - MoqBroadcastPublisher(relay_url="") + MoqBroadcastPublisher(MoqBroadcastPublisherConfig(relay_url="")) def test_the_relay_refusal_says_where_a_draft_16_token_goes(): @@ -199,14 +210,22 @@ def test_the_relay_refusal_says_where_a_draft_16_token_goes(): path, so a bare host is not a usable endpoint and the message has to say so — there is nowhere else to learn it.""" with pytest.raises(ValueError, match="token"): - MoqBroadcastSubscriber(relay_url="", broadcast=A_BROADCAST, video_track="1.m4s") + MoqBroadcastSubscriber( + MoqBroadcastSubscriberConfig( + relay_url="", broadcast=A_BROADCAST, video_track="1.m4s" + ) + ) def test_a_subscriber_naming_no_track_at_all_is_refused_by_name(): """Three static output ports and no track named for any would subscribe to nothing and produce nothing, which reads from outside as a hang.""" with pytest.raises(ValueError, match=r"video_track.*audio_track.*data_track"): - MoqBroadcastSubscriber(relay_url=A_RELAY, broadcast=A_BROADCAST) + MoqBroadcastSubscriber( + MoqBroadcastSubscriberConfig( + relay_url=A_RELAY, broadcast=A_BROADCAST + ) + ) @pytest.mark.parametrize("config", ["video_track", "audio_track", "data_track"]) @@ -215,12 +234,12 @@ def test_a_track_named_as_the_empty_string_is_refused_by_name_at_construction(co refusal is retried with backoff — so said here, or a config mistake reads from outside as a subscriber that never connects.""" with pytest.raises(ValueError, match=config): - MoqBroadcastSubscriber( + MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, container_format="streamlib_bag", **{config: ""}, - ) + )) @pytest.mark.parametrize( @@ -229,18 +248,18 @@ def test_a_track_named_as_the_empty_string_is_refused_by_name_at_construction(co ) def test_one_name_given_to_two_tracks_is_refused_by_name_at_construction(first, second): with pytest.raises(ValueError, match=rf"{first}.*{second}"): - MoqBroadcastSubscriber( + MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, container_format="streamlib_bag", **{first: "both", second: "both"}, - ) + )) def test_a_subscriber_may_name_one_track_and_leave_the_other_ports_silent(): - MoqBroadcastSubscriber( + MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="1.m4s" - ) + )) def test_a_subscriber_naming_only_a_data_track_is_added_like_any_other(runtime): @@ -259,14 +278,14 @@ def test_the_subscribers_data_bags_port_wires_to_a_processor_that_reads_it(runti def test_a_data_track_beside_both_media_tracks_is_accepted_under_streamlib_bag(): - MoqBroadcastSubscriber( + MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, container_format="streamlib_bag", video_track="video", audio_track="audio", data_track="telemetry", - ) + )) def test_a_data_track_under_cmaf_is_refused_by_name_at_construction(): @@ -274,7 +293,11 @@ def test_a_data_track_under_cmaf_is_refused_by_name_at_construction(): `cmaf` — so a subscriber that named one and nothing else must hear it here, not as a broadcast that never produces.""" with pytest.raises(ValueError, match=r"data_track.*cmaf"): - MoqBroadcastSubscriber(relay_url=A_RELAY, broadcast=A_BROADCAST, data_track="telemetry") + MoqBroadcastSubscriber( + MoqBroadcastSubscriberConfig( + relay_url=A_RELAY, broadcast=A_BROADCAST, data_track="telemetry" + ) + ) def test_the_documented_envelope_decodes_to_its_three_parts_with_the_bag_whole(): @@ -394,12 +417,12 @@ def write(self, port: str, bag: "dict[str, Any]", timestamp_ns: "int | None" = N def _a_data_track_subscriber() -> MoqBroadcastSubscriber: - return MoqBroadcastSubscriber( + return MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, container_format="streamlib_bag", data_track="telemetry", - ) + )) def _an_envelope_of(sequence_index: int) -> bytes: @@ -468,7 +491,11 @@ def test_stop_says_what_the_data_track_wrote_and_lost_even_when_the_cadence_neve def test_a_subscriber_naming_no_data_track_says_nothing_about_one_at_stop(): - subscriber = MoqBroadcastSubscriber(relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="1.m4s") + subscriber = MoqBroadcastSubscriber( + MoqBroadcastSubscriberConfig( + relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="1.m4s" + ) + ) said: "list[str]" = [] with mock.patch.object(log, "info", said.append): @@ -646,9 +673,9 @@ def test_a_bag_past_the_link_ceiling_is_reported_once_and_not_every_frame(): subscriber that said nothing would look like a stream that just stopped — but saying it per frame would bury the log of a stream that never recovers.""" - subscriber = MoqBroadcastSubscriber( + subscriber = MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="1.m4s" - ) + )) bag = {**A_VIDEO_BAG, "bitstream": b"x" * (HELPER_LINK_PAYLOAD_CEILING_BYTES + 1)} said: "list[str]" = [] with mock.patch.object(log, "error", said.append): @@ -677,7 +704,11 @@ def test_a_delivery_deadline_that_is_not_a_count_of_milliseconds_is_refused_by_n """`bool` is an `int` in Python, so `True` would otherwise read as a one-millisecond deadline that sheds every frame but the sync points.""" with pytest.raises(ValueError, match="delivery_deadline_ms"): - MoqBroadcastPublisher(relay_url=A_RELAY, delivery_deadline_ms=not_a_deadline) + MoqBroadcastPublisher( + MoqBroadcastPublisherConfig( + relay_url=A_RELAY, delivery_deadline_ms=not_a_deadline + ) + ) def test_a_run_that_shed_nothing_says_so_rather_than_saying_nothing(): @@ -836,7 +867,11 @@ def close(self) -> "str | None": def _a_streamlib_bag_publisher_over( session: _SessionRecordingWhatWasPublished, ) -> MoqBroadcastPublisher: - publisher = MoqBroadcastPublisher(relay_url=A_RELAY, container_format="streamlib_bag") + publisher = MoqBroadcastPublisher( + MoqBroadcastPublisherConfig( + relay_url=A_RELAY, container_format="streamlib_bag" + ) + ) publisher._session = session # type: ignore[assignment] return publisher @@ -860,7 +895,7 @@ def __init__(self, inbound_links: "list[str]") -> None: def _drive_bags_through( reaches_the_transport: bool, bag_count: int = 1 ) -> "tuple[list[str], int]": - publisher = MoqBroadcastPublisher(relay_url=A_RELAY) + publisher = MoqBroadcastPublisher(MoqBroadcastPublisherConfig(relay_url=A_RELAY)) session = _SessionThatAnswers(reaches_the_transport) publisher._session = session # type: ignore[assignment] said: "list[str]" = [] @@ -1080,36 +1115,40 @@ def test_track_names_that_are_not_a_sequence_of_names_are_refused_by_name(not_na """A bare string is a `Sequence[str]` to Python, so it is refused by name rather than read as one track per character.""" with pytest.raises(ValueError, match="track_names"): - MoqBroadcastPublisher( + MoqBroadcastPublisher(MoqBroadcastPublisherConfig( relay_url=A_RELAY, container_format="streamlib_bag", track_names=not_names - ) + )) def test_track_names_unequal_in_count_to_the_inbound_links_are_refused_by_name_at_setup(): """Links are known at `setup()` and not before, so the count is checked there — by the wheel's Rust, which is the one place the names are declared.""" - publisher = MoqBroadcastPublisher( + publisher = MoqBroadcastPublisher(MoqBroadcastPublisherConfig( relay_url=A_RELAY, container_format="streamlib_bag", track_names=["video"] - ) + )) with pytest.raises(ValueError, match="track_names"): publisher.setup(_SetupContextWiredTo(["encoder/video", "probe/telemetry"])) # type: ignore[arg-type] def test_track_names_under_cmaf_are_refused_by_name_at_setup(): - publisher = MoqBroadcastPublisher(relay_url=A_RELAY, track_names=["video"]) + publisher = MoqBroadcastPublisher( + MoqBroadcastPublisherConfig( + relay_url=A_RELAY, track_names=["video"] + ) + ) with pytest.raises(ValueError, match="cmaf"): publisher.setup(_SetupContextWiredTo(["encoder/video"])) # type: ignore[arg-type] def test_track_names_matching_the_links_are_declared_and_said_at_setup(): - publisher = MoqBroadcastPublisher( + publisher = MoqBroadcastPublisher(MoqBroadcastPublisherConfig( relay_url=A_RELAY, container_format="streamlib_bag", track_names=["video", "telemetry"], - ) + )) said: "list[str]" = [] with mock.patch.object(log, "info", said.append): publisher.setup(_SetupContextWiredTo(["encoder/video", "probe/telemetry"])) # type: ignore[arg-type] @@ -1122,9 +1161,9 @@ def test_the_oversize_guard_charges_the_framed_encoded_bag_not_the_bitstream_alo the ceiling, so a bitstream just under it is still dropped — and a guard reading `len(bitstream)` would have stayed silent about it.""" bag = {**A_VIDEO_BAG, "bitstream": b"\x00" * 100} - subscriber = MoqBroadcastSubscriber( + subscriber = MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="video" - ) + )) said: "list[str]" = [] with ( mock.patch.object(processors_module, "HELPER_LINK_PAYLOAD_CEILING_BYTES", 150), @@ -1142,9 +1181,9 @@ def test_the_oversize_guard_measures_the_framed_size_only_near_the_ceiling(): """The exact measure is an encode of the whole bag on the reader thread, so a bag whose bitstream leaves it far under the ceiling is never encoded twice — the engine's own write is the only encode it gets.""" - subscriber = MoqBroadcastSubscriber( + subscriber = MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="video" - ) + )) outputs = _OutputsRecordingWrites() with mock.patch.object( processors_module, "encode_bag_to_msgpack_bytes" @@ -1159,9 +1198,9 @@ def test_the_oversize_guard_measures_the_framed_size_only_near_the_ceiling(): def test_the_oversize_guard_stops_measuring_once_it_has_reported(): - subscriber = MoqBroadcastSubscriber( + subscriber = MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url=A_RELAY, broadcast=A_BROADCAST, video_track="video" - ) + )) subscriber._reported_an_oversized_bag = True with mock.patch.object( processors_module, "encode_bag_to_msgpack_bytes" diff --git a/packages/streamlib-moq/tests/test_wire_contract.py b/packages/streamlib-moq/tests/test_wire_contract.py index b8a9582a3..e7094b7c4 100644 --- a/packages/streamlib-moq/tests/test_wire_contract.py +++ b/packages/streamlib-moq/tests/test_wire_contract.py @@ -23,7 +23,7 @@ from streamlib import EncodedAudioPacket, EncodedVideoFrame, encode_bag_to_msgpack_bytes, log from streamlib._engine import ProcessorLinkDataAccess -from streamlib_moq import MoqBroadcastSubscriber, _native +from streamlib_moq import MoqBroadcastSubscriber, MoqBroadcastSubscriberConfig, _native from streamlib_moq.processors import ( DATA_BAGS_OUTPUT_PORT, encoded_audio_packet_bag, @@ -96,12 +96,12 @@ def write( def a_data_track_subscriber() -> MoqBroadcastSubscriber: - return MoqBroadcastSubscriber( + return MoqBroadcastSubscriber(MoqBroadcastSubscriberConfig( relay_url="https://relay.invalid/a-token", broadcast="a-broadcast", container_format="streamlib_bag", data_track=DATA_TRACK, - ) + )) def an_envelope_stating(**overrides: Any) -> bytes: diff --git a/packages/streamlib-webrtc/pyproject.toml b/packages/streamlib-webrtc/pyproject.toml index 729a41e16..16f91c6f3 100644 --- a/packages/streamlib-webrtc/pyproject.toml +++ b/packages/streamlib-webrtc/pyproject.toml @@ -29,13 +29,19 @@ classifiers = [ # The engine is a binary dependency, never a source one: this wheel links no # streamlib crate and speaks no engine internals. # -# 0.18.49 is the earliest release carrying both halves this wheel needs. The -# capability-extension mechanism `extension.py:load` is declared against arrived -# in 0.18.48; `read_from_inbound_link_with_timestamp`, which every bag is read -# through, arrived in 0.18.49 — in the same commit as this wheel, which is why -# the floor could only be stated once the release wiring existed. Both resolved -# by `git describe --contains` on the commit that introduced each into -# `_engine.pyi`, not from a changelog. +# What this wheel needs arrived across two releases: the capability-extension +# mechanism `extension.py:load` is declared against in 0.18.48, and +# `read_from_inbound_link_with_timestamp`, which every bag is read through, in +# 0.18.49. Both resolved by `git describe --contains` on the commit that +# introduced each into `_engine.pyi`, not from a changelog. +# +# The binding half is newer than both and cannot be named. These processors take +# a config class, which only an engine whose helper constructs one can host, and +# that engine ships in the release *after* this floor was last touched — a +# version the ceiling rule below forbids naming. So 0.20.0 is the highest floor +# that is legal here, not the floor that is true: an install against exactly +# 0.20.0 resolves and then fails in the helper child. Bump this to the release +# carrying config-class hosting once it exists. # # The ceiling is the last *published* engine version, not the one in the tree. # Two lanes install this wheel and they resolve differently: the PR lane @@ -43,7 +49,7 @@ classifiers = [ # lane installs from the published index, which is rebuilt only after the # extension wheels are built. So a floor naming the engine version released in # the same run would fail the release smoke test. -dependencies = ["streamlib>=0.18.49"] +dependencies = ["streamlib>=0.20.0"] # What pip records at install and the engine reads back through # `importlib.metadata` when a process takes an engine role. diff --git a/packages/streamlib-webrtc/python/streamlib_webrtc/__init__.py b/packages/streamlib-webrtc/python/streamlib_webrtc/__init__.py index 3aa6fd8f9..d41eb6301 100644 --- a/packages/streamlib-webrtc/python/streamlib_webrtc/__init__.py +++ b/packages/streamlib-webrtc/python/streamlib_webrtc/__init__.py @@ -10,6 +10,8 @@ """ from .processors import WhepPlayer as WhepPlayer +from .processors import WhepPlayerConfig as WhepPlayerConfig from .processors import WhipPublisher as WhipPublisher +from .processors import WhipPublisherConfig as WhipPublisherConfig -__all__ = ["WhepPlayer", "WhipPublisher"] +__all__ = ["WhepPlayer", "WhepPlayerConfig", "WhipPublisher", "WhipPublisherConfig"] diff --git a/packages/streamlib-webrtc/python/streamlib_webrtc/processors.py b/packages/streamlib-webrtc/python/streamlib_webrtc/processors.py index a3bc1465e..36e47b6fc 100644 --- a/packages/streamlib-webrtc/python/streamlib_webrtc/processors.py +++ b/packages/streamlib-webrtc/python/streamlib_webrtc/processors.py @@ -12,9 +12,10 @@ from __future__ import annotations +import dataclasses import threading from collections.abc import Mapping -from typing import Any, Literal, Protocol +from typing import Annotated, Any, Literal, Protocol from streamlib import ( EncodedAudioPacket, @@ -211,6 +212,16 @@ def _optional_bearer_token(bearer_token: Any) -> "str | None": return bearer_token if isinstance(bearer_token, str) and bearer_token else None +@dataclasses.dataclass +class WhipPublisherConfig: + """What a `WhipPublisher` is configured with.""" + + url: Annotated[str, "The WHIP endpoint to publish to."] + bearer_token: Annotated[ + "str | None", "Sent as `Authorization: Bearer` when the endpoint wants one." + ] = None + + @processor( description=( "Publishes encoded video and audio to a WHIP endpoint, " @@ -222,8 +233,8 @@ class WhipPublisher: The `Mp4Sink` shape: one fan-in input, and each inbound link is one track whose medium the link's first bag settles by its `codec`. Its settings are - ordinary constructor parameters — `url`, and an optional `bearer_token` — - which is what `rt.add(WhipPublisher, config={"url": ...})` passes. + a `WhipPublisherConfig` — `url`, and an optional `bearer_token` — which is + what `rt.add(WhipPublisher, config={"url": ...})` is constructed into. The session opens on the first bag rather than in `setup()`, because a relay round trip inside `setup()` spends the helper's start-up budget and a @@ -238,9 +249,9 @@ class WhipPublisher: zero. """ - def __init__(self, url: str, bearer_token: "str | None" = None) -> None: - self._url = _required_url(url, "WhipPublisher") - self._bearer_token = _optional_bearer_token(bearer_token) + def __init__(self, config: WhipPublisherConfig) -> None: + self._url = _required_url(config.url, "WhipPublisher") + self._bearer_token = _optional_bearer_token(config.bearer_token) self._session: "_native.WhipSession | None" = None self._inbound_links: "list[str]" = [] self._kind_by_inbound_link: "dict[str, VideoOrAudio]" = {} @@ -355,6 +366,16 @@ def _connected_session(self) -> "_native.WhipSession": return self._session +@dataclasses.dataclass +class WhepPlayerConfig: + """What a `WhepPlayer` is configured with.""" + + url: Annotated[str, "The WHEP endpoint to play from."] + bearer_token: Annotated[ + "str | None", "Sent as `Authorization: Bearer` when the endpoint wants one." + ] = None + + @processor( execution="manual", description="Plays encoded video and audio back from a WHEP endpoint", @@ -375,9 +396,9 @@ class WhepPlayer: decoder trims nothing. """ - def __init__(self, url: str, bearer_token: "str | None" = None) -> None: - self._url = _required_url(url, "WhepPlayer") - self._bearer_token = _optional_bearer_token(bearer_token) + def __init__(self, config: WhepPlayerConfig) -> None: + self._url = _required_url(config.url, "WhepPlayer") + self._bearer_token = _optional_bearer_token(config.bearer_token) self._stop = threading.Event() self._reader: "threading.Thread | None" = None self._reported_an_oversized_bag = False diff --git a/packages/streamlib-webrtc/tests/test_processors.py b/packages/streamlib-webrtc/tests/test_processors.py index b2c206612..19f2479fe 100644 --- a/packages/streamlib-webrtc/tests/test_processors.py +++ b/packages/streamlib-webrtc/tests/test_processors.py @@ -24,7 +24,7 @@ ) from streamlib._engine import ProcessorLinkDataAccess from streamlib._processor_hosting import construct_processor_instance -from streamlib_webrtc import WhepPlayer, WhipPublisher +from streamlib_webrtc import WhepPlayer, WhepPlayerConfig, WhipPublisher from streamlib_webrtc.processors import ( FIRST_RECONNECT_DELAY_SECONDS, HELPER_LINK_PAYLOAD_CEILING_BYTES, @@ -136,7 +136,7 @@ def test_an_oversized_bag_is_reported_once_rather_than_silently_dropped( per-frame condition reported per frame is noise, so it says it once.""" reported: "list[str]" = [] monkeypatch.setattr(log, "error", reported.append) - player = WhepPlayer(url="https://example.invalid/whep") + player = WhepPlayer(WhepPlayerConfig(url="https://example.invalid/whep")) over_the_ceiling = b"\x00" * (HELPER_LINK_PAYLOAD_CEILING_BYTES + 1) player._report_a_bag_the_link_will_drop("encoded_video", over_the_ceiling) @@ -151,7 +151,11 @@ def test_a_bag_inside_the_ceiling_is_not_reported(monkeypatch): reported: "list[str]" = [] monkeypatch.setattr(log, "error", reported.append) - WhepPlayer(url="https://example.invalid/whep")._report_a_bag_the_link_will_drop( + WhepPlayer( + WhepPlayerConfig( + url="https://example.invalid/whep" + ) + )._report_a_bag_the_link_will_drop( "encoded_video", b"\x00" * 4096 ) @@ -196,8 +200,8 @@ def set_up_with( context = RuntimeContextFullAccess.open_for_helper_process( {}, link_data_access, "runtime-under-test", "processor-under-test" ) - # Constructed the way the helper constructs it: config is the class's - # own keyword arguments, not something read off the context. + # Constructed the way the helper constructs it: the mapping becomes + # the class's config object, not something read off the context. construct_processor_instance(WhipPublisher, config, link_data_access).setup( context ) @@ -235,21 +239,21 @@ def test_a_publisher_whose_endpoint_is_not_an_address_is_refused_by_name( _PublisherUnderTest.set_up_with(request, 1, config) -def test_a_publisher_added_with_no_endpoint_at_all_is_refused_by_the_engine(request): - """`url` has no default, so the engine's own construction refusal names it - before any of this wheel's validation runs.""" +def test_a_publisher_added_with_no_endpoint_at_all_is_refused_by_its_config_class(request): + """`url` has no default, so the config class refuses the empty mapping and + names the missing key before any of this wheel's own validation runs.""" with pytest.raises(TypeError, match="missing 1 required positional argument"): _PublisherUnderTest.set_up_with(request, 1, {}) @pytest.mark.parametrize("processor_class", [WhipPublisher, WhepPlayer]) -def test_config_reaches_a_processor_as_its_own_constructor_keywords(processor_class): +def test_config_reaches_a_processor_as_its_own_config_object(processor_class): """The shape `rt.add(cls, config={...})` actually delivers. - `rt.add` records the config and the helper constructs the class from it, so - a class whose settings are not constructor parameters passes every - graph-building test and then fails in the child on the first run. This is - the engine's own mapping, called directly. + `rt.add` records the config and the helper constructs the class's config + class from it, so a class whose settings are not on that config class + passes every graph-building test and then fails in the child on the first + run. This is the engine's own mapping, called directly. """ constructed = construct_processor_instance( processor_class, @@ -370,7 +374,7 @@ def scripted_whep_session(monkeypatch): def _player_for_a_drain_that_returns_immediately() -> WhepPlayer: - player = WhepPlayer(url="https://example.invalid/whep") + player = WhepPlayer(WhepPlayerConfig(url="https://example.invalid/whep")) # The drain would otherwise spin on `next_media` returning None; stopping # the player makes each attempt reach the backoff at once. return player diff --git a/runtime/streamlib-api-server/src/mcp.rs b/runtime/streamlib-api-server/src/mcp.rs index e329562cc..bce8ddbcf 100644 --- a/runtime/streamlib-api-server/src/mcp.rs +++ b/runtime/streamlib-api-server/src/mcp.rs @@ -300,7 +300,7 @@ fn tool_definitions() -> Vec { "type": "object", "properties": { "type": { "type": "string", "description": "The processor class import path, e.g. `processors.grayscale_effect:GrayscaleEffect`." }, - "config": { "type": "object", "description": "The processor's configuration — for a Python class, the keyword arguments its constructor takes. Omit for none." }, + "config": { "type": "object", "description": "The processor's configuration, as the keys its config schema declares. Omit for none." }, "display_name": { "type": "string", "description": "Human-facing label; defaults to the class's short name, disambiguated within the graph." } }, "required": ["type"], diff --git a/runtime/streamlib-engine/src/core/json_schema.rs b/runtime/streamlib-engine/src/core/json_schema.rs index fecf96fcc..5fd9b2273 100644 --- a/runtime/streamlib-engine/src/core/json_schema.rs +++ b/runtime/streamlib-engine/src/core/json_schema.rs @@ -876,9 +876,9 @@ mod config_schema_rendering_tests { assert_eq!(rendered["config_schema"], document); } - /// A descriptor built without one — every Python descriptor today — - /// renders no key at all rather than a null an agent would have to read - /// as "no config". + /// A descriptor built without one — which no declared processor is, in + /// either language — renders no key at all rather than a null an agent + /// would have to read as "no config". #[test] fn a_descriptor_carrying_no_config_schema_renders_no_key_rather_than_a_null() { let rendered = diff --git a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi index 57629973e..8745b3abc 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi +++ b/sdk/streamlib-python-wheel/python/streamlib/_engine.pyi @@ -465,7 +465,15 @@ class Runtime: config: dict[str, Any] | None = None, display_name: str | None = None, ) -> AddedProcessor: - """Add a processor class to the graph, configured with `config`.""" + """Add a processor class to the graph, configured with `config`. + + `config` is the mapping the processor's config type is built from: for + a Python class, the class named by the annotation on its `__init__`'s + `config` parameter; for a native built-in, its Rust config struct. + Either way a processor that declares no config refuses a non-empty one, + and the keys it takes — with their types, defaults and descriptions — + are published as its config schema in the processor catalog. + """ def connect( self, source: ProcessorOutputPortReference, destination: ProcessorInputPortReference diff --git a/sdk/streamlib-python-wheel/python/streamlib/_processor_config_schema.py b/sdk/streamlib-python-wheel/python/streamlib/_processor_config_schema.py new file mode 100644 index 000000000..d86388681 --- /dev/null +++ b/sdk/streamlib-python-wheel/python/streamlib/_processor_config_schema.py @@ -0,0 +1,465 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""Deriving a processor's config schema from the config class its author wrote. + +The document is JSON Schema draft 2020-12 with no `$schema` key — the dialect +`sdk/streamlib-processor-schema/src/config_schema_document.rs` emits for a Rust +config type, so a node serves one dialect whichever language declared the +processor. Nested classes are inlined and `Optional[T]` is an `anyOf` with null, +so nothing here emits a `$ref`. One document can still carry `$defs`: a model +handed in as the config class contributes its own schema, and a pydantic model +writes its nested types that way. A model nested under a *property* is flattened +instead, because its pointers are root-relative and resolve against nothing once +the document is no longer the root. + +The dialect is shared; two spellings inside it are not, and both are valid +2020-12. The Rust side writes a nullable as `"type": [T, "null"]` and stamps a +root `title` from the config type's name, because `schemars` does. Neither is +worth converting on either side, and a reader who takes one for drift would +break the other. + +Stdlib only. A model is recognised by the `model_json_schema` method it carries +rather than by importing pydantic, which the wheel does not depend on. + +`additionalProperties` is stated only where the config class refuses unknown +keys: a dataclass does, a TypedDict does not, and a model's own document speaks +for itself. +""" + +from __future__ import annotations + +import collections.abc +import dataclasses +import enum +import inspect +import math +import types +import typing +from typing import Any + +__all__ = [ + "derive_config_class_json_schema", + "json_schema_for_a_processor_declaring_no_config", +] + +_SCALAR_JSON_TYPES = { + bool: "boolean", + int: "integer", + float: "number", + str: "string", +} + +_SEQUENCE_ORIGINS = ( + list, + set, + frozenset, + tuple, + collections.abc.Sequence, + collections.abc.MutableSequence, + collections.abc.Set, +) + +_MAPPING_ORIGINS = ( + dict, + collections.abc.Mapping, + collections.abc.MutableMapping, +) + +# `get_type_hints` keeps a TypedDict key's requiredness qualifier, which says +# nothing about the value's type and hides it from the mapper. Recognised by +# the special form's own name rather than by identity, because the 3.10 floor +# has these from `typing_extensions` and 3.11+ from `typing`, and a config +# class may be written against either. +_REQUIREDNESS_QUALIFIER_NAMES = ("Required", "NotRequired") + +# Where a model's own document points at its own definitions. +_DEFINITION_POINTER_PREFIX = "#/$defs/" + + +def _is_a_typed_dict(candidate: Any) -> bool: + """Whether `candidate` is a TypedDict under either spelling. + + `typing.is_typeddict` recognises `typing.TypedDict` alone, and a + `typing_extensions.TypedDict` subclass — what an author on the 3.10 floor + reaches for, and what `Required` / `NotRequired` need there — is invisible + to it. The pair of key sets is the structural signature of one; nothing + else carries both. + """ + return typing.is_typeddict(candidate) or ( + isinstance(candidate, type) + and hasattr(candidate, "__required_keys__") + and hasattr(candidate, "__optional_keys__") + ) + + +def json_schema_for_a_processor_declaring_no_config() -> "dict[str, Any]": + """The document a processor that takes no configuration publishes. + + Mirrors what `EmptyConfig` publishes on the Rust side, description + included, so the catalog reads the same for either language. + """ + return { + "type": "object", + "description": "This processor declares no configuration.", + "additionalProperties": False, + } + + +def derive_config_class_json_schema(config_class: type) -> "dict[str, Any]": + """The JSON Schema of `config_class`, derived from what its author wrote.""" + return _document_for_class(config_class, ()) + + +def _document_for_class( + config_class: type, classes_being_inlined: "tuple[type, ...]" +) -> "dict[str, Any]": + if config_class in classes_being_inlined: + # A config class that reaches itself. Inlining is the only nesting this + # module emits, so a cycle has no fixed point — the walk stops here and + # the key says only that it is an object. Without this the recursion + # exhausts the stack at decoration, which is import time. + return {"type": "object"} + + model_json_schema = getattr(config_class, "model_json_schema", None) + if callable(model_json_schema): + return _document_the_model_carries(config_class, model_json_schema) + + ancestry = classes_being_inlined + (config_class,) + if _is_a_typed_dict(config_class): + return _typed_dict_document(config_class, ancestry) + if dataclasses.is_dataclass(config_class): + return _dataclass_document(config_class, ancestry) + # An open object says the configuration is a mapping and claims nothing + # about its keys, which is all that can honestly be derived from a class of + # a kind the deriver does not recognise. + return {"type": "object"} + + +def _resolved_against_its_own_definitions(document: Any) -> Any: + """`document` with each `#/$defs/` pointer replaced by what it points at. + + A model nested under a property carries pointers written when it was the + root, so they name a `$defs` the enclosing document does not have. Nothing + else here emits a `$ref`, so this only ever has a model's document to walk. + """ + if not isinstance(document, dict) or "$defs" not in document: + return document + definitions = document["$defs"] + resolved = {key: value for key, value in document.items() if key != "$defs"} + return _with_pointers_followed(resolved, definitions, ()) + + +def _with_pointers_followed( + node: Any, definitions: "dict[str, Any]", pointers_being_followed: "tuple[str, ...]" +) -> Any: + if isinstance(node, list): + return [ + _with_pointers_followed(entry, definitions, pointers_being_followed) + for entry in node + ] + if not isinstance(node, dict): + return node + + pointer = node.get("$ref") + if isinstance(pointer, str) and pointer.startswith(_DEFINITION_POINTER_PREFIX): + name = pointer[len(_DEFINITION_POINTER_PREFIX) :] + if name in pointers_being_followed or name not in definitions: + # A definition that reaches itself, or one the document never + # carried: an open object beats a pointer to nothing. + return {"type": "object"} + followed = _with_pointers_followed( + definitions[name], definitions, pointers_being_followed + (name,) + ) + beside_the_pointer = { + key: _with_pointers_followed(value, definitions, pointers_being_followed) + for key, value in node.items() + if key != "$ref" + } + return {**followed, **beside_the_pointer} + + return { + key: _with_pointers_followed(value, definitions, pointers_being_followed) + for key, value in node.items() + } + + +def _document_the_model_carries( + config_class: type, model_json_schema: "typing.Callable[[], Any]" +) -> "dict[str, Any]": + """A model's own schema, verbatim minus the two keys the catalog owns.""" + document = model_json_schema() + if not isinstance(document, dict): + raise TypeError( + f"{config_class.__name__}.model_json_schema() returned " + f"{type(document).__name__} rather than a dict, so it cannot be a " + f"processor's config class. A config class is a TypedDict, a dataclass, " + f"or a model whose `model_json_schema()` returns a JSON Schema document." + ) + return { + key: value for key, value in document.items() if key not in ("$schema", "title") + } + + +def _typed_dict_document( + config_class: type, ancestry: "tuple[type, ...]" +) -> "dict[str, Any]": + annotations = _resolved_class_annotations(config_class) + required_keys = getattr(config_class, "__required_keys__", frozenset()) + document: "dict[str, Any]" = { + "type": "object", + "properties": { + key: _json_schema_for_annotation(annotation, ancestry) + for key, annotation in annotations.items() + }, + } + # Declaration order rather than the set's, so deriving one class twice + # gives one document. + required = [key for key in annotations if key in required_keys] + if required: + document["required"] = required + return document + + +def _dataclass_document( + config_class: type, ancestry: "tuple[type, ...]" +) -> "dict[str, Any]": + annotations = _resolved_class_annotations(config_class) + fields_by_name = {field.name: field for field in dataclasses.fields(config_class)} + # An `InitVar` is a constructor input that `dataclasses.fields()` omits. Left + # out it would be absent from `properties` while `additionalProperties: false` + # forbade it — a catalog telling an agent that a required key is illegal. + init_parameters = inspect.signature(config_class).parameters + # And the constructor has the final say on what a configuration may carry, + # because that is what `config_class(**configuration)` calls. A generated + # `__init__` takes exactly the `init=True` fields, so this narrows nothing + # for an ordinary dataclass; `init=False` or a hand-written constructor is + # where the field list and the callable disagree, and documenting a key the + # class refuses is the same lie as omitting one it requires. + accepts_any_key = any( + parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in init_parameters.values() + ) + + properties: "dict[str, Any]" = {} + required: "list[str]" = [] + for name, annotation in annotations.items(): + if not accepts_any_key and name not in init_parameters: + continue + field = fields_by_name.get(name) + if field is not None: + # An `init=False` field is not a constructor input, so a + # configuration cannot carry it and it is not documented. + if not field.init: + continue + declared_default = ( + field.default if field.default is not dataclasses.MISSING else _ABSENT + ) + has_default_factory = field.default_factory is not dataclasses.MISSING + annotation = annotations.get(name, field.type) + elif isinstance(annotation, dataclasses.InitVar): + parameter = init_parameters.get(name) + declared_default = ( + _ABSENT + if parameter is None or parameter.default is inspect.Parameter.empty + else parameter.default + ) + has_default_factory = False + annotation = annotation.type + else: + # A `ClassVar` or a bare annotation the dataclass machinery ignored: + # not a constructor input either way. + continue + + field_schema = _json_schema_for_annotation(annotation, ancestry) + if declared_default is not _ABSENT: + rendered_default = _json_representable(declared_default) + if rendered_default is not _NOT_JSON_REPRESENTABLE: + field_schema["default"] = rendered_default + elif not has_default_factory: + required.append(name) + properties[name] = field_schema + + document: "dict[str, Any]" = { + "type": "object", + "properties": properties, + "additionalProperties": False, + } + if required: + document["required"] = required + return document + + +def _resolved_class_annotations(config_class: type) -> "dict[str, Any]": + """Every annotated field of `config_class`, with forward references resolved.""" + try: + return typing.get_type_hints(config_class, include_extras=True) + except Exception as unresolvable: + raise TypeError( + f"{config_class.__name__} carries an annotation that cannot be resolved, " + f"so its config schema cannot be derived: {unresolvable}. Every " + f"annotation on a config class must name something importable at run " + f"time, not only under `TYPE_CHECKING`." + ) from unresolvable + + +def _json_schema_for_annotation( + annotation: Any, ancestry: "tuple[type, ...]" = () +) -> "dict[str, Any]": + """The schema of one annotated field. + + An annotation the deriver does not recognise renders as an empty schema + rather than refusing the class: a config class is worth publishing long + before every type in it is describable. + """ + origin_name = getattr(typing.get_origin(annotation), "_name", None) + if origin_name in _REQUIREDNESS_QUALIFIER_NAMES: + return _json_schema_for_annotation(typing.get_args(annotation)[0], ancestry) + + metadata = getattr(annotation, "__metadata__", None) + if metadata is not None: + described = _json_schema_for_annotation(typing.get_args(annotation)[0], ancestry) + description = next((entry for entry in metadata if isinstance(entry, str)), None) + if description is not None: + described["description"] = description + return described + + if annotation is None or annotation is type(None): + return {"type": "null"} + if annotation is Any: + return {} + + origin = typing.get_origin(annotation) + if origin is typing.Union or origin is types.UnionType: + return { + "anyOf": [ + _json_schema_for_annotation(member, ancestry) + for member in typing.get_args(annotation) + ] + } + if origin is typing.Literal: + return _enumerated_schema(typing.get_args(annotation)) + if origin is not None: + if origin in _SEQUENCE_ORIGINS: + return _sequence_schema(annotation, origin, ancestry) + if origin in _MAPPING_ORIGINS: + return {"type": "object"} + # A parameterized generic the deriver does not know. This branch is + # also what keeps one out of the nested-class branch below on Python + # 3.10, where `isinstance(dict[str, int], type)` is still True. + return {} + + if isinstance(annotation, type): + if annotation in _SCALAR_JSON_TYPES: + return {"type": _SCALAR_JSON_TYPES[annotation]} + if annotation in _SEQUENCE_ORIGINS: + return {"type": "array"} + if annotation in _MAPPING_ORIGINS: + return {"type": "object"} + if issubclass(annotation, enum.Enum): + return _enumerated_schema(tuple(member.value for member in annotation)) + if ( + _is_a_typed_dict(annotation) + or dataclasses.is_dataclass(annotation) + or callable(getattr(annotation, "model_json_schema", None)) + ): + # A model's own document points at its own `$defs` with a + # root-relative pointer, which resolves against nothing once the + # document sits under a property. Every other kind is already + # self-contained, so this is a no-op for them. + return _resolved_against_its_own_definitions( + _document_for_class(annotation, ancestry) + ) + + return {} + + +def _enumerated_schema(members: "tuple[Any, ...]") -> "dict[str, Any]": + rendered = [_json_representable(member) for member in members] + if any(member is _NOT_JSON_REPRESENTABLE for member in rendered): + return {} + return {"enum": rendered} + + +def _sequence_schema( + annotation: Any, origin: Any, ancestry: "tuple[type, ...]" +) -> "dict[str, Any]": + document: "dict[str, Any]" = {"type": "array"} + element_annotations = typing.get_args(annotation) + if origin is tuple: + # `tuple[T, ...]` is a homogeneous sequence; every other tuple is + # positional, which 2020-12 spells `prefixItems`. + if len(element_annotations) == 2 and element_annotations[1] is Ellipsis: + document["items"] = _json_schema_for_annotation( + element_annotations[0], ancestry + ) + elif element_annotations: + positional_item_schemas = [ + _json_schema_for_annotation(element, ancestry) + for element in element_annotations + ] + document["prefixItems"] = positional_item_schemas + # `prefixItems` says what each position holds and nothing about how + # many there are, so a fixed-length tuple that stated only that + # would validate a shorter or longer array. The Rust seam bounds + # its tuples the same way. + document["minItems"] = len(positional_item_schemas) + document["maxItems"] = len(positional_item_schemas) + return document + if len(element_annotations) == 1: + document["items"] = _json_schema_for_annotation( + element_annotations[0], ancestry + ) + return document + + +class _NotJsonRepresentableSentinel: + """The type of [`_NOT_JSON_REPRESENTABLE`] — never constructed by an author.""" + + __slots__ = () + + def __repr__(self) -> str: + return "_NOT_JSON_REPRESENTABLE" + + +# Distinct from `None`, which is itself a representable default. +_NOT_JSON_REPRESENTABLE = _NotJsonRepresentableSentinel() + +# Distinct from `None` for the same reason: a field may default to it. +_ABSENT = object() + +# The document crosses into the descriptor through the msgpack value tree the +# data plane uses, which carries no integer wider than 64 bits and would refuse +# the whole declaration rather than the one default. +_WIDEST_REPRESENTABLE_INTEGERS = range(-(2**63), 2**64) + + +def _json_representable(value: Any) -> Any: + """`value` as JSON, or [`_NOT_JSON_REPRESENTABLE`] if it is not expressible. + + The derived document crosses into Rust as JSON, so a default the wire + cannot carry is dropped rather than left to fail the whole declaration. + """ + if isinstance(value, enum.Enum): + return _json_representable(value.value) + if value is None or isinstance(value, (bool, str)): + return value + if isinstance(value, int): + return value if value in _WIDEST_REPRESENTABLE_INTEGERS else _NOT_JSON_REPRESENTABLE + if isinstance(value, float): + # JSON has no infinity and no NaN, so a default that is one cannot + # cross into the descriptor and is dropped rather than rewritten. + return value if math.isfinite(value) else _NOT_JSON_REPRESENTABLE + if isinstance(value, (list, tuple)): + rendered = [_json_representable(entry) for entry in value] + if any(entry is _NOT_JSON_REPRESENTABLE for entry in rendered): + return _NOT_JSON_REPRESENTABLE + return rendered + if isinstance(value, dict): + if not all(isinstance(key, str) for key in value): + return _NOT_JSON_REPRESENTABLE + rendered = {key: _json_representable(entry) for key, entry in value.items()} + if any(entry is _NOT_JSON_REPRESENTABLE for entry in rendered.values()): + return _NOT_JSON_REPRESENTABLE + return rendered + return _NOT_JSON_REPRESENTABLE diff --git a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py index ebfb7ed0b..d6c092ea9 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py +++ b/sdk/streamlib-python-wheel/python/streamlib/_processor_declaration.py @@ -18,8 +18,15 @@ from __future__ import annotations import dataclasses +import inspect +import typing from typing import Any, Callable, Optional, TypeVar +from ._processor_config_schema import ( + derive_config_class_json_schema, + json_schema_for_a_processor_declaring_no_config, +) + __all__ = [ "AudioWindowContract", @@ -340,6 +347,14 @@ def processor( input port, and is required for one that declares none — a source has nothing to react to, so defaulting it there would produce a processor that silently never runs. + + Configuration is one class, named by the annotation on the `config` + parameter of `__init__` — a TypedDict, a dataclass or a model. A class + whose `__init__` takes nothing beyond `self` declares no config and refuses + one. Any other signature is refused here, at decoration. The class's JSON + Schema is derived from its annotations and defaults and published in the + processor catalog, which is how an agent learns the keys before adding the + node. """ if isinstance(processor_class, type): return _declare_processor( @@ -381,8 +396,15 @@ def _declare_processor( description: str, ) -> ProcessorClass: input_ports, output_ports = _collect_declared_ports(processor_class) + config_class = _config_class_named_by_the_init_annotation(processor_class) processor_class.__streamlib_processor_declared__ = True # type: ignore[attr-defined] + processor_class.__streamlib_processor_config_class__ = config_class # type: ignore[attr-defined] + processor_class.__streamlib_processor_config_schema__ = ( # type: ignore[attr-defined] + json_schema_for_a_processor_declaring_no_config() + if config_class is None + else derive_config_class_json_schema(config_class) + ) processor_class.__streamlib_processor_description__ = description # type: ignore[attr-defined] processor_class.__streamlib_processor_execution__ = _resolve_execution( # type: ignore[attr-defined] execution, interval_ms, processor_class, has_input_ports=bool(input_ports) @@ -395,6 +417,102 @@ def _declare_processor( return processor_class +def _config_class_named_by_the_init_annotation( + processor_class: type, +) -> "Optional[type]": + """The config class `processor_class.__init__` names, or `None` for no config. + + Every other signature is refused here rather than at the first `add`: the + decorator runs at import, which is the last moment an author is still + looking at the class. + """ + # A class defining no `__init__` inherits `object`'s, whose signature is + # `(self, /, *args, **kwargs)` — a shape that would otherwise be refused. + if processor_class.__init__ is object.__init__: + return None + + parameters = [ + parameter + for name, parameter in inspect.signature(processor_class.__init__).parameters.items() + if name != "self" + ] + if not parameters: + return None + + how_to_declare_a_config_class = ( + f"declare one parameter named `config`, annotated with the class its settings " + f"live on — `def __init__(self, config: {processor_class.__name__}Config) -> " + f"None` — where that class is a TypedDict, a dataclass or a model. " + f"`rt.add(cls, config={{...}})` still passes a dict; the helper constructs the " + f"class from it and hands the object in." + ) + + if len(parameters) > 1: + raise TypeError( + f"{processor_class.__name__}.__init__ takes {len(parameters)} parameters " + f"besides `self` ({', '.join(parameter.name for parameter in parameters)}); " + f"a processor's config is one class, not a parameter list. To fix: {how_to_declare_a_config_class}" + ) + + parameter = parameters[0] + if parameter.kind is inspect.Parameter.VAR_KEYWORD: + raise TypeError( + f"{processor_class.__name__}.__init__ takes `**{parameter.name}`; " + f"keyword-argument configuration is not how a processor is configured. " + f"To fix: {how_to_declare_a_config_class}" + ) + if parameter.kind is inspect.Parameter.VAR_POSITIONAL: + raise TypeError( + f"{processor_class.__name__}.__init__ takes `*{parameter.name}`; a " + f"processor's config is one object, not a variadic. To fix: {how_to_declare_a_config_class}" + ) + if parameter.name != "config": + raise TypeError( + f"{processor_class.__name__}.__init__ takes `{parameter.name}`, but a " + f"processor's config parameter must be named `config`. To fix: {how_to_declare_a_config_class}" + ) + if parameter.kind is inspect.Parameter.POSITIONAL_ONLY: + raise TypeError( + f"{processor_class.__name__}.__init__ takes `config` positionally only, but " + f"the helper constructs a processor as `cls(config=...)`. To fix: drop the " + f"`/` so `config` can be passed by name." + ) + + annotation = _resolved_init_annotations(processor_class).get("config") + if annotation is None: + raise TypeError( + f"{processor_class.__name__}.__init__ takes `config` with no annotation, so " + f"nothing names its config class and no schema can be derived. " + f"To fix: {how_to_declare_a_config_class}" + ) + if annotation is Any: + raise TypeError( + f"{processor_class.__name__}.__init__ annotates `config` as `Any`, which " + f"names no class, so the helper has nothing to construct and no schema can " + f"be derived. To fix: {how_to_declare_a_config_class}" + ) + # The origin check, not the class check, is what refuses `dict[str, Any]` on + # Python 3.10, where `isinstance(dict[str, Any], type)` is still True. + if typing.get_origin(annotation) is not None or not isinstance(annotation, type): + raise TypeError( + f"{processor_class.__name__}.__init__ annotates `config` as {annotation!r}, " + f"which is not a class. A processor's config is one class the helper " + f"constructs. To fix: {how_to_declare_a_config_class}" + ) + return annotation + + +def _resolved_init_annotations(processor_class: type) -> "dict[str, Any]": + try: + return typing.get_type_hints(processor_class.__init__, include_extras=True) + except Exception as unresolvable: + raise TypeError( + f"{processor_class.__name__}.__init__ carries an annotation that cannot be " + f"resolved, so its config class cannot be read: {unresolvable}. The config " + f"class must be importable at run time, not only under `TYPE_CHECKING`." + ) from unresolvable + + def _collect_declared_ports( processor_class: type, ) -> "tuple[list[dict[str, Any]], list[dict[str, Any]]]": diff --git a/sdk/streamlib-python-wheel/python/streamlib/_processor_hosting.py b/sdk/streamlib-python-wheel/python/streamlib/_processor_hosting.py index 4dcf21d93..3b62382a5 100644 --- a/sdk/streamlib-python-wheel/python/streamlib/_processor_hosting.py +++ b/sdk/streamlib-python-wheel/python/streamlib/_processor_hosting.py @@ -4,8 +4,8 @@ """Turning a declared processor class into a running processor object. The engine calls into this module rather than constructing the object itself: -mapping configuration onto constructor keywords is Python's job. App code -never calls anything here. +building a processor's config class out of the configuration is Python's job. +App code never calls anything here. """ from __future__ import annotations @@ -22,21 +22,19 @@ def construct_processor_instance( ) -> Any: """Instantiate `processor_class` with its configuration. - Configuration arrives as the keyword arguments the class was added with, so - a processor's settings are ordinary constructor parameters with ordinary - Python defaults — there is no configuration object to learn. The link data - access argument is unused: ports are reached through `ctx.inputs` / - `ctx.outputs`, but the host still passes it, so the arity stays. + The configuration arrives as the mapping the class was added with and is + constructed into the config class `__init__` names. Construction is the + only check performed here: how strict it is is the author's choice of + config class, the same dial `read(port, into=T)` is. The link data access + argument is unused — ports are reached through `ctx.inputs` / `ctx.outputs` + — but the host still passes it, so the arity stays. """ - keyword_arguments = _as_keyword_arguments(processor_class, configuration) - try: - return processor_class(**keyword_arguments) - except TypeError as construction_failure: - raise TypeError( - f"{processor_class.__name__}({_render_call(keyword_arguments)}) failed: " - f"{construction_failure}. `rt.add(cls, config={{...}})` passes config as " - f"keyword arguments to the class." - ) from construction_failure + config_class = getattr(processor_class, "__streamlib_processor_config_class__", None) + configuration = _as_configuration_mapping(processor_class, configuration) + if config_class is None: + _refuse_a_configuration_with_nowhere_to_go(processor_class, configuration) + return processor_class() + return processor_class(config=config_class(**configuration)) def apply_configuration(processor_instance: Any, configuration: Optional[Any]) -> None: @@ -45,33 +43,55 @@ def apply_configuration(processor_instance: Any, configuration: Optional[Any]) - Only processors that define `configure` accept one; for anything else a config change means a new pipeline, which is what re-running `dev` does. """ + processor_class = type(processor_instance) reconfigure = getattr(processor_instance, "configure", None) if reconfigure is None: raise TypeError( - f"{type(processor_instance).__name__} cannot be reconfigured while running: " - f"define `configure(self, **config)` on it to accept updates." + f"{processor_class.__name__} cannot be reconfigured while running: " + f"define `configure(self, config)` on it to take one." ) - reconfigure(**_as_keyword_arguments(type(processor_instance), configuration)) + config_class = getattr(processor_class, "__streamlib_processor_config_class__", None) + configuration = _as_configuration_mapping(processor_class, configuration) + if config_class is None: + _refuse_a_configuration_with_nowhere_to_go(processor_class, configuration) + reconfigure(None) + return + reconfigure(config_class(**configuration)) + +def _refuse_a_configuration_with_nowhere_to_go( + processor_class: type, configuration: "dict[str, Any]" +) -> None: + """Refuse a configuration handed to a class that declared none. + + Named rather than discarded, and named the same way the Rust `EmptyConfig` + names it: a processor that declares no config cannot act on one, and + silently dropping it hides a wiring mistake. + """ + if not configuration: + return + refused_key = next(iter(configuration)) + raise TypeError( + f"{processor_class.__name__} declares no config and takes none, so " + f"`{refused_key}` has nowhere to go. To fix: give its `__init__` a `config` " + f"parameter annotated with the class those settings live on, or drop the key." + ) -def _as_keyword_arguments( + +def _as_configuration_mapping( processor_class: type, configuration: Optional[Any] ) -> "dict[str, Any]": if configuration is None: return {} if not isinstance(configuration, dict): raise TypeError( - f"config for {processor_class.__name__} must be a dict of keyword arguments, " - f"got {type(configuration).__name__}" + f"config for {processor_class.__name__} must be a dict, got " + f"{type(configuration).__name__}" ) non_string_keys = [key for key in configuration if not isinstance(key, str)] if non_string_keys: raise TypeError( - f"config keys for {processor_class.__name__} must be strings — they become " - f"keyword arguments; got {non_string_keys!r}" + f"config keys for {processor_class.__name__} must be strings — they name " + f"the config class's fields; got {non_string_keys!r}" ) return dict(configuration) - - -def _render_call(keyword_arguments: "dict[str, Any]") -> str: - return ", ".join(f"{name}={value!r}" for name, value in keyword_arguments.items()) diff --git a/sdk/streamlib-python-wheel/src/python_bag_conversion.rs b/sdk/streamlib-python-wheel/src/python_bag_conversion.rs index cae8b9348..f3a86c032 100644 --- a/sdk/streamlib-python-wheel/src/python_bag_conversion.rs +++ b/sdk/streamlib-python-wheel/src/python_bag_conversion.rs @@ -263,8 +263,8 @@ pub(crate) fn python_type_name_for_error_message( ) } -/// Convert a processor's JSON configuration into the keyword arguments its -/// class is constructed with. +/// Convert a processor's JSON configuration into the mapping its config class +/// is constructed from. /// /// Routed through the same msgpack value tree the data plane uses rather than /// growing a second converter: the engine stores configuration as JSON, and one diff --git a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs index 94ff90e3a..cf2bfade3 100644 --- a/sdk/streamlib-python-wheel/src/python_processor_declaration.rs +++ b/sdk/streamlib-python-wheel/src/python_processor_declaration.rs @@ -17,6 +17,9 @@ use streamlib::sdk::descriptors::{ }; use streamlib::sdk::execution::{ExecutionConfig, ProcessExecution, ThreadPriority}; +use crate::python_bag_conversion::{ + python_object_to_json_value, python_type_name_for_error_message, +}; use crate::python_processor_import_path::processor_class_import_path; /// Everything the engine needs to register and instantiate one Python @@ -46,7 +49,8 @@ impl PythonProcessorDeclaration { .with_entrypoint(class_import_path) .with_scheduling(ProcessorScheduling { priority: read_thread_priority(processor_class)?, - }); + }) + .with_config_schema(read_config_schema_document(processor_class)?); descriptor.inputs = read_port_descriptors(processor_class, PortDirection::Input)?; descriptor.outputs = read_port_descriptors(processor_class, PortDirection::Output)?; @@ -78,6 +82,27 @@ fn read_class_short_name(processor_class: &Bound<'_, PyAny>) -> PyResult) -> PyResult { + let stamped = processor_class.getattr("__streamlib_processor_config_schema__")?; + // Refused here rather than by the converter, whose own messages are + // written for a bag on the data plane and would tell a processor author + // about GPU frames. + let document = stamped.clone().cast_into::().map_err(|_| { + PyTypeError::new_err(format!( + "__streamlib_processor_config_schema__ must be a JSON object, got a {} — the \ + decorator derives this document, so a class reaching here was built by hand \ + rather than by @streamlib.processor", + python_type_name_for_error_message(&stamped, "value of unknown type") + )) + })?; + python_object_to_json_value(document.as_any()) +} + fn read_execution_config(processor_class: &Bound<'_, PyAny>) -> PyResult { let execution = processor_class .getattr("__streamlib_processor_execution__")? @@ -406,6 +431,8 @@ fn read_dict_string(dictionary: &Bound<'_, PyDict>, key: &str) -> PyResult) -> Bound<'_, PyDict> { + install_stand_in_streamlib_package(python); let namespace = PyDict::new(python); + namespace.set_item("__package__", "streamlib").unwrap(); python .run( &std::ffi::CString::new(PROCESSOR_DECLARATION_MODULE_SOURCE).unwrap(), @@ -470,6 +504,45 @@ class BlurProcessor: namespace } + /// The wheel's own Python directory, where the decorator module's siblings + /// live. + const WHEEL_PYTHON_PACKAGE_DIRECTORY: &str = + concat!(env!("CARGO_MANIFEST_DIR"), "/python/streamlib"); + + /// Put a `streamlib` package on `sys.modules` whose search path is the real + /// source directory, so the decorator module's relative imports resolve + /// without an installed wheel. + /// + /// A package object already on `sys.modules` is never initialised again, so + /// `__init__.py` — which imports the compiled `_engine` a `cargo test` run + /// does not have — is not executed. Only the siblings actually imported are + /// loaded, and each is the same file `include_str!` above reads. + fn install_stand_in_streamlib_package(python: Python<'_>) { + let sys_modules = python + .import("sys") + .unwrap() + .getattr("modules") + .unwrap() + .cast_into::() + .unwrap(); + if sys_modules.contains("streamlib").unwrap() { + return; + } + + let package = python + .import("types") + .unwrap() + .call_method1("ModuleType", ("streamlib",)) + .unwrap(); + package + .setattr( + "__path__", + PyList::new(python, [WHEEL_PYTHON_PACKAGE_DIRECTORY]).unwrap(), + ) + .unwrap(); + sys_modules.set_item("streamlib", package).unwrap(); + } + /// Run the real decorator module, run `class_body_source` against it, and /// read the resulting class through the bridge the engine uses at `rt.add`. /// @@ -502,6 +575,130 @@ class BlurProcessor: .inputs } + /// The catalog's whole point: an agent reads a processor's keys off the + /// descriptor before the class is ever in a graph, so the document the + /// decorator derived has to survive the trip into Rust intact. + #[test] + fn the_descriptor_carries_the_config_classs_derived_schema() { + let declaration = read_python_declaration( + "\ +import dataclasses + + +@dataclasses.dataclass +class AudioConsumerConfig: + gain: float + label: str = 'unlabelled' + + +@processor(execution='manual') +class AudioConsumer: + def __init__(self, config: AudioConsumerConfig) -> None: + self.config = config +", + ) + .expect("the declaration reads"); + + let document = declaration + .descriptor + .config_schema + .expect("a declared Python class always carries a config schema"); + assert_eq!(document["type"], "object"); + assert_eq!(document["properties"]["gain"]["type"], "number"); + assert_eq!(document["properties"]["label"]["type"], "string"); + assert_eq!(document["properties"]["label"]["default"], "unlabelled"); + assert_eq!(document["required"], serde_json::json!(["gain"])); + assert_eq!(document["additionalProperties"], false); + } + + /// What `/api/registry` actually serializes for a Python class. + /// + /// The endpoint's own test registers a descriptor by hand, and the test + /// above stops at the descriptor, so without this nothing in a GPU-free CI + /// run carries a Python class's schema as far as the served shape — and the + /// end-to-end proof needs a running graph, which needs a GPU. + #[test] + fn the_served_rendering_of_a_python_class_carries_its_config_schema() { + let declaration = read_python_declaration( + "\ +import dataclasses +import typing + + +@dataclasses.dataclass +class AudioConsumerConfig: + gain: float + label: str = 'unlabelled' + fallback: typing.Optional[str] = None + + +@processor(execution='manual') +class AudioConsumer: + def __init__(self, config: AudioConsumerConfig) -> None: + self.config = config +", + ) + .expect("the declaration reads"); + + let served = serde_json::to_value( + streamlib::sdk::json_schema::ProcessorDescriptorOutput::from(&declaration.descriptor), + ) + .expect("the rendering serializes"); + + assert_eq!( + served["config_schema"]["properties"]["gain"]["type"], + "number" + ); + assert_eq!( + served["config_schema"]["properties"]["label"]["default"], + "unlabelled" + ); + assert_eq!( + served["config_schema"]["required"], + serde_json::json!(["gain"]) + ); + + // The null leg of the hop the document takes into Rust: a dropped key + // would read as "no default" rather than as the default the author + // wrote, and nothing else in a GPU-free run crosses a nil. + let fallback = &served["config_schema"]["properties"]["fallback"]; + assert!( + fallback.get("default").is_some(), + "the null default was dropped: {fallback}" + ); + assert_eq!(fallback["default"], serde_json::Value::Null); + } + + /// A processor declaring no config publishes what `EmptyConfig` publishes + /// in Rust, so one catalog reads one way whichever language declared the + /// processor. + /// + /// `schemars` stamps a root `title` from the config type's name and no + /// Python document carries one, so the comparison drops it. + #[test] + fn a_class_declaring_no_config_carries_the_same_document_rust_publishes() { + let declaration = read_python_declaration( + "\ +@processor(execution='manual') +class AudioConsumer: + def __init__(self) -> None: + self.frames = 0 +", + ) + .expect("the declaration reads"); + + let mut what_rust_publishes = EmptyConfig::processor_config_schema_document(); + what_rust_publishes + .as_object_mut() + .expect("the document is an object") + .remove("title"); + + assert_eq!( + declaration.descriptor.config_schema, + Some(what_rust_publishes) + ); + } + /// The message a refused Python declaration hands a user. /// /// `expect_err` is not available here: the success type is a production @@ -645,6 +842,7 @@ class AudioConsumer: __streamlib_processor_description__ = '' __streamlib_processor_execution__ = {{'mode': 'reactive'}} __streamlib_processor_scheduling_priority__ = None + __streamlib_processor_config_schema__ = {{'type': 'object'}} __streamlib_processor_input_ports__ = [{{ 'name': 'audio', 'description': '', @@ -843,6 +1041,7 @@ class AudioConsumer: __streamlib_processor_description__ = '' __streamlib_processor_execution__ = {'mode': 'manual'} __streamlib_processor_scheduling_priority__ = None + __streamlib_processor_config_schema__ = {'type': 'object'} __streamlib_processor_input_ports__ = [] __streamlib_processor_output_ports__ = [{ 'name': 'windows', diff --git a/sdk/streamlib-python-wheel/src/python_runtime_lifecycle.rs b/sdk/streamlib-python-wheel/src/python_runtime_lifecycle.rs index 6a81e8321..ceb2fb340 100644 --- a/sdk/streamlib-python-wheel/src/python_runtime_lifecycle.rs +++ b/sdk/streamlib-python-wheel/src/python_runtime_lifecycle.rs @@ -269,11 +269,12 @@ impl PythonRuntimeHandle { /// Add a processor class to the graph. /// - /// Takes the class, not an instance. `config` becomes the keyword arguments - /// the class is constructed with — which happens later, on the engine's - /// compile thread as `run()` brings the graph up, so a failing `__init__` - /// surfaces from `run()` rather than from here. Adding the same class twice - /// gives two processors, each with its own instance and configuration. + /// Takes the class, not an instance. `config` is the mapping the class's + /// config class is constructed from — which happens later, on the engine's + /// compile thread as `run()` brings the graph up, so a config the class + /// refuses surfaces from `run()` rather than from here. Adding the same + /// class twice gives two processors, each with its own instance and + /// configuration. #[pyo3(signature = (processor_class, *, config = None, display_name = None))] fn add( &self, diff --git a/sdk/streamlib-python-wheel/tests/capability_context_probes.py b/sdk/streamlib-python-wheel/tests/capability_context_probes.py index 307396420..3b51a9987 100644 --- a/sdk/streamlib-python-wheel/tests/capability_context_probes.py +++ b/sdk/streamlib-python-wheel/tests/capability_context_probes.py @@ -13,6 +13,7 @@ in it can be a live object. """ +import dataclasses import json import os import threading @@ -93,14 +94,27 @@ def observe() -> dict: # --------------------------------------------------------------------------- +@dataclasses.dataclass +class ConfigProbeConfig: + gain: float = 0.0 + label: str = "" + + @processor(execution="manual") class ConfigProbe: - def __init__(self, gain: float = 0.0, label: str = "") -> None: - self.gain = gain - self.label = label + def __init__(self, config: ConfigProbeConfig) -> None: + self.config = config def setup(self, ctx: RuntimeContextFullAccess) -> None: - _report(lambda: {"config": ctx.config}) + # Both halves of the contract from inside the helper: the object the + # helper constructed, and `ctx.config` still being the raw mapping. + _report( + lambda: { + "config": ctx.config, + "constructed": dataclasses.asdict(self.config), + "constructed_type": type(self.config).__name__, + } + ) @processor(execution="manual") diff --git a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py index dcec81bad..b22a444f9 100644 --- a/sdk/streamlib-python-wheel/tests/helper_placement_processors.py +++ b/sdk/streamlib-python-wheel/tests/helper_placement_processors.py @@ -9,17 +9,23 @@ instance, which is the shape the ban forbids. """ +import dataclasses import os from streamlib import input, log, output, processor +@dataclasses.dataclass +class ReportsItsOwnProcessSourceConfig: + label: str = "unlabelled" + + @processor(execution="continuous", interval_ms=10) class ReportsItsOwnProcessSource: """Stamps every bag with the pid it was produced in.""" - def __init__(self, label: str = "unlabelled") -> None: - self.label = label + def __init__(self, config: ReportsItsOwnProcessSourceConfig) -> None: + self.label = config.label self.announced = False @output() diff --git a/sdk/streamlib-python-wheel/tests/helper_process_probes.py b/sdk/streamlib-python-wheel/tests/helper_process_probes.py index dd2f409a0..2d5349e98 100644 --- a/sdk/streamlib-python-wheel/tests/helper_process_probes.py +++ b/sdk/streamlib-python-wheel/tests/helper_process_probes.py @@ -10,16 +10,23 @@ """ import time +from typing import TypedDict from streamlib import input, output, processor +class PassThroughProbeConfig(TypedDict, total=False): + """A TypedDict config, so a real helper run covers that kind too.""" + + tag: str + + @processor class PassThroughProbe: """Copies every bag from its input to its output.""" - def __init__(self, tag: str = "untagged") -> None: - self.tag = tag + def __init__(self, config: PassThroughProbeConfig) -> None: + self.tag = config.get("tag", "untagged") @input(delivery_profile="newest") def frames_from_upstream(self) -> None: ... diff --git a/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py b/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py new file mode 100644 index 000000000..0af4ba084 --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/processor_config_catalog_app.py @@ -0,0 +1,69 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""The processor catalog a running node serves, read over its own control plane. + +Run as a real `python app.py`: four processors are added, each hosted in its own +helper process, and this app then reads `GET /api/registry` off itself — the +exact payload an agent gets before deciding which keys a processor takes. +""" + +import json +import os +import sys +import threading +import urllib.request + +import streamlib +from streamlib._node_registry import live_nodes + +import processor_config_catalog_probes as probes + +READ_TIMEOUT_SECONDS = 30.0 +GRAPH_READY_TIMEOUT_SECONDS = 90.0 + + +def _this_processes_control_url() -> str: + return next(node.control_url for node in live_nodes() if node.pid == os.getpid()) + + +def main() -> None: + runtime = streamlib.Runtime() + runtime.host_control_plane() + runtime.add(probes.TypedDictConfiguredProbe, config={"width": 320}) + runtime.add(probes.DataclassConfiguredProbe, config={"width": 640, "label": "left"}) + runtime.add(probes.ModelConfiguredProbe, config={"width": 1280}) + runtime.add(probes.UnconfiguredProbe) + + def read_the_catalog_this_node_serves() -> None: + try: + runtime.wait_until_every_processor_is_running( + timeout=GRAPH_READY_TIMEOUT_SECONDS + ) + registry_url = f"{_this_processes_control_url()}/api/registry" + # A loopback URL this app minted, read back off itself. + with urllib.request.urlopen(registry_url, timeout=READ_TIMEOUT_SECONDS) as response: + served = json.load(response) + catalog = { + entry["processor_class_import_path"]: entry.get("config_schema") + for entry in served["processors"] + if entry["processor_class_import_path"].startswith( + "processor_config_catalog_probes:" + ) + } + print(f"MARKER:CATALOG {json.dumps(catalog)}", flush=True) + except Exception as read_failure: + # Said rather than swallowed: this runs on a daemon thread, where an + # unhandled raise leaves the test waiting on a marker that never comes. + print(f"MARKER:CATALOG_FAILED {read_failure!r}", flush=True) + finally: + runtime.shutdown() + + threading.Thread(target=read_the_catalog_this_node_serves, daemon=True).start() + runtime.run() + print("MARKER:CLEAN_EXIT", flush=True) + + +if __name__ == "__main__": + main() + sys.exit(0) diff --git a/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py b/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py new file mode 100644 index 000000000..1177043ad --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/processor_config_catalog_probes.py @@ -0,0 +1,75 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""Four processors — three kinds of config class and one declaring none. + +Their own module, not the test module, because each runs in a helper process +that reaches its class by importing the module it was declared in. + +Each reports, from inside its helper, the object its `__init__` was handed — +which is the half of the contract a served schema cannot show. +""" + +import dataclasses +import json +from typing import Annotated, Optional, TypedDict + +import pydantic + +from streamlib import RuntimeContextFullAccess, log, processor + + +def _report(processor_name: str, config: object) -> None: + log.info( + f"MARKER:CONSTRUCTED {json.dumps({'processor': processor_name, 'config_type': type(config).__name__})}" + ) + + +class TypedDictProbeConfig(TypedDict, total=False): + width: Annotated[int, "How wide the probe pretends its frames are."] + + +@dataclasses.dataclass +class DataclassProbeConfig: + width: Annotated[int, "How wide the probe pretends its frames are."] = 640 + label: Annotated[str, "What to call this probe."] = "unlabelled" + # A null default has to survive the msgpack hop the document takes into + # Rust, which is the one value on this class that could be dropped there. + fallback: Annotated[Optional[str], "Where the probe falls back to."] = None + + +class ModelProbeConfig(pydantic.BaseModel): + width: int = 1280 + + +@processor(execution="manual", description="Configured by a TypedDict") +class TypedDictConfiguredProbe: + def __init__(self, config: TypedDictProbeConfig) -> None: + self.config = config + + def setup(self, ctx: RuntimeContextFullAccess) -> None: + _report("TypedDictConfiguredProbe", self.config) + + +@processor(execution="manual", description="Configured by a dataclass") +class DataclassConfiguredProbe: + def __init__(self, config: DataclassProbeConfig) -> None: + self.config = config + + def setup(self, ctx: RuntimeContextFullAccess) -> None: + _report("DataclassConfiguredProbe", self.config) + + +@processor(execution="manual", description="Configured by a model") +class ModelConfiguredProbe: + def __init__(self, config: ModelProbeConfig) -> None: + self.config = config + + def setup(self, ctx: RuntimeContextFullAccess) -> None: + _report("ModelConfiguredProbe", self.config) + + +@processor(execution="manual", description="Takes no configuration at all") +class UnconfiguredProbe: + def setup(self, ctx: RuntimeContextFullAccess) -> None: + _report("UnconfiguredProbe", None) diff --git a/sdk/streamlib-python-wheel/tests/single_processor_under_test.py b/sdk/streamlib-python-wheel/tests/single_processor_under_test.py index 4bdcd3e7e..da9775732 100644 --- a/sdk/streamlib-python-wheel/tests/single_processor_under_test.py +++ b/sdk/streamlib-python-wheel/tests/single_processor_under_test.py @@ -9,6 +9,8 @@ class declared inside a pytest module would have the child import the test suite. """ +import dataclasses + import numpy from streamlib import AudioBlock, RuntimeContextLimitedAccess, input, output, processor @@ -31,12 +33,17 @@ def process(self, ctx: RuntimeContextLimitedAccess) -> None: ctx.outputs.write("numbers_to_downstream", {"value": bag["value"] * 2}) +@dataclasses.dataclass +class ConfiguredScalerConfig: + factor: int = 1 + + @processor class ConfiguredScaler: """Reads its factor from config, so the harness's `config=` is exercised.""" - def __init__(self, factor: int = 1) -> None: - self.factor = factor + def __init__(self, config: ConfiguredScalerConfig) -> None: + self.factor = config.factor @input(delivery_profile="ordered") def numbers_from_upstream(self) -> None: ... diff --git a/sdk/streamlib-python-wheel/tests/test_capability_contexts.py b/sdk/streamlib-python-wheel/tests/test_capability_contexts.py index a23bf730f..0dcecd5ba 100644 --- a/sdk/streamlib-python-wheel/tests/test_capability_contexts.py +++ b/sdk/streamlib-python-wheel/tests/test_capability_contexts.py @@ -84,6 +84,10 @@ def test_process_receives_the_limited_context_without_gpu_full_access( def test_ctx_config_is_the_dict_the_processor_was_added_with(start_app_under_test): observation = run_probe(start_app_under_test, "configured_probe") assert observation["config"] == {"gain": 2.5, "label": "left"} + # The helper built the config class out of that mapping and handed the + # object to `__init__`; `ctx.config` above is still the mapping itself. + assert observation["constructed"] == {"gain": 2.5, "label": "left"} + assert observation["constructed_type"] == "ConfigProbeConfig" def test_ctx_config_is_an_empty_dict_when_nothing_was_passed(start_app_under_test): diff --git a/sdk/streamlib-python-wheel/tests/test_live_graph_mutation.py b/sdk/streamlib-python-wheel/tests/test_live_graph_mutation.py index 11053b226..32e2bb658 100644 --- a/sdk/streamlib-python-wheel/tests/test_live_graph_mutation.py +++ b/sdk/streamlib-python-wheel/tests/test_live_graph_mutation.py @@ -59,6 +59,8 @@ def setup(rt: Runtime) -> None: LIVE_ADDED_EFFECT_SOURCE = '''\ """An effect that announces its frames, written after the node started.""" +import dataclasses + from streamlib import ( # noqa: A004 — `input` is streamlib's port decorator ProcessorOutputTextureRing, RuntimeContextFullAccess, @@ -71,12 +73,17 @@ def setup(rt: Runtime) -> None: ) +@dataclasses.dataclass +class LiveAddedEffectConfig: + marker: str = "LIVE_FRAME" + + @processor class LiveAddedEffect: """Republishes each frame on a texture of its own and counts them.""" - def __init__(self, marker: str = "LIVE_FRAME") -> None: - self.marker = marker + def __init__(self, config: LiveAddedEffectConfig) -> None: + self.marker = config.marker self.frames = 0 @input(delivery_profile="newest") diff --git a/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py b/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py new file mode 100644 index 000000000..f2eda8b05 --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/test_processor_config_catalog.py @@ -0,0 +1,141 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""What a running node tells an agent about a processor's config, end to end. + +The declaration suite proves the document is derived, and the wheel's Rust tests +prove it reaches the shape `/api/registry` serializes. This is the whole path on +a real node with each processor in its own helper process — which needs a +running graph, and a running graph initializes a GPU context, so it runs on the +rig like every other live proof in this suite. +""" + +import json +import re +from pathlib import Path +from typing import Any, Iterator + +import pytest + +from app_under_test import start_app + +pytestmark = pytest.mark.requires_gpu + +APP = Path(__file__).parent / "processor_config_catalog_app.py" + +CATALOG = re.compile(r"MARKER:CATALOG (\{.*\})\s*$", re.MULTILINE) +CONSTRUCTED = re.compile(r"MARKER:CONSTRUCTED (\{.*?\})") + + +@pytest.fixture(scope="module") +def catalog_app_output() -> "Iterator[str]": + """One node, one run: four helper spawns are the cost, so they are paid once. + + Module-scoped, so it reaps its own process group rather than reaching for + the function-scoped fixture every other app suite uses. Leaving that to a + failed assertion would strand a live engine holding a socket and an + iceoryx2 node. + """ + app = start_app(APP) + try: + app.await_output_containing("MARKER:CATALOG", "the served catalog") + app.await_marker("CLEAN_EXIT") + app.await_clean_exit() + yield app.output + finally: + app.kill_process_group() + + +@pytest.fixture(scope="module") +def served_catalog(catalog_app_output: str) -> "dict[str, Any]": + match = CATALOG.search(catalog_app_output) + assert match is not None, f"no catalog line:\n{catalog_app_output}" + return json.loads(match.group(1)) + + +def schema_for(served_catalog: "dict[str, Any]", probe: str) -> "dict[str, Any]": + document = served_catalog[f"processor_config_catalog_probes:{probe}"] + assert document is not None, f"{probe} served a null config schema" + return document + + +def test_a_dataclass_config_reaches_the_registry_with_types_defaults_and_descriptions( + served_catalog, +): + document = schema_for(served_catalog, "DataclassConfiguredProbe") + + assert document["properties"]["width"] == { + "type": "integer", + "description": "How wide the probe pretends its frames are.", + "default": 640, + } + assert document["properties"]["label"]["default"] == "unlabelled" + assert document["additionalProperties"] is False + + +def test_a_null_default_survives_the_hop_into_the_descriptor(served_catalog): + """The document crosses into Rust through the msgpack value tree the data + plane uses, where `None` is the one value that could arrive as absent.""" + fallback = schema_for(served_catalog, "DataclassConfiguredProbe")["properties"][ + "fallback" + ] + + assert fallback["anyOf"] == [{"type": "string"}, {"type": "null"}] + assert "default" in fallback, f"the null default was dropped: {fallback}" + assert fallback["default"] is None + + +def test_a_typed_dict_config_reaches_the_registry(served_catalog): + document = schema_for(served_catalog, "TypedDictConfiguredProbe") + + assert document["properties"]["width"]["type"] == "integer" + assert document["properties"]["width"]["description"] + # `total=False`, so nothing is required and the class admits an unknown key. + assert "required" not in document + assert "additionalProperties" not in document + + +def test_a_model_config_reaches_the_registry_as_the_model_describes_itself( + served_catalog, +): + document = schema_for(served_catalog, "ModelConfiguredProbe") + + assert document["properties"]["width"] == { + "default": 1280, + "title": "Width", + "type": "integer", + } + assert "$schema" not in document + assert "title" not in document, "the catalog entry names the processor already" + + +def test_a_processor_declaring_no_config_serves_an_empty_object_not_a_null( + served_catalog, +): + """A null would read as "this node does not know", which is a different + claim from "this processor takes nothing".""" + assert schema_for(served_catalog, "UnconfiguredProbe") == { + "type": "object", + "description": "This processor declares no configuration.", + "additionalProperties": False, + } + + +def test_every_helper_constructed_the_config_class_its_processor_named( + catalog_app_output, +): + """The half a served document cannot show: the object really arrived in the + child, built from the mapping `rt.add` recorded.""" + constructed = { + report["processor"]: report["config_type"] + for report in ( + json.loads(match.group(1)) for match in CONSTRUCTED.finditer(catalog_app_output) + ) + } + + assert constructed == { + "TypedDictConfiguredProbe": "dict", + "DataclassConfiguredProbe": "DataclassProbeConfig", + "ModelConfiguredProbe": "ModelProbeConfig", + "UnconfiguredProbe": "NoneType", + } diff --git a/sdk/streamlib-python-wheel/tests/test_processor_config_class.py b/sdk/streamlib-python-wheel/tests/test_processor_config_class.py new file mode 100644 index 000000000..2c205a215 --- /dev/null +++ b/sdk/streamlib-python-wheel/tests/test_processor_config_class.py @@ -0,0 +1,794 @@ +# Copyright (c) 2025 Jonathan Fontanez +# SPDX-License-Identifier: BUSL-1.1 + +"""A processor's config class: how it is declared, derived and constructed. + +Three seams, none of which boots an engine. `@processor` reads the config class +off `__init__` and refuses every other signature; the deriver turns that class +into the JSON Schema the catalog publishes; and the hosting module constructs +the class from the mapping `rt.add` recorded, which is what a helper process +does on the compile thread. +""" + +import dataclasses +from typing import Annotated, Any, Literal, Optional, TypedDict + +import pydantic +import pytest + +# The 3.10 floor spells per-key requiredness from here, and it is what the +# wheel's own test environment installs; `typing.Required` arrives only at 3.11. +from typing_extensions import NotRequired, Required + +from streamlib import processor +from streamlib._processor_hosting import ( + apply_configuration, + construct_processor_instance, +) + +# --------------------------------------------------------------------------- +# The config classes the cases below are declared against +# --------------------------------------------------------------------------- + + +class BlurConfigTypedDict(TypedDict): + """Admits anything at run time — the loosest dial an author can pick.""" + + width: int + label: Annotated[str, "What to call this blur."] + + +class PartialConfigTypedDict(TypedDict, total=False): + width: int + + +@dataclasses.dataclass +class BlurConfigDataclass: + """Refuses an unknown key but not a mistyped value.""" + + width: int + label: str = "unlabelled" + tags: "list[str]" = dataclasses.field(default_factory=list) + quality: Literal["fast", "good"] = "fast" + fallback: Optional[str] = None + derived: int = dataclasses.field(init=False, default=7) + + +class BlurConfigModel(pydantic.BaseModel): + """Validates values — the strictest dial, and it carries its own schema.""" + + width: int + label: str = "unlabelled" + + +@dataclasses.dataclass +class NestedConfig: + inner: BlurConfigDataclass + count: int = 0 + + +# At module scope, not inside their tests: `typing.get_type_hints` resolves a +# class's forward references against its module's globals and never against an +# enclosing function's locals, so a self-reference written in a test body cannot +# resolve at all. +@dataclasses.dataclass +class TreeConfig: + child: "Optional[TreeConfig]" = None + depth: int = 0 + + +@dataclasses.dataclass +class LeftConfig: + right: "Optional[RightConfig]" = None + + +@dataclasses.dataclass +class RightConfig: + left: "Optional[LeftConfig]" = None + + +# --------------------------------------------------------------------------- +# Declaration: which `__init__` signatures name a config class +# --------------------------------------------------------------------------- + + +def test_an_annotated_config_parameter_names_the_config_class(): + @processor(execution="manual") + class Blur: + def __init__(self, config: BlurConfigDataclass) -> None: + self.config = config + + assert Blur.__streamlib_processor_config_class__ is BlurConfigDataclass + + +def test_an_init_taking_nothing_beyond_self_declares_no_config(): + @processor(execution="manual") + class Counter: + def __init__(self) -> None: + self.seen = 0 + + assert Counter.__streamlib_processor_config_class__ is None + + +def test_a_class_defining_no_init_at_all_declares_no_config(): + """`object.__init__` reports `(self, /, *args, **kwargs)`. + + Read literally that is a variadic signature, which the rule below refuses — + so the commonest class in the suite would be refused for a signature its + author never wrote. + """ + + @processor(execution="manual") + class Bare: + pass + + assert Bare.__streamlib_processor_config_class__ is None + + +def test_a_keyword_parameter_is_refused_with_the_fix_named(): + with pytest.raises(TypeError) as refusal: + + @processor(execution="manual") + class Blur: + def __init__(self, width: int = 1) -> None: + self.width = width + + assert "Blur" in str(refusal.value) + assert "`width`" in str(refusal.value) + assert "must be named `config`" in str(refusal.value) + assert "BlurConfig" in str(refusal.value), "the fix must show the shape wanted" + + +def test_several_parameters_are_refused_and_all_of_them_named(): + with pytest.raises(TypeError, match="width, height"): + + @processor(execution="manual") + class Blur: + def __init__(self, width: int, height: int) -> None: + self.size = (width, height) + + +def test_an_unannotated_config_parameter_is_refused(): + with pytest.raises(TypeError, match="with no annotation"): + + @processor(execution="manual") + class Blur: + def __init__(self, config) -> None: # noqa: ANN001 + self.config = config + + +def test_a_config_annotated_as_a_parameterized_generic_is_refused(): + """`dict[str, Any]` is a mapping, not a class the helper can construct. + + The guard is on the annotation's origin rather than on `isinstance(_, type)` + because on Python 3.10 — the wheel's floor — `isinstance(dict[str, Any], + type)` is still True. + """ + with pytest.raises(TypeError, match="is not a class"): + + @processor(execution="manual") + class Blur: + def __init__(self, config: "dict[str, Any]") -> None: + self.config = config + + +def test_keyword_variadic_configuration_is_refused_by_name(): + with pytest.raises(TypeError, match=r"\*\*config"): + + @processor(execution="manual") + class Blur: + def __init__(self, **config: Any) -> None: + self.config = config + + +def test_a_positional_only_config_is_refused_because_the_helper_passes_it_by_name(): + with pytest.raises(TypeError, match="positionally only"): + + @processor(execution="manual") + class Blur: + def __init__(self, config: BlurConfigDataclass, /) -> None: + self.config = config + + +def test_an_unresolvable_annotation_is_refused_where_the_author_can_see_it(): + with pytest.raises(TypeError, match="cannot be resolved"): + + @processor(execution="manual") + class Blur: + def __init__( + self, + config: "NeverImported", # noqa: F821 # pyright: ignore[reportUndefinedVariable] + ) -> None: + self.config = config + + +# --------------------------------------------------------------------------- +# Derivation: the document each kind of config class publishes +# --------------------------------------------------------------------------- + + +def schema_of(config_class: "Optional[type]") -> "dict[str, Any]": + """The document a processor taking `config_class` publishes.""" + if config_class is None: + + @processor(execution="manual") + class SubjectDeclaringNoConfig: + def __init__(self) -> None: + self.seen = 0 + + return SubjectDeclaringNoConfig.__streamlib_processor_config_schema__ + + @processor(execution="manual") + class SubjectTakingAConfigClass: + def __init__(self, config: config_class) -> None: # type: ignore[valid-type] + self.config = config + + return SubjectTakingAConfigClass.__streamlib_processor_config_schema__ + + +def test_a_typed_dict_yields_its_annotations_and_its_required_keys(): + document = schema_of(BlurConfigTypedDict) + + assert document["type"] == "object" + assert document["properties"]["width"] == {"type": "integer"} + assert document["properties"]["label"] == { + "type": "string", + "description": "What to call this blur.", + } + assert document["required"] == ["width", "label"] + # A TypedDict admits an unknown key at run time, so claiming otherwise + # would be a promise the class does not keep. + assert "additionalProperties" not in document + + +def test_a_total_false_typed_dict_requires_nothing(): + assert "required" not in schema_of(PartialConfigTypedDict) + + +def test_a_dataclass_yields_its_init_fields_defaults_and_required_names(): + document = schema_of(BlurConfigDataclass) + + assert document["properties"]["width"] == {"type": "integer"} + assert document["properties"]["label"] == {"type": "string", "default": "unlabelled"} + assert document["properties"]["quality"] == { + "enum": ["fast", "good"], + "default": "fast", + } + assert document["properties"]["fallback"] == { + "anyOf": [{"type": "string"}, {"type": "null"}], + "default": None, + } + assert document["required"] == ["width"] + # A dataclass raises on an unknown key, so the document may say so. + assert document["additionalProperties"] is False + + +def test_a_field_with_a_default_factory_is_optional_and_carries_no_default(): + """A factory's result is not a default — calling one to document it would + run the author's code at import.""" + tags = schema_of(BlurConfigDataclass)["properties"]["tags"] + + assert tags == {"type": "array", "items": {"type": "string"}} + assert "tags" not in schema_of(BlurConfigDataclass)["required"] + + +def test_an_init_false_field_is_not_documented_because_it_is_not_an_input(): + assert "derived" not in schema_of(BlurConfigDataclass)["properties"] + + +def test_a_model_contributes_its_own_document_without_the_two_catalog_keys(): + document = schema_of(BlurConfigModel) + + assert document["properties"]["width"]["type"] == "integer" + assert document["properties"]["label"]["default"] == "unlabelled" + assert document["required"] == ["width"] + assert "$schema" not in document + assert "title" not in document, "the catalog names a processor, not its config type" + + +def test_a_nested_config_class_is_inlined_rather_than_referenced(): + """Nothing here emits a `$ref`, so no document carries a `$defs` for one to + point into.""" + document = schema_of(NestedConfig) + + assert document["properties"]["inner"]["properties"]["width"] == {"type": "integer"} + assert "$defs" not in document + assert "$ref" not in repr(document) + + +def test_an_annotation_the_deriver_does_not_know_renders_as_an_open_schema(): + """A config class is worth publishing long before every type in it is + describable.""" + + @dataclasses.dataclass + class WithAnOpaqueField: + handle: complex + width: int = 2 + + document = schema_of(WithAnOpaqueField) + + assert document["properties"]["handle"] == {} + assert document["properties"]["width"] == {"type": "integer", "default": 2} + assert document["required"] == ["handle"], "an opaque field is still an input" + + +def test_a_self_referential_config_class_stops_rather_than_exhausting_the_stack(): + """Inlining is the only nesting the deriver emits, so a cycle has no fixed + point. Unrecognised, the walk runs out of stack at decoration — which is + import time, where the traceback names typing internals and not the class.""" + document = schema_of(TreeConfig) + + assert document["properties"]["child"]["anyOf"] == [ + {"type": "object"}, + {"type": "null"}, + ] + assert document["properties"]["depth"] == {"type": "integer", "default": 0} + + +def test_two_config_classes_that_reach_each_other_stop_at_the_second_pass(): + left = schema_of(LeftConfig)["properties"]["right"]["anyOf"][0] + + assert left["properties"]["left"]["anyOf"] == [{"type": "object"}, {"type": "null"}] + + +def test_the_same_class_nested_twice_without_a_cycle_is_inlined_both_times(): + """The guard is on an ancestry, not on a visited set: a diamond is not a + cycle and must not be truncated.""" + + @dataclasses.dataclass + class LeafConfig: + width: int = 1 + + @dataclasses.dataclass + class BranchConfig: + first: LeafConfig = dataclasses.field(default_factory=LeafConfig) + second: LeafConfig = dataclasses.field(default_factory=LeafConfig) + + document = schema_of(BranchConfig) + + assert document["properties"]["first"]["properties"]["width"]["type"] == "integer" + assert document["properties"]["second"]["properties"]["width"]["type"] == "integer" + + +def test_a_frozen_slotted_dataclass_derives_like_any_other(): + @dataclasses.dataclass(frozen=True) + class FrozenConfig: + width: int = 1 + + assert schema_of(FrozenConfig)["properties"]["width"] == { + "type": "integer", + "default": 1, + } + + +def test_a_typed_dict_inheriting_another_carries_both_key_sets(): + class BaseConfig(TypedDict): + width: int + + class DerivedConfig(BaseConfig, total=False): + label: str + + document = schema_of(DerivedConfig) + + assert set(document["properties"]) == {"width", "label"} + assert document["required"] == ["width"] + + +def test_a_processor_declaring_no_config_publishes_what_rust_publishes(): + """One catalog reads one way whichever language declared the processor.""" + assert schema_of(None) == { + "type": "object", + "description": "This processor declares no configuration.", + "additionalProperties": False, + } + + +def test_the_document_is_2020_12_with_no_meta_schema_key(): + for config_class in (BlurConfigTypedDict, BlurConfigDataclass, BlurConfigModel): + assert "$schema" not in schema_of(config_class), config_class + + +# --------------------------------------------------------------------------- +# Hosting: constructing the class from the mapping, and reconfiguring +# --------------------------------------------------------------------------- + + +@processor(execution="manual") +class DataclassConfigured: + def __init__(self, config: BlurConfigDataclass) -> None: + self.config = config + + def configure(self, config: BlurConfigDataclass) -> None: + self.config = config + + +@processor(execution="manual") +class TypedDictConfigured: + def __init__(self, config: BlurConfigTypedDict) -> None: + self.config = config + + +@processor(execution="manual") +class ModelConfigured: + def __init__(self, config: BlurConfigModel) -> None: + self.config = config + + +@processor(execution="manual") +class Unconfigured: + def __init__(self) -> None: + self.seen = 0 + + +def test_a_dataclass_config_reaches_the_processor_as_an_object(): + built = construct_processor_instance( + DataclassConfigured, {"width": 4, "label": "left"}, None + ) + + assert built.config == BlurConfigDataclass(width=4, label="left") + + +def test_a_typed_dict_config_reaches_the_processor_as_the_mapping_itself(): + built = construct_processor_instance(TypedDictConfigured, {"width": 4}, None) + + assert built.config == {"width": 4} + + +def test_a_model_config_reaches_the_processor_validated(): + built = construct_processor_instance(ModelConfigured, {"width": "4"}, None) + + assert built.config.width == 4, "the model coerced it; the wheel added no opinion" + + +def test_whatever_the_config_class_raises_is_what_the_author_sees(): + """Construction is the only check the wheel performs: how strict it is is + the author's choice of config class, the same dial `read(port, into=T)` is.""" + with pytest.raises(TypeError, match="bogus"): + construct_processor_instance(DataclassConfigured, {"bogus": 1}, None) + + with pytest.raises(pydantic.ValidationError): + construct_processor_instance(ModelConfigured, {"width": "wide"}, None) + + +def test_a_processor_declaring_no_config_refuses_a_non_empty_one_by_name(): + with pytest.raises(TypeError, match="`width` has nowhere to go"): + construct_processor_instance(Unconfigured, {"width": 1}, None) + + +def test_a_processor_declaring_no_config_takes_an_empty_one(): + assert construct_processor_instance(Unconfigured, {}, None).seen == 0 + assert construct_processor_instance(Unconfigured, None, None).seen == 0 + + +def test_reconfiguration_hands_configure_the_same_kind_of_object(): + built = construct_processor_instance(DataclassConfigured, {"width": 4}, None) + + apply_configuration(built, {"width": 9, "label": "right"}) + + assert built.config == BlurConfigDataclass(width=9, label="right") + + +def test_a_processor_without_configure_is_refused_by_the_hook_it_needs(): + built = construct_processor_instance(TypedDictConfigured, {"width": 4}, None) + + with pytest.raises(TypeError, match=r"configure\(self, config\)"): + apply_configuration(built, {"width": 9}) + + +def test_a_configuration_that_is_not_a_mapping_is_refused_before_construction(): + with pytest.raises(TypeError, match="must be a dict"): + construct_processor_instance(DataclassConfigured, ["width", 4], None) + + +def test_a_typing_extensions_typed_dict_is_recognised_as_one(): + """`typing.is_typeddict` sees `typing.TypedDict` alone. + + On the 3.10 floor `Required` / `NotRequired` come from `typing_extensions`, + so its spelling is the one an author reaches for — and unrecognised it + would fall through to an open object with no keys and no refusal to say so. + """ + typing_extensions = pytest.importorskip("typing_extensions") + + class ExtensionSpelledConfig(typing_extensions.TypedDict): # pyright: ignore[reportGeneralTypeIssues] + width: int + + document = schema_of(ExtensionSpelledConfig) + + assert document["properties"]["width"] == {"type": "integer"} + assert document["required"] == ["width"] + + +def test_a_config_annotated_as_any_is_refused_the_same_on_every_version(): + """`isinstance(typing.Any, type)` is False on 3.10 and True on 3.11+, so + without naming `Any` the rule would differ across the wheel's own range.""" + with pytest.raises(TypeError, match="`Any`"): + + @processor(execution="manual") + class Blur: + def __init__(self, config: Any) -> None: + self.config = config + + +# --------------------------------------------------------------------------- +# The migrated fixtures, guarded where CI can see them +# --------------------------------------------------------------------------- + +# Every other test that runs these five is `requires_gpu` and so runs on the rig +# alone. Decoration is where a bad migration raises, so importing them here is +# what puts the migration in front of CI at all. +MIGRATED_FIXTURES = [ + ("capability_context_probes", "ConfigProbe", "ConfigProbeConfig"), + ("helper_placement_processors", "ReportsItsOwnProcessSource", "ReportsItsOwnProcessSourceConfig"), + ("helper_process_probes", "PassThroughProbe", "PassThroughProbeConfig"), + ("single_processor_under_test", "ConfiguredScaler", "ConfiguredScalerConfig"), + ("texture_ring_producer_probes", "TextureRingPublishingVideoSource", "TextureRingPublishingVideoSourceConfig"), +] + + +@pytest.mark.parametrize( + ("module_name", "processor_name", "config_name"), + MIGRATED_FIXTURES, + ids=[processor_name for _, processor_name, _ in MIGRATED_FIXTURES], +) +def test_a_migrated_fixture_declares_the_config_class_beside_it( + module_name, processor_name, config_name +): + module = __import__(module_name) + processor_class = getattr(module, processor_name) + + assert processor_class.__streamlib_processor_config_class__ is getattr( + module, config_name + ) + + +def test_the_live_mutation_fixture_written_as_a_source_string_still_declares(): + """`LiveAddedEffect` lives as a triple-quoted literal, so no import, no + linter and no AST sweep reaches it — running it here is the only way a bad + migration of it fails anywhere but on the rig.""" + from test_live_graph_mutation import LIVE_ADDED_EFFECT_SOURCE + + namespace: "dict[str, Any]" = {"__name__": "processors.live_added_effect"} + exec(compile(LIVE_ADDED_EFFECT_SOURCE, "live_added_effect.py", "exec"), namespace) + + effect = namespace["LiveAddedEffect"] + assert effect.__streamlib_processor_config_class__ is namespace["LiveAddedEffectConfig"] + assert effect.__streamlib_processor_config_schema__["properties"]["marker"] == { + "type": "string", + "default": "LIVE_FRAME", + } + + +# --------------------------------------------------------------------------- +# Shapes the deriver has to describe rather than drop +# --------------------------------------------------------------------------- + + +def test_a_requiredness_qualifier_does_not_hide_the_type_it_wraps(): + """`get_type_hints` keeps `Required` / `NotRequired`, and unrecognised they + swallow the key's type — the one thing the catalog exists to publish.""" + + class QualifiedConfig(TypedDict, total=False): + width: Required[Annotated[int, "How wide."]] + label: NotRequired[str] + + document = schema_of(QualifiedConfig) + + assert document["properties"]["width"] == {"type": "integer", "description": "How wide."} + assert document["properties"]["label"] == {"type": "string"} + assert document["required"] == ["width"] + + +def test_an_init_var_is_documented_because_it_is_a_constructor_input(): + """`dataclasses.fields()` omits an InitVar. Left out, it is absent from + `properties` while `additionalProperties: false` forbids it — a catalog + telling an agent that a required key is illegal.""" + + @dataclasses.dataclass + class SeededConfig: + width: int + seed: dataclasses.InitVar[int] = 3 + + def __post_init__(self, seed: int) -> None: + self.scaled_width = self.width * seed + + assert SeededConfig(width=2, seed=5).scaled_width == 10, "an InitVar is an input" + document = schema_of(SeededConfig) + + assert document["properties"]["seed"] == {"type": "integer", "default": 3} + assert "seed" not in document["required"] + assert document["additionalProperties"] is False + + +def test_a_nested_models_pointers_are_followed_rather_than_left_dangling(): + """A model writes `#/$defs/...` pointers as the root. Inlined under a + property they name a `$defs` the enclosing document does not have, so a + reader resolves them against nothing.""" + + class InnerModel(pydantic.BaseModel): + depth: int = 1 + + class OuterModel(pydantic.BaseModel): + inner: InnerModel = InnerModel() + + @dataclasses.dataclass + class HoldingAModel: + model: OuterModel + + document = schema_of(HoldingAModel) + nested = document["properties"]["model"] + + assert "$ref" not in repr(nested), f"a pointer survived into a nested document: {nested}" + assert "$defs" not in nested + assert nested["properties"]["inner"]["properties"]["depth"]["type"] == "integer" + + +def test_a_root_model_keeps_the_defs_it_wrote_for_itself(): + """As the root its pointers resolve, so its document is taken verbatim.""" + + class InnerModel(pydantic.BaseModel): + depth: int = 1 + + class RootModel(pydantic.BaseModel): + inner: InnerModel = InnerModel() + + document = schema_of(RootModel) + + assert document["properties"]["inner"]["$ref"] == "#/$defs/InnerModel" + assert document["$defs"]["InnerModel"]["properties"]["depth"]["type"] == "integer" + + +@pytest.mark.parametrize( + ("annotation", "default", "why"), + [ + (float, float("inf"), "JSON has no infinity"), + (float, float("nan"), "JSON has no NaN"), + (int, 2**64, "msgpack carries no integer wider than 64 bits"), + ], +) +def test_a_default_the_wire_cannot_carry_is_dropped_not_rewritten( + annotation, default, why +): + """The document crosses into Rust through the msgpack value tree, which + turns a non-finite float into a null and refuses a wider integer outright — + losing the whole declaration over one default the author can live without.""" + OddlyDefaultedConfig = dataclasses.make_dataclass( + "OddlyDefaultedConfig", + [("setting", annotation, dataclasses.field(default=default))], + ) + + document = schema_of(OddlyDefaultedConfig) + + assert "default" not in document["properties"]["setting"], why + assert "setting" not in document.get("required", []), "it still has a default" + + +# --------------------------------------------------------------------------- +# The rest of the construction contract +# --------------------------------------------------------------------------- + + +def test_the_helper_constructs_the_processor_by_the_config_keyword(): + """Positionally would work for every class in this suite and break the + moment an author writes a keyword-only `config`.""" + constructed_with: "dict[str, Any]" = {} + + @processor(execution="manual") + class KeywordOnlyConfigured: + def __init__(self, *, config: BlurConfigDataclass) -> None: + constructed_with["config"] = config + + construct_processor_instance(KeywordOnlyConfigured, {"width": 2}, None) + + assert constructed_with["config"] == BlurConfigDataclass(width=2) + + +def test_a_variadic_positional_config_is_refused_by_name(): + with pytest.raises(TypeError, match=r"\*config"): + + @processor(execution="manual") + class Blur: + def __init__(self, *config: Any) -> None: + self.config = config + + +def test_reconfiguring_a_processor_that_declares_no_config_refuses_the_keys(): + @processor(execution="manual") + class UnconfiguredButReconfigurable: + def __init__(self) -> None: + self.configured_with: Any = "never" + + def configure(self, config: None) -> None: + self.configured_with = config + + built = construct_processor_instance(UnconfiguredButReconfigurable, {}, None) + + with pytest.raises(TypeError, match="`width` has nowhere to go"): + apply_configuration(built, {"width": 1}) + + # An empty update is not a mistake, so it reaches the hook with the nothing + # the class declared. + apply_configuration(built, {}) + assert built.configured_with is None + + +def test_a_config_class_of_a_kind_the_deriver_cannot_read_is_accepted_and_open(): + """A plain annotated class constructs fine and describes nothing. + + Pinned rather than left to drift, because it is the one shape where the + catalog goes quiet on a class that works: an agent reading this entry learns + that configuration is a mapping and nothing about its keys. Whether such a + class should instead be refused at decoration, or read off its `__init__`, + is an open question for the owner — the plan says "any class constructible + from the config's keys with annotated fields" while the change enumerates + three kinds. + """ + + class PlainlyAnnotatedConfig: + def __init__(self, width: int = 3, label: str = "x") -> None: + self.width = width + self.label = label + + @processor(execution="manual") + class PlainlyConfigured: + def __init__(self, config: PlainlyAnnotatedConfig) -> None: + self.config = config + + assert PlainlyConfigured.__streamlib_processor_config_schema__ == {"type": "object"} + built = construct_processor_instance(PlainlyConfigured, {"width": 9}, None) + assert built.config.width == 9, "it constructs; only the description is missing" + + +def test_a_dataclass_whose_constructor_takes_less_than_its_fields_documents_the_constructor(): + """`config_class(**configuration)` is what a configuration meets, so the + constructor has the final say. Documenting a key it refuses is the same lie + as omitting one it requires.""" + + @dataclasses.dataclass(init=False) + class NarrowerThanItsFieldsConfig: + width: int = 1 + label: str = "x" + + def __init__(self, width: int = 1) -> None: + self.width = width + self.label = "derived" + + document = schema_of(NarrowerThanItsFieldsConfig) + + assert set(document["properties"]) == {"width"} + with pytest.raises(TypeError, match="label"): + NarrowerThanItsFieldsConfig(width=2, label="refused") # pyright: ignore[reportCallIssue] + + +def test_a_dataclass_with_no_generated_constructor_documents_no_keys_at_all(): + @dataclasses.dataclass(init=False) + class TakesNothingConfig: + width: int = 1 + + assert schema_of(TakesNothingConfig) == { + "type": "object", + "properties": {}, + "additionalProperties": False, + } + + +def test_a_fixed_length_tuple_states_its_length_not_only_its_positions(): + """2020-12 reads `prefixItems` as what each position holds and nothing about + how many there are, so on its own it validates a shorter or longer array. + The Rust seam bounds its tuples the same way.""" + + @dataclasses.dataclass + class CroppedConfig: + crop: "tuple[int, int, int, int]" = (0, 0, 0, 0) + tail: "tuple[int, ...]" = () + + document = schema_of(CroppedConfig) + + assert document["properties"]["crop"]["minItems"] == 4 + assert document["properties"]["crop"]["maxItems"] == 4 + assert len(document["properties"]["crop"]["prefixItems"]) == 4 + # A homogeneous tuple has no length to state. + assert document["properties"]["tail"] == { + "type": "array", + "items": {"type": "integer"}, + "default": [], + } diff --git a/sdk/streamlib-python-wheel/tests/texture_ring_producer_probes.py b/sdk/streamlib-python-wheel/tests/texture_ring_producer_probes.py index e4858f515..ebd6089a5 100644 --- a/sdk/streamlib-python-wheel/tests/texture_ring_producer_probes.py +++ b/sdk/streamlib-python-wheel/tests/texture_ring_producer_probes.py @@ -12,6 +12,7 @@ the other probes use, tagged with `probe` because a scenario runs two of them. """ +import dataclasses import json import os import traceback @@ -60,6 +61,11 @@ def pixel_value_of_frame(frame_index: int) -> int: return 10 + frame_index +@dataclasses.dataclass +class TextureRingPublishingVideoSourceConfig: + frames_to_publish: int = RING_DEPTH + + @processor(execution="continuous", interval_ms=10) class TextureRingPublishingVideoSource: """Publishes frames from its own output ring, one slot per frame.""" @@ -67,11 +73,11 @@ class TextureRingPublishingVideoSource: @output() def frames_to_downstream(self) -> None: ... - def __init__(self, frames_to_publish: int = RING_DEPTH) -> None: + def __init__(self, config: TextureRingPublishingVideoSourceConfig) -> None: self._output_texture_ring = ProcessorOutputTextureRing( RING_TEXTURE_FORMAT, RING_TEXTURE_USAGE, depth=RING_DEPTH ) - self._frames_to_publish = frames_to_publish + self._frames_to_publish = config.frames_to_publish self._surface_ids_published_so_far: "list[str]" = [] def process(self, ctx) -> None: