diff --git a/backend/fluksio/ai/client.py b/backend/fluksio/ai/client.py
index d9607a6..3c68aa5 100644
--- a/backend/fluksio/ai/client.py
+++ b/backend/fluksio/ai/client.py
@@ -6,16 +6,22 @@ OpenRouter, Ollama, vLLM — so one client covers them without a provider SDK.
The admin types the URL; the model never chooses where a request goes.
The call is streamed, so the answer reaches the browser as it is written.
+
+Speech is the same dialect's audio half: a realtime transcription session, a
+whole clip transcribed, and a sentence spoken, as `parrot serve` answers them.
"""
from __future__ import annotations
+import asyncio
+import base64
import json
from collections.abc import AsyncIterator
from typing import Any
from urllib.parse import urlparse
import httpx
+from websockets.asyncio.client import connect
class ProviderError(Exception):
@@ -152,3 +158,113 @@ async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> lis
raise ProviderError(_detail(response))
data = response.json().get("data") or []
return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m)
+
+
+async def transcribe(
+ api_base: str,
+ audio: bytes,
+ filename: str,
+ media_type: str,
+ *,
+ model: str = "whisper-1",
+ timeout: float = 30.0,
+) -> str:
+ """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says."""
+ async with httpx.AsyncClient(timeout=timeout) as http:
+ response = await http.post(
+ api_base.rstrip("/") + "/audio/transcriptions",
+ files={"file": (filename, audio, media_type)},
+ data={"model": model, "response_format": "json"},
+ )
+ if response.is_error:
+ raise ProviderError(_detail(response))
+ return str(response.json().get("text") or "")
+
+
+async def speak(
+ api_base: str,
+ text: str,
+ voice: str,
+ *,
+ model: str = "tts-1",
+ timeout: float = 30.0,
+) -> bytes:
+ """``POST {api_base}/audio/speech``: ``text`` spoken, as WAV."""
+ async with httpx.AsyncClient(timeout=timeout) as http:
+ response = await http.post(
+ api_base.rstrip("/") + "/audio/speech",
+ json={
+ "model": model,
+ "input": text,
+ "voice": voice,
+ "response_format": "wav",
+ },
+ )
+ if response.is_error:
+ raise ProviderError(_detail(response))
+ return response.content
+
+
+async def transcribe_stream(
+ api_base: str,
+ audio: asyncio.Queue[bytes | None],
+ *,
+ rate: int = 16000,
+ idle_s: float = 10.0,
+) -> AsyncIterator[tuple[bool, str]]:
+ """``WS {api_base}/realtime``: one utterance, transcribed while it is spoken.
+
+ Sends each chunk of PCM16 taken off ``audio`` and commits at ``None``.
+ Yields ``(False, delta)`` as words are agreed on, then ``(True,
+ transcript)`` once. A chunk that takes longer than ``idle_s`` to come
+ raises ``TimeoutError``, so an utterance nobody commits ends.
+ """
+ url = "ws" + api_base.rstrip("/").removeprefix("http") + "/realtime"
+ async with connect(url, open_timeout=idle_s) as ws:
+ update = {"type": "audio/pcm", "rate": rate}
+ await ws.send(
+ json.dumps(
+ {
+ "type": "session.update",
+ "session": {
+ "type": "transcription",
+ "audio": {"input": {"format": update}},
+ },
+ }
+ )
+ )
+
+ async def append() -> None:
+ try:
+ while (
+ chunk := await asyncio.wait_for(audio.get(), idle_s)
+ ) is not None:
+ data = base64.b64encode(chunk).decode()
+ await ws.send(
+ json.dumps({"type": "input_audio_buffer.append", "audio": data})
+ )
+ await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
+ except BaseException:
+ # Ends the reading below, which then says why.
+ await ws.close()
+ raise
+
+ sender = asyncio.create_task(append())
+ sender.add_done_callback(lambda done: done.cancelled() or done.exception())
+ try:
+ async for raw in ws:
+ event = json.loads(raw)
+ kind = str(event.get("type") or "")
+ if kind == "error":
+ raise ProviderError(str((event.get("error") or {}).get("message")))
+ if kind.endswith("input_audio_transcription.delta"):
+ yield False, str(event.get("delta") or "")
+ elif kind.endswith("input_audio_transcription.completed"):
+ yield True, str(event.get("transcript") or "")
+ return
+ finally:
+ sender.cancel()
+ failure = (
+ sender.exception() if sender.done() and not sender.cancelled() else None
+ )
+ raise failure or ProviderError("the session closed before its transcript")
diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py
index 8a5602c..d343fa9 100644
--- a/backend/fluksio/api/routes/flows.py
+++ b/backend/fluksio/api/routes/flows.py
@@ -46,6 +46,7 @@ from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
from fluksio.flow.dashboards import DashboardStore
from fluksio.flow.demand import demand
from fluksio.flow.devices import (
+ Binding,
DeviceDef,
DeviceStore,
DeviceUnavailable,
@@ -1362,6 +1363,7 @@ FRAME_REFUSALS = {
"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",
+ "not_scalar": "the port it names is not wired as a bool, int, float or str",
"undeclared": "no flow declares its message",
"refused": "its message refused the value",
}
@@ -1393,14 +1395,7 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
controller: FlowController | None = getattr(app.state, "flow_controller", None)
if devices is None or store is None or controller is None:
return "no_engine"
- binding = next(
- (
- b
- for b in devices.bound(device.id)
- if b.direction == "in" and b.message == name
- ),
- None,
- )
+ binding = _bound_in(devices, device, name)
if binding is None:
return "not_wired"
if binding.dtype not in (DType.IMAGE, DType.AUDIO):
@@ -1420,6 +1415,59 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
return ""
+def _bound_in(devices: DeviceStore, device: DeviceDef, name: str) -> Binding | None:
+ """The device's own ``in`` port wired to ``name``, if one is."""
+ return next(
+ (
+ b
+ for b in devices.bound(device.id)
+ if b.direction == "in" and b.message == name
+ ),
+ None,
+ )
+
+
+def published(frame: str) -> dict[str, Any] | None:
+ """A value a device publishes on its socket, or None for anything else."""
+ try:
+ payload = orjson.loads(frame)
+ except orjson.JSONDecodeError:
+ return None
+ if isinstance(payload, dict) and payload.get("type") == "publish":
+ return payload
+ return None
+
+
+def ingest_publish(app: Any, device: DeviceDef, payload: dict[str, Any]) -> str:
+ """Publish a value a device sent on its socket, as ``POST /messages`` would.
+
+ Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it
+ is published. Only a message one of the device's own ``in`` ports is wired
+ to as a scalar, credited to the device node holding that port. Taken in
+ turn with the device's binary frames, so a ``listening`` false lands after
+ the last chunk of what was heard.
+ """
+ name = str(payload.get("name") or "")
+ devices: DeviceStore | None = getattr(app.state, "device_store", None)
+ controller: FlowController | None = getattr(app.state, "flow_controller", None)
+ if devices is None or controller is None:
+ return "no_engine"
+ binding = _bound_in(devices, device, name)
+ if binding is None:
+ return "not_wired"
+ if binding.dtype not in (DType.BOOL, DType.INT, DType.FLOAT, DType.STR):
+ return "not_scalar"
+ try:
+ controller.publish_message(
+ name, payload.get("value"), port_source(devices, device, name)
+ )
+ except KeyError:
+ return "undeclared"
+ except TypeError:
+ return "refused"
+ return ""
+
+
async def _receive(websocket: WebSocket) -> str | bytes:
"""The client's next frame, text or binary."""
message = await websocket.receive()
@@ -1465,8 +1513,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
A client says which media it is drawing (``{"type": "media", "names":
[...]}``) and is pushed those frames. A device also sends its camera's
- frames here, as binary frames in the layout it is pushed them in, and is
- told when its on-demand ports have demand.
+ frames here, as binary frames in the layout it is pushed them in, and
+ the values of its scalar ports as ``publish`` messages, and is told when
+ its on-demand ports have demand.
A live view's signalling rides here too (``fluksio.flow.live``): ``live``
messages on a person's socket, ``signal`` messages on a device's.
@@ -1628,6 +1677,23 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
if asked is not None:
wanted = asked
demand.want(viewer, wanted)
+ elif (
+ device_def is not None
+ and (value := published(frame)) is not None
+ ):
+ if credential is not None and await credential.revoked():
+ await websocket.close(code=1008)
+ return
+ reason = await run_in_threadpool(
+ ingest_publish, websocket.app, device_def, value
+ )
+ if reason and reason not in refused:
+ refused.add(reason)
+ logger.warning(
+ "Dropped a value from %s: %s",
+ device,
+ FRAME_REFUSALS[reason],
+ )
elif (signal := live_view.parse(frame)) is not None:
live_view.sessions.heard(
signal,
diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index 6a37aa7..7115779 100644
--- a/backend/fluksio/flow/controller.py
+++ b/backend/fluksio/flow/controller.py
@@ -57,7 +57,9 @@ from fluksio.flow.nodes import (
NtfyNode,
RbeNode,
ServiceNode,
+ SttNode,
SwitchNode,
+ TtsNode,
WebSocketNode,
WrenNode,
)
@@ -383,6 +385,18 @@ NODE_TYPES: dict[str, NodeType] = {
cls=WrenNode,
params_schema=_schema_of(WrenNode),
),
+ "stt": NodeType(
+ title="Speech to text",
+ description="Turn a microphone's audio into text, through a speech server.",
+ cls=SttNode,
+ params_schema=_schema_of(SttNode),
+ ),
+ "tts": NodeType(
+ title="Text to speech",
+ description="Speak text as audio for a speaker, through a speech server.",
+ cls=TtsNode,
+ params_schema=_schema_of(TtsNode),
+ ),
"inject": NodeType(
title="Inject",
description="Emit a value on request, on a timer, or when the flow starts.",
diff --git a/backend/fluksio/flow/nodes/__init__.py b/backend/fluksio/flow/nodes/__init__.py
index 3cef5e0..02cffb5 100644
--- a/backend/fluksio/flow/nodes/__init__.py
+++ b/backend/fluksio/flow/nodes/__init__.py
@@ -18,6 +18,7 @@ from fluksio.flow.nodes.mlp import MLPNode
from fluksio.flow.nodes.mqtt import MqttNode
from fluksio.flow.nodes.ntfy import NtfyNode
from fluksio.flow.nodes.service import ServiceNode
+from fluksio.flow.nodes.speech import SttNode, TtsNode
from fluksio.flow.nodes.websocket import WebSocketNode
from fluksio.flow.nodes.wren import WrenNode
@@ -39,7 +40,9 @@ __all__ = [
"RESERVED_SETTINGS",
"RbeNode",
"ServiceNode",
+ "SttNode",
"SwitchNode",
+ "TtsNode",
"WebSocketNode",
"WrenNode",
]
diff --git a/backend/fluksio/flow/nodes/speech.py b/backend/fluksio/flow/nodes/speech.py
new file mode 100644
index 0000000..ca90ced
--- /dev/null
+++ b/backend/fluksio/flow/nodes/speech.py
@@ -0,0 +1,353 @@
+"""Speech: a microphone's audio to text, and text to audio a speaker plays.
+
+Both talk to an OpenAI-compatible speech server, such as `parrot serve` kept
+running by a Service node, whose address arrives on a ``url`` input. An empty
+url is a server that is not answering: the node reads degraded and drops what
+it gets.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import io
+import logging
+import re
+import threading
+import time
+import wave
+from collections.abc import Coroutine, Iterable
+from concurrent.futures import Future
+from typing import TYPE_CHECKING, Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from fluksio.ai import client
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes.base import Node
+
+if TYPE_CHECKING:
+ from fastapi import FastAPI
+
+logger = logging.getLogger(__name__)
+
+#: Seconds a session waits for its next chunk, and the most one utterance may
+#: take from its first chunk to its text. Past either it is given up.
+IDLE_S = 10.0
+UTTERANCE_S = 60.0
+#: The rate of a microphone's PCM chunks, the panels' (openpanel V3).
+RATE = 16000
+#: How many spoken sentences a tts node keeps.
+SPOKEN = 256
+#: How long after its estimated end a sentence still counts as playing: the
+#: device fetches it before it starts.
+PLAYING_SLACK_S = 1.0
+
+
+def _check(specs: list[MessageSpec], ports: dict[str, DType], node: str) -> None:
+ """Refuse a port the node does not have, or one of another type."""
+ for spec in specs:
+ dtype = ports.get(spec.port)
+ if dtype is None:
+ raise ValueError(
+ f"'{spec.port}' is not a port of {node}. Its ports are "
+ f"{', '.join(ports)}."
+ )
+ if spec.dtype is not dtype:
+ raise ValueError(f"{node}'s '{spec.port}' is {dtype.value}.")
+
+
+def _is_pcm(audio: dict[str, Any]) -> bool:
+ """A microphone chunk rather than a whole clip."""
+ media_type = str(audio.get("media_type") or "").lower()
+ return media_type.startswith(("audio/l16", "audio/pcm"))
+
+
+def sentences(text: str) -> list[str]:
+ """``text`` cut after each ``.``, ``!`` and ``?``, one call to speak each."""
+ return [s for s in (p.strip() for p in re.split(r"(?<=[.!?])\s+", text)) if s]
+
+
+class _Speech(Node):
+ """What both nodes share: the url, the artifact store and their health."""
+
+ __slots__ = ("_artifacts", "_loop")
+
+ def __init__(
+ self,
+ f: Any,
+ requires: Iterable[MessageSpec],
+ provides: Iterable[MessageSpec],
+ params: dict[str, Any] | None,
+ name: str,
+ inputs: dict[str, DType],
+ outputs: dict[str, DType],
+ ):
+ # Set on start; a test sets them directly.
+ self._artifacts: Any = None
+ self._loop: asyncio.AbstractEventLoop | None = None
+ super().__init__(
+ f, requires=requires, provides=provides, params=params, name=name
+ )
+ _check(self.input_ports, inputs, name)
+ _check(self.output_ports, outputs, name)
+ if not any(spec.port == "url" for spec in self.input_ports):
+ raise ValueError(
+ f"{name} needs a 'url' input: the speech server's address."
+ )
+
+ async def start(self, app: FastAPI | None = None) -> None:
+ self._loop = asyncio.get_running_loop()
+ self._artifacts = getattr(app.state, "artifact_store", None) if app else None
+
+ def _reachable(self, url: str) -> bool:
+ """Whether there is a server to ask, said as the node's health."""
+ if not url:
+ if self._health != "degraded":
+ self.report_health("degraded", "no speech server: its url is empty")
+ return False
+ if self._health != "ok":
+ self.report_health("ok")
+ return True
+
+ def _load(self, reference: dict[str, Any]) -> bytes:
+ return b"".join(self._artifacts.read(str(reference["digest"])))
+
+ def _say(self, **values: Any) -> None:
+ """Publish on the ports that are wired."""
+ wired = {spec.port for spec in self.output_ports if spec.name}
+ self.emit({port: value for port, value in values.items() if port in wired})
+
+
+class SttNode(_Speech):
+ """Speech to text: a microphone's audio in, what was said out.
+
+ Inputs: ``audio``, a stream of 16 kHz PCM16 chunks or a whole clip,
+ ``listening``, true while the device listens, and ``url``.
+ Outputs: ``partial``, the words agreed on so far, and ``text``.
+
+ One realtime session per utterance, opened when ``listening`` turns true
+ or at the first chunk, and committed when it turns false. A clip goes to
+ the server whole. Either way the node returns at once, since a device's
+ next frame waits for it, and publishes once the server answers. An
+ utterance that fails publishes nothing and is logged.
+ """
+
+ __slots__ = ("_heard", "_listening", "_lock", "_session", "_tasks")
+
+ def __init__(
+ self,
+ requires: MessageSpec | Iterable[MessageSpec] = (),
+ provides: MessageSpec | Iterable[MessageSpec] = (),
+ params: dict[str, Any] | None = None,
+ name: str | None = None,
+ ):
+ self._lock = threading.Lock()
+ #: The chunks of the utterance being spoken, None between utterances.
+ self._session: asyncio.Queue[bytes | None] | None = None
+ #: The digest last heard, and whether the device was listening.
+ self._heard = ""
+ self._listening: bool | None = None
+ self._tasks: set[Future[None]] = set()
+ inputs = [
+ # The url is read, not waited for: its changing is nothing to hear.
+ spec.model_copy(update={"trigger": spec.port != "url"})
+ for spec in self._normalize_ports(requires)
+ ]
+ if not any(spec.port == "audio" for spec in inputs):
+ raise ValueError("stt needs an 'audio' input: the microphone.")
+ super().__init__(
+ self._hear,
+ inputs,
+ self._normalize_ports(provides),
+ params,
+ name or "stt",
+ {"audio": DType.AUDIO, "listening": DType.BOOL, "url": DType.STR},
+ {"partial": DType.STR, "text": DType.STR},
+ )
+
+ async def start(self, app: FastAPI | None = None) -> None:
+ await super().start(app)
+ if self._pipeline is None:
+ return
+ # What the inputs hold already was heard before this start.
+ held = await asyncio.to_thread(
+ self._pipeline.state.get_present, list(self.requires)
+ )
+ for spec in self.input_ports:
+ value = held.get(spec.name)
+ if spec.port == "audio" and isinstance(value, dict):
+ self._heard = str(value.get("digest") or "")
+ elif spec.port == "listening" and isinstance(value, bool):
+ self._listening = value
+
+ async def stop(self, app: FastAPI | None = None) -> None:
+ for task in list(self._tasks):
+ task.cancel()
+ self._session = None
+
+ def _hear(
+ self,
+ params: dict[str, Any],
+ audio: dict[str, Any] | None = None,
+ listening: bool | None = None,
+ url: str = "",
+ ) -> None:
+ with self._lock:
+ digest = str((audio or {}).get("digest") or "")
+ fresh = bool(digest) and digest != self._heard
+ self._heard = digest or self._heard
+ was, self._listening = self._listening, listening
+ if not self._reachable(url):
+ self._session = None
+ return None
+ if fresh and audio is not None and not _is_pcm(audio):
+ self._spawn(self._clip(url, audio, self._load(audio)))
+ elif fresh and audio is not None:
+ if self._session is None:
+ self._session = self._open(url)
+ self._feed(self._session, self._load(audio))
+ elif listening and not was and self._session is None:
+ self._session = self._open(url)
+ elif was and not listening and self._session is not None:
+ self._feed(self._session, None)
+ self._session = None
+ return None
+
+ def _spawn(self, work: Coroutine[Any, Any, None]) -> None:
+ """Run on the engine's loop, so this run can return at once."""
+ if self._loop is None:
+ work.close()
+ raise RuntimeError("stt only hears in a live flow")
+ task = asyncio.run_coroutine_threadsafe(work, self._loop)
+ self._tasks.add(task)
+ task.add_done_callback(self._tasks.discard)
+
+ def _open(self, url: str) -> asyncio.Queue[bytes | None]:
+ chunks: asyncio.Queue[bytes | None] = asyncio.Queue()
+ self._spawn(self._transcribe(url, chunks))
+ return chunks
+
+ def _feed(self, chunks: asyncio.Queue[bytes | None], chunk: bytes | None) -> None:
+ assert self._loop is not None
+ self._loop.call_soon_threadsafe(chunks.put_nowait, chunk)
+
+ async def _transcribe(self, url: str, chunks: asyncio.Queue[bytes | None]) -> None:
+ words = ""
+ try:
+ async with asyncio.timeout(UTTERANCE_S):
+ async for final, text in client.transcribe_stream(
+ url + "/v1", chunks, rate=RATE, idle_s=IDLE_S
+ ):
+ words = text.strip() if final else words + text
+ if final and words:
+ await asyncio.to_thread(self._say, partial=words, text=words)
+ elif words:
+ await asyncio.to_thread(self._say, partial=words)
+ except Exception as exc:
+ logger.warning(
+ "'%s' lost an utterance: %s: %s", self.id, type(exc).__name__, exc
+ )
+ finally:
+ # Ended without a commit, by a timeout or a failure: the next chunk
+ # opens a session of its own rather than feeding this dead one.
+ with self._lock:
+ if self._session is chunks:
+ self._session = None
+
+ async def _clip(self, url: str, audio: dict[str, Any], data: bytes) -> None:
+ try:
+ text = await client.transcribe(
+ url + "/v1", data, "speech.wav", str(audio.get("media_type"))
+ )
+ except Exception as exc:
+ logger.warning("'%s' lost a clip: %s: %s", self.id, type(exc).__name__, exc)
+ return
+ if text.strip():
+ await asyncio.to_thread(self._say, partial=text.strip(), text=text.strip())
+
+
+class TtsNode(_Speech):
+ """Text to speech: each sentence of ``text`` spoken, out on ``speech``.
+
+ One call to the server per sentence, each sentence published as an
+ ``audio`` reference as soon as it is spoken, so a reply plays before its
+ last sentence is. A sentence spoken before in the same voice is not spoken
+ again.
+
+ A reference is marked ``"append": true`` while the ones before it are
+ still playing, which queues it on the device. The first sentence of a
+ reply carries no mark and cuts off whatever was playing, however its
+ sentences arrive: as one text, or one value each while they are written.
+ """
+
+ class Params(BaseModel):
+ model_config = ConfigDict(extra="allow")
+
+ voice: str = Field(
+ default="af_heart",
+ description="The voice, one of Kokoro's such as af_heart or bm_george.",
+ )
+
+ __slots__ = ("cfg", "_lock", "_until")
+
+ def __init__(
+ self,
+ requires: MessageSpec | Iterable[MessageSpec] = (),
+ provides: MessageSpec | Iterable[MessageSpec] = (),
+ params: dict[str, Any] | None = None,
+ name: str | None = None,
+ ):
+ self.cfg = self.Params.model_validate(params or {})
+ # One reply at a time, so its sentences go out in order.
+ self._lock = threading.Lock()
+ #: When what was published so far stops playing, on the monotonic clock.
+ self._until = 0.0
+ inputs = [
+ spec.model_copy(update={"trigger": spec.port != "url"})
+ for spec in self._normalize_ports(requires)
+ ]
+ if not any(spec.port == "text" for spec in inputs):
+ raise ValueError("tts needs a 'text' input: what to say.")
+ super().__init__(
+ self._speak,
+ inputs,
+ self._normalize_ports(provides),
+ params,
+ name or "tts",
+ {"text": DType.STR, "url": DType.STR},
+ {"speech": DType.AUDIO},
+ )
+
+ def _speak(self, params: dict[str, Any], text: str = "", url: str = "") -> None:
+ if not text.strip() or not self._reachable(url):
+ return None
+ with self._lock:
+ for sentence in sentences(text):
+ reference = self._spoken(url, sentence)
+ now = time.monotonic()
+ if now < self._until + PLAYING_SLACK_S:
+ reference = {**reference, "append": True}
+ self._until = max(now, self._until) + reference["meta"]["seconds"]
+ self._say(speech=reference)
+ return None
+
+ def _spoken(self, url: str, sentence: str) -> dict[str, Any]:
+ """The reference to ``sentence`` spoken, from the cache or the server."""
+ key = hashlib.sha256(f"{self.cfg.voice}\n{sentence}".encode()).hexdigest()
+ # Kept in state, so the sweep keeps the bytes it refers to.
+ spoken = dict(self.recall("spoken") or {})
+ reference = spoken.get(key)
+ if reference is not None and self._artifacts.has(reference["digest"]):
+ return dict(reference)
+ data = asyncio.run(client.speak(url + "/v1", sentence, self.cfg.voice))
+ with wave.open(io.BytesIO(data)) as clip:
+ seconds = round(clip.getnframes() / clip.getframerate(), 3)
+ stored: dict[str, Any] = self._artifacts.put(
+ [data], name="speech.wav", media_type="audio/wav"
+ )
+ reference = stored | {"meta": {"seconds": seconds}}
+ spoken.pop(key, None)
+ spoken[key] = reference
+ self.remember("spoken", dict(list(spoken.items())[-SPOKEN:]))
+ return reference
diff --git a/backend/fluksio/flow/nodes/wren.py b/backend/fluksio/flow/nodes/wren.py
index 2048508..80e1d71 100644
--- a/backend/fluksio/flow/nodes/wren.py
+++ b/backend/fluksio/flow/nodes/wren.py
@@ -30,7 +30,11 @@ def _wren() -> Any:
try:
import wren
except ImportError:
- raise RuntimeError("the wren node needs the model: pip install wren") from None
+ # Not `pip install wren`: that name on PyPI is another package.
+ raise RuntimeError(
+ "the wren node needs the model: pip install git+https://github.com/"
+ "stroblme/wren@eae4d6ff9ded68e36624972fd3f4a8f8d210b0c2"
+ ) from None
return wren
@@ -62,8 +66,11 @@ INPUTS = {
"dark": {DType.BOOL},
}
-#: The output the calls are reported on. Every other output is a device.
+#: The output the calls are reported on, and the two that say what came of
+#: the command. Every other output is a device.
ACTIONS = "actions"
+REPLY = "reply"
+UNHANDLED = "unhandled"
_TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR}
@@ -93,6 +100,21 @@ _VALUES: dict[str, tuple[bool | None, float | None, str | None]] = {
"watering.stop": (False, None, "off"),
}
+#: An applied call, said: "kitchen light on". `heating.set` says its number.
+_SAID = {
+ "lights.on": "light on",
+ "lights.off": "light off",
+ "media.play": "media playing",
+ "media.stop": "media stopped",
+ "heating.boost": "heating boosted",
+ "heating.off": "heating off",
+ "shutters.up": "shutters up",
+ "shutters.down": "shutters down",
+ "shutters.stop": "shutters stopped",
+ "watering.start": "watering on",
+ "watering.stop": "watering off",
+}
+
#: What every area held in training: a light, and in the living room, which
#: always existed, media and heating besides. A house missing them reads as
#: unfamiliar and the confidence drops with it, so they are filled in; a call
@@ -150,6 +172,18 @@ def _value(call: Any, dtype: DType) -> Any:
return int(number) if dtype is DType.INT else float(number)
+def _said(call: Any, area: str) -> str:
+ """An applied call as the reply says it: "garage light on in 5 minutes"."""
+ if call.tool == "heating.set":
+ said = f"heating to {_value(call, DType.FLOAT):g} degrees"
+ else:
+ said = _SAID.get(call.tool, call.tool)
+ if call.delay_s:
+ minutes = round(call.delay_s / 60)
+ said += f" in {minutes} minute{'s' * (minutes != 1)}"
+ return f"{area} {said}"
+
+
def _text(call: Any) -> str:
"""The call as the chat shows it, less the device it names."""
args = ", ".join(f"{k}={v}" for k, v in call.args.items() if k != "id")
@@ -163,9 +197,11 @@ class WrenNode(Node):
``presence`` (the area the user is in), ``outdoor`` (°C) and ``dark``
(otherwise read off the clock: before 07:00 or from 18:00).
- Outputs: ``actions``, a list of records, one per call and target, and
- any number of targets named ``_``. Targets of one area and kind
- are one device to the model; each takes what the call means for its type.
+ Outputs: ``actions``, a list of records, one per call and target,
+ ``reply``, a sentence saying what was applied ("Kitchen light on."),
+ ``unhandled``, the prompt when nothing was, and any number of targets
+ named ``_``. Targets of one area and kind are one device to
+ the model; each takes what the call means for its type.
A delayed call is held by the engine's work queue, like the Trigger
node's second value.
"""
@@ -224,6 +260,10 @@ class WrenNode(Node):
if spec.dtype is not DType.LIST or spec.item_dtype is not DType.RECORD:
raise ValueError(f"'{ACTIONS}' is a list of records.")
continue
+ if spec.port in (REPLY, UNHANDLED):
+ if spec.dtype is not DType.STR:
+ raise ValueError(f"'{spec.port}' is a str.")
+ continue
device = _device(spec.port, areas)
if device is None:
raise ValueError(
@@ -316,6 +356,7 @@ class WrenNode(Node):
write = self.cfg.apply and confidence >= self.cfg.min_confidence
writes: dict[str, Any] = {}
actions: list[dict[str, Any]] = []
+ said: list[str] = []
for call in calls:
area, _, kind = str(call.args.get("id", "")).partition(":")
ports = self.devices.get((area, kind), [])
@@ -336,6 +377,7 @@ class WrenNode(Node):
}
if not hits:
actions.append(record)
+ done = False
for spec, value in hits:
applied = write
if applied and call.delay_s:
@@ -347,7 +389,16 @@ class WrenNode(Node):
actions.append(
record | {"message": spec.name, "value": value, "applied": applied}
)
+ done = done or applied
+ if done:
+ said.append(_said(call, area))
- if any(spec.port == ACTIONS for spec in self.output_ports):
+ wired = {spec.port for spec in self.output_ports}
+ if ACTIONS in wired:
writes[ACTIONS] = actions
+ if said and REPLY in wired:
+ reply = ", ".join(said)
+ writes[REPLY] = reply[0].upper() + reply[1:] + "."
+ if not said and UNHANDLED in wired:
+ writes[UNHANDLED] = prompt
return writes or None
diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index 0c074c4..0eaf098 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/backend/fluksio/flow/pipeline.py
@@ -1717,7 +1717,12 @@ class Pipeline:
targets.add(consumer)
targets.update(self._get_downstream(consumer))
if targets:
- self._execute_parallel(targets, self._state, check_ready=True)
+ # Only what these values reach, as in a cascade: a node further
+ # down whose producer published nothing would otherwise run again
+ # on what it already read, once per microphone chunk.
+ self._execute_parallel(
+ targets, self._state, check_ready=True, changed=set(values)
+ )
def defer(
self,
diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py
index c5fadc8..d6be659 100644
--- a/backend/tests/flow/test_node_types.py
+++ b/backend/tests/flow/test_node_types.py
@@ -105,6 +105,22 @@ FIXTURES: dict[str, dict] = {
"requires": [MessageSpec(name="prompt", dtype=DType.STR)],
"provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)],
},
+ "stt": {
+ "params": {},
+ "requires": [
+ MessageSpec(name="mic", port="audio", dtype=DType.AUDIO),
+ MessageSpec(name="url", dtype=DType.STR),
+ ],
+ "provides": [MessageSpec(name="text", dtype=DType.STR)],
+ },
+ "tts": {
+ "params": {"voice": "bm_george"},
+ "requires": [
+ MessageSpec(name="reply", port="text", dtype=DType.STR),
+ MessageSpec(name="url", dtype=DType.STR),
+ ],
+ "provides": [MessageSpec(name="speech", dtype=DType.AUDIO)],
+ },
"device": {
"params": {"device": "hall", "config": {"gpio.4": {"pull": "up"}}},
"requires": [
diff --git a/backend/tests/flow/test_speech.py b/backend/tests/flow/test_speech.py
new file mode 100644
index 0000000..c7de53a
--- /dev/null
+++ b/backend/tests/flow/test_speech.py
@@ -0,0 +1,147 @@
+"""The speech nodes: one session per utterance, and each sentence spoken once."""
+
+import asyncio
+import hashlib
+import io
+import wave
+from typing import Any
+
+from fluksio.ai import client
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes.speech import SttNode, TtsNode, sentences
+from fluksio.flow.pipeline import Pipeline
+
+URL = "http://speech:8100"
+
+
+class Store:
+ """The artifact store's surface the nodes use."""
+
+ def __init__(self) -> None:
+ self.data: dict[str, bytes] = {}
+
+ def put(self, chunks: list[bytes], name: str, media_type: str) -> dict[str, Any]:
+ data = b"".join(chunks)
+ digest = "sha256:" + hashlib.sha256(data).hexdigest()
+ self.data[digest] = data
+ return {"digest": digest, "size": len(data), "media_type": media_type}
+
+ def has(self, digest: str) -> bool:
+ return digest in self.data
+
+ def read(self, digest: str) -> list[bytes]:
+ return [self.data[digest]]
+
+
+def test_sentences_are_cut_after_their_ending():
+ assert sentences(" Kitchen light on. Heating off! Is it?\nYes") == [
+ "Kitchen light on.",
+ "Heating off!",
+ "Is it?",
+ "Yes",
+ ]
+
+
+def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch):
+ heard: list[bytes] = []
+
+ async def session(api_base, audio, *, rate, idle_s):
+ assert (api_base, rate) == (f"{URL}/v1", 16000)
+ while (chunk := await audio.get()) is not None:
+ heard.append(chunk)
+ yield False, "turn on"
+ yield False, " the"
+ yield True, " Turn on the light."
+
+ monkeypatch.setattr(client, "transcribe_stream", session)
+
+ async def scenario() -> None:
+ node = SttNode(
+ requires=[
+ MessageSpec(name="mic", port="audio", dtype=DType.AUDIO),
+ MessageSpec(name="listening", dtype=DType.BOOL),
+ MessageSpec(name="url", dtype=DType.STR),
+ ],
+ provides=[
+ MessageSpec(name="partial", dtype=DType.STR),
+ MessageSpec(name="text", dtype=DType.STR),
+ ],
+ )
+ node.assign_flow("voice", "stt")
+ pipeline = Pipeline([node])
+ store = Store()
+ chunks = [
+ store.put([bytes([n]) * 3200], "voice.mic", "audio/L16;rate=16000")
+ for n in range(3)
+ ]
+ # Left over from before the engine started: not part of what is said.
+ pipeline.state["voice.mic"] = chunks[0]
+ await node.start()
+ node._artifacts = store
+
+ async def hear(audio: dict, listening: bool, url: str = URL) -> None:
+ inputs = {"voice.mic": audio, "voice.listening": listening}
+ await asyncio.to_thread(node.execute, inputs | {"voice.url": url})
+
+ await hear(chunks[0], True)
+ await hear(chunks[1], True)
+ await hear(chunks[2], True)
+ await hear(chunks[2], False)
+ for _ in range(200):
+ if "voice.text" in pipeline.state:
+ break
+ await asyncio.sleep(0.01)
+
+ assert heard == [bytes([1]) * 3200, bytes([2]) * 3200]
+ assert pipeline.state["voice.text"] == "Turn on the light."
+ assert pipeline.state["voice.partial"] == "Turn on the light."
+ assert node._session is None
+
+ # No server, no session: the node says so instead.
+ await hear(chunks[1], True, url="")
+ assert node._session is None
+ assert node._health == "degraded"
+
+ asyncio.run(scenario())
+
+
+def _wav(seconds: float) -> bytes:
+ out = io.BytesIO()
+ with wave.open(out, "wb") as clip:
+ clip.setnchannels(1)
+ clip.setsampwidth(2)
+ clip.setframerate(24000)
+ clip.writeframes(b"\0\0" * int(24000 * seconds))
+ return out.getvalue()
+
+
+def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch):
+ spoken: list[str] = []
+ published: list[dict] = []
+
+ async def speak(api_base, text, voice):
+ spoken.append(text)
+ return _wav(len(spoken))
+
+ monkeypatch.setattr(client, "speak", speak)
+ monkeypatch.setattr(TtsNode, "_say", lambda self, speech: published.append(speech))
+ node = TtsNode(
+ requires=[
+ MessageSpec(name="reply", port="text", dtype=DType.STR),
+ MessageSpec(name="url", dtype=DType.STR),
+ ],
+ provides=[MessageSpec(name="speech", dtype=DType.AUDIO)],
+ )
+ node.assign_flow("voice", "tts")
+ Pipeline([node])
+ node._artifacts = Store()
+
+ node.execute({"voice.reply": "Kitchen light on. Heating off.", "voice.url": URL})
+ # Long after both stopped playing.
+ node._until = -10.0
+ node.execute({"voice.reply": "Kitchen light on.", "voice.url": URL})
+
+ assert spoken == ["Kitchen light on.", "Heating off."]
+ assert ["append" in speech for speech in published] == [False, True, False]
+ assert published[2] == published[0]
+ assert published[1]["meta"] == {"seconds": 2.0}
diff --git a/backend/tests/flow/test_wren.py b/backend/tests/flow/test_wren.py
index 44c048b..18a8b03 100644
--- a/backend/tests/flow/test_wren.py
+++ b/backend/tests/flow/test_wren.py
@@ -165,3 +165,27 @@ def test_the_model_runs_on_one_blas_thread(fake_wren, monkeypatch):
assert threads
assert set(threads) == {1}
+
+
+def test_a_reply_says_what_was_applied_and_the_rest_is_unhandled(fake_wren):
+ node = WrenNode(
+ requires=[MessageSpec(name="command", port="prompt", dtype=DType.STR)],
+ provides=[
+ MessageSpec(name="kitchen_light", dtype=DType.BOOL),
+ MessageSpec(name="reply", dtype=DType.STR),
+ MessageSpec(name="unhandled", dtype=DType.STR),
+ ],
+ params={"apply": True},
+ )
+ node.assign_flow("house", "control")
+ fake_wren.answer = ([FakeCall("lights.on", {"id": "kitchen:light"})], 0.99)
+
+ out = node.execute({"house.command": "turn on the kitchen light"})
+
+ assert out == {"house.kitchen_light": True, "house.reply": "Kitchen light on."}
+
+ # Nothing wired for the call, or nothing to call at all.
+ for calls in ([FakeCall("lights.on", {"id": "garage:light"})], []):
+ fake_wren.answer = (calls, 0.99)
+ out = node.execute({"house.command": "what is the weather tomorrow"})
+ assert out == {"house.unhandled": "what is the weather tomorrow"}
diff --git a/docs/code/api.md b/docs/code/api.md
index 8f7e3ed..04ccedd 100644
--- a/docs/code/api.md
+++ b/docs/code/api.md
@@ -286,8 +286,12 @@ 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
+open. A value of one of its `in` ports wired as `bool`, `int`, `float` or `str`
+goes as the text message `{"type": "publish", "name": "", "value":
+...}`, taken in turn with the frames, so a microphone's `listening` false
+arrives after its last chunk. With `http` it uploads with
+`PUT /artifacts?volatile=true&name=` and publishes with
+`POST /messages/`. A device that is unpaired or
deleted while connected is closed with 1008, at the latest with its next frame.
### Live view
diff --git a/docs/code/nodes.md b/docs/code/nodes.md
index 32803e4..faff69c 100644
--- a/docs/code/nodes.md
+++ b/docs/code/nodes.md
@@ -179,7 +179,10 @@ and ARM, so nothing has to be installed into the image.
openpanel publishes a 16 kHz mono WAV on its `mic` port after its wake word,
and plays any `audio` reference sent to its `speaker` port (WAV, MP3 or AAC).
-Two Function nodes close that loop with local models.
+The [Speech to text](../reference/node-types.md#speech-to-text) and
+[Text to speech](../reference/node-types.md#text-to-speech) nodes close that
+loop through a speech server on a GPU. Without one, two Function nodes close it
+with local models.
[faster-whisper](https://github.com/SYSTRAN/faster-whisper) turns speech into
text and [piper](https://github.com/OHF-Voice/piper1-gpl) turns text into
speech.
diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md
index f488d75..eb1ece8 100644
--- a/docs/interface/dashboards.md
+++ b/docs/interface/dashboards.md
@@ -514,8 +514,10 @@ to 640×360. Re-encode other clips with the [clip recipe](../code/nodes.md#makin
The device decides what its [port](#the-devices-own-ports) names
mean. openpanel [reports](#what-a-device-reports) the ones it serves, such as
`gpio.`, `backlight`, `touch`, `mic` and `speaker`, with the settings each
-one reads, and ignores any other name. A browser screen serves none of them. The [speech recipe](../code/nodes.md#speech-to-and-from-a-panel)
-turns `mic` into text and text into `speaker`.
+one reads, and ignores any other name. A browser screen serves none of them. The
+[Speech to text](../reference/node-types.md#speech-to-text) and
+[Text to speech](../reference/node-types.md#text-to-speech) nodes turn `mic`
+into text and text into `speaker`, and `caption` shows what is being heard.
### The device's own ports
diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md
index 4afa892..1c6b1f9 100644
--- a/docs/reference/node-types.md
+++ b/docs/reference/node-types.md
@@ -344,7 +344,9 @@ use a [batch flow](../concepts/runs.md) and a Function node.
[Wren](https://github.com/stroblme/wren) is an optional local language model
for device commands. Install it in the engine's environment with
-`pip install wren`; the node reports a missing dependency if it is absent.
+`pip install git+https://github.com/stroblme/wren@eae4d6ff9ded68e36624972fd3f4a8f8d210b0c2`.
+The `wren` package on PyPI is an unrelated one. Without the model the node
+reports a missing dependency.
| Input | Type | Notes |
|---|---|---|
@@ -353,7 +355,8 @@ for device commands. Install it in the engine's environment with
| `outdoor` | `float` or `int` | °C; 20 without it, clamped to −5…30 |
| `dark` | `bool` | without it, dark before 07:00 and from 18:00 |
-The outputs are `actions`, a list of records, and any number of **targets**. A
+The outputs are `actions`, a list of records, `reply`, `unhandled`, and any
+number of **targets**. A
target is an output named `_` and bound to the message its device
listens on: `living_room_setpoint` bound to `home.setpoint` is the living
room's heating. The wiring is the allowlist, so the model can write nothing
@@ -393,6 +396,10 @@ Each record in `actions` contains `message`, `value`, `applied`, `delay_s`,
`call`, and `confidence`. Unwired calls have an empty `message` and are not
applied. Keep `apply=false` while checking how the model interprets commands.
+`reply` is one sentence saying what was applied, such as "Kitchen light on."
+or "Garage light on in 5 minutes.". When nothing was applied, `unhandled`
+carries the prompt instead, for something else to answer.
+
A call with a delay ("in 10m") is held by the engine's work queue, like the
Debounce node's second value: it survives a restart where the queue is in
Redis, and not otherwise.
@@ -408,6 +415,57 @@ They are reported but cannot receive writes unless wired.
The chat's [Control tab](../code/agents.md#control) sends commands by hand.
+### Speech to text
+
+**`stt`** — turn a microphone's audio into text, through a speech server.
+
+| Input | Type | Notes |
+|---|---|---|
+| `audio` | `audio` | 16 kHz mono PCM16 chunks (`audio/L16`), or a whole clip such as a WAV |
+| `listening` | `bool` | true while the device listens; false ends the utterance |
+| `url` | `str` | the speech server, from a [Service](#service) node; read, never waited for |
+
+| Output | Type | Notes |
+|---|---|---|
+| `partial` | `str` | the words agreed on so far, as they come, then the final text |
+| `text` | `str` | what was said, once per utterance |
+
+The server is any that speaks OpenAI's audio API, such as `parrot serve`. Each
+utterance is one realtime transcription session (`{url}/v1/realtime`), opened
+when `listening` turns true or at the first chunk and committed when it turns
+false. A clip is sent whole to `{url}/v1/audio/transcriptions`. The node
+returns at once and publishes when the server answers, since a device's next
+chunk waits for it. An utterance that fails, or gets no chunk for 10 seconds,
+publishes nothing and is logged.
+
+`listening` has to hold a value before a chunk wakes the node. Declare it as a
+flow input with the initial value `false` when a device may never send it,
+such as one that sends whole clips.
+
+### Text to speech
+
+**`tts`** — speak text as audio for a speaker, through a speech server.
+
+| Input | Type | Notes |
+|---|---|---|
+| `text` | `str` | what to say, a whole reply or one sentence at a time |
+| `url` | `str` | the speech server, as for Speech to text |
+
+| Output | Type | Notes |
+|---|---|---|
+| `speech` | `audio` | one WAV reference per sentence |
+
+| Setting | Default | Notes |
+|---|---|---|
+| `voice` | `af_heart` | one of Kokoro's voices, such as `bm_george` |
+
+Each sentence is one call to `{url}/v1/audio/speech` and published as soon as
+it is spoken. A sentence spoken before in the same voice is reused rather than
+spoken again, for the last 256 sentences. A reference carries `"append": true`
+while the sentences before it are still playing, which queues it on a panel's
+`speaker`. The first sentence of a reply carries none and cuts off whatever
+was playing.
+
## Devices
### Device
diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx
index 123a976..fc90041 100644
--- a/frontend/src/components/Flow/FlowNode.tsx
+++ b/frontend/src/components/Flow/FlowNode.tsx
@@ -12,6 +12,7 @@ import {
Filter,
Globe,
Merge,
+ Mic,
MonitorSmartphone,
Play,
Plug,
@@ -22,6 +23,7 @@ import {
Square,
Terminal,
Timer,
+ Volume2,
} from "lucide-react"
import { memo, useEffect, useRef, useState } from "react"
@@ -63,6 +65,8 @@ const NODE_ICONS = {
file: FileText,
ntfy: Bell,
wren: Bird,
+ stt: Mic,
+ tts: Volume2,
// The glyph the Devices tab goes by.
device: MonitorSmartphone,
} as const