auracaster/auracast-translator
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refractoring and more tests

Browse Source
This commit is contained in:
pstruebi
2024-12-19 14:01:32 +01:00
parent daa2d92257
commit 5d2f0f4fdf
10 changed files with 185 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
multilang_translator/backend_controller/broadcaster_config.py
View File

@@ -2,6 +2,20 @@ import serial
import time
import logging as log
SAMPLING_RATE_KHZ = 16
FRAME_DUR_MS = 10
PRESET = f'{SAMPLING_RATE_KHZ}_2_1'
if SAMPLING_RATE_KHZ == 8:
BITRATE_KBPS = 16
elif SAMPLING_RATE_KHZ == 16:
BITRATE_KBPS = 32
elif SAMPLING_RATE_KHZ == 24:
BITRATE_KBPS = 48
else:
raise NotImplemented()
def write_to_serial_read_respone(port, cmd, timeout = 2):
# Initialize serial connection
ser = serial.Serial(timeout = timeout)
@@ -40,7 +54,7 @@ def write_to_serial_read_respone(port, cmd, timeout = 2):
def gen_broadcast_config_cmd(serial_port, preset, broadcast_config: dict):
def gen_broadcast_config_cmd(preset, broadcast_config: dict):
"""
Writes broadcaster configuration to the given serial port.
@@ -50,45 +64,55 @@ def gen_broadcast_config_cmd(serial_port, preset, broadcast_config: dict):
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
for ch, file_name in broadcast_config.items():
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 broadcast_name broadcast{ch} {ch}")
cmds.append(f"nac file select_play_once {file_name} {ch} 0 0")
cmds.append(f"nac num_bises 1 {ch} 0")
#cmds.append(f"nac immediate 1 {ch} 0")
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"
}
# TODO: Advertising interval wird ungelmäßig bei mehr als 3 broadcasts 10ms -> 1s< bei 24kHz sampling rate
BROADCAST_CONFIG = {
0: f"announcement_{SAMPLING_RATE_KHZ}_{FRAME_DUR_MS}_{BITRATE_KBPS}_de.lc3",
1: f"announcement_{SAMPLING_RATE_KHZ}_{FRAME_DUR_MS}_{BITRATE_KBPS}_en.lc3",
2: f"announcement_{SAMPLING_RATE_KHZ}_{FRAME_DUR_MS}_{BITRATE_KBPS}_fr.lc3",
#3: f"announcement_{SAMPLING_RATE_KHZ}_{FRAME_DUR_MS}_{BITRATE_KBPS}_es.lc3",
#4: f"announcement_{SAMPLING_RATE_KHZ}_{FRAME_DUR_MS}_{BITRATE_KBPS}_it.lc3"
}
def broadcaster_config():
import subprocess
PORT = "/dev/ttyACM0"
total_ret= ""
cmds = gen_broadcast_config_cmd(PRESET, BROADCAST_CONFIG)
subprocess.run(["nrfjprog", "--reset", "-s", "1050109484"], check=True)
time.sleep(2)
ret = write_to_serial_read_respone(PORT, f"nac en_usb_mass", timeout=0.1)
total_ret += "\n".join(ret)
log.info("\n".join(ret))
time.sleep(1)
#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)
ret = write_to_serial_read_respone(PORT, cmd, timeout=0.1)
log.info("\n".join(ret))
ret = write_to_serial_read_respone(serial_port, "nac show", timeout=0.1)
ret_str = "\n".join(ret)
log.info(ret_str)
total_ret += "\n".join(ret)
time.sleep(1)
for i in BROADCAST_CONFIG.keys():
ret = write_to_serial_read_respone(PORT, f"nac start_idx {i}", timeout=0.1)
total_ret += "\n".join(ret)
log.info("\n".join(ret))
time.sleep(0.2)
return total_ret
return ret_str

19
multilang_translator/backend_controller/broadcaster_play_once.py
View File

@@ -3,21 +3,26 @@ from .broadcaster_config import write_to_serial_read_respone
import time
import logging as log
def broadcaster_play_file(broadcast_ch, file):
def broadcaster_play_file(broadcast_ch, file, wait_after_stop = 1):
serial_port = "/dev/ttyACM0"
ret_all_str = ""
for i in range(3):
ret = write_to_serial_read_respone(serial_port, f"nac file stream_close {broadcast_ch} 0 0", timeout=0.1)
time.sleep(0.5)
if wait_after_stop is not None:
time.sleep(wait_after_stop)
ret += "\n"
ret += write_to_serial_read_respone(serial_port, f"nac file select_play_once {file} {broadcast_ch} 0 0", timeout=0.1)
ret_str = "\n".join(ret)
log.info(ret_str)
ret = "\n".join(ret)
ret_all_str += ret
if not "Failed" in ret:
if (not "Failed" in ret) and (not "err" in ret):
log.info("Breaking after %s retries.", i)
break
log.info(ret_all_str)
return ret
return ret_str
log.error("Failed to play file after 3 retries.")
log.error(ret_all_str)
return ret_all_str

0
multilang_translator/config.py Normal file
View File

94
multilang_translator/main.py
View File

@@ -11,6 +11,9 @@ from examples import custom_style_2
import os
from copy import copy
import time
import logging as log
from .translator import llm_translator
from .text_to_speech import text_to_speech, resample
from .backend_controller.broadcaster_config import broadcaster_config
@@ -20,60 +23,71 @@ from .backend_controller.broadcaster_copy_files import copy_to_broadcaster
from .encode import encode_lc3
ANNOUNCEMENT_DIR = os.path.join(os.path.dirname(__file__), 'announcements')
N_MAX_BIS = 4
SAMPLING_RATE = int(8e3)
SAMPLING_RATE = int(16e3)
FRAME_DUR_MS = 10
BPS = int(16e3)
FILENAMES = {
"de": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_de",
"en": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_en",
"fr": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_fr",
"es": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_es",
"it": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_it",
BPS = int(32e3) # TODO test 16khz 16kbps
CONFIG = {
"de": {
"file": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_de",
"tts": 'de_DE-kerstin-low',
},
"en": {
"file": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_en",
"tts": 'en_US-lessac-medium'
},
"fr": {
"file": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_fr",
"tts": 'fr_FR-siwis-medium'
},
"es": {
"file": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_es",
"tts": 'es_ES-sharvard-medium'
},
"it": {
"file": f"{ANNOUNCEMENT_DIR}/announcement_{SAMPLING_RATE//1000}_{FRAME_DUR_MS}_{BPS//1000}_it",
"tts": 'it_IT-paola-medium'
}
}
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)
audio_dur = text_to_speech.synthesize(text, tts_model, output_file)
resample.resample(output_file, output_file, target_rate=SAMPLING_RATE)
encode_lc3.encode_lc3(output_file, bps=BPS, frame_dur_ms=FRAME_DUR_MS)
return audio_dur
def translate_from_german_and_encode(text_de):
file = FILENAMES['de']
synthesize_resample_encode(text_de, 'de_DE-kerstin-low', f'{file}.wav')
config = copy(CONFIG)
base_lang = "de"
text_en = llm_translator.translator_de_en(text_de)
file = FILENAMES['en']
synthesize_resample_encode(text_en, 'en_US-lessac-medium', f'{file}.wav')
text_fr = llm_translator.translator_de_fr(text_de)
file = FILENAMES['fr']
synthesize_resample_encode(text_fr, 'fr_FR-siwis-medium', f'{file}.wav')
text_es = llm_translator.translator_de_es(text_de)
file = FILENAMES['es']
synthesize_resample_encode(text_es, 'es_ES-sharvard-medium', f'{file}.wav')
text_it = llm_translator.translator_de_it(text_de)
file = FILENAMES['it']
synthesize_resample_encode(text_it, 'it_IT-paola-medium', f'{file}.wav')
file = config[base_lang]["file"]
audio_dur_s [base_lang] = synthesize_resample_encode(text_de, config['de']["tts"], f'{file}.wav')
del config[base_lang]
audio_dur_s = {}
for key, val in config.items():
text = llm_translator.translate_de_to_x(key, text_de)
file = val['file']
audio_dur_s[key] = synthesize_resample_encode(text, val['tts'], f'{file}.wav')
return audio_dur_s
def announcement_from_german_text(text_de):
translate_from_german_and_encode(text_de)
audio_durs = translate_from_german_and_encode(text_de)
# Transfer the files to broadcaster memory
start = time.time()
for val in FILENAMES.values():
copy_to_broadcaster(f'{val}.lc3')
for val in CONFIG.values():
copy_to_broadcaster(f'{val["file"]}.lc3')
log.info("Transfering files to broadcaster took %s s", round(time.time() - start, 3))
# Instruct the broadcaster to stream the files
for i, val in enumerate(list(FILENAMES.values())[:N_MAX_BIS]):
time.sleep(1)
broadcaster_play_file(i, f'{os.path.basename(val)}.lc3')
for i, d in enumerate(list(CONFIG.items())):
key, val = d
broadcaster_play_file(i, f'{os.path.basename(val["file"])}.lc3')
time.sleep(audio_durs[key] + 1)
log.info("Starting all broadcasts %s s", round(time.time() - start, 3))
@@ -92,15 +106,3 @@ def announcement_from_german_text(text_de):
# answers = prompt(questions, style=custom_style_2)
# pprint(answers)
if __name__ == '__main__':
import time
from translator import test_content
import logging as log
log.basicConfig(level=log.INFO)
start= time.time()
#broadcaster_config()
announcement_from_german_text(test_content.TESTSENTENCE_DE_RAINBOW)
print("Generating and starting the announcement took", time.time() - start)

15
multilang_translator/text_to_speech/text_to_speech.py
View File

@@ -2,6 +2,7 @@ import os
import subprocess
import time
import logging as log
import wave
TTS_DIR = os.path.join(os.path.dirname(__file__))
@@ -11,9 +12,15 @@ def synthesize(text, model="en_US-lessac-medium", output_file="out.wav"):
os.chdir(TTS_DIR)
start = time.time()
ret = subprocess.run(['piper', '--model', model, '--output_file', output_file], input=text.encode('utf-8'), check=True)
log.info("Running piper took %s s", round(time.time() - start, 3))
with wave.open(output_file, "rb") as wf:
frames = wf.getnframes()
rate = wf.getframerate()
length_in_seconds = round(frames / rate, 1)
log.info(f"Audio length: {length_in_seconds} s")
os.chdir(pwd)
log.info("Running piper took %s s", round(time.time() - start, 3))
if __name__ == "__main__":
synthesize("Hello, how are you?", "en_US-lessac-medium", "hello.wav")
return length_in_seconds

15
multilang_translator/translator/llm_translator.py
View File

@@ -6,7 +6,7 @@ import time
from . import credentials
from . import syspromts
def translate(model, query):
def query_model(model, query):
url = f'{credentials.BASE_URL}/api/chat/completions'
headers = {
'Authorization': f'Bearer {credentials.TOKEN}',
@@ -21,22 +21,27 @@ def translate(model, query):
return response.json()
def translate_de_to_x(target_language: str, text:str, model ='llama3.2:3b-instruct-q4_0'):
s = getattr(syspromts, f"TRANSLATOR_DE_{target_language.upper()}")
return query_model(model, s + text)['choices'][0]['message']['content']
def translator_de_en(query):
MODEL = 'llama3.2:3b-instruct-q4_0'
#MODEL = 'llama3.1:8b-instruct-q4_0'
return translate(MODEL, syspromts.TRANSLATOR_DE_EN + query)['choices'][0]['message']['content']
return query_model(MODEL, syspromts.TRANSLATOR_DE_EN + query)['choices'][0]['message']['content']
def translator_de_fr(query):
MODEL = 'llama3.2:3b-instruct-q4_0'
return translate(MODEL, syspromts.TRANSLATOR_DE_FR + query)['choices'][0]['message']['content']
return query_model(MODEL, syspromts.TRANSLATOR_DE_FR + query)['choices'][0]['message']['content']
def translator_de_es(query):
MODEL = 'llama3.2:3b-instruct-q4_0'
return translate(MODEL, syspromts.TRANSLATOR_DE_ES + query)['choices'][0]['message']['content']
return query_model(MODEL, syspromts.TRANSLATOR_DE_ES + query)['choices'][0]['message']['content']
def translator_de_it(query):
MODEL = 'llama3.2:3b-instruct-q4_0'
return translate(MODEL, syspromts.TRANSLATOR_DE_IT + query)['choices'][0]['message']['content']
return query_model(MODEL, syspromts.TRANSLATOR_DE_IT + query)['choices'][0]['message']['content']
if __name__ == "__main__":

19
tests/conftest.py
View File

@@ -1,8 +1,25 @@
import logging as log
import pytest
import time
import os
import subprocess
from multilang_translator.backend_controller.broadcaster_config import broadcaster_config, BROADCAST_CONFIG
from multilang_translator.backend_controller.broadcaster_play_once import broadcaster_play_file
from multilang_translator.backend_controller.broadcaster_copy_files import copy_to_broadcaster
# Set the logging level to DEBUG (most verbose) at the root logger
log.basicConfig(
level=log.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
@pytest.fixture(scope="session")
def ft_configure_broadcaster():
log.info("Configuring Broadcaster...")
start = time.time()
ret = broadcaster_config()
log.info(f"Configuration took {round(time.time() - start, 3)} seconds")
assert "err" not in ret
assert "Failed" not in ret
yield ret

39
tests/test_backend.py
View File

@@ -4,47 +4,32 @@ import time
import os
import subprocess
from multilang_translator.backend_controller.broadcaster_config import broadcaster_config
from multilang_translator.backend_controller.broadcaster_config import broadcaster_config, BROADCAST_CONFIG
from multilang_translator.backend_controller.broadcaster_play_once import broadcaster_play_file
from multilang_translator.backend_controller.broadcaster_copy_files import copy_to_broadcaster
@pytest.fixture(scope="session")
def ft_reset_broadcaster():
subprocess.run(["nrfjprog", "--reset", "-s", "1050109484"], check=True)
log.info("Resetting Broadcaster...")
time.sleep(2.)
@pytest.fixture(scope="session")
def ft_configure_broadcaster(ft_reset_broadcaster):
log.info("Configuring Broadcaster...")
start = time.time()
ret = broadcaster_config()
log.info(f"Configuration took {round(time.time() - start, 3)} seconds")
assert "err" not in ret
assert "Failed" not in ret
yield ret
def test_config_broadcaster(ft_configure_broadcaster):
ret = ft_configure_broadcaster
def test_play_files(
ft_configure_broadcaster
):
def test_copy_to_broadcaster():
for key, val in BROADCAST_CONFIG.items():
ret = broadcaster_play_file(key, val)
assert "err" not in ret
assert "Failed" not in ret
time.sleep(9)
def test_copy_to_broadcaster(ft_configure_broadcaster):
log.info("Current working directory is: " + os.getcwd())
start = time.time()
copy_to_broadcaster('./tests/announcement_de.lc3')
log.info(f"Copy to broadcaster took {round(time.time() - start, 3)} seconds")
def test_play_files():
ret = broadcaster_play_file(0, 'announcement_de.lc3')
assert "err" not in ret
assert "Failed" not in ret
time.sleep(1)
ret = broadcaster_play_file(1, 'announcement_en.lc3')
assert "err" not in ret
assert "Failed" not in ret
def test_copy_and_play():
copy_to_broadcaster('./tests/announcement_de.lc3')
copy_to_broadcaster('./tests/announcement_en.lc3')
time.sleep(0.5)

14
tests/test_system.py Normal file
View File

@@ -0,0 +1,14 @@
from multilang_translator.main import announcement_from_german_text
from multilang_translator.translator import test_content
def test_announcement_from_german_text(
ft_configure_broadcaster
):
announcement_from_german_text(test_content.TESTSENTENCE_DE_RAINBOW)
def test_announcement_from_german_text_without_config():
announcement_from_german_text(test_content.TESTSENTENCE_DE_RAINBOW)

4
tests/test_tts.py Normal file
View File

@@ -0,0 +1,4 @@
from multilang_translator.text_to_speech.text_to_speech import synthesize
def test_synthesize():
synthesize("Hello, how are you?", "en_US-lessac-medium", "hello.wav")