"""
Module de traitement du signal audio et gestion des flux PCM.
Fournit des utilitaires de décodage multi-formats (WebM, WAV, MP4, OGG),
de calcul de puissance RMS, d'enregistrement cyclique basé sur VAD et de
normalisation des niveaux crêtes.
"""
import collections
import io
import time
from collections.abc import Callable
import av
import numpy as np
import sounddevice as sd
import soundfile as sf
from ..config import AudioConfig
from ..core.logger import getMirzaLogger
from .vadDetector import SileroVadDetector
logger = getMirzaLogger("mirza.audio.processor")
#: Sample rate every STT engine expects. All decoding resamples to this rate, so
#: it is a hard requirement of the ASR models rather than a tunable preference.
TARGET_SAMPLE_RATE = 16000
[docs]
class AudioProcessor:
"""Traitement du signal audio : capture, detection d'activite vocale, restitution.
Attributes
----------
config : AudioConfig
Configuration acoustique du systeme.
"""
[docs]
def __init__(self, config: AudioConfig | None = None):
"""
Initialise le processeur audio avec sa configuration acoustique.
Parameters
----------
config : AudioConfig, optional
Configuration audio du système. Si None, les valeurs par défaut sont appliquées.
"""
self.config = config or AudioConfig()
self.vadDetector = SileroVadDetector(
sampleRate=self.config.sampleRate,
speechThreshold=self.config.speechThreshold,
)
[docs]
@staticmethod
def decodeAudioBytes(audioBytes: bytes) -> tuple[np.ndarray, int]:
"""
Decode n'importe quel conteneur audio en tableau numpy float32 16kHz mono.
Parameters
----------
audioBytes : bytes
Flux binaire d'un conteneur audio (WAV, WebM, MP4, AAC, OGG).
Returns
-------
tuple[np.ndarray, int]
Tableau numpy contenant les echantillons audio float32 et la frequence d'echantillonnage.
"""
if not audioBytes or len(audioBytes) == 0:
return np.array([], dtype=np.float32), TARGET_SAMPLE_RATE
# Essai 1 : Soundfile rapide (pour WAV standard)
try:
audioArray, sampleRate = sf.read(io.BytesIO(audioBytes), dtype="float32")
if len(audioArray.shape) > 1:
audioArray = audioArray.mean(axis=1)
if sampleRate != TARGET_SAMPLE_RATE and len(audioArray) > 0:
targetLen = int(len(audioArray) * (TARGET_SAMPLE_RATE / sampleRate))
audioArray = np.interp(
np.linspace(0, len(audioArray), targetLen, endpoint=False),
np.arange(len(audioArray)),
audioArray,
).astype(np.float32)
sampleRate = TARGET_SAMPLE_RATE
return audioArray.astype(np.float32), sampleRate
except Exception as exc:
# Best-effort decode chain: the next reader gets a chance. Note we
# must not touch `sampleRate` here, it is unbound when the very
# first reader is the one that failed.
logger.debug("Lecture audio directe impossible, essai du decodeur conteneur : %s", exc)
# Essai 2 : Decodage universel avec PyAV (WebM, Opus, MP4, AAC)
container = None
try:
container = av.open(io.BytesIO(audioBytes))
stream = next((s for s in container.streams if s.type == "audio"), None)
if stream is None:
return np.array([], dtype=np.float32), TARGET_SAMPLE_RATE
resampler = av.AudioResampler(format="fltp", layout="mono", rate=TARGET_SAMPLE_RATE)
samplesList = []
for frame in container.decode(stream):
resampledFrames = resampler.resample(frame)
for rf in resampledFrames:
samplesList.append(rf.to_ndarray()[0])
if not samplesList:
return np.array([], dtype=np.float32), TARGET_SAMPLE_RATE
fullAudio = np.concatenate(samplesList).astype(np.float32)
return fullAudio, TARGET_SAMPLE_RATE
except Exception as exc:
logger.warning("Erreur de decodage audio binaire : %s", exc)
return np.array([], dtype=np.float32), TARGET_SAMPLE_RATE
finally:
if container is not None:
try:
container.close()
except Exception as exc:
logger.debug("Fermeture du conteneur audio impossible : %s", exc)
[docs]
@staticmethod
def normalizeAudio(
audioSamples: np.ndarray,
targetPeak: float = 0.95,
minSignalThreshold: float = 0.005,
allowAmplification: bool = False,
maxGainFactor: float = 3.5,
) -> np.ndarray:
"""
Centre le signal audio (suppression DC) et evite l'ecretage numerique.
Si allowAmplification est actif, calibre les signaux faibles avec un gain plafonne.
Parameters
----------
audioSamples : np.ndarray
Echantillons audio a normaliser.
targetPeak : float, optional
Valeur crete maximale absolue souhaitee, default is 0.95.
minSignalThreshold : float, optional
Seuil minimal d'amplitude crete pour l'amplification optionnelle, default is 0.005.
allowAmplification : bool, optional
Si True, calibre les signaux faibles jusqu'a targetPeak (plafonne a maxGainFactor), default is False.
maxGainFactor : float, optional
Facteur maximal de gain applique aux signaux faibles (+11 dB max), default is 3.5.
Returns
-------
np.ndarray
Tableau numpy contenant les echantillons audio traites en float32.
"""
if len(audioSamples) == 0:
return audioSamples
# Suppression de l'offset DC (centrage)
audio = audioSamples - np.mean(audioSamples)
peak = float(np.max(np.abs(audio)))
if peak <= 0:
return audio.astype(np.float32)
# 1. Attenuation si ecretage / depassement de la crete
if peak > targetPeak:
return ((audio / peak) * targetPeak).astype(np.float32)
# 2. Calibration des signaux vocaux faibles (micro distant ou smartphone)
if allowAmplification and peak >= minSignalThreshold:
desiredGain = targetPeak / peak
effectiveGain = min(desiredGain, maxGainFactor)
return (audio * effectiveGain).astype(np.float32)
return audio.astype(np.float32)
[docs]
@staticmethod
def recordSpeechWithVad(
micId: int | None,
vadDetector: SileroVadDetector,
audioConfig: AudioConfig,
onAudioLevel: Callable[[float, float], None] | None = None,
stopSignal: Callable[[], bool] | None = None,
) -> np.ndarray | None:
"""
Capture la voix avec coupure rapide par Silero VAD et streaming des niveaux.
Parameters
----------
micId : int or None
Identifiant du peripherique audio d'entree ou None pour le peripherique par defaut.
vadDetector : SileroVadDetector
Detecteur neuronal d'activite vocale.
audioConfig : AudioConfig
Configuration acoustique et de coupure du silence.
onAudioLevel : callable, optional
Fonction de rappel recevant le niveau RMS et la probabilite de parole.
stopSignal : callable, optional
Fonction indiquant si l'enregistrement doit s'interrompre immediatement.
Returns
-------
np.ndarray or None
Tableau des echantillons de parole enregistres ou None si aucun signal vocal detecte.
"""
vadDetector.reset()
chunkSamples = audioConfig.chunkSamples
preRollBuffer = collections.deque(maxlen=audioConfig.preRollChunks)
speechChunks = []
hasStartedSpeaking = False
silenceStartTime = None
recordingStartTime = time.time()
lastLevelReportTime = 0.0
wasSpeech = False
with sd.InputStream(
samplerate=audioConfig.sampleRate,
channels=1,
dtype="float32",
device=micId,
) as stream:
while True:
if stopSignal and stopSignal():
return None
chunkData, _ = stream.read(chunkSamples)
chunkFlat = chunkData.flatten()
isSpeech, prob = vadDetector.predict(chunkFlat)
rms = float(np.sqrt(np.mean(chunkFlat**2)))
currentTime = time.time()
elapsedTotal = currentTime - recordingStartTime
if onAudioLevel and (currentTime - lastLevelReportTime >= 0.05 or isSpeech != wasSpeech):
lastLevelReportTime = currentTime
wasSpeech = isSpeech
onAudioLevel(rms, prob)
if not hasStartedSpeaking:
preRollBuffer.append(chunkFlat)
if isSpeech:
hasStartedSpeaking = True
speechChunks.extend(list(preRollBuffer))
elif elapsedTotal > audioConfig.initialWaitTimeout:
return None
else:
speechChunks.append(chunkFlat)
if not isSpeech:
if silenceStartTime is None:
silenceStartTime = currentTime
elif (currentTime - silenceStartTime) >= audioConfig.silenceDurationThreshold:
break
else:
silenceStartTime = None
if elapsedTotal > audioConfig.maxTotalDuration:
break
if not speechChunks or not hasStartedSpeaking:
return None
return np.concatenate(speechChunks)