Joaquin Madrid Belando
3 weeks ago
parent
30bea31ef8
commit
b7aea01a71
@ -0,0 +1,224 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
###############################################################################
|
||||
# Copyright (C) 2026 Joaquin Madrid Belando, EA5GVK <ea5gvk@gmail.com>
|
||||
#
|
||||
# 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
|
||||
Loading…
Reference in new issue