128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
list prompt example
|
|
"""
|
|
from __future__ import print_function, unicode_literals
|
|
|
|
from dataclasses import asdict
|
|
import asyncio
|
|
from copy import copy
|
|
import time
|
|
import logging as log
|
|
import aioconsole
|
|
|
|
from utils import resample
|
|
from translator import llm_translator, test_content
|
|
from text_to_speech import text_to_speech
|
|
from encode import encode_lc3
|
|
from auracast import multicast_control
|
|
from auracast import auracast_config
|
|
from config import LANG_CONFIG, SAMPLING_RATE_HZ
|
|
from translator.test_content import TESTSENTENCE
|
|
|
|
# TODO: look for a end to end translation solution
|
|
|
|
def transcribe():
|
|
pass # TODO: Implement transcribing input audio e.g. with whisper
|
|
|
|
|
|
def syntesize_resample(text, tts_model, file_wav, file_wav_resamp):
|
|
audio_dur = text_to_speech.synthesize(text, tts_model, file_wav)
|
|
resample.resample_file(file_wav, file_wav_resamp, target_rate=SAMPLING_RATE_HZ)
|
|
return audio_dur
|
|
|
|
|
|
async def announcement_from_german_text(
|
|
caster: multicast_control.Multicaster,
|
|
text_de
|
|
):
|
|
TRANSLATOR_LLM = 'llama3.2:3b-instruct-q4_0'
|
|
|
|
config = copy(LANG_CONFIG)
|
|
base_lang = "de"
|
|
|
|
for i, d in enumerate(config.items()):
|
|
key, val = d
|
|
if key == base_lang:
|
|
text = text_de
|
|
else:
|
|
text = llm_translator.translate_de_to_x(text_de, key, model=TRANSLATOR_LLM)
|
|
|
|
log.info('%s', text)
|
|
|
|
lc3_audio = text_to_speech.synthesize(
|
|
text,
|
|
SAMPLING_RATE_HZ,
|
|
'piper',
|
|
val['tts'],
|
|
return_lc3=True
|
|
)
|
|
caster.big_conf[i].audio_source = lc3_audio
|
|
|
|
start = time.time()
|
|
caster.start_streaming()
|
|
|
|
log.info("Starting all broadcasts took %s s", round(time.time() - start, 3))
|
|
|
|
|
|
async def command_line_ui(caster: multicast_control.Multicaster):
|
|
while True:
|
|
# make a list of all available testsentence
|
|
sentence_list = list(asdict(TESTSENTENCE).values())
|
|
|
|
prompt = "Enter your Announcement|quit or choose:] > \n"
|
|
prompt += "\n".join([f"{i}: {s}" for i,s in enumerate(sentence_list)])
|
|
prompt += "\n"
|
|
command = await aioconsole.ainput(prompt)
|
|
|
|
if command.strip().lower() == "quit":
|
|
print("👋 Exiting...")
|
|
if caster.device:
|
|
caster.stop_streaming()
|
|
await caster.shutdown()
|
|
break # Exit loop
|
|
elif command.strip() == '':
|
|
print('Nothing to Announce')
|
|
# Check if command is a single number
|
|
elif command.strip().isdigit():
|
|
ind = int(command.strip())
|
|
await announcement_from_german_text(caster, sentence_list[ind])
|
|
await asyncio.wait([caster.streamer.task])
|
|
# Interpret the command as announcement
|
|
else:
|
|
await announcement_from_german_text(caster, command)
|
|
await asyncio.wait([caster.streamer.task])
|
|
|
|
async def main():
|
|
log.basicConfig(
|
|
level=log.INFO,
|
|
format='%(module)s.py:%(lineno)d %(levelname)s: %(message)s'
|
|
)
|
|
|
|
global_conf = auracast_config.global_base_config
|
|
#global_conf.transport='serial:/dev/serial/by-id/usb-SEGGER_J-Link_001057705357-if02,1000000,rtscts' # transport for nrf54l15dk
|
|
global_conf.transport='serial:/dev/serial/by-id/usb-ZEPHYR_Zephyr_HCI_UART_sample_81BD14B8D71B5662-if00,115200,rtscts' #nrf52dongle hci_uart usb cdc
|
|
|
|
big_conf = [
|
|
auracast_config.broadcast_de,
|
|
auracast_config.broadcast_en,
|
|
auracast_config.broadcast_fr,
|
|
#auracast_config.broadcast_es,
|
|
#auracast_config.broadcast_it,
|
|
]
|
|
for i, conf in enumerate(big_conf):
|
|
conf.loop = False
|
|
|
|
caster = multicast_control.Multicaster(global_conf, big_conf)
|
|
await caster.init_broadcast()
|
|
|
|
#await announcement_from_german_text(caster, test_content.TESTSENTENCE.DE_HELLO)
|
|
#await asyncio.wait([caster.streamer.task])
|
|
await command_line_ui(caster)
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|
|
|
|
# TODO: integrate this in the LANG_CONFIG dict, better: make a hierachy of dataclasses
|
|
# TODO: remove the nececcity for files
|
|
# TODO: add support for multiple radios |