restructure the project and start using pytest

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true,
+    "python.testing.pytestArgs": [
+        "."
+    ]
+}

multilang_translator/backend_controller/broadcaster_config.py

@@ -0,0 +1,97 @@
+import serial
+import time
+
+
+def write_to_serial_read_respone(port, cmd, timeout = 2):
+    # Initialize serial connection
+    ser = serial.Serial(timeout = timeout)
+    ser.port = port
+    ser.baudrate = 115200
+    ser.bytesize = serial.EIGHTBITS
+    ser.parity = serial.PARITY_NONE
+    ser.stopbits = serial.STOPBITS_ONE
+
+    try:
+        # Try to open the serial connection
+        #if not ser.is_open:
+        ser.open()
+
+        # Send string to serial port and get response
+        command = f"{cmd.strip()}\r\n"
+        ser.write(command.encode())
+        time.sleep(1)  # wait a bit for response
+
+        readlines = []
+        for _ in range(20):
+            line = ser.readline().decode('utf-8').strip()
+            if not line:
+                ser.close()
+                break
+            else:
+               readlines.append(line)
+
+    except serial.SerialException as e:
+        print(f"Error communicating with serial port: {e}")
+    finally:
+        # Close serial connection before returning
+        if ser.is_open:
+            ser.close()
+        return readlines
+
+
+
+def gen_broadcast_config_cmd(serial_port, preset, broadcast_config: dict):
+    """
+    Writes broadcaster configuration to the given serial port.
+
+    Args:
+        serial_port (str): Device path of the serial port (e.g., '/dev/ttyACM0')
+        preset (str): Preset string used in nac preset line
+        broadcast_names (list): List of names for each broadcast group
+    """
+    cmds = []
+    with serial.Serial(serial_port, 115200) as ser:
+        ser.write(f"nac stop".encode() + b'\r\n')
+        ser.write(f"nac clear".encode() + b'\r\n')
+
+    time.sleep(.5)
+
+    for ch, d in enumerate(broadcast_config.items()):
+        broadcast_name, file_name = d
+    
+        left_channel_file = file_name
+        right_channel_file = file_name
+        
+        cmds.append(f"nac preset {preset} {ch}") 
+        cmds.append(f"nac broadcast_name {broadcast_name} {ch}")
+        cmds.append(f"nac file select_play_once {left_channel_file} {ch} 0 0")
+        cmds.append(f"nac file select_play_once {right_channel_file} {ch} 0 1")
+    
+    cmds.append(f"nac start")
+
+    return cmds
+
+def broadcaster_config():
+    serial_port = "/dev/ttyACM0"
+    broadcast_config = {
+        "broadcast1": "left24kHz_48kbps.lc3",
+        "broadcast2": "right-channel_24kHz_left_48kbps_10ms.lc3"
+    }
+
+    #write_config_to_tty(serial_port, "24_2_1", broadcast_config)
+    # Example usage:
+    cmds = gen_broadcast_config_cmd(serial_port, "24_2_1", broadcast_config)
+
+    for cmd in cmds:
+        ret = write_to_serial_read_respone(serial_port, cmd, timeout=0.1)
+        print("\n".join(ret))
+    
+    ret = write_to_serial_read_respone(serial_port, "nac show", timeout=0.1)
+    print("\n".join(ret))
+
+
+if __name__ == "__main__":
+    broadcaster_config()
+
+
+

multilang_translator/backend_controller/broadcaster_play_once.py

