Binary uplink takes audio chunks beside image frames (openpanel VS4)

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:

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:

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": "<message>",
-"media_type": "image/jpeg", "ts": <device ms>}`. 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": <device ms>}`. 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=<message>`
 and publishes with `POST /messages/<message>`. A device that is unpaired or