Utilitaires de conversion audio en Python

Bibliothèque d’utilitaires audio

Pour construire une application Voice Agent robuste, vous avez besoin de fonctions utilitaires pour convertir l’audio entre les différents formats. Cette leçon rassemble les outils Python essentiels pour le traitement audio en temps réel.

Conversion PCM16 vers base64 et inversement

Les deux fonctions les plus utilisées : encoder l’audio capturé en base64 pour l’envoyer, et décoder l’audio reçu en base64 pour le jouer.

import base64
import numpy as np

def audio_to_base64(audio_data: np.ndarray) -> str:
    """Convertit un tableau numpy float32 [-1, 1] en base64 PCM16."""
    audio_int16 = (audio_data * 32767).astype(np.int16)
    return base64.b64encode(audio_int16.tobytes()).decode("utf-8")

def base64_to_audio(base64_audio: str) -> np.ndarray:
    """Convertit une chaîne base64 PCM16 en tableau numpy float32."""
    audio_bytes = base64.b64decode(base64_audio)
    audio_int16 = np.frombuffer(audio_bytes, dtype=np.int16)
    return audio_int16.astype(np.float32) / 32768.0

Utilisation dans le flux Voice Agent

import pyaudio
import json

# Capture micro → envoi
def capture_and_send(ws, stream, chunk_size=4096):
    raw_data = stream.read(chunk_size, exception_on_overflow=False)
    audio_array = np.frombuffer(raw_data, dtype=np.int16)
    audio_float = audio_array.astype(np.float32) / 32768.0
    b64 = audio_to_base64(audio_float)

    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

# Réception → lecture
def play_received(player_stream, base64_audio):
    audio_float = base64_to_audio(base64_audio)
    audio_int16 = (audio_float * 32767).astype(np.int16)
    player_stream.write(audio_int16.tobytes())

Rééchantillonnage (resampling)

Si votre microphone capture à une fréquence différente de celle attendue par l’API, vous devez rééchantillonner :

from scipy.signal import resample

def resample_audio(
    audio: np.ndarray,
    original_rate: int,
    target_rate: int
) -> np.ndarray:
    """Rééchantillonne l'audio d'une fréquence à une autre."""
    if original_rate == target_rate:
        return audio
    num_samples = int(len(audio) * target_rate / original_rate)
    return resample(audio, num_samples)

# Exemple : micro à 44100 Hz → API à 24000 Hz
audio_44100 = capture_microphone()
audio_24000 = resample_audio(audio_44100, 44100, 24000)
b64 = audio_to_base64(audio_24000)

Conversion entre formats G.711 et PCM

Pour les intégrations téléphoniques, vous devrez convertir entre G.711 et PCM linéaire :

import audioop

def pcm_to_mulaw(pcm_data: bytes) -> bytes:
    """Convertit PCM16 en G.711 mu-law."""
    return audioop.lin2ulaw(pcm_data, 2)

def mulaw_to_pcm(mulaw_data: bytes) -> bytes:
    """Convertit G.711 mu-law en PCM16."""
    return audioop.ulaw2lin(mulaw_data, 2)

def pcm_to_alaw(pcm_data: bytes) -> bytes:
    """Convertit PCM16 en G.711 A-law."""
    return audioop.lin2alaw(pcm_data, 2)

def alaw_to_pcm(alaw_data: bytes) -> bytes:
    """Convertit G.711 A-law en PCM16."""
    return audioop.alaw2lin(alaw_data, 2)

Relais téléphonique complet

Voici un exemple de relais entre un flux SIP G.711 et l’API Voice Agent :

async def relay_sip_to_voice_agent(sip_stream, ws):
    """Relaie l'audio d'un appel SIP vers l'API Voice Agent."""
    async for mulaw_chunk in sip_stream:
        # Convertir mu-law → PCM → base64
        pcm_data = mulaw_to_pcm(mulaw_chunk)
        b64 = base64.b64encode(pcm_data).decode("utf-8")

        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": b64
        }))

async def relay_voice_agent_to_sip(ws, sip_stream):
    """Relaie la réponse audio de l'API vers le flux SIP."""
    async for message in ws:
        event = json.loads(message)
        if event["type"] == "response.output_audio.delta":
            pcm_data = base64.b64decode(event["delta"])
            mulaw_data = pcm_to_mulaw(pcm_data)
            await sip_stream.send(mulaw_data)

Détection de silence côté client

Si vous gérez le push-to-talk ou si vous voulez une détection de silence locale avant d’envoyer, cette fonction mesure l’énergie du signal :

def is_silence(
    audio: np.ndarray,
    threshold: float = 0.01
) -> bool:
    """Détecte si un chunk audio est du silence."""
    rms = np.sqrt(np.mean(audio ** 2))
    return rms < threshold

# Usage
audio_chunk = capture_microphone()
if not is_silence(audio_chunk, threshold=0.02):
    b64 = audio_to_base64(audio_chunk)
    ws.send(json.dumps({
        "type": "input_audio_buffer.append",
        "audio": b64
    }))

Sauvegarde et chargement de fichiers WAV

Pour le débogage et les tests, vous pouvez sauvegarder et charger des fichiers WAV :

import wave

def save_wav(
    filename: str,
    audio: np.ndarray,
    sample_rate: int = 24000
):
    """Sauvegarde un tableau numpy en fichier WAV."""
    audio_int16 = (audio * 32767).astype(np.int16)
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(audio_int16.tobytes())

def load_wav(filename: str) -> tuple[np.ndarray, int]:
    """Charge un fichier WAV en tableau numpy float32."""
    with wave.open(filename, "rb") as wf:
        sample_rate = wf.getframerate()
        frames = wf.readframes(wf.getnframes())
        audio_int16 = np.frombuffer(frames, dtype=np.int16)
        return audio_int16.astype(np.float32) / 32768.0, sample_rate

Points clés à retenir

• audio_to_base64 et base64_to_audio sont les deux fonctions essentielles pour le flux Voice Agent
• Le rééchantillonnage est nécessaire quand la fréquence du microphone ne correspond pas à celle de l’API
• Le module audioop de Python gère nativement les conversions G.711 mu-law et A-law
• La détection de silence côté client permet d’optimiser la bande passante en n’envoyant que l’audio utile
• Les utilitaires WAV facilitent le débogage en permettant de sauvegarder et rejouer les flux audio