commit 469ce9c3e852cdbe733065cb3ebe3d568ab6ff2f Author: Michael Hansen Date: Tue Aug 5 13:46:07 2025 -0500 Initial commit diff --git a/hav_sat/__init__.py b/hav_sat/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/hav_sat/__main__.py b/hav_sat/__main__.py new file mode 100644 index 0000000..9faa68c --- /dev/null +++ b/hav_sat/__main__.py @@ -0,0 +1,382 @@ +#!/usr/bin/env python3 + +import asyncio +import argparse +import logging +import threading +import time +from collections.abc import Iterable +from enum import Enum, auto +from pathlib import Path +from dataclasses import dataclass +from typing import TYPE_CHECKING, Optional, List, Union, Dict +from queue import Queue + +from google.protobuf import message +import sounddevice as sd +from aioesphomeapi.api_pb2 import ( + DeviceInfoRequest, + DeviceInfoResponse, + ListEntitiesRequest, + ListEntitiesDoneResponse, + VoiceAssistantConfigurationRequest, + VoiceAssistantConfigurationResponse, + VoiceAssistantWakeWord, + SubscribeVoiceAssistantRequest, + VoiceAssistantEventResponse, + VoiceAssistantRequest, + VoiceAssistantAudio, + ListEntitiesMediaPlayerResponse, + MediaPlayerStateResponse, + MediaPlayerCommandRequest, + SubscribeHomeAssistantStatesRequest, + VoiceAssistantAnnounceFinished, + VoiceAssistantAnnounceRequest, + VoiceAssistantTimerEventResponse, +) +from aioesphomeapi.model import ( + VoiceAssistantFeature, + VoiceAssistantEventType, + VoiceAssistantTimerEventType, +) + +from .api_server import APIServer +from .microwakeword import MicroWakeWord +from .entity import ESPHomeEntity, MediaPlayerEntity +from .util import get_mac, call_all +from .mpv_player import MpvMediaPlayer + +_LOGGER = logging.getLogger(__name__) + + +class Sound(Enum): + WAKEUP = auto() + TIMER_FINISHED = auto() + + +_DIR = Path(__file__).parent +_SOUNDS_DIR = _DIR / "sounds" + +DEFAULT_SOUNDS: Dict[Sound, Path] = { + Sound.WAKEUP: _SOUNDS_DIR / "wake_word_triggered.flac", + Sound.TIMER_FINISHED: _SOUNDS_DIR / "timer_finished.flac", +} + + +@dataclass +class ServerState: + name: str + mac_address: str + audio_queue: "Queue[Optional[bytes]]" + entities: List[ESPHomeEntity] + wake_word: MicroWakeWord + stop_word: MicroWakeWord + music_player: MpvMediaPlayer + announce_player: MpvMediaPlayer + tts_player: MpvMediaPlayer + media_player_entity: Optional[MediaPlayerEntity] = None + satellite: "Optional[VoiceSatelliteProtocol]" = None + + +class VoiceSatelliteProtocol(APIServer): + + def __init__(self, state: ServerState) -> None: + super().__init__(state.name) + + self.state = state + self.state.satellite = self + + if self.state.media_player_entity is None: + self.state.media_player_entity = MediaPlayerEntity( + server=self, + key=len(state.entities), + name="Media Player", + object_id="hav_sat_media_player", + music_player=state.music_player, + announce_player=state.announce_player, + ) + self.state.entities.append(self.state.media_player_entity) + + self._is_streaming_audio = False + self._tts_url: Optional[str] = None + self._tts_played = False + self._continue_conversation = False + self._timer_finished = False + + def handle_voice_event( + self, event_type: VoiceAssistantEventType, data: Dict[str, str] + ) -> None: + _LOGGER.debug("%s: %s", event_type.name, data) + + if event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_START: + self._tts_url = data.get("url") + self._tts_played = False + self._continue_conversation = False + elif event_type in ( + VoiceAssistantEventType.VOICE_ASSISTANT_STT_VAD_END, + VoiceAssistantEventType.VOICE_ASSISTANT_STT_END, + ): + self._is_streaming_audio = False + elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_PROGRESS: + if data.get("tts_start_streaming") == "1": + # Start streaming early + self.play_tts() + elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_END: + if data.get("continue_conversation") == "1": + self._continue_conversation + elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_TTS_END: + self._tts_url = data.get("url") + self.play_tts() + elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_END: + self._is_streaming_audio = False + if not self._tts_played: + self._tts_finished() + + # TODO: handle error + + def handle_timer_event( + self, + event_type: VoiceAssistantTimerEventType, + msg: VoiceAssistantTimerEventResponse, + ) -> None: + if event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_FINISHED: + if not self._timer_finished: + self.state.stop_word.is_active = True + self._timer_finished = True + self._play_timer_finished() + + def handle_message(self, msg: message.Message) -> Iterable[message.Message]: + if isinstance(msg, VoiceAssistantEventResponse): + # Pipeline event + data: Dict[str, str] = {} + for arg in msg.data: + data[arg.name] = arg.value + + self.handle_voice_event(VoiceAssistantEventType(msg.event_type), data) + elif isinstance(msg, VoiceAssistantAnnounceRequest): + _LOGGER.debug("Announcing: %s", msg.text) + + assert self.state.media_player_entity is not None + + urls = [] + if msg.preannounce_media_id: + urls.append(msg.preannounce_media_id) + + urls.append(msg.media_id) + + self.state.stop_word.is_active = True + self._continue_conversation = msg.start_conversation + + self.duck() + yield from self.state.media_player_entity.play( + urls, announcement=True, done_callback=self._tts_finished + ) + elif isinstance(msg, VoiceAssistantTimerEventResponse): + self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg) + elif isinstance(msg, DeviceInfoRequest): + yield DeviceInfoResponse( + uses_password=False, + name=self.state.name, + mac_address=self.state.mac_address, + voice_assistant_feature_flags=( + VoiceAssistantFeature.VOICE_ASSISTANT + | VoiceAssistantFeature.API_AUDIO + | VoiceAssistantFeature.ANNOUNCE + | VoiceAssistantFeature.START_CONVERSATION + | VoiceAssistantFeature.TIMERS + ), + ) + elif isinstance( + msg, + ( + ListEntitiesRequest, + SubscribeHomeAssistantStatesRequest, + MediaPlayerCommandRequest, + ), + ): + for entity in self.state.entities: + yield from entity.handle_message(msg) + + if isinstance(msg, ListEntitiesRequest): + yield ListEntitiesDoneResponse() + elif isinstance(msg, VoiceAssistantConfigurationRequest): + yield VoiceAssistantConfigurationResponse( + available_wake_words=[ + VoiceAssistantWakeWord( + id=self.state.wake_word.id, + wake_word=self.state.wake_word.wake_word, + trained_languages=self.state.wake_word.trained_languages, + ) + ], + active_wake_words=[self.state.wake_word.id], + max_active_wake_words=1, + ) + _LOGGER.debug("Registered voice assistant") + + def handle_audio(self, audio_chunk: bytes) -> None: + + if not self._is_streaming_audio: + return + + self.send_messages([VoiceAssistantAudio(data=audio_chunk)]) + + def wakeup(self) -> None: + if self._timer_finished: + # Stop timer instead + self._timer_finished = False + return + + wake_word_phrase = self.state.wake_word.wake_word + _LOGGER.debug("Detected wake word: %s", wake_word_phrase) + self.send_messages( + [VoiceAssistantRequest(start=True, wake_word_phrase=wake_word_phrase)] + ) + self.duck() + self._is_streaming_audio = True + self.state.tts_player.play(str(DEFAULT_SOUNDS[Sound.WAKEUP])) + + def stop(self) -> None: + self.state.stop_word.is_active = False + + if self._timer_finished: + self._timer_finished = False + self.state.tts_player.stop() + self.state.music_player.resume() + else: + _LOGGER.debug("TTS response stopped manually") + self.state.tts_player.stop() + self.state.announce_player.stop() + self._tts_finished() + + def play_tts(self) -> None: + if (not self._tts_url) or self._tts_played: + return + + self._tts_played = True + _LOGGER.debug("Playing TTS response: %s", self._tts_url) + + self.state.stop_word.is_active = True + self.state.tts_player.play(self._tts_url, done_callback=self._tts_finished) + + def duck(self) -> None: + _LOGGER.debug("Ducking music") + self.state.music_player.set_volume(50) + + def unduck(self) -> None: + _LOGGER.debug("Unducking music") + self.state.music_player.set_volume(100) + + def _tts_finished(self) -> None: + self.state.stop_word.is_active = False + self.send_messages([VoiceAssistantAnnounceFinished()]) + + if self._continue_conversation: + self.send_messages([VoiceAssistantRequest(start=True)]) + self._is_streaming_audio = True + _LOGGER.debug("Continuing conversation") + else: + self.unduck() + + _LOGGER.debug("TTS response finished") + + def _play_timer_finished(self) -> None: + if not self._timer_finished: + return + + self.state.tts_player.play( + str(DEFAULT_SOUNDS[Sound.TIMER_FINISHED]), + done_callback=lambda: call_all( + lambda: time.sleep(1.0), self._play_timer_finished + ), + ) + + +def process_audio(state: ServerState): + + try: + while True: + audio_chunk = state.audio_queue.get() + if audio_chunk is None: + break + + if state.satellite is None: + continue + + try: + state.satellite.handle_audio(audio_chunk) + + if state.wake_word.is_active and state.wake_word.process_streaming( + audio_chunk + ): + state.satellite.wakeup() + + if state.stop_word.is_active and state.stop_word.process_streaming( + audio_chunk + ): + state.satellite.stop() + except: + _LOGGER.exception("Unexpected error handling audio") + + except: + _LOGGER.exception("Unexpected error processing audio") + + +async def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--audio-input-device", default="default") + parser.add_argument("--audio-output-device") + args = parser.parse_args() + + logging.basicConfig(level=logging.DEBUG) + + state = ServerState( + name="test", + mac_address=get_mac(), + audio_queue=Queue(), + entities=[], + wake_word=MicroWakeWord.from_config("wakewords/okay_nabu.json"), + stop_word=MicroWakeWord.from_config("wakewords/stop.json"), + music_player=MpvMediaPlayer(device=args.audio_output_device), + announce_player=MpvMediaPlayer(device=args.audio_output_device), + tts_player=MpvMediaPlayer(device=args.audio_output_device), + ) + + process_audio_thread = threading.Thread( + target=process_audio, args=(state,), daemon=True + ) + process_audio_thread.start() + + def sd_callback(indata, frames, time, status): + state.audio_queue.put_nowait(bytes(indata)) + + # for idx, dev in enumerate(sd.query_devices()): + # _LOGGER.debug(f"{idx}: {dev['name']}") + + loop = asyncio.get_running_loop() + server = await loop.create_server( + lambda: VoiceSatelliteProtocol(state), host="127.0.0.1", port=6053 + ) + + try: + with sd.RawInputStream( + samplerate=16000, + blocksize=1024, + device=args.audio_input_device, + dtype="int16", + channels=1, + callback=sd_callback, + ): + async with server: + _LOGGER.info("Server started") + await server.serve_forever() + except KeyboardInterrupt: + pass + finally: + state.audio_queue.put_nowait(None) + process_audio_thread.join() + + _LOGGER.debug("Server stopped") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/hav_sat/api_server.py b/hav_sat/api_server.py new file mode 100644 index 0000000..3a36f79 --- /dev/null +++ b/hav_sat/api_server.py @@ -0,0 +1,181 @@ +"""Partial ESPHome server implementation.""" + +import logging +import asyncio +from typing import Optional, TYPE_CHECKING, List, Union + + +from abc import abstractmethod +from collections.abc import Iterable +from aioesphomeapi._frame_helper.packets import make_plain_text_packets +from google.protobuf import message +from aioesphomeapi.core import MESSAGE_TYPE_TO_PROTO +from aioesphomeapi.api_pb2 import ( + HelloRequest, + HelloResponse, + ConnectRequest, + ConnectResponse, + DisconnectRequest, + DisconnectResponse, + PingRequest, + PingResponse, +) + +PROTO_TO_MESSAGE_TYPE = {v: k for k, v in MESSAGE_TYPE_TO_PROTO.items()} + +_LOGGER = logging.getLogger(__name__) + + +class APIServer(asyncio.Protocol): + + def __init__(self, name: str) -> None: + self.name = name + + self._buffer: Optional[bytes] = None + self._buffer_len: int = 0 + self._pos: int = 0 + self._transport = None + self._writelines = None + + @abstractmethod + def handle_message(self, msg: message.Message) -> Iterable[message.Message]: + pass + + def process_packet(self, msg_type: int, packet_data: bytes) -> None: + msg_class = MESSAGE_TYPE_TO_PROTO[msg_type] + msg_inst = msg_class.FromString(packet_data) + + if isinstance(msg_inst, HelloRequest): + self.send_messages( + [ + HelloResponse( + api_version_major=1, + api_version_minor=10, + name=self.name, + ) + ] + ) + return + + if isinstance(msg_inst, ConnectRequest): + self.send_messages([ConnectResponse()]) + elif isinstance(msg_inst, DisconnectRequest): + self.send_messages([DisconnectResponse()]) + _LOGGER.debug("Disconnect requested") + if self.transport: + self.transport.close() + self.transport = None + self.writelines = None + elif isinstance(msg_inst, PingRequest): + self.send_messages([PingResponse()]) + elif msgs := self.handle_message(msg_inst): + if isinstance(msgs, message.Message): + msgs = [msgs] + + self.send_messages(msgs) + + def send_messages(self, msgs: List[message.Message]): + if self.writelines is None: + return + + packets = [ + (PROTO_TO_MESSAGE_TYPE[msg.__class__], msg.SerializeToString()) + for msg in msgs + ] + packet_bytes = make_plain_text_packets(packets) + self.writelines(packet_bytes) + + def connection_made(self, transport) -> None: + self.transport = transport + self.writelines = self.transport.writelines + + def data_received(self, data: bytes): + if self._buffer is None: + self._buffer = data + self._buffer_len = len(data) + else: + self._buffer += data + self._buffer_len += len(data) + + while self._buffer_len >= 3: + self._pos = 0 + # Read preamble, which should always 0x00 + if (preamble := self._read_varuint()) != 0x00: + _LOGGER.error("Incorrect preamble: %s", preamble) + return + + if (length := self._read_varuint()) == -1: + _LOGGER.error("Incorrect length: %s", length) + return + + if (msg_type := self._read_varuint()) == -1: + _LOGGER.error("Incorrect message type: %s", msg_type) + return + + if length == 0: + # Empty message (allowed) + self._remove_from_buffer() + self.process_packet(msg_type, b"") + continue + + if (packet_data := self._read(length)) is None: + return + + self._remove_from_buffer() + self.process_packet(msg_type, packet_data) + + def _read(self, length: int) -> bytes | None: + """Read exactly length bytes from the buffer or None if all the bytes are not yet available.""" + new_pos = self._pos + length + if self._buffer_len < new_pos: + return None + original_pos = self._pos + self._pos = new_pos + if TYPE_CHECKING: + assert self._buffer is not None, "Buffer should be set" + cstr = self._buffer + # Important: we must keep the bounds check (self._buffer_len < new_pos) + # above to verify we never try to read past the end of the buffer + return cstr[original_pos:new_pos] + + def connection_lost(self, exc): + self.transport = None + self.writelines = None + _LOGGER.debug("Connection closed") + + def _read_varuint(self) -> int: + """Read a varuint from the buffer or -1 if the buffer runs out of bytes.""" + if not self._buffer: + return -1 + + result = 0 + bitpos = 0 + cstr = self._buffer + while self._buffer_len > self._pos: + val = cstr[self._pos] + self._pos += 1 + result |= (val & 0x7F) << bitpos + if (val & 0x80) == 0: + return result + bitpos += 7 + return -1 + + def _remove_from_buffer(self) -> None: + """Remove data from the buffer.""" + end_of_frame_pos = self._pos + self._buffer_len -= end_of_frame_pos + if self._buffer_len == 0: + # This is the best case scenario, we can just set the buffer to None + # and don't have to copy the data. This is the most common case as well. + self._buffer = None + return + if TYPE_CHECKING: + assert self._buffer is not None, "Buffer should be set" + # This is the worst case scenario, we have to copy the data + # and can't just use the buffer directly. This should only happen + # when we read multiple frames at once because the event loop + # is blocked and we cannot pull the data out of the buffer fast enough. + cstr = self._buffer + # Important: we must use the explicit length for the slice + # since Cython will stop at any '\0' character if we don't + self._buffer = cstr[end_of_frame_pos : self._buffer_len + end_of_frame_pos] diff --git a/hav_sat/entity.py b/hav_sat/entity.py new file mode 100644 index 0000000..2788512 --- /dev/null +++ b/hav_sat/entity.py @@ -0,0 +1,132 @@ +from abc import abstractmethod +from collections.abc import Iterable +from typing import Callable, Optional, List, Union + +from aioesphomeapi import MediaPlayerState +from google.protobuf import message +from aioesphomeapi.api_pb2 import ( + ListEntitiesRequest, + MediaPlayerStateResponse, + ListEntitiesMediaPlayerResponse, + SubscribeHomeAssistantStatesRequest, + MediaPlayerCommandRequest, +) +from aioesphomeapi.model import MediaPlayerCommand, MediaPlayerState + +from .api_server import APIServer +from .mpv_player import MpvMediaPlayer +from .util import call_all + + +class ESPHomeEntity: + def __init__(self, server: APIServer) -> None: + self.server = server + + @abstractmethod + def handle_message(self, msg: message.Message) -> Iterable[message.Message]: + pass + + +# ----------------------------------------------------------------------------- + + +class MediaPlayerEntity(ESPHomeEntity): + def __init__( + self, + server: APIServer, + key: int, + name: str, + object_id: str, + music_player: MpvMediaPlayer, + announce_player: MpvMediaPlayer, + ) -> None: + ESPHomeEntity.__init__(self, server) + + self.key = key + self.name = name + self.object_id = object_id + self.state = MediaPlayerState.IDLE + self.volume = 1.0 + self.muted = False + self.music_player = music_player + self.announce_player = announce_player + + def play( + self, + url: Union[str, List[str]], + announcement: bool = False, + done_callback: Optional[Callable[[], None]] = None, + ) -> Iterable[message.Message]: + if announcement: + if self.music_player.is_playing: + # Announce, resume music + self.music_player.pause() + self.announce_player.play( + url, + done_callback=lambda: call_all( + self.music_player.resume, done_callback + ), + ) + else: + # Announce, idle + self.announce_player.play( + url, + done_callback=lambda: call_all( + self.server.send_messages( + [self._update_state(MediaPlayerState.IDLE)] + ), + done_callback, + ), + ) + else: + # Music + self.music_player.play( + url, + done_callback=lambda: call_all( + self.server.send_messages( + [self._update_state(MediaPlayerState.IDLE)] + ), + done_callback, + ), + ) + + yield self._update_state(MediaPlayerState.PLAYING) + + def handle_message(self, msg: message.Message) -> Iterable[message.Message]: + if isinstance(msg, MediaPlayerCommandRequest) and (msg.key == self.key): + if msg.has_media_url: + announcement = msg.has_announcement and msg.announcement + yield from self.play(msg.media_url, announcement=announcement) + elif msg.has_command: + if msg.command == MediaPlayerCommand.PAUSE: + if self.music_player.is_playing: + self.music_player.pause() + yield self._update_state(MediaPlayerState.PAUSED) + elif msg.command == MediaPlayerCommand.PLAY: + if not self.music_player.is_playing: + self.player.resume() + yield self._update_state(MediaPlayerState.PLAYING) + # TODO + # elif msg.has_volume: + # self.player.set_volume() + elif isinstance(msg, ListEntitiesRequest): + yield ListEntitiesMediaPlayerResponse( + object_id=self.object_id, + key=self.key, + name=self.name, + supports_pause=True, + ) + elif isinstance(msg, SubscribeHomeAssistantStatesRequest): + yield self._get_state_message() + + def _update_state(self, new_state: MediaPlayerState) -> MediaPlayerStateResponse: + self.state = new_state + return self._get_state_message() + + def _get_state_message(self) -> MediaPlayerStateResponse: + return MediaPlayerStateResponse( + key=self.key, + state=self.state, + volume=self.volume, + muted=self.muted, + ) diff --git a/hav_sat/microwakeword.py b/hav_sat/microwakeword.py new file mode 100644 index 0000000..0fd8219 --- /dev/null +++ b/hav_sat/microwakeword.py @@ -0,0 +1,270 @@ +import ctypes +import json +import numpy as np +import statistics +from collections import deque +from pathlib import Path +from typing import Union, List, Deque + +from pymicro_features import MicroFrontend + +_SAMPLES_PER_SECOND = 16000 +_SAMPLES_PER_CHUNK = 160 # 10ms +_BYTES_PER_SAMPLE = 2 # 16-bit +_BYTES_PER_CHUNK = _SAMPLES_PER_CHUNK * _BYTES_PER_SAMPLE +_SECONDS_PER_CHUNK = _SAMPLES_PER_CHUNK / _SAMPLES_PER_SECOND +_STRIDE = 3 +_DEFAULT_REFRACTORY = 2 # seconds + + +class TfLiteQuantizationParams(ctypes.Structure): + _fields_ = [("scale", ctypes.c_float), ("zero_point", ctypes.c_int32)] + + +class MicroWakeWord: + def __init__( + self, + id: str, + wake_word: str, + tflite_model: Union[str, Path], + probability_cutoff: float, + sliding_window_size: int, + refractory_seconds: float, + trained_languages: List[str], + ): + self.id = id + self.wake_word = wake_word + self.tflite_model = tflite_model + self.probability_cutoff = probability_cutoff + self.sliding_window_size = sliding_window_size + self.refractory_seconds = refractory_seconds + self.trained_languages = trained_languages + + self.is_active = True + + # Load the shared library + self.lib = ctypes.cdll.LoadLibrary( + str(Path("libtensorflowlite_c.so").resolve()) + ) + + # Define required argument/return types for C API + self.lib.TfLiteModelCreateFromFile.argtypes = [ctypes.c_char_p] + self.lib.TfLiteModelCreateFromFile.restype = ctypes.c_void_p + + self.lib.TfLiteInterpreterCreate.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + self.lib.TfLiteInterpreterCreate.restype = ctypes.c_void_p + + self.lib.TfLiteInterpreterAllocateTensors.argtypes = [ctypes.c_void_p] + self.lib.TfLiteInterpreterInvoke.argtypes = [ctypes.c_void_p] + + self.lib.TfLiteInterpreterGetInputTensor.argtypes = [ + ctypes.c_void_p, + ctypes.c_int, + ] + self.lib.TfLiteInterpreterGetInputTensor.restype = ctypes.c_void_p + + self.lib.TfLiteInterpreterGetOutputTensor.argtypes = [ + ctypes.c_void_p, + ctypes.c_int, + ] + self.lib.TfLiteInterpreterGetOutputTensor.restype = ctypes.c_void_p + + self.lib.TfLiteTensorCopyFromBuffer.argtypes = [ + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_size_t, + ] + self.lib.TfLiteTensorCopyToBuffer.argtypes = [ + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_size_t, + ] + + self.lib.TfLiteTensorType.restype = ctypes.c_int + + self.lib.TfLiteTensorByteSize.argtypes = [ctypes.c_void_p] + self.lib.TfLiteTensorByteSize.restype = ctypes.c_size_t + + self.lib.TfLiteTensorQuantizationParams.restype = ctypes.Structure + + self.lib.TfLiteTensorQuantizationParams.argtypes = [ctypes.c_void_p] + self.lib.TfLiteTensorQuantizationParams.restype = TfLiteQuantizationParams + + # Load the model and create interpreter + self.model_path = str(Path(tflite_model).resolve()).encode("utf-8") + self._load_model() + + self._frontend = MicroFrontend() + self._features: List[np.ndarray] = [] + self._probabilities: Deque[float] = deque(maxlen=self.sliding_window_size) + self._audio_buffer = bytes() + self._ignore_seconds: float = 0 + + def _load_model(self) -> None: + self.model = self.lib.TfLiteModelCreateFromFile(self.model_path) + self.interpreter = self.lib.TfLiteInterpreterCreate(self.model, None) + self.lib.TfLiteInterpreterAllocateTensors(self.interpreter) + + # Access input and output tensor + self.input_tensor = self.lib.TfLiteInterpreterGetInputTensor( + self.interpreter, 0 + ) + self.output_tensor = self.lib.TfLiteInterpreterGetOutputTensor( + self.interpreter, 0 + ) + + # Get quantization parameters + input_q = self.lib.TfLiteTensorQuantizationParams(self.input_tensor) + output_q = self.lib.TfLiteTensorQuantizationParams(self.output_tensor) + + self.input_scale, self.input_zero_point = input_q.scale, input_q.zero_point + self.output_scale, self.output_zero_point = output_q.scale, output_q.zero_point + + def reset(self) -> None: + """Reload model and clear state. + + This must be done between audio clips when not streaming. + """ + self._audio_buffer = bytes() + self._features.clear() + self._probabilities.clear() + self._ignore_seconds = 0 + + # Clear out residual features + self._frontend = MicroFrontend() + + # Need to reload model to reset intermediary results + # reset_all_variables() doesn't work. + self._load_model() + + @property + def samples_per_chunk(self) -> int: + """Number of samples in a streaming audio chunk.""" + return _SAMPLES_PER_CHUNK + + @property + def bytes_per_chunk(self) -> int: + """Number of bytes in a streaming audio chunk. + + Assumes 16-bit mono samples at 16Khz. + """ + return _BYTES_PER_CHUNK + + @staticmethod + def from_config( + config_path: Union[str, Path], + refractory_seconds: float = _DEFAULT_REFRACTORY, + ) -> "MicroWakeWord": + """Load a microWakeWord model from a JSON config file. + + Parameters + ---------- + config_path: str or Path + Path to JSON configuration file + refractory_seconds: float + Number of seconds to ignore after detection + """ + config_path = Path(config_path) + with open(config_path, "r", encoding="utf-8") as config_file: + config = json.load(config_file) + + micro_config = config["micro"] + + return MicroWakeWord( + id=Path(config["model"]).stem, + wake_word=config["wake_word"], + tflite_model=config_path.parent / config["model"], + probability_cutoff=micro_config["probability_cutoff"], + sliding_window_size=micro_config["sliding_window_size"], + refractory_seconds=refractory_seconds, + trained_languages=micro_config.get("trained_languages", []), + ) + + def process_streaming(self, audio_bytes: bytes) -> bool: + """Process a chunk of audio in streaming mode. + + Parameters + ---------- + audio_bytes: bytes + Raw 16-bit mono audio samples at 16Khz + + Returns True if the wake word was detected. + """ + self._audio_buffer += audio_bytes + + if len(self._audio_buffer) < _BYTES_PER_CHUNK: + # Not enough audio to get features + return False + + detected = False + audio_buffer_idx = 0 + while (audio_buffer_idx + _BYTES_PER_CHUNK) <= len(self._audio_buffer): + # Process chunk + chunk_bytes = self._audio_buffer[ + audio_buffer_idx : audio_buffer_idx + _BYTES_PER_CHUNK + ] + frontend_result = self._frontend.ProcessSamples(chunk_bytes) + audio_buffer_idx += frontend_result.samples_read * _BYTES_PER_SAMPLE + self._ignore_seconds = max(0, self._ignore_seconds - _SECONDS_PER_CHUNK) + + if not frontend_result.features: + # Not enough audio for a full window + continue + + self._features.append( + np.array(frontend_result.features).reshape( + (1, 1, len(frontend_result.features)) + ) + ) + + if len(self._features) < _STRIDE: + # Not enough windows + continue + + # Allocate and quantize input data + quant_features = np.round( + np.concatenate(self._features, axis=1) / self.input_scale + + self.input_zero_point + ).astype(np.uint8) + + # Stride instead of rolling + self._features.clear() + + # Set tensor + quant_ptr = quant_features.ctypes.data_as(ctypes.c_void_p) + self.lib.TfLiteTensorCopyFromBuffer( + self.input_tensor, quant_ptr, quant_features.nbytes + ) + + # Run inference + self.lib.TfLiteInterpreterInvoke(self.interpreter) + + # Read output + output_bytes = self.lib.TfLiteTensorByteSize(self.output_tensor) + output_data = np.empty(output_bytes, dtype=np.uint8) + self.lib.TfLiteTensorCopyToBuffer( + self.output_tensor, + output_data.ctypes.data_as(ctypes.c_void_p), + output_bytes, + ) + + # Dequantize output + result = ( + output_data.astype(np.float32) - self.output_zero_point + ) * self.output_scale + + self._probabilities.append(result.item()) + + if len(self._probabilities) < self.sliding_window_size: + # Not enough probabilities + continue + + if statistics.mean(self._probabilities) > self.probability_cutoff: + if self._ignore_seconds <= 0: + detected = True + self._ignore_seconds = self.refractory_seconds + + # Remove processed audio + self._audio_buffer = self._audio_buffer[audio_buffer_idx:] + + return detected diff --git a/hav_sat/mpv_player.py b/hav_sat/mpv_player.py new file mode 100644 index 0000000..551f0c0 --- /dev/null +++ b/hav_sat/mpv_player.py @@ -0,0 +1,174 @@ +"""Media player using mpv in a subprocess.""" + +import subprocess +import logging +import socket +import os +import uuid +import threading +import json +import time +import tempfile +from collections.abc import Callable +from typing import Optional, Union, List, Any +from pathlib import Path + +_LOGGER = logging.getLogger(__name__) + + +class MpvMediaPlayer: + def __init__( + self, + device: Optional[str] = None, + socket_path: Optional[Union[str, Path]] = None, + ) -> None: + self.device = device + self.is_playing = False + + if socket_path: + self._socket_path = str(Path(socket_path).resolve()) + else: + base_path = ( + Path(os.environ.get("XDG_RUNTIME_DIR") or "/dev/shm") / "mpv-sockets" + ) + base_path.mkdir(parents=True, exist_ok=True) + self._socket_path = str(base_path / f"mpv-{uuid.uuid4().hex}.sock") + + self._playlist_file = tempfile.NamedTemporaryFile( + "w+", suffix=".m3u8", encoding="utf-8" + ) + + cmd = [ + "mpv", + f"--input-ipc-server={self._socket_path}", + "--no-video", + "--quiet", + "--term-playing-msg=", + "--msg-level=all=no", + "--idle=yes", # keep alive + ] + if device: + cmd.append(f"--audio-device={device}") + + _LOGGER.debug(cmd) + + self._proc = subprocess.Popen(cmd) + self._socket_connected = False + + self._done_callback: Optional[Callable[[], None]] = None + + threading.Thread(target=self._read_socket, daemon=True).start() + + def play( + self, + url: Union[str, List[str]], + done_callback: Optional[Callable[[], None]] = None, + ) -> None: + if isinstance(url, str): + url = [url] + + self._done_callback = done_callback + self.is_playing = True + + if len(url) == 1: + self.send_command(["loadfile", url[0], "replace"]) + else: + self._playlist_file.seek(0) + for item in url: + print(item, file=self._playlist_file, flush=True) + + self.send_command(["loadlist", self._playlist_file.name, "replace"]) + + def stop(self) -> None: + self.send_command(["stop"]) + self.is_playing = False + + def pause(self) -> None: + if not self.is_playing: + return + + self.send_command(["set_property", "pause", True]) + self.is_playing = False + + def resume(self) -> None: + if self.is_playing: + return + + self.send_command(["set_property", "pause", False]) + self.is_playing = True + + def mute(self) -> None: + self.send_command(["set_property", "mute", True]) + + def unmute(self) -> None: + self.send_command(["set_property", "mute", False]) + + def set_volume(self, volume: int) -> None: + self.send_command(["set_property", "volume", volume]) + + def send_command(self, cmd: List[Any]): + if not self._socket_connected: + _LOGGER.warning("Socket not connected: %s", cmd) + return + + payload = json.dumps({"command": cmd}).encode("utf-8") + b"\n" + + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as mpv_socket: + mpv_socket.connect(self._socket_path) + mpv_socket.sendall(payload) + + _LOGGER.debug("Sent command: %s", payload) + + def _read_socket(self): + # Connect to socket + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as mpv_socket: + timeout = time.monotonic() + 5 + while time.monotonic() < timeout: + try: + mpv_socket.connect(self._socket_path) + self._socket_connected = True + break + except (ConnectionRefusedError, FileNotFoundError, OSError): + pass + + if not self._socket_connected: + _LOGGER.warning("Failed to connect to socket: %s", self._socket_path) + return + + _LOGGER.debug("Connected to socket: %s", self._socket_path) + + # Read from socket + mpv_buffer = bytes() + + try: + while True: + chunk = mpv_socket.recv(4096) + if not chunk: + break + + mpv_buffer += chunk + while b"\n" in mpv_buffer: + line, mpv_buffer = mpv_buffer.split(b"\n", 1) + event = json.loads(line.decode()) + _LOGGER.debug("Received event: %s", event) + + event_type = event.get("event") + + if event_type == "start-file": + self.is_playing = True + continue + + if event_type != "end-file": + continue + + self.is_playing = False + + if not self._done_callback: + continue + + try: + self._done_callback() + except: + _LOGGER.exception("Unexpected error in callback") + except: + _LOGGER.exception("Unexpected error reading MPV socket") diff --git a/hav_sat/util.py b/hav_sat/util.py new file mode 100644 index 0000000..8e247bd --- /dev/null +++ b/hav_sat/util.py @@ -0,0 +1,16 @@ +"""Utility methods.""" + +from collections.abc import Callable +import uuid +from typing import Optional + + +def get_mac() -> str: + mac = uuid.getnode() + mac_str = ":".join(f"{(mac >> i) & 0xff:02x}" for i in range(40, -1, -8)) + return mac_str + + +def call_all(*callables: Optional[Callable[[], None]]) -> None: + for item in filter(None, callables): + item()