refactor code into modules

echo_crafter/config/__init__.py

@@ -0,0 +1,28 @@
+import os
+from pathlib import Path
+from typing import TypedDict
+
+PROJECT_ROOT=Path(os.getenv('EC_ROOT') or '')
+DATA_DIR=Path(os.getenv('EC_DATA_DIR') or '')
+
+class _Config(TypedDict):
+    CHEETAH_MODEL_FILE: str
+    PICOVOICE_API_KEY: str
+    FRAME_LENGTH: int
+    ENDPOINT_DURATION_SEC: float
+    TRANSCRIPT_BEGIN_WAV: str
+    TRANSCRIPT_SUCCESS_WAV: str
+    SOCKET_PATH: str
+
+
+Config: _Config = dict(
+    CHEETAH_MODEL_FILE=str(DATA_DIR/"speech-command-cheetah-v1.pv"),
+    PICOVOICE_API_KEY=os.environ.get('PICOVOICE_API_KEY', ''),
+    FRAME_LENGTH=512,
+    ENDPOINT_DURATION_SEC=1.3,
+    TRANSCRIPT_BEGIN_WAV=str(DATA_DIR/"transcript_begin.wav"),
+    TRANSCRIPT_SUCCESS_WAV=str(DATA_DIR/"transcript_success.wav"),
+    SOCKET_PATH=str(Path(os.getenv('EC_SOCKET_FILE') or '/tmp/echo-crafter.sock'))
+)
+
+__all__ = ['Config']

echo_crafter/listener/listener_with_wake_word.py

@@ -0,0 +1,40 @@
+from echo_crafter.listener.utils import (
+    socket_connection,
+    Microphone
+)
+from echo_crafter.logger import setup_logger
+from echo_crafter.config import Config
+
+import subprocess
+import traceback
+
+
+def play_sound(wav_file):
+    """Play a ding sound to indicate that the wake word was detected."""
+    subprocess.Popen(["aplay", "-q", str(wav_file)])
+
+
+def main():
+    """Upon detection of a wake word, transcribe speech until endpoint is detected."""
+    logger = setup_logger()
+
+    with Microphone() as mic:
+        try:
+            while True:
+                mic.wait_for_wake_word()
+                play_sound(Config['TRANSCRIPT_BEGIN_WAV'])
+
+                with socket_connection(Config['SOCKET_PATH']) as client:
+                    mic.process_and_transmit_utterance(client)
+                    play_sound(Config['TRANSCRIPT_SUCCESS_WAV'])
+
+        except KeyboardInterrupt:
+            pass
+
+        except Exception as e:
+            logger.error(f"An error occured:\n{e}")
+            logger.error(traceback.format_exc())
+
+
+if __name__ == '__main__':
+    main()

listener/socket_read.py → echo_crafter/listener/socket_read.py

@@ -1,26 +1,18 @@
-#!/usr/bin/env python3
+from echo_crafter.config import Config
 
-import os
 from pathlib import Path
 import socket
 import subprocess
 
-SOCKET_PATH = Path(os.getenv('XDG_RUNTIME_DIR')) / "transcription"
-n_connections = 0
-
-
 def handle_partial_transcript(partial_transcript):
     """Send partial transcript to the active window."""
