use_bumble (#1)
Adapt the project to use the bumble auracaster Reviewed-on: https://gitea.pstruebi.xyz/auracaster/multilang-translator-local/pulls/1
This commit was merged in pull request #1.
This commit is contained in:
@@ -4,80 +4,112 @@ list prompt example
|
||||
"""
|
||||
from __future__ import print_function, unicode_literals
|
||||
|
||||
from pprint import pprint
|
||||
|
||||
from PyInquirer import prompt, Separator
|
||||
from examples import custom_style_2
|
||||
|
||||
|
||||
import os
|
||||
import asyncio
|
||||
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
|
||||
from .backend_controller.broadcaster_play_once import broadcaster_play_file
|
||||
from .backend_controller.broadcaster_copy_files import copy_to_broadcaster
|
||||
from .encode import encode_lc3
|
||||
import aioconsole
|
||||
|
||||
from .config import LANG_CONFIG, BITRATE_BPS, SAMPLING_RATE_HZ, FRAME_DUR_MS
|
||||
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, BITRATE_BPS, SAMPLING_RATE_HZ, FRAME_DUR_MS
|
||||
|
||||
# TODO: look for a end to end translation solution
|
||||
|
||||
def transcribe():
|
||||
pass # TODO: Implement transcribing input audio e.g. with whisper
|
||||
|
||||
|
||||
def synthesize_resample_encode(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_HZ)
|
||||
encode_lc3.encode_lc3(output_file, bps=BITRATE_BPS, frame_dur_ms=FRAME_DUR_MS)
|
||||
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
|
||||
|
||||
def translate_from_german_and_encode(text_de):
|
||||
|
||||
def translate_from_german(text_de, model):
|
||||
config = copy(LANG_CONFIG)
|
||||
base_lang = "de"
|
||||
|
||||
file = config[base_lang]["file"]
|
||||
audio_dur_s = {}
|
||||
audio_dur_s [base_lang] = synthesize_resample_encode(text_de, config['de']["tts"], f'{file}.wav')
|
||||
|
||||
del config[base_lang]
|
||||
file = config[base_lang]["filepath_wav"]
|
||||
file_resamp = config[base_lang]['filepath_wav_resamp']
|
||||
tts_json = {}
|
||||
|
||||
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
|
||||
if key == base_lang:
|
||||
text = text_de
|
||||
else:
|
||||
text = llm_translator.translate_de_to_x(text_de, key, model=model)
|
||||
|
||||
log.info('%s', text)
|
||||
file = val['filepath_wav']
|
||||
file_resamp = val['filepath_wav_resamp']
|
||||
tts_json[key] = syntesize_resample(text, val['tts'], file, file_resamp)
|
||||
|
||||
def announcement_from_german_text(text_de):
|
||||
return tts_json
|
||||
|
||||
|
||||
async def announcement_from_german_text(caster:multicast_control.Multicaster, text_de):
|
||||
|
||||
tts_json = translate_from_german(text_de, model = 'llama3.2:3b-instruct-q4_0')
|
||||
|
||||
audio_durs = translate_from_german_and_encode(text_de)
|
||||
# Transfer the files to broadcaster memory
|
||||
start = time.time()
|
||||
for val in LANG_CONFIG.values():
|
||||
copy_to_broadcaster(f'{val["file"]}.lc3')
|
||||
log.info("Transfering files to broadcaster took %s s", round(time.time() - start, 3))
|
||||
await caster.init_audio()
|
||||
caster.start_streaming()
|
||||
|
||||
time.sleep(2)
|
||||
log.info("Starting all broadcasts took %s s", round(time.time() - start, 3))
|
||||
|
||||
# Instruct the broadcaster to stream the files
|
||||
for i, d in enumerate(list(LANG_CONFIG.items())):
|
||||
key, val = d
|
||||
broadcaster_play_file(i, f'{os.path.basename(val["file"])}.lc3')
|
||||
time.sleep(audio_durs[key])
|
||||
|
||||
log.info("Starting all broadcasts %s s", round(time.time() - start, 3))
|
||||
|
||||
|
||||
# questions = [
|
||||
# {
|
||||
# 'type': 'list',
|
||||
# 'name': 'theme',
|
||||
# 'message': 'What type of annoucement would you like to make?',
|
||||
# 'choices': [
|
||||
# 'predefined',
|
||||
# 'custom',
|
||||
# 'audio'
|
||||
# ]
|
||||
# },
|
||||
# ]
|
||||
async def command_line_ui(caster: multicast_control.Multicaster):
|
||||
while True:
|
||||
command = await aioconsole.ainput("\nEnter your Announcement|quit] > ")
|
||||
|
||||
if command.strip().lower() == "quit":
|
||||
print("👋 Exiting...")
|
||||
if caster.device:
|
||||
caster.stop_streaming()
|
||||
await caster.shutdown()
|
||||
break # Exit loop
|
||||
# TODO: Implement predefined announcements
|
||||
|
||||
# answers = prompt(questions, style=custom_style_2)
|
||||
# pprint(answers)
|
||||
elif command.strip() == '':
|
||||
print('Nothing to Announce')
|
||||
else:
|
||||
await announcement_from_german_text(caster, command)
|
||||
|
||||
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 = [ # TODO: integrate this in the LANG_CONFIG dict, better: make a hirachry of dataclasses
|
||||
auracast_config.broadcast_de,
|
||||
auracast_config.broadcast_en,
|
||||
auracast_config.broadcast_fr,
|
||||
#auracast_config.broadcast_es,
|
||||
#auracast_config.broadcast_it,
|
||||
]
|
||||
files = [v['filepath_wav_resamp'] for v in LANG_CONFIG.values()]
|
||||
for i, conf in enumerate(big_conf):
|
||||
conf.loop_wav = False
|
||||
conf.audio_source = f'file:{files[i]}'
|
||||
|
||||
caster = multicast_control.Multicaster(global_conf, big_conf)
|
||||
await caster.init_broadcast()
|
||||
|
||||
#await announcement_from_german_text(caster, test_content.TESTSENTENCE_DE_HELLO)
|
||||
|
||||
await command_line_ui(caster)
|
||||
#await asyncio.wait([caster.streamer.task])
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user