@@ -0,0 +1,23 @@
+#import broadcaster_config
+from .broadcaster_config import write_to_serial_read_respone
+import time
+
+def broadcaster_play_file(broadcast_ch, file):
+    serial_port = "/dev/ttyACM0"
+    ret = write_to_serial_read_respone(serial_port, f"nac file stream_close {broadcast_ch} 0 0", timeout=0.1)
+    print("\n".join(ret))
+
+    ret = write_to_serial_read_respone(serial_port, f"nac file select_play_once {file} {broadcast_ch} 0 0", timeout=0.1)
+    print("\n".join(ret))
+
+
+if __name__ == "__main__":
+    broadcast_ch = 0
+    file = "left24kHz_48kbps.lc3"
+    broadcaster_play_file(broadcast_ch, file)
+
+    time.sleep(1)
+    broadcast_ch = 1
+    file = "right-channel_24kHz_left_48kbps_10ms.lc3"
+
+    broadcaster_play_file(broadcast_ch, file)

multilang_translator/main.py

@@ -16,6 +16,7 @@ from translator import llm_translator
 from encode import encode_lc3
 
 ANNOUNCEMENT_DIR = os.path.join(os.path.dirname(__file__), 'announcements')
+os.makedirs(ANNOUNCEMENT_DIR, exist_ok=True)
 
 def synthesize_resample_encode(text, tts_model, output_file):
     text_to_speech.synthesize(text, tts_model, output_file)
@@ -23,16 +24,16 @@ def synthesize_resample_encode(text, tts_model, output_file):
     encode_lc3.encode_lc3(output_file)
 
 
-def announcement_from_german_text(test_de):
+def announcement_from_german_text(text_de):
-    synthesize_resample_encode(test_de, 'de_DE-kerstin-low', f'{ANNOUNCEMENT_DIR}/announcement_de.wav')
+    synthesize_resample_encode(text_de, 'de_DE-kerstin-low', f'{ANNOUNCEMENT_DIR}/announcement_de.wav')
 
-    text_en = llm_translator.translator_de_en(test_de)
+    text_en = llm_translator.translator_de_en(text_de)
     synthesize_resample_encode(text_en, 'en_US-lessac-medium', f'{ANNOUNCEMENT_DIR}/announcement_en.wav')
 
-    text_fr = llm_translator.translator_de_fr(test_de)
+    text_fr = llm_translator.translator_de_fr(text_de)
     synthesize_resample_encode(text_fr, 'fr_FR-siwis-medium', f'{ANNOUNCEMENT_DIR}/announcement_fr.wav')
 
-    text_fr = llm_translator.translator_de_es(test_de)
+    text_fr = llm_translator.translator_de_es(text_de)
     synthesize_resample_encode(text_fr, 'es_ES-sharvard-medium', f'{ANNOUNCEMENT_DIR}/announcement_fr.wav')
 
 # questions = [

multilang_translator/translator/llm_translator.py

@@ -1,5 +1,7 @@
 import requests
 import json
+import logging as log
+import time
 
 from . import credentials
 from . import syspromts
@@ -13,7 +15,9 @@ def translate(model, query):
         'model': model,
         'messages': [{'role': 'user', 'content': query}],
     }
+    start = time.time()
     response = requests.post(url, headers=headers, json=payload)
+    log.info("Translating the text took %s s", round(time.time() - start, 2))
     return response.json()
 
 

pyproject.toml

@@ -0,0 +1,12 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "multilang_translator"
+version = '0.1'
+
+[tool.pytest.ini_options]
+addopts = [
+    "--import-mode=importlib", "-s", "-v"
+]

readme.md

@@ -2,7 +2,7 @@
 sudo apt install liblc3-tools
 
 use python3.9
-pip install piper-tts soundfile librosa
+pip install piper-tts soundfile librosa pyserial pytest
 
 # Piper update voices
 piper --update-voices -m en_US-lessac-medium

tests/test_backend.py

@@ -0,0 +1,10 @@
+from multilang_translator.backend_controller.broadcaster_config import broadcaster_config
+from multilang_translator.backend_controller.broadcaster_play_once import broadcaster_play_file
+
+
+def test_config_broadcaster():
+    broadcaster_config()
+    broadcaster_play_file(0, 'announcement_de.wav')
+
+def test_play_file():
+    broadcaster_play_file(0, 'announcement_de.wav')