-    subprocess.run(
+    subprocess.Popen(
         ['xdotool', 'type', '--clearmodifiers', '--delay', '0', partial_transcript]
     )
 
 
-def handle_client(client_socket, client_address):
+def handle_client(client_socket, _):
     """Handle a client connection."""
-    global n_connections
-    n_connections += 1
-
     try:
         while True:
             partial_transcript = client_socket.recv(1024)
@@ -35,28 +27,25 @@ def handle_client(client_socket, client_address):
                 handle_partial_transcript(partial_transcript_s)
     finally:
         client_socket.close()
-        n_connections -= 1
 
 
 def main():
     """Listen for connections and handle them."""
+    SOCKET_PATH = Path(Config['SOCKET_PATH'])
     try:
         SOCKET_PATH.unlink()
     except OSError:
         if SOCKET_PATH.exists():
             raise
 
     with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as server_socket:
-        server_socket.bind(str(SOCKET_PATH))
+        server_socket.bind(Config['SOCKET_PATH'])
         server_socket.listen(1)
 
         try:
             while True:
                 client_socket, client_address = server_socket.accept()
-                if n_connections > 0:
-                    client_socket.close()
-                else:
-                    handle_client(client_socket, client_address)
+                handle_client(client_socket, client_address)
         finally:
             SOCKET_PATH.unlink()
 

echo_crafter/listener/utils/__init__.py

@@ -0,0 +1,7 @@
+from .porcupine import Microphone
+from .sockets import socket_connection
+
+__all__ = [
+    'socket_connection',
+    'Microphone'
+]

echo_crafter/listener/utils/porcupine.py

@@ -0,0 +1,88 @@
+from echo_crafter.config import Config
+
+from contextlib import contextmanager
+import pvcheetah
+import pvporcupine
+import pvrecorder
+
+@contextmanager
+def porcupine_context_manager():
+    """Create a Porcupine instance and yield it. Delete the instance upon exit."""
+    porcupine_instance = None
+    try:
+        porcupine_instance = pvporcupine.create(
+            keywords=['computer'],
+            sensitivities=[0.1],
+            access_key=Config['PICOVOICE_API_KEY']
+        )
+        yield porcupine_instance
+    finally:
+        if porcupine_instance is not None:
+            porcupine_instance.delete()
+
+
+@contextmanager
+def cheetah_context_manager():
+    """Create a Cheetah instance and yield it. Delete the instance upon exit."""
+    cheetah_instance = None
+    try:
+        cheetah_instance = pvcheetah.create(
+            access_key=Config['PICOVOICE_API_KEY'],
+            endpoint_duration_sec=Config['ENDPOINT_DURATION_SEC'],
+            model_path=Config['CHEETAH_MODEL_FILE']
+        )
+        yield cheetah_instance
+    finally:
+        if cheetah_instance is not None:
+            cheetah_instance.delete()
+
+
+@contextmanager
+def recorder_context_manager():
+    """Create a PvRecorder instance and yield it. Delete the instance upon exit."""
+    recorder_instance = None
+    try:
+        recorder_instance = pvrecorder.PvRecorder(
+            frame_length=Config['FRAME_LENGTH']
+        )
+        recorder_instance.start()
+        yield recorder_instance
+    finally:
+        if recorder_instance is not None:
+            recorder_instance.delete()
+
+
+class _Microphone:
+    """A context manager for the recorder_context_manager."""
+
+    def __init__(self, recorder, porcupine, cheetah):
+        self.recorder = recorder
+        self.porcupine = porcupine
+        self.cheetah = cheetah
+
+    def wait_for_wake_word(self):
+        """Wait for the wake word to be detected."""
+        while True:
+            pcm_frame = self.recorder.read()
+            keyword_index = self.porcupine.process(pcm_frame)
+            if keyword_index >= 0:
+                break
+
+    def process_and_transmit_utterance(self, client):
+        """Process the utterance and transmit the partial transcript to the client."""
+        while True:
+            pcm_frame = self.recorder.read()
+            partial_transcript, is_endpoint = self.cheetah.process(pcm_frame)
+            client.sendall((partial_transcript).encode())
+            if is_endpoint:
+                partial_transcript = self.cheetah.flush()
+                client.sendall((partial_transcript + 'STOP').encode())
+                break
+
+
+@contextmanager
+def Microphone():
+    """Create a Microphone instance and yield it. Delete the instance upon exit."""
+    with recorder_context_manager() as recorder, porcupine_context_manager() as porcupine, cheetah_context_manager() as cheetah:
+        mic = _Microphone(recorder, porcupine, cheetah)
+        yield mic

echo_crafter/listener/utils/sockets.py

@@ -0,0 +1,12 @@
+import socket
+import contextlib
+
+# Define a context manager for the socket connection
+@contextlib.contextmanager
+def socket_connection(socket_path):
+    client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        client.connect(socket_path)
+        yield client
+    finally:
+        client.close()

echo_crafter/logger/__init__.py

@@ -0,0 +1,45 @@
+"""Logging module for the Echo Crafter application."""
+
+import logging
+import json
+import time
+import os
+from pathlib import Path
+
+LOG_FILE = Path(os.environ.get('EC_LOG_DIR', '')) / "transcripts.jsonl"
+
+class CustomRecord(logging.LogRecord):
+    """Custom LogRecord class with a timestamp attribute."""
+
+    def __init__(self, name, level, pathname, lineno, msg, args,
+                 exc_info, func=None, sinfo=None, **kwargs):
+        """Initialize a CustomRecord instance."""
+        super().__init__(name, level, pathname, lineno, msg, args,
+                         exc_info, func=func, sinfo=sinfo)
+        self.timestamp = time.time()
+        self.intent = kwargs.get('intent', '')
+        self.slots = kwargs.get('slots', {})
+
+
+class JsonFormatter(logging.Formatter):
+    """JSON formatter for log records."""
+
+    def format(self, record):
+        """Format a log record as a JSON string."""
+        log_dict = record.__dict__.copy()
+        log_dict['msg'] = record.getMessage()
+        return json.dumps(log_dict)
+
+
+def setup_logger(name=__name__, level=logging.INFO):
+    """Set up a logger with a JSON formatter."""
+    logger = logging.getLogger(name)
+    logger.setLevel(level)
+    logging.setLogRecordFactory(CustomRecord)
+    handler = logging.FileHandler(LOG_FILE, encoding='utf-8')
+    handler.setFormatter(JsonFormatter())
+    logger.addHandler(handler)
+    return logger
+
+
+__all__ = ['setup_logger']