diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 775a6ff..bbdd280 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -55,7 +55,7 @@ from fluksio.flow.devices import ( presence, ) from fluksio.flow.events import event_bus -from fluksio.flow.messages import flow_of, qualify +from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify from fluksio.flow.pipeline import ValidationIssue from fluksio.flow.runs import RunRejected from fluksio.flow.schemas import ( @@ -1322,8 +1322,9 @@ def media_frames( return frames -#: What a device may say a frame is. -_IMAGE_TYPE = re.compile(r"^image/[\w.+-]+$") +#: What a device may say a frame is: a media type, parameters allowed +#: (``audio/L16;rate=16000``). +_MEDIA_TYPE = re.compile(r"^[\w.+-]+/[\w.+-]+(;[\w.+-]+=[\w.+-]+)*$") def parse_frame(data: bytes) -> tuple[dict[str, Any], bytes] | None: @@ -1353,8 +1354,8 @@ FRAME_REFUSALS = { "not_a_device": "only a device sends frames", "no_engine": "the engine is not running", "not_wired": "it names no message one of the device's ports publishes", - "not_image": "the port it names is not wired as an image", - "media_type": "its media type is not an image type", + "not_media": "the port it names is not wired as an image or audio", + "media_type": "its media type does not match its port", "too_large": "it is larger than MAX_FRAME_BYTES", "undeclared": "no flow declares its message", "refused": "its message refused the value", @@ -1370,8 +1371,8 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: The same frame over HTTP is two requests, each authenticated: the upload, then the publish. Here the socket was authenticated once, at its handshake, and the frame is held and published in one step. Only a - message one of the device's own ``in`` ports is wired to, as an ``image``, - and the value is credited to the device node holding that port, as an + message one of the device's own ``in`` ports is wired to, as an ``image`` + or an ``audio`` chunk, and the value is credited to the device node holding that port, as an HTTP publish is. The frame is named after that message. """ cap = settings.MAX_FRAME_BYTES @@ -1397,10 +1398,12 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: ) if binding is None: return "not_wired" - if binding.dtype != "image": - return "not_image" + if binding.dtype not in (DType.IMAGE, DType.AUDIO): + return "not_media" media_type = str(header.get("media_type") or "") - if not _IMAGE_TYPE.match(media_type): + if not _MEDIA_TYPE.match(media_type) or not media_type.startswith( + MEDIA_FAMILIES[DType(binding.dtype)] + ): return "media_type" reference = store.put([payload], name=name, media_type=media_type, volatile=True) try: diff --git a/backend/tests/api/routes/test_cameras.py b/backend/tests/api/routes/test_cameras.py index f027187..2f69c95 100644 --- a/backend/tests/api/routes/test_cameras.py +++ b/backend/tests/api/routes/test_cameras.py @@ -248,7 +248,7 @@ def test_a_frame_the_device_may_not_send_is_dropped( # By fixed keys, never in the words the device chose. said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message] - for key in ("not_image", "not_wired", "media_type", "not_a_frame"): + for key in ("not_media", "not_wired", "media_type", "not_a_frame"): assert f"Dropped a frame from cam1: {FRAME_REFUSALS[key]}" in said assert any(FRAME_REFUSALS["not_a_device"] in line for line in said) assert not any("elsewhere" in line or "text/html" in line for line in said) @@ -268,6 +268,35 @@ def test_an_unpaired_device_is_closed_and_its_frames_stop( assert closed.value.code == 1008 +def test_a_device_streams_audio_chunks_on_its_socket( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A microphone's chunks take the camera's path, in their own family.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "mic1"}]}) + device = _pair(client, superuser_token_headers, "mic1") + port = {"port": "mic", "direction": "in", "dtype": "audio"} + reported = client.put( + f"{PREFIX}/mic1/report", headers=device, json={**REPORT, "ports": [port]} + ) + assert reported.status_code == 200, reported.text + node = _device_node( + "mic1", + node="panel", + provides=[{"name": "mic", "port": "mic", "dtype": "audio"}], + ) + _publish_flow(client, superuser_token_headers, "voice", [node]) + controller = client.app.state.flow_controller + with client.websocket_connect(SOCKET, headers=device) as ws: + ws.receive_json() + ws.send_bytes(frame("voice.mic", b"\x00\x01")) # an image on an audio port + ws.send_bytes(frame("voice.mic", b"\x00\x02", "audio/L16;rate=16000")) + value = wait_for( + lambda: (controller.values().get("voice.mic") or {}).get("value") + ) + assert value["media_type"] == "audio/L16;rate=16000" + assert client.app.state.artifact_store.volatile.get(value["digest"]) == b"\x00\x02" + + def test_a_frame_past_the_cap_is_dropped( client: TestClient, camera, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/docs/code/api.md b/docs/code/api.md index 38e530e..8f7e3ed 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -279,10 +279,12 @@ fresh snapshot. With a `binary` uplink the device sends a camera frame as a binary message in the same layout, with the header `{"type": "frame", "name": "", -"media_type": "image/jpeg", "ts": }`. The message has to be one of -its `in` ports wired as `image`, and the whole message at most -`MAX_FRAME_BYTES`. The frame is held in memory and published in one step, -named after the message and credited to its device node. A frame that does not +"media_type": "image/jpeg", "ts": }`. A microphone's chunk travels +the same way with a media type such as `audio/L16;rate=16000`. The message has +to be one of its `in` ports wired as `image` or `audio`, the media type of +that family, and the whole message at most `MAX_FRAME_BYTES`. The frame is +held in memory and published in one step, named after the message and credited +to its device node. A frame that does not qualify is dropped, the reason logged once per socket, and the socket stays open. With `http` it uploads with `PUT /artifacts?volatile=true&name=` and publishes with `POST /messages/`. A device that is unpaired or