Abstract base for streaming audio effects, defining async iterator interface for sample rate and chunk data.
# Source: https://github.com/neural-maze/realtime-phone-agents-course | License: MIT | Copyright 2025 The Neural Maze
# Provided via justclaw.it.com; keep this notice with the code.
from abc import ABC, abstractmethod
from typing import AsyncIterator, Tuple
import numpy as np
AudioChunk = Tuple[int, np.ndarray] # (sample_rate, samples)
class BaseVoiceEffect(ABC):
"""
Base class for any audio effect that streams audio chunks.
Examples: keyboard clicks, hold music, notifications.
"""
@abstractmethod
async def stream(self) -> AsyncIterator[AudioChunk]:
"""
Yields (sample_rate, chunk) pairs.
Chunk is expected to be a float32 numpy array in [-1, 1].
"""
...
See the full source, get the GitHub permalink, and search for more like it.
Get a free API key