From ae180e6e71e47b6a37722bdc8d05e32fba07e455 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 13:11:20 +0200 Subject: [PATCH 1/9] fix(rebase): rebase conflicts --- Makefile | 5 +++-- src/core/client.py | 7 +++++++ src/modules/gesture/emage/modeling.py | 1 + src/modules/gesture/gesture.py | 16 ++++++++++++---- src/modules/rag/rag.py | 6 ++---- 5 files changed, 25 insertions(+), 10 deletions(-) diff --git a/Makefile b/Makefile index a373f79..fc63860 100644 --- a/Makefile +++ b/Makefile @@ -1,10 +1,11 @@ lint: black . isort . - flake8 . + + mypy . --check-untyped-defs test: pytest -check: lint test \ No newline at end of file +check: lint test diff --git a/src/core/client.py b/src/core/client.py index 9df62d0..68f955c 100644 --- a/src/core/client.py +++ b/src/core/client.py @@ -113,6 +113,13 @@ async def _receive_loop(self, ws: websockets.ClientConnection): try: while True: msg = await ws.recv() + if isinstance(msg, bytes): + if len(msg) < 2: + print(f"<< bytes ({len(msg)}B, no topic)") + continue + (topic_len,) = struct.unpack(">H", msg[:2]) + topic = msg[2 : 2 + topic_len].decode() + payload = msg[2 + topic_len :] if isinstance(msg, bytes): topic_len = struct.unpack(">H", msg[:2])[0] diff --git a/src/modules/gesture/emage/modeling.py b/src/modules/gesture/emage/modeling.py index f58b967..3fa1da9 100644 --- a/src/modules/gesture/emage/modeling.py +++ b/src/modules/gesture/emage/modeling.py @@ -14,6 +14,7 @@ VQEncoderV5, VQEncoderV6, WavEncoder, + axis_angle_to_matrix, axis_angle_to_rotation_6d, recover_from_mask_ts, rotation_6d_to_axis_angle, diff --git a/src/modules/gesture/gesture.py b/src/modules/gesture/gesture.py index a39704a..e0c2582 100644 --- a/src/modules/gesture/gesture.py +++ b/src/modules/gesture/gesture.py @@ -88,19 +88,27 @@ def __init__( print("[Gesture] loading upper_vq...") upper_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/upper" - ).to(self.device) # type: ignore[arg-type] + ).to( + self.device + ) # type: ignore[arg-type] print("[Gesture] loading lower_vq...") lower_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/lower" - ).to(self.device) # type: ignore[arg-type] + ).to( + self.device + ) # type: ignore[arg-type] print("[Gesture] loading hands_vq...") hands_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/hands" - ).to(self.device) # type: ignore[arg-type] + ).to( + self.device + ) # type: ignore[arg-type] print("[Gesture] loading global_ae...") global_ae = EmageVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/global" - ).to(self.device) # type: ignore[arg-type] + ).to( + self.device + ) # type: ignore[arg-type] self.motion_vq = EmageVQModel( face_model=face_vq, diff --git a/src/modules/rag/rag.py b/src/modules/rag/rag.py index 23fddee..6ae9b9d 100644 --- a/src/modules/rag/rag.py +++ b/src/modules/rag/rag.py @@ -426,9 +426,7 @@ def base_strength(payload: dict) -> float: vec = await self._embed(merged) vector_size = len(vec) now = datetime.now().isoformat() - imp = min( - max((p.payload or {}).get("importance", 3) for p in weak) + 1, 10 - ) + imp = min(max((p.payload or {}).get("importance", 3) for p in weak) + 1, 10) self._qdrant.upsert( collection_name=self._cfg.memory_collection, points=[ @@ -580,7 +578,7 @@ def _build_prompt( # slot for a formatting rule. open-mistral-nemo skips the persona-level # no-Ah/Oh rule often enough that we restate it right at the tail. user_prompt += ( - '\n\nStart your answer straight on the substance — do not open with ' + "\n\nStart your answer straight on the substance — do not open with " '"Ah" or "Oh".' ) From 49243624555a24508a42adccbe78d767810d4b61 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 19:08:28 +0200 Subject: [PATCH 2/9] evol(events): added RawBytes event and serialization function to EventData --- src/core/events.py | 27 +++++++++++++++++---------- 1 file changed, 17 insertions(+), 10 deletions(-) diff --git a/src/core/events.py b/src/core/events.py index 862760b..1f99218 100644 --- a/src/core/events.py +++ b/src/core/events.py @@ -1,7 +1,7 @@ import asyncio import logging from collections import defaultdict -from dataclasses import dataclass +from dataclasses import dataclass, asdict from typing import Any, Mapping import numpy as np @@ -18,16 +18,23 @@ class EventData: @classmethod def from_wire(cls, data: Mapping[str, Any]) -> "EventData": - """Build an event from a JSON payload sent by a client. - - Default: keyword-splat the payload onto the dataclass. Events whose fields - are nested dataclasses (or that accept a simpler external shape than the - in-pipeline one) override this — e.g. RAGQuestion accepts a bare - ``{"text": ...}`` typed question. In-process producers construct the - dataclass directly and never go through this path. - """ return cls(**data) + def to_wire(self) -> Mapping[str, Any] | bytes: + return asdict(self) + + +@dataclass +class RawBytes(EventData): + data: bytes + + @classmethod + def from_wire(cls, data: bytes) -> "RawBytes": + return cls(data=data) + + def to_wire(self) -> bytes: + return self.data + class EventGraph: """ @@ -112,7 +119,7 @@ async def _run(self, module: Module, data): ) -def _summarize(item) -> str: +def _summarize(item) -> str: # TODO event data summarize function """Short repr that avoids dumping full numpy arrays into the log.""" cls = type(item).__name__ data = getattr(item, "data", None) From add87adcca6e386224fbb845df6f4967ec416777 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 19:09:59 +0200 Subject: [PATCH 3/9] feat(events): events are abstractly serialized and deserialized --- src/core/client.py | 53 ++++---------------- src/core/huri.py | 2 +- src/core/module.py | 5 +- src/interfaces/cli_interface.py | 13 ++--- src/modules/events.py | 12 ++--- src/modules/factory.py | 19 +++---- src/modules/gesture/gesture.py | 2 +- src/modules/speech_to_text/microphone_vad.py | 11 ++-- src/modules/text_to_speech/text_to_speech.py | 2 +- src/modules/utils/sender.py | 43 +++++----------- 10 files changed, 55 insertions(+), 107 deletions(-) diff --git a/src/core/client.py b/src/core/client.py index 68f955c..55fdb59 100644 --- a/src/core/client.py +++ b/src/core/client.py @@ -4,7 +4,7 @@ import struct from collections import defaultdict from dataclasses import asdict -from typing import Any, Dict, Generic, List, Type, TypeVar +from typing import Any, Dict, Generic, List, Type, TypeVar, Mapping import numpy as np import websockets @@ -12,7 +12,7 @@ from src.core.dataclasses.config import ClientConfig from src.core.events import EventData -T = TypeVar("T", bound=EventData | bytes) +T = TypeVar("T", bound=EventData) class ClientSender(Generic[T]): @@ -45,16 +45,20 @@ async def _send_bytes(self, ws: websockets.ClientConnection, data: bytes): await ws.send(packet) - async def _send_event_data(self, ws: websockets.ClientConnection, data: EventData): - packet = json.dumps({"topic": self.topic, "data": asdict(data)}) + async def _send_event_data( + self, ws: websockets.ClientConnection, data: Mapping[str, Any] + ): + packet = json.dumps({"topic": self.topic, "data": data}) await ws.send(packet) async def send(self, ws: websockets.ClientConnection, data: T): - if isinstance(data, bytes): - await self._send_bytes(ws, data) + wire = data.to_wire() + + if isinstance(wire, bytes): + await self._send_bytes(ws, wire) else: - await self._send_event_data(ws, data) + await self._send_event_data(ws, wire) class ClientHook(Generic[T]): @@ -113,13 +117,6 @@ async def _receive_loop(self, ws: websockets.ClientConnection): try: while True: msg = await ws.recv() - if isinstance(msg, bytes): - if len(msg) < 2: - print(f"<< bytes ({len(msg)}B, no topic)") - continue - (topic_len,) = struct.unpack(">H", msg[:2]) - topic = msg[2 : 2 + topic_len].decode() - payload = msg[2 + topic_len :] if isinstance(msg, bytes): topic_len = struct.unpack(">H", msg[:2])[0] @@ -127,40 +124,12 @@ async def _receive_loop(self, ws: websockets.ClientConnection): topic = msg[2 : 2 + topic_len].decode() data = msg[2 + topic_len :] - if topic == "audio" and len(data) >= 13: - sample_rate, end, pts = struct.unpack(">IBd", data[:13]) - # Samples are native-endian float32 - # (Sender uses ndarray.tobytes()). - samples = np.frombuffer(data[13:], dtype=np.float32) - data = { - "sample_rate": sample_rate, - "end": end, - "pts": pts, - "data": samples, - } - elif topic == "motion" and len(data) >= 16: - pts, fps, n_frames = struct.unpack(">dII", data[:16]) - print(f"<< motion: pts={pts:.3f}s frames={n_frames} @ {fps}fps") - data = { - "poses": np.ndarray(0), - "expressions": np.ndarray(0), - "trans": np.ndarray(0), - "fps": fps, - "pts": pts, - } - else: - print(f"<< {topic}: bytes ({len(data)}B)") else: event = json.loads(msg) topic = event["topic"] data = event["data"] for hook in self.hooks[topic]: - # Hydrate via from_wire (not a bare **data splat) so events - # with nested EventData fields — e.g. RAGQuestion.transcript / - # .emotion — are rebuilt as dataclasses, mirroring the server's - # EventDataFactory. Build a fresh instance per hook so a topic - # with several hooks doesn't re-splat an already-built event. hook_data = ( data if isinstance(data, bytes) diff --git a/src/core/huri.py b/src/core/huri.py index 6b92e12..a38d71b 100644 --- a/src/core/huri.py +++ b/src/core/huri.py @@ -32,7 +32,7 @@ def __init__( self, modules: Dict[str, Type[Module]], handles: Dict[str, handle.DeploymentHandle], - events: Dict[str, Type[EventData | bytes]], + events: Dict[str, Type[EventData]], ) -> None: self.module_factory = ModuleFactory(handles) self.event_factory = EventDataFactory() diff --git a/src/core/module.py b/src/core/module.py index 281a68b..964ca85 100644 --- a/src/core/module.py +++ b/src/core/module.py @@ -1,6 +1,7 @@ from typing import Any, AsyncGenerator, Coroutine, Optional, Type from ray.serve import handle +from .events import EventData class Module: @@ -24,7 +25,9 @@ class Module: input_type: str output_type: Optional[str] - def process(self, _) -> Coroutine[Any, Any, Any] | AsyncGenerator[Any, None]: + def process( + self, _ + ) -> Coroutine[Any, Any, EventData] | AsyncGenerator[EventData, None]: raise NotImplementedError diff --git a/src/interfaces/cli_interface.py b/src/interfaces/cli_interface.py index 3cf2282..0f647d6 100644 --- a/src/interfaces/cli_interface.py +++ b/src/interfaces/cli_interface.py @@ -11,13 +11,14 @@ from scipy.signal import resample from src.core.client import ClientHook, ClientSender +from src.core.events import RawBytes from src.core.interface import Interface from src.modules.rag.events import RAGQuestion, RAGResult from src.modules.speech_to_text.events import Transcript from src.modules.text_to_speech.events import Audio, Token -class AudioSender(ClientSender[bytes]): +class AudioSender(ClientSender[RawBytes]): def __init__( self, sample_rate: int = 16000, frame_duration: float = 0.030, **kwargs ): @@ -43,7 +44,7 @@ def callback(indata: np.ndarray, frames, time, status): ): while True: chunk = await queue.get() - await self.send(ws, chunk.tobytes()) + await self.send(ws, RawBytes(data=chunk.tobytes())) class TextSender(ClientSender[RAGQuestion]): @@ -162,14 +163,14 @@ async def hook(self, data: Audio): self._collect_audio(data.data, data.sample_rate, bool(data.end)) -class TextHook(ClientHook[RAGResult]): - input_type = RAGResult +class TextHook(ClientHook[RAGQuestion]): + input_type = RAGQuestion def __init__(self, **kwargs): super().__init__(**kwargs) - async def hook(self, data: RAGResult): - print("<<", data.answer) + async def hook(self, data: RAGQuestion): + print("<<", data.transcript, data.emotion) class TokenHook(ClientHook[Token]): diff --git a/src/modules/events.py b/src/modules/events.py index 8cdc066..0ecb219 100644 --- a/src/modules/events.py +++ b/src/modules/events.py @@ -1,17 +1,17 @@ from typing import Dict, Type -from src.core.events import EventData +from src.core.events import EventData, RawBytes from src.modules.emotion.events import Emotion from src.modules.gesture.events import Motion from src.modules.rag.events import PartialQuestion, RAGQuestion from src.modules.speech_to_text.events import Transcript, Voice -from src.modules.text_to_speech.events import Token +from src.modules.text_to_speech.events import Token, Audio -def get_events() -> Dict[str, Type[EventData | bytes]]: - events: Dict[str, Type[EventData | bytes]] = { - "audio_in": bytes, # inbound mic frames (raw int16 PCM) - "audio": bytes, +def get_events() -> Dict[str, Type[EventData]]: + events: Dict[str, Type[EventData]] = { + "audio.in": RawBytes, # inbound mic frames (raw int16 PCM) + "audio.out": Audio, "voice": Voice, "transcript": Transcript, "emotion": Emotion, diff --git a/src/modules/factory.py b/src/modules/factory.py index 315dd6e..1d843ee 100644 --- a/src/modules/factory.py +++ b/src/modules/factory.py @@ -9,9 +9,9 @@ class EventDataFactory: def __init__(self): - self._registry: Dict[str, Type[EventData | bytes]] = {} + self._registry: Dict[str, Type[EventData]] = {} - def register(self, topic: str, event_cls: Type[EventData | bytes] | None) -> None: + def register(self, topic: str, event_cls: Type[EventData] | None) -> None: if topic in self._registry: if event_cls is None or event_cls == self._registry[topic]: return @@ -23,23 +23,16 @@ def register(self, topic: str, event_cls: Type[EventData | bytes] | None) -> Non self._registry[topic] = event_cls - def create(self, topic: str, data: Mapping[str, Any] | bytes) -> EventData | bytes: + def create(self, topic: str, data: Mapping[str, Any]) -> EventData: if topic not in self._registry: raise RuntimeError(f"unknown event topic {topic}") event_cls = self._registry[topic] - if isinstance(data, bytes): - if issubclass(event_cls, bytes): - return data - else: - raise RuntimeError(f"mismatched event data type: \ -{event_cls} is not type bytes but should be.") + if issubclass(event_cls, EventData): + return event_cls.from_wire(data) else: - if issubclass(event_cls, EventData): - return event_cls.from_wire(data) - else: - raise RuntimeError(f"mismatched event data type: \ + raise RuntimeError(f"mismatched event data type: \ {event_cls} is not derived from EventData but should be.") diff --git a/src/modules/gesture/gesture.py b/src/modules/gesture/gesture.py index e0c2582..8b51dcc 100644 --- a/src/modules/gesture/gesture.py +++ b/src/modules/gesture/gesture.py @@ -307,7 +307,7 @@ class Gesture(ModuleWithHandle): """ _handle_cls = GestureDeployment - input_type = "audio" + input_type = "audio.out" output_type = "motion" def __init__( diff --git a/src/modules/speech_to_text/microphone_vad.py b/src/modules/speech_to_text/microphone_vad.py index eb52f74..99c2972 100644 --- a/src/modules/speech_to_text/microphone_vad.py +++ b/src/modules/speech_to_text/microphone_vad.py @@ -4,6 +4,7 @@ import webrtcvad from src.core.module import Module +from src.core.events import RawBytes from .events import Voice @@ -13,7 +14,7 @@ class MIC(Module): Detect voice and silence using WebRTC VAD. - input: audio_in, + input: audio.in, output: voice :vad_agressiveness: from 0 (low) to 3 (high, can distord audio). @@ -26,7 +27,7 @@ class MIC(Module): # Inbound microphone frames travel on their own topic so the TTS-output # "audio" topic (consumed by Gesture and the client Sender) never collides # with mic input — otherwise raw mic bytes get echoed back to the client. - input_type = "audio_in" + input_type = "audio.in" output_type = "voice" def __init__( @@ -49,11 +50,11 @@ def __init__( self.vad = webrtcvad.Vad(vad_agressiveness) - async def process(self, data: bytes) -> Optional[Voice]: - if self.vad.is_speech(data, self.sample_rate) is True: + async def process(self, data: RawBytes) -> Optional[Voice]: + if self.vad.is_speech(data.data, self.sample_rate) is True: self.silence_frames_count = 0 - audio_array = np.frombuffer(data, dtype=np.int16) + audio_array = np.frombuffer(data.data, dtype=np.int16) audio_array_float = audio_array.astype(np.float32) / 32768.0 return Voice(audio_array_float) diff --git a/src/modules/text_to_speech/text_to_speech.py b/src/modules/text_to_speech/text_to_speech.py index 63e1b46..322a6cd 100644 --- a/src/modules/text_to_speech/text_to_speech.py +++ b/src/modules/text_to_speech/text_to_speech.py @@ -232,7 +232,7 @@ class TTS(ModuleWithHandle): _handle_cls = TTSDeployment input_type = "token" - output_type = "audio" + output_type = "audio.out" def __init__(self, _handle: handle.DeploymentHandle): super().__init__(_handle) diff --git a/src/modules/utils/sender.py b/src/modules/utils/sender.py index 434f457..7e516e5 100644 --- a/src/modules/utils/sender.py +++ b/src/modules/utils/sender.py @@ -41,40 +41,21 @@ def __init__(self, ws: WebSocket, type: str): self.ws: WebSocket = ws self.input_type = type - async def process(self, data: EventData | bytes): + async def process(self, data: EventData): logger.info("[Sender:%s] received %s", self.input_type, type(data).__name__) - if isinstance(data, bytes): - await self.ws.send_bytes(self._prefix(data)) - elif isinstance(data, Audio): - logger.info( - "[Sender:%s] Audio samples=%d sr=%d end=%s pts=%.3fs", - self.input_type, - data.data.shape[0], - data.sample_rate, - data.end, - data.pts, - ) - header = struct.pack(">IBd", data.sample_rate, int(data.end), data.pts) - await self.ws.send_bytes(self._prefix(header + data.data.tobytes())) - elif isinstance(data, Motion): - n_frames = data.poses.shape[0] - logger.info( - "[Sender:%s] Motion frames=%d fps=%d pts=%.3fs", - self.input_type, - n_frames, - data.fps, - data.pts, - ) - header = struct.pack(">dII", data.pts, data.fps, n_frames) - body = ( - data.poses.astype(np.float32).tobytes() - + data.expressions.astype(np.float32).tobytes() - + data.trans.astype(np.float32).tobytes() - ) - await self.ws.send_bytes(self._prefix(header + body)) + + wire = data.to_wire() + if isinstance(wire, bytes): + await self.ws.send_bytes(self._prefix(wire)) else: - await self.ws.send_json({"topic": self.input_type, "data": asdict(data)}) + await self.ws.send_json( + { + "topic": self.input_type, + "data": wire, + } + ) def _prefix(self, payload: bytes) -> bytes: + """Encode topic and topic len and adds it as a prefix to the payload""" topic_bytes = self.input_type.encode() return struct.pack(">H", len(topic_bytes)) + topic_bytes + payload From 5a3a3042987ddeb3cc90f874d9608dc0836132ac Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 19:10:50 +0200 Subject: [PATCH 4/9] evol(tts): Audio event serialization/deserialization --- src/modules/text_to_speech/events.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/src/modules/text_to_speech/events.py b/src/modules/text_to_speech/events.py index dceb269..1270ff6 100644 --- a/src/modules/text_to_speech/events.py +++ b/src/modules/text_to_speech/events.py @@ -1,8 +1,12 @@ from dataclasses import dataclass +import logging import numpy as np from src.core.events import EventData +import struct + +logger = logging.getLogger("ray.serve") @dataclass @@ -17,3 +21,21 @@ class Audio(EventData): sample_rate: int end: bool = False pts: float = 0.0 # presentation timestamp in seconds from utterance start + + def to_wire(self) -> bytes: + logger.info( + "Audio samples=%d sr=%d end=%s pts=%.3fs", + self.data.shape[0], + self.sample_rate, + self.end, + self.pts, + ) + header = struct.pack(">IBd", self.sample_rate, int(self.end), self.pts) + return header + self.data.tobytes() + + @classmethod + def from_wire(cls, data: bytes) -> "Audio": + sample_rate, end, pts = struct.unpack(">IBd", data[:13]) + samples = np.frombuffer(data[13:], dtype=np.float32) + + return cls(sample_rate=sample_rate, end=end, pts=pts, samples=samples) From cef8100fd55d5e2e6a1755e59286719ba4173876 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 19:10:59 +0200 Subject: [PATCH 5/9] evol(gesture): Motion event serialization/deserialization --- src/modules/gesture/events.py | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/src/modules/gesture/events.py b/src/modules/gesture/events.py index 7793de1..150f203 100644 --- a/src/modules/gesture/events.py +++ b/src/modules/gesture/events.py @@ -1,11 +1,14 @@ from dataclasses import dataclass import numpy as np - +import struct from src.core.events import EventData +import logging _EMAGE_FPS = 30 +logger = logging.getLogger("ray.serve") + @dataclass class Motion(EventData): @@ -14,3 +17,32 @@ class Motion(EventData): trans: np.ndarray # (t, 3) global root translation fps: int = _EMAGE_FPS pts: float = 0.0 # presentation timestamp in seconds, paired with Audio.pts + + def to_wire(self) -> bytes: + n_frames = self.poses.shape[0] + logger.info( + "Motion frames=%d fps=%d pts=%.3fs", + n_frames, + self.fps, + self.pts, + ) + header = struct.pack(">dII", self.pts, self.fps, n_frames) + body = ( + self.poses.astype(np.float32).tobytes() + + self.expressions.astype(np.float32).tobytes() + + self.trans.astype(np.float32).tobytes() + ) + return header + body + + @classmethod + def from_wire(cls, data: bytes) -> "Motion": + pts, fps, n_frames = struct.unpack(">dII", data[:16]) + print(f"<< motion: pts={pts:.3f}s frames={n_frames} @ {fps}fps") + + return cls( + poses=np.ndarray(0), + expressions=np.ndarray(0), + trans=np.ndarray(0), + fps=fps, + pts=pts, + ) From 275b5a2a716c501f13bcd6e0ec126b181951727f Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 19:30:12 +0200 Subject: [PATCH 6/9] evol(EventData): added summarize function for cleaner logs per event type --- src/core/events.py | 29 ++++++++++++++-------------- src/core/module.py | 2 +- src/modules/speech_to_text/events.py | 9 +++++++++ src/modules/text_to_speech/events.py | 6 ++++++ 4 files changed, 31 insertions(+), 15 deletions(-) diff --git a/src/core/events.py b/src/core/events.py index 1f99218..119e165 100644 --- a/src/core/events.py +++ b/src/core/events.py @@ -2,7 +2,7 @@ import logging from collections import defaultdict from dataclasses import dataclass, asdict -from typing import Any, Mapping +from typing import Any, Mapping, cast, AsyncGenerator, Coroutine import numpy as np @@ -23,6 +23,10 @@ def from_wire(cls, data: Mapping[str, Any]) -> "EventData": def to_wire(self) -> Mapping[str, Any] | bytes: return asdict(self) + def summarize(self) -> str: + cls = type(self).__name__ + return f"{cls}({self!r})" + @dataclass class RawBytes(EventData): @@ -35,6 +39,10 @@ def from_wire(cls, data: bytes) -> "RawBytes": def to_wire(self) -> bytes: return self.data + def summarize(self) -> str: + cls = type(self).__name__ + return f"{cls}(len:{len(self.data)})" + class EventGraph: """ @@ -82,14 +90,15 @@ async def _run(self, module: Module, data): if hasattr(result, "__aiter__"): try: - async for item in result: + generator = cast(AsyncGenerator[EventData | None, None], result) + async for item in generator: if item is None: continue logger.info( "[GRAPH] %s -> %r: %s", type(module).__name__, module.output_type, - _summarize(item), + item.summarize(), ) await self.publish(module.output_type, item) except Exception: @@ -98,14 +107,15 @@ async def _run(self, module: Module, data): ) else: + coroutine = cast(Coroutine[Any, Any, EventData | None], result) try: - value = await result + value = await coroutine if value is not None: logger.info( "[GRAPH] %s -> %r: %s", type(module).__name__, module.output_type, - _summarize(value), + value.summarize(), ) await self.publish(module.output_type, value) except Exception: @@ -117,12 +127,3 @@ async def _run(self, module: Module, data): logger.exception( "[GRAPH] process() call failed in %s", type(module).__name__ ) - - -def _summarize(item) -> str: # TODO event data summarize function - """Short repr that avoids dumping full numpy arrays into the log.""" - cls = type(item).__name__ - data = getattr(item, "data", None) - if isinstance(data, np.ndarray): - return f"{cls}(shape={data.shape}, dtype={data.dtype})" - return f"{cls}({item!r})" diff --git a/src/core/module.py b/src/core/module.py index 964ca85..22200d7 100644 --- a/src/core/module.py +++ b/src/core/module.py @@ -27,7 +27,7 @@ class Module: def process( self, _ - ) -> Coroutine[Any, Any, EventData] | AsyncGenerator[EventData, None]: + ) -> Coroutine[Any, Any, EventData | None] | AsyncGenerator[EventData | None, None]: raise NotImplementedError diff --git a/src/modules/speech_to_text/events.py b/src/modules/speech_to_text/events.py index fc04674..17ab092 100644 --- a/src/modules/speech_to_text/events.py +++ b/src/modules/speech_to_text/events.py @@ -15,3 +15,12 @@ class Transcript(EventData): @dataclass class Voice(EventData): data: Optional[np.ndarray] + + def summarize(self) -> str: + """Short repr that avoids dumping full numpy arrays into the log.""" + + if self.data: + cls = type(self).__name__ + return f"{cls}(shape={self.data.shape}, dtype={self.data.dtype})" + else: + return super().summarize() diff --git a/src/modules/text_to_speech/events.py b/src/modules/text_to_speech/events.py index 1270ff6..fbf62c5 100644 --- a/src/modules/text_to_speech/events.py +++ b/src/modules/text_to_speech/events.py @@ -39,3 +39,9 @@ def from_wire(cls, data: bytes) -> "Audio": samples = np.frombuffer(data[13:], dtype=np.float32) return cls(sample_rate=sample_rate, end=end, pts=pts, samples=samples) + + def summarize(self) -> str: + """Short repr that avoids dumping full numpy arrays into the log.""" + + cls = type(self).__name__ + return f"{cls}(shape={self.data.shape}, dtype={self.data.dtype})" From 3352e3d0d2ca826e7af5a7c21016ffc1b1ba87d8 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 20:21:04 +0200 Subject: [PATCH 7/9] lint(isort): make lint --- src/core/client.py | 2 +- src/core/events.py | 4 ++-- src/core/module.py | 1 + src/modules/events.py | 2 +- src/modules/gesture/events.py | 5 +++-- src/modules/speech_to_text/microphone_vad.py | 2 +- src/modules/text_to_speech/events.py | 4 ++-- 7 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/core/client.py b/src/core/client.py index 55fdb59..42d7036 100644 --- a/src/core/client.py +++ b/src/core/client.py @@ -4,7 +4,7 @@ import struct from collections import defaultdict from dataclasses import asdict -from typing import Any, Dict, Generic, List, Type, TypeVar, Mapping +from typing import Any, Dict, Generic, List, Mapping, Type, TypeVar import numpy as np import websockets diff --git a/src/core/events.py b/src/core/events.py index 119e165..444a248 100644 --- a/src/core/events.py +++ b/src/core/events.py @@ -1,8 +1,8 @@ import asyncio import logging from collections import defaultdict -from dataclasses import dataclass, asdict -from typing import Any, Mapping, cast, AsyncGenerator, Coroutine +from dataclasses import asdict, dataclass +from typing import Any, AsyncGenerator, Coroutine, Mapping, cast import numpy as np diff --git a/src/core/module.py b/src/core/module.py index 22200d7..f3144da 100644 --- a/src/core/module.py +++ b/src/core/module.py @@ -1,6 +1,7 @@ from typing import Any, AsyncGenerator, Coroutine, Optional, Type from ray.serve import handle + from .events import EventData diff --git a/src/modules/events.py b/src/modules/events.py index 0ecb219..46f86aa 100644 --- a/src/modules/events.py +++ b/src/modules/events.py @@ -5,7 +5,7 @@ from src.modules.gesture.events import Motion from src.modules.rag.events import PartialQuestion, RAGQuestion from src.modules.speech_to_text.events import Transcript, Voice -from src.modules.text_to_speech.events import Token, Audio +from src.modules.text_to_speech.events import Audio, Token def get_events() -> Dict[str, Type[EventData]]: diff --git a/src/modules/gesture/events.py b/src/modules/gesture/events.py index 150f203..44679d7 100644 --- a/src/modules/gesture/events.py +++ b/src/modules/gesture/events.py @@ -1,9 +1,10 @@ +import logging +import struct from dataclasses import dataclass import numpy as np -import struct + from src.core.events import EventData -import logging _EMAGE_FPS = 30 diff --git a/src/modules/speech_to_text/microphone_vad.py b/src/modules/speech_to_text/microphone_vad.py index 99c2972..c1d4579 100644 --- a/src/modules/speech_to_text/microphone_vad.py +++ b/src/modules/speech_to_text/microphone_vad.py @@ -3,8 +3,8 @@ import numpy as np import webrtcvad -from src.core.module import Module from src.core.events import RawBytes +from src.core.module import Module from .events import Voice diff --git a/src/modules/text_to_speech/events.py b/src/modules/text_to_speech/events.py index fbf62c5..1d9a5e5 100644 --- a/src/modules/text_to_speech/events.py +++ b/src/modules/text_to_speech/events.py @@ -1,10 +1,10 @@ -from dataclasses import dataclass import logging +import struct +from dataclasses import dataclass import numpy as np from src.core.events import EventData -import struct logger = logging.getLogger("ray.serve") From 0c74d7750c81a46c5a6d8477180083baac505ead Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 20:22:20 +0200 Subject: [PATCH 8/9] evol(events): correctly typed events --- src/core/bus.py | 94 +++++++++++++++++++++ src/core/events.py | 119 +++++---------------------- src/core/session.py | 2 +- src/interfaces/cli_interface.py | 10 +-- src/modules/events.py | 4 +- src/modules/rag/events.py | 8 +- src/modules/text_to_speech/events.py | 8 +- 7 files changed, 130 insertions(+), 115 deletions(-) create mode 100644 src/core/bus.py diff --git a/src/core/bus.py b/src/core/bus.py new file mode 100644 index 0000000..003abb5 --- /dev/null +++ b/src/core/bus.py @@ -0,0 +1,94 @@ +import asyncio +import logging +from collections import defaultdict +from typing import Any, AsyncGenerator, Coroutine, cast + +from .events import EventData +from .module import Module + +logger = logging.getLogger("ray.serve") + + +class EventGraph: + """ + Asynchronous event routing system for HuRI modules. + + The EventGraph is responsible for: + - Registering module subscribers + - Routing events between modules + - Executing module pipelines asynchronously + - Handling coroutine and async generator outputs + + Modules subscribe to events through their `input_type`. + When an event is published, all subscribed modules are executed + concurrently. + + Supports: + - Coroutine-based modules + - Async generator streaming modules + - Recursive event propagation + + :subscribers: + Dictionary mapping event topics to subscribed modules. + """ + + def __init__(self): + self.subscribers = defaultdict(list) + + def register(self, module: Module): + self.subscribers[module.input_type].append(module) + + async def publish(self, event_topic, data): + subs = self.subscribers[event_topic] + if event_topic not in ("audio_in",): # skip mic-frame spam + logger.info( + "[GRAPH] publish topic=%r subscribers=%s", + event_topic, + [type(m).__name__ for m in subs], + ) + for module in subs: + asyncio.create_task(self._run(module, data)) + + async def _run(self, module: Module, data): + try: + result = module.process(data) + + if hasattr(result, "__aiter__"): + try: + generator = cast(AsyncGenerator[EventData | None, None], result) + async for item in generator: + if item is None: + continue + logger.info( + "[GRAPH] %s -> %r: %s", + type(module).__name__, + module.output_type, + item.summarize(), + ) + await self.publish(module.output_type, item) + except Exception: + logger.exception( + "[GRAPH] async generator failed in %s", type(module).__name__ + ) + + else: + coroutine = cast(Coroutine[Any, Any, EventData | None], result) + try: + value = await coroutine + if value is not None: + logger.info( + "[GRAPH] %s -> %r: %s", + type(module).__name__, + module.output_type, + value.summarize(), + ) + await self.publish(module.output_type, value) + except Exception: + logger.exception( + "[GRAPH] coroutine failed in %s", type(module).__name__ + ) + + except Exception: + logger.exception( + "[GRAPH] process() call failed in %s", type(module).__name__ + ) diff --git a/src/core/events.py b/src/core/events.py index 444a248..68ee7bd 100644 --- a/src/core/events.py +++ b/src/core/events.py @@ -1,27 +1,23 @@ -import asyncio import logging -from collections import defaultdict from dataclasses import asdict, dataclass -from typing import Any, AsyncGenerator, Coroutine, Mapping, cast - -import numpy as np - -from .module import Module +from typing import Any, Generic, Mapping, TypeVar logger = logging.getLogger("ray.serve") +WireT = TypeVar("WireT", Mapping[str, Any], bytes) + @dataclass -class EventData: +class EventData(Generic[WireT]): """An event data must be derived from this class, and use @dataclass decorator. Or they can be bytes.""" @classmethod - def from_wire(cls, data: Mapping[str, Any]) -> "EventData": - return cls(**data) + def from_wire(cls, data: WireT) -> "EventData[WireT]": + raise NotImplementedError - def to_wire(self) -> Mapping[str, Any] | bytes: - return asdict(self) + def to_wire(self) -> WireT: + raise NotImplementedError def summarize(self) -> str: cls = type(self).__name__ @@ -29,11 +25,21 @@ def summarize(self) -> str: @dataclass -class RawBytes(EventData): +class JsonEvent(EventData[Mapping[str, Any]]): + @classmethod + def from_wire(cls, data: Mapping[str, Any]) -> "JsonEvent": + return cls(**data) + + def to_wire(self) -> Mapping[str, Any]: + return asdict(self) + + +@dataclass +class BytesEvent(EventData[bytes]): data: bytes @classmethod - def from_wire(cls, data: bytes) -> "RawBytes": + def from_wire(cls, data: bytes) -> "BytesEvent": return cls(data=data) def to_wire(self) -> bytes: @@ -42,88 +48,3 @@ def to_wire(self) -> bytes: def summarize(self) -> str: cls = type(self).__name__ return f"{cls}(len:{len(self.data)})" - - -class EventGraph: - """ - Asynchronous event routing system for HuRI modules. - - The EventGraph is responsible for: - - Registering module subscribers - - Routing events between modules - - Executing module pipelines asynchronously - - Handling coroutine and async generator outputs - - Modules subscribe to events through their `input_type`. - When an event is published, all subscribed modules are executed - concurrently. - - Supports: - - Coroutine-based modules - - Async generator streaming modules - - Recursive event propagation - - :subscribers: - Dictionary mapping event topics to subscribed modules. - """ - - def __init__(self): - self.subscribers = defaultdict(list) - - def register(self, module: Module): - self.subscribers[module.input_type].append(module) - - async def publish(self, event_topic, data): - subs = self.subscribers[event_topic] - if event_topic not in ("audio_in",): # skip mic-frame spam - logger.info( - "[GRAPH] publish topic=%r subscribers=%s", - event_topic, - [type(m).__name__ for m in subs], - ) - for module in subs: - asyncio.create_task(self._run(module, data)) - - async def _run(self, module: Module, data): - try: - result = module.process(data) - - if hasattr(result, "__aiter__"): - try: - generator = cast(AsyncGenerator[EventData | None, None], result) - async for item in generator: - if item is None: - continue - logger.info( - "[GRAPH] %s -> %r: %s", - type(module).__name__, - module.output_type, - item.summarize(), - ) - await self.publish(module.output_type, item) - except Exception: - logger.exception( - "[GRAPH] async generator failed in %s", type(module).__name__ - ) - - else: - coroutine = cast(Coroutine[Any, Any, EventData | None], result) - try: - value = await coroutine - if value is not None: - logger.info( - "[GRAPH] %s -> %r: %s", - type(module).__name__, - module.output_type, - value.summarize(), - ) - await self.publish(module.output_type, value) - except Exception: - logger.exception( - "[GRAPH] coroutine failed in %s", type(module).__name__ - ) - - except Exception: - logger.exception( - "[GRAPH] process() call failed in %s", type(module).__name__ - ) diff --git a/src/core/session.py b/src/core/session.py index 2b64daf..4f09c22 100644 --- a/src/core/session.py +++ b/src/core/session.py @@ -1,6 +1,6 @@ from typing import List -from .events import EventGraph +from .bus import EventGraph from .module import Module diff --git a/src/interfaces/cli_interface.py b/src/interfaces/cli_interface.py index 0f647d6..10949ec 100644 --- a/src/interfaces/cli_interface.py +++ b/src/interfaces/cli_interface.py @@ -11,14 +11,14 @@ from scipy.signal import resample from src.core.client import ClientHook, ClientSender -from src.core.events import RawBytes +from src.core.events import BytesEvent from src.core.interface import Interface -from src.modules.rag.events import RAGQuestion, RAGResult +from src.modules.rag.events import RAGQuestion from src.modules.speech_to_text.events import Transcript from src.modules.text_to_speech.events import Audio, Token -class AudioSender(ClientSender[RawBytes]): +class AudioSender(ClientSender[BytesEvent]): def __init__( self, sample_rate: int = 16000, frame_duration: float = 0.030, **kwargs ): @@ -44,7 +44,7 @@ def callback(indata: np.ndarray, frames, time, status): ): while True: chunk = await queue.get() - await self.send(ws, RawBytes(data=chunk.tobytes())) + await self.send(ws, BytesEvent(data=chunk.tobytes())) class TextSender(ClientSender[RAGQuestion]): @@ -75,9 +75,9 @@ class AudioHook(ClientHook[Audio]): def __init__( self, + save_audio_dir: str, sample_rate=48000, incoming_sample_rate=16000, - save_audio_dir: Optional[str] = None, **kwargs, ): super().__init__(**kwargs) diff --git a/src/modules/events.py b/src/modules/events.py index 46f86aa..099174d 100644 --- a/src/modules/events.py +++ b/src/modules/events.py @@ -1,6 +1,6 @@ from typing import Dict, Type -from src.core.events import EventData, RawBytes +from src.core.events import BytesEvent, EventData from src.modules.emotion.events import Emotion from src.modules.gesture.events import Motion from src.modules.rag.events import PartialQuestion, RAGQuestion @@ -10,7 +10,7 @@ def get_events() -> Dict[str, Type[EventData]]: events: Dict[str, Type[EventData]] = { - "audio.in": RawBytes, # inbound mic frames (raw int16 PCM) + "audio.in": BytesEvent, # inbound mic frames (raw int16 PCM) "audio.out": Audio, "voice": Voice, "transcript": Transcript, diff --git a/src/modules/rag/events.py b/src/modules/rag/events.py index 0873376..955ff40 100644 --- a/src/modules/rag/events.py +++ b/src/modules/rag/events.py @@ -1,13 +1,13 @@ from dataclasses import dataclass, field from typing import Any, Mapping, Optional -from src.core.events import EventData +from src.core.events import JsonEvent from src.modules.emotion.events import Emotion from src.modules.speech_to_text.events import Transcript @dataclass -class RAGResult(EventData): +class RAGResult(JsonEvent): """What RAGHandle returns.""" answer: str @@ -15,7 +15,7 @@ class RAGResult(EventData): @dataclass -class PartialQuestion(EventData): +class PartialQuestion(JsonEvent): """Partial question used to aggregate a sentence to an emotion.""" transcript: Optional[Transcript] @@ -23,7 +23,7 @@ class PartialQuestion(EventData): @dataclass -class RAGQuestion(EventData): +class RAGQuestion(JsonEvent): """Fully aggregated question to send to the RAG.""" transcript: Transcript diff --git a/src/modules/text_to_speech/events.py b/src/modules/text_to_speech/events.py index 1d9a5e5..434c14f 100644 --- a/src/modules/text_to_speech/events.py +++ b/src/modules/text_to_speech/events.py @@ -4,19 +4,19 @@ import numpy as np -from src.core.events import EventData +from src.core.events import EventData, JsonEvent logger = logging.getLogger("ray.serve") @dataclass -class Token(EventData): +class Token(JsonEvent): text: str end: bool @dataclass -class Audio(EventData): +class Audio(EventData[bytes]): data: np.ndarray sample_rate: int end: bool = False @@ -38,7 +38,7 @@ def from_wire(cls, data: bytes) -> "Audio": sample_rate, end, pts = struct.unpack(">IBd", data[:13]) samples = np.frombuffer(data[13:], dtype=np.float32) - return cls(sample_rate=sample_rate, end=end, pts=pts, samples=samples) + return cls(sample_rate=sample_rate, end=end, pts=pts, data=samples) def summarize(self) -> str: """Short repr that avoids dumping full numpy arrays into the log.""" From 5c7aaaf67ec25074519b5221945af5c99667c296 Mon Sep 17 00:00:00 2001 From: Popochounet Date: Tue, 25 Aug 2026 20:22:47 +0200 Subject: [PATCH 9/9] lint(mypy/flake8): make lint --- Makefile | 3 +-- pyproject.toml | 3 +++ src/core/client.py | 1 - src/modules/gesture/emage/modeling.py | 1 - src/modules/gesture/events.py | 2 +- src/modules/gesture/gesture.py | 26 ++++++-------------- src/modules/rag/rag.py | 4 +-- src/modules/speech_to_text/events.py | 6 ++--- src/modules/speech_to_text/microphone_vad.py | 4 +-- src/modules/text_to_speech/text_to_speech.py | 4 +-- src/modules/utils/sender.py | 4 --- tests/core/test_events.py | 26 ++------------------ 12 files changed, 21 insertions(+), 63 deletions(-) diff --git a/Makefile b/Makefile index fc63860..4d8b7d9 100644 --- a/Makefile +++ b/Makefile @@ -1,8 +1,7 @@ lint: black . isort . - - + flake8 . mypy . --check-untyped-defs test: diff --git a/pyproject.toml b/pyproject.toml index 72f1390..e2558d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,9 @@ strict_optional = true warn_unused_ignores = true warn_return_any = true warn_unused_configs = true +exclude = [ + "^tests/", +] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/src/core/client.py b/src/core/client.py index 42d7036..cfd84e2 100644 --- a/src/core/client.py +++ b/src/core/client.py @@ -6,7 +6,6 @@ from dataclasses import asdict from typing import Any, Dict, Generic, List, Mapping, Type, TypeVar -import numpy as np import websockets from src.core.dataclasses.config import ClientConfig diff --git a/src/modules/gesture/emage/modeling.py b/src/modules/gesture/emage/modeling.py index 3fa1da9..f58b967 100644 --- a/src/modules/gesture/emage/modeling.py +++ b/src/modules/gesture/emage/modeling.py @@ -14,7 +14,6 @@ VQEncoderV5, VQEncoderV6, WavEncoder, - axis_angle_to_matrix, axis_angle_to_rotation_6d, recover_from_mask_ts, rotation_6d_to_axis_angle, diff --git a/src/modules/gesture/events.py b/src/modules/gesture/events.py index 44679d7..754fde3 100644 --- a/src/modules/gesture/events.py +++ b/src/modules/gesture/events.py @@ -12,7 +12,7 @@ @dataclass -class Motion(EventData): +class Motion(EventData[bytes]): poses: np.ndarray # (t, 165) SMPL-X axis-angle, 55 joints × 3 expressions: np.ndarray # (t, 100) facial expression coefficients trans: np.ndarray # (t, 3) global root translation diff --git a/src/modules/gesture/gesture.py b/src/modules/gesture/gesture.py index 8b51dcc..7c533a2 100644 --- a/src/modules/gesture/gesture.py +++ b/src/modules/gesture/gesture.py @@ -83,32 +83,24 @@ def __init__( print("[Gesture] loading face_vq...") face_vq = EmageVQVAEConv.from_pretrained(hf_repo, subfolder="emage_vq/face").to( - self.device # type: ignore[arg-type] + self.device ) print("[Gesture] loading upper_vq...") upper_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/upper" - ).to( - self.device - ) # type: ignore[arg-type] + ).to(self.device) print("[Gesture] loading lower_vq...") lower_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/lower" - ).to( - self.device - ) # type: ignore[arg-type] + ).to(self.device) print("[Gesture] loading hands_vq...") hands_vq = EmageVQVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/hands" - ).to( - self.device - ) # type: ignore[arg-type] + ).to(self.device) print("[Gesture] loading global_ae...") global_ae = EmageVAEConv.from_pretrained( hf_repo, subfolder="emage_vq/global" - ).to( - self.device - ) # type: ignore[arg-type] + ).to(self.device) self.motion_vq = EmageVQModel( face_model=face_vq, @@ -120,9 +112,7 @@ def __init__( self.motion_vq.eval() print("[Gesture] loading EmageAudioModel...") - self.model = EmageAudioModel.from_pretrained(hf_repo).to( - self.device # type: ignore[arg-type] - ) + self.model = EmageAudioModel.from_pretrained(hf_repo).to(self.device) self.model.eval() self._warmup() @@ -349,9 +339,7 @@ def _end_utterance(self) -> None: self._buf_start = 0 self._emitted = 0 - async def process( # type: ignore[override] - self, audio: Audio - ) -> AsyncGenerator[Motion, None]: + async def process(self, audio: Audio) -> AsyncGenerator[Motion, None]: # Each chunk arrives as its own process() task on the shared per-session # instance, so serialise under a lock to keep the buffer ordered. async with self._lock: diff --git a/src/modules/rag/rag.py b/src/modules/rag/rag.py index 6ae9b9d..782ca8b 100644 --- a/src/modules/rag/rag.py +++ b/src/modules/rag/rag.py @@ -778,9 +778,7 @@ def __init__( self._max_history_turns = max_history_turns self.history: list[dict] = [] - async def process( # type: ignore[override] - self, data: RAGQuestion - ) -> AsyncGenerator[Token, None]: + async def process(self, data: RAGQuestion) -> AsyncGenerator[Token, None]: """ Called when a "question" event arrives through the event bus. Packages _user_id + question, sends to the stateless RAGHandle. diff --git a/src/modules/speech_to_text/events.py b/src/modules/speech_to_text/events.py index 17ab092..e53fb9d 100644 --- a/src/modules/speech_to_text/events.py +++ b/src/modules/speech_to_text/events.py @@ -3,17 +3,17 @@ import numpy as np -from src.core.events import EventData +from src.core.events import JsonEvent @dataclass -class Transcript(EventData): +class Transcript(JsonEvent): text: str end: bool @dataclass -class Voice(EventData): +class Voice(JsonEvent): data: Optional[np.ndarray] def summarize(self) -> str: diff --git a/src/modules/speech_to_text/microphone_vad.py b/src/modules/speech_to_text/microphone_vad.py index c1d4579..2142198 100644 --- a/src/modules/speech_to_text/microphone_vad.py +++ b/src/modules/speech_to_text/microphone_vad.py @@ -3,7 +3,7 @@ import numpy as np import webrtcvad -from src.core.events import RawBytes +from src.core.events import BytesEvent from src.core.module import Module from .events import Voice @@ -50,7 +50,7 @@ def __init__( self.vad = webrtcvad.Vad(vad_agressiveness) - async def process(self, data: RawBytes) -> Optional[Voice]: + async def process(self, data: BytesEvent) -> Optional[Voice]: if self.vad.is_speech(data.data, self.sample_rate) is True: self.silence_frames_count = 0 diff --git a/src/modules/text_to_speech/text_to_speech.py b/src/modules/text_to_speech/text_to_speech.py index 322a6cd..c080792 100644 --- a/src/modules/text_to_speech/text_to_speech.py +++ b/src/modules/text_to_speech/text_to_speech.py @@ -248,9 +248,7 @@ def __init__(self, _handle: handle.DeploymentHandle): # and silently drop trailing words). self._push_lock = asyncio.Lock() - async def process( # type: ignore[override] - self, token: Token - ) -> AsyncGenerator[Audio, None]: + async def process(self, token: Token) -> AsyncGenerator[Audio, None]: # Acquire BEFORE any await so lock-acquisition order matches token order. # Setup + push happen under the lock; only the first token of an # utterance goes on to drain/yield audio (outside the lock, so pushes of diff --git a/src/modules/utils/sender.py b/src/modules/utils/sender.py index 7e516e5..255678a 100644 --- a/src/modules/utils/sender.py +++ b/src/modules/utils/sender.py @@ -1,14 +1,10 @@ import logging import struct -from dataclasses import asdict -import numpy as np from fastapi import WebSocket from src.core.events import EventData from src.core.module import Module -from src.modules.gesture.events import Motion -from src.modules.text_to_speech.events import Audio logger = logging.getLogger("ray.serve") diff --git a/tests/core/test_events.py b/tests/core/test_events.py index c4485b1..9c2430b 100644 --- a/tests/core/test_events.py +++ b/tests/core/test_events.py @@ -1,11 +1,11 @@ import asyncio -import logging from dataclasses import dataclass import numpy as np import pytest -from src.core.events import EventData, EventGraph, _summarize +from src.core.bus import EventGraph +from src.core.events import EventData # --------------------------------------------------------------------------- # Helpers @@ -206,25 +206,3 @@ async def process(self, data): await graph.publish("a", DummyEvent(1)) await asyncio.wait_for(done.wait(), timeout=1) - - -# --------------------------------------------------------------------------- -# summarize() -# --------------------------------------------------------------------------- - - -def test_summarize_numpy(): - event = ArrayEvent(np.zeros((3, 4), dtype=np.float32)) - - text = _summarize(event) - - assert "shape=(3, 4)" in text - assert "float32" in text - - -def test_summarize_regular(): - event = DummyEvent(123) - - text = _summarize(event) - - assert "DummyEvent" in text