137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
list prompt example
|
|
"""
|
|
from __future__ import print_function, unicode_literals
|
|
|
|
from typing import List
|
|
from dataclasses import asdict
|
|
import asyncio
|
|
import time
|
|
import logging as log
|
|
import aioconsole
|
|
|
|
from auracast import multicast_control
|
|
from auracast import auracast_config
|
|
import multilang_translator.translator_config as translator_config
|
|
from translator import llm_translator
|
|
from translator.test_content import TESTSENTENCE
|
|
from voice_provider import text_to_speech
|
|
|
|
# TODO: look for a end to end translation solution
|
|
|
|
def transcribe():
|
|
pass # TODO: Implement transcribing input audio e.g. with whisper
|
|
|
|
|
|
async def announcement_from_german_text(
|
|
global_config: auracast_config.AuracastGlobalConfig,
|
|
translator_config: List[translator_config.TranslatorConfigDe],
|
|
caster: multicast_control.Multicaster,
|
|
text_de
|
|
):
|
|
base_lang = "deu"
|
|
|
|
for i, trans in enumerate(translator_config):
|
|
if trans.big.language == base_lang:
|
|
text = text_de
|
|
else:
|
|
text = llm_translator.translate_de_to_x(
|
|
text_de,
|
|
trans.big.language,
|
|
model=trans.translator_llm,
|
|
client = trans.llm_client,
|
|
host=trans.llm_host_url,
|
|
token=trans.llm_host_token
|
|
)
|
|
|
|
log.info('%s', text)
|
|
|
|
lc3_audio = text_to_speech.synthesize(
|
|
text,
|
|
global_config.auracast_sampling_rate_hz,
|
|
trans.tts_system,
|
|
trans.tts_model,
|
|
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(global_conf, translator_conf, 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(
|
|
global_conf,
|
|
translator_conf,
|
|
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.AuracastGlobalConfig()
|
|
#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
|
|
|
|
|
|
translator_conf = [
|
|
translator_config.TranslatorConfigDe(),
|
|
translator_config.TranslatorConfigEn(),
|
|
translator_config.TranslatorConfigFr(),
|
|
#auracast_config.broadcast_es,
|
|
#auracast_config.broadcast_it,
|
|
]
|
|
for conf in translator_conf:
|
|
conf.big.loop = False
|
|
conf.llm_client = 'openwebui' # comment out for local llm
|
|
conf.llm_host_url = 'https://ollama.pstruebi.xyz'
|
|
conf.llm_host_token = 'sk-17124cb84df14cc6ab2d9e17d0724d13'
|
|
|
|
caster = multicast_control.Multicaster(global_conf, [conf.big for conf in translator_conf])
|
|
await caster.init_broadcast()
|
|
|
|
# await announcement_from_german_text(
|
|
# global_conf,
|
|
# translator_conf,
|
|
# caster,
|
|
# test_content.TESTSENTENCE.DE_HELLO
|
|
# )
|
|
# await asyncio.wait([caster.streamer.task])
|
|
await command_line_ui(global_conf, translator_conf, caster)
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|
|
|
|
# TODO: add support for multiple radios |