diff --git a/tts_engine.py b/tts_engine.py new file mode 100644 index 0000000..c36d19e --- /dev/null +++ b/tts_engine.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python +# +############################################################################### +# Copyright (C) 2026 Joaquin Madrid Belando, EA5GVK +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software Foundation, +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +############################################################################### + +''' +TTS Engine for ADN-DMR-Peer-Server +Converts text files (.txt) to AMBE audio files (.ambe) for DMR transmission. + +Pipeline: .txt -> gTTS -> .mp3 -> ffmpeg -> .wav (8kHz mono 16-bit) -> vocoder -> .ambe + +Requires: + - gTTS (pip install gTTS) + - ffmpeg (system package) + - External AMBE vocoder (configured via TTS_VOCODER_CMD in voice.cfg) + Example vocoders: md380-vocoder, DV3000/ThumbDV serial interface +''' + +import os +import subprocess +import logging + +logger = logging.getLogger(__name__) + +_LANG_MAP = { + 'es_ES': 'es', 'en_GB': 'en', 'en_US': 'en', 'fr_FR': 'fr', + 'de_DE': 'de', 'it_IT': 'it', 'pt_PT': 'pt', 'pt_BR': 'pt', + 'pl_PL': 'pl', 'nl_NL': 'nl', 'da_DK': 'da', 'sv_SE': 'sv', + 'no_NO': 'no', 'el_GR': 'el', 'th_TH': 'th', 'cy_GB': 'cy', + 'ca_ES': 'ca', 'gl_ES': 'gl', 'eu_ES': 'eu', +} + + +def _get_tts_lang(announcement_language): + if announcement_language in _LANG_MAP: + return _LANG_MAP[announcement_language] + return announcement_language[:2] + + +def _generate_tts_audio(text, lang, mp3_path): + try: + from gtts import gTTS + except ImportError: + logger.error('(TTS) gTTS no esta instalado. Ejecuta: pip install gTTS') + return False + + try: + tts = gTTS(text=text, lang=lang, slow=False) + tts.save(mp3_path) + logger.info('(TTS) Audio TTS generado: %s', mp3_path) + return True + except Exception as e: + logger.error('(TTS) Error generando audio TTS: %s', e) + return False + + +def _convert_to_wav(mp3_path, wav_path): + try: + result = subprocess.run( + ['ffmpeg', '-y', '-i', mp3_path, + '-ar', '8000', '-ac', '1', '-sample_fmt', 's16', + '-f', 'wav', wav_path], + capture_output=True, timeout=60 + ) + if result.returncode != 0: + logger.error('(TTS) Error ffmpeg: %s', result.stderr.decode('utf-8', errors='ignore')[:500]) + return False + logger.info('(TTS) Audio convertido a WAV 8kHz mono: %s', wav_path) + return True + except FileNotFoundError: + logger.error('(TTS) ffmpeg no encontrado. Instala ffmpeg en el sistema') + return False + except subprocess.TimeoutExpired: + logger.error('(TTS) Timeout en conversion ffmpeg') + return False + except Exception as e: + logger.error('(TTS) Error en conversion de audio: %s', e) + return False + + +def _encode_ambe(wav_path, ambe_path, vocoder_cmd): + if not vocoder_cmd: + logger.error('(TTS) No hay vocoder configurado (TTS_VOCODER_CMD vacio). ' + 'Configura la ruta al vocoder en voice.cfg o coloca un archivo .ambe pre-convertido') + return False + + cmd = vocoder_cmd.replace('{wav}', wav_path).replace('{ambe}', ambe_path) + + try: + result = subprocess.run( + cmd, shell=True, capture_output=True, timeout=120 + ) + if result.returncode != 0: + logger.error('(TTS) Error del vocoder: %s', result.stderr.decode('utf-8', errors='ignore')[:500]) + return False + if not os.path.isfile(ambe_path): + logger.error('(TTS) El vocoder no genero el archivo AMBE: %s', ambe_path) + return False + logger.info('(TTS) Audio codificado a AMBE: %s', ambe_path) + return True + except FileNotFoundError: + logger.error('(TTS) Comando vocoder no encontrado: %s', cmd.split()[0] if cmd else '(vacio)') + return False + except subprocess.TimeoutExpired: + logger.error('(TTS) Timeout en codificacion AMBE') + return False + except Exception as e: + logger.error('(TTS) Error ejecutando vocoder: %s', e) + return False + + +def text_to_ambe(txt_path, ambe_path, language, vocoder_cmd): + if not os.path.isfile(txt_path): + logger.warning('(TTS) Archivo de texto no encontrado: %s', txt_path) + return False + + if os.path.isfile(ambe_path): + txt_mtime = os.path.getmtime(txt_path) + ambe_mtime = os.path.getmtime(ambe_path) + if ambe_mtime > txt_mtime: + logger.info('(TTS) Usando AMBE cacheado (mas reciente que .txt): %s', ambe_path) + return True + + with open(txt_path, 'r', encoding='utf-8') as f: + text = f.read().strip() + + if not text: + logger.warning('(TTS) Archivo de texto vacio: %s', txt_path) + return False + + logger.info('(TTS) Convirtiendo texto a AMBE: %s (%d caracteres, idioma: %s)', txt_path, len(text), language) + + _dir = os.path.dirname(ambe_path) + if _dir: + os.makedirs(_dir, exist_ok=True) + + _base = os.path.splitext(ambe_path)[0] + _mp3_path = _base + '.mp3' + _wav_path = _base + '.wav' + + _tts_lang = _get_tts_lang(language) + + if not _generate_tts_audio(text, _tts_lang, _mp3_path): + return False + + if not _convert_to_wav(_mp3_path, _wav_path): + _cleanup([_mp3_path]) + return False + + if not vocoder_cmd: + logger.warning('(TTS) No hay vocoder configurado. Archivos intermedios disponibles:') + logger.warning('(TTS) MP3: %s', _mp3_path) + logger.warning('(TTS) WAV: %s', _wav_path) + logger.warning('(TTS) Convierte manualmente el WAV a AMBE y guardalo como: %s', ambe_path) + return False + + if not _encode_ambe(_wav_path, ambe_path, vocoder_cmd): + _cleanup([_mp3_path]) + return False + + _cleanup([_mp3_path, _wav_path]) + + logger.info('(TTS) Conversion completada: %s -> %s', txt_path, ambe_path) + return True + + +def _cleanup(files): + for f in files: + try: + if os.path.isfile(f): + os.remove(f) + except Exception: + pass + + +def ensure_tts_ambe(config, tts_num): + _prefix = 'TTS_ANNOUNCEMENT{}'.format(tts_num) + + if not config['GLOBAL'].get('{}_ENABLED'.format(_prefix), False): + return None + + _file = config['GLOBAL']['{}_FILE'.format(_prefix)] + _lang = config['GLOBAL']['{}_LANGUAGE'.format(_prefix)] + _vocoder_cmd = config['GLOBAL'].get('TTS_VOCODER_CMD', '') + + _txt_path = './Audio/{}/ondemand/{}.txt'.format(_lang, _file) + _ambe_path = './Audio/{}/ondemand/{}.ambe'.format(_lang, _file) + + if os.path.isfile(_ambe_path): + if not os.path.isfile(_txt_path): + logger.info('(TTS-%d) Usando archivo AMBE existente (sin .txt): %s', tts_num, _ambe_path) + return _ambe_path + txt_mtime = os.path.getmtime(_txt_path) + ambe_mtime = os.path.getmtime(_ambe_path) + if ambe_mtime > txt_mtime: + logger.debug('(TTS-%d) Usando AMBE cacheado: %s', tts_num, _ambe_path) + return _ambe_path + + if not os.path.isfile(_txt_path): + logger.warning('(TTS-%d) Archivo de texto no encontrado: %s', tts_num, _txt_path) + return None + + if text_to_ambe(_txt_path, _ambe_path, _lang, _vocoder_cmd): + return _ambe_path + else: + if os.path.isfile(_ambe_path): + logger.warning('(TTS-%d) Usando AMBE anterior (conversion fallo): %s', tts_num, _ambe_path) + return _ambe_path + return None