Skip to content

Audio & Hardware Reference

This section documents the hardware discovery, multi-platform rate negotiation, non-blocking microphone streaming, and low-latency audio playback components.

AudioDeviceManager

edgevoiceai.audio.device

Audio Device Management for Voice AI Framework. Supports cross-platform hardware discovery and split input/output device routing on Raspberry Pi, Linux, macOS, and Windows.

AudioDeviceManager

Cross-platform manager for querying, matching, and configuring independent input and output audio hardware.

Source code in edgevoiceai/audio/device.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
class AudioDeviceManager:
    """
    Cross-platform manager for querying, matching, and configuring
    independent input and output audio hardware.
    """

    @classmethod
    def list_devices(cls) -> list[AudioDeviceInfo]:
        """Returns all audio devices discovered on the host system."""
        if not PYAUDIO_AVAILABLE or pyaudio is None:
            return []

        pa = pyaudio.PyAudio()
        devices: list[AudioDeviceInfo] = []
        try:
            count = pa.get_device_count()
            for i in range(count):
                try:
                    info = pa.get_device_info_by_index(i)
                    host_api_idx = int(info.get("hostApi", 0))
                    host_api_info = pa.get_host_api_info_by_index(host_api_idx)
                    api_name = host_api_info.get("name", "Unknown") if host_api_info else "Unknown"

                    devices.append(
                        AudioDeviceInfo(
                            index=i,
                            name=str(info.get("name", f"Device {i}")),
                            max_input_channels=int(info.get("maxInputChannels", 0)),
                            max_output_channels=int(info.get("maxOutputChannels", 0)),
                            default_sample_rate=float(info.get("defaultSampleRate", 16000.0)),
                            host_api=str(api_name),
                        )
                    )
                except Exception:
                    continue
        finally:
            pa.terminate()

        return devices

    @classmethod
    def find_device_index(
        cls,
        spec: int | str | None,
        is_input: bool = True,
    ) -> int | None:
        """
        Resolves an audio device specification (integer index or fuzzy name string)
        to an actual PortAudio device index.

        Supports regex and substring matching (e.g., 'USB', 'bcm2835', 'AirPods').
        Returns None to use the system default if spec is None or device not found.
        """
        if spec is None:
            return None

        # Direct integer index
        if isinstance(spec, int):
            return spec

        spec_str = str(spec).strip()
        if not spec_str:
            return None

        # Check if spec is an integer formatted as string
        if spec_str.isdigit():
            return int(spec_str)

        all_devices = cls.list_devices()
        candidates = [d for d in all_devices if (d.is_input if is_input else d.is_output)]

        # 1. Exact case-insensitive match
        for dev in candidates:
            if dev.name.lower() == spec_str.lower():
                return dev.index

        # 2. Substring match
        for dev in candidates:
            if spec_str.lower() in dev.name.lower():
                return dev.index

        # 3. Regex match
        try:
            pattern = re.compile(spec_str, re.IGNORECASE)
            for dev in candidates:
                if pattern.search(dev.name):
                    return dev.index
        except re.error:
            pass

        return None

    @classmethod
    def find_best_input_device(cls, pa: Any, preferred_index: int | None = None) -> int | None:
        """
        Finds the most suitable working audio input device index (USB mic).
        Falls back to any input-capable device if the default is not marked.
        """
        count = pa.get_device_count()
        if count == 0:
            return None

        if preferred_index is not None:
            try:
                info = pa.get_device_info_by_index(preferred_index)
                if int(info.get("maxInputChannels", 0)) > 0:
                    return preferred_index
            except Exception:
                pass

        # 1. Try system default
        try:
            default_in = pa.get_default_input_device_info()
            if int(default_in.get("maxInputChannels", 0)) > 0:
                return int(default_in["index"])
        except Exception:
            pass

        # 2. Search for USB / mic device
        usb_candidate = None
        first_input = None
        for i in range(count):
            try:
                info = pa.get_device_info_by_index(i)
                if int(info.get("maxInputChannels", 0)) > 0:
                    if first_input is None:
                        first_input = i
                    name_lower = str(info.get("name", "")).lower()
                    if "usb" in name_lower or "mic" in name_lower:
                        usb_candidate = i
                        break
            except Exception:
                continue

        return usb_candidate if usb_candidate is not None else first_input

    @classmethod
    def find_best_output_device(cls, pa: Any, preferred_index: int | None = None) -> int | None:
        """
        Finds the most suitable working audio output device index (Speaker / Bluetooth).
        """
        count = pa.get_device_count()
        if count == 0:
            return None

        if preferred_index is not None:
            try:
                info = pa.get_device_info_by_index(preferred_index)
                if int(info.get("maxOutputChannels", 0)) > 0:
                    return preferred_index
            except Exception:
                pass

        # 1. Try system default
        try:
            default_out = pa.get_default_output_device_info()
            if int(default_out.get("maxOutputChannels", 0)) > 0:
                return int(default_out["index"])
        except Exception:
            pass

        # 2. Search for preferred output devices (Pulse, bluez, default, headphone, speaker)
        first_output = None
        preferred_candidate = None
        for i in range(count):
            try:
                info = pa.get_device_info_by_index(i)
                if int(info.get("maxOutputChannels", 0)) > 0:
                    if first_output is None:
                        first_output = i
                    name_lower = str(info.get("name", "")).lower()
                    if any(k in name_lower for k in ("pulse", "default", "bluez", "headphone", "speaker", "usb", "bcm2835", "jack", "audio", "alsa")):
                        preferred_candidate = i
                        break
            except Exception:
                continue

        return preferred_candidate if preferred_candidate is not None else first_output

    @classmethod
    def print_diagnostics(cls) -> None:
        """Prints a user-friendly terminal table of available audio devices."""
        devices = cls.list_devices()
        print("\n" + "=" * 65)
        print("🔊 [VOICE FRAMEWORK - AUDIO HARDWARE DIAGNOSTICS]")
        print("=" * 65)

        inputs = [d for d in devices if d.is_input]
        outputs = [d for d in devices if d.is_output]

        print("\n🎤 Microphone / Input Devices:")
        if not inputs:
            print("  (None detected)")
        for d in inputs:
            print(f"  • [{d.index}] {d.name} ({int(d.default_sample_rate)}Hz, {d.host_api})")

        print("\n🔈 Speaker / Output Devices:")
        if not outputs:
            print("  (None detected)")
        for d in outputs:
            print(f"  • [{d.index}] {d.name} ({int(d.default_sample_rate)}Hz, {d.host_api})")
        print("=" * 65 + "\n")

find_best_input_device classmethod 

find_best_input_device(pa, preferred_index=None)

Finds the most suitable working audio input device index (USB mic). Falls back to any input-capable device if the default is not marked.

Source code in edgevoiceai/audio/device.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@classmethod
def find_best_input_device(cls, pa: Any, preferred_index: int | None = None) -> int | None:
    """
    Finds the most suitable working audio input device index (USB mic).
    Falls back to any input-capable device if the default is not marked.
    """
    count = pa.get_device_count()
    if count == 0:
        return None

    if preferred_index is not None:
        try:
            info = pa.get_device_info_by_index(preferred_index)
            if int(info.get("maxInputChannels", 0)) > 0:
                return preferred_index
        except Exception:
            pass

    # 1. Try system default
    try:
        default_in = pa.get_default_input_device_info()
        if int(default_in.get("maxInputChannels", 0)) > 0:
            return int(default_in["index"])
    except Exception:
        pass

    # 2. Search for USB / mic device
    usb_candidate = None
    first_input = None
    for i in range(count):
        try:
            info = pa.get_device_info_by_index(i)
            if int(info.get("maxInputChannels", 0)) > 0:
                if first_input is None:
                    first_input = i
                name_lower = str(info.get("name", "")).lower()
                if "usb" in name_lower or "mic" in name_lower:
                    usb_candidate = i
                    break
        except Exception:
            continue

    return usb_candidate if usb_candidate is not None else first_input

find_best_output_device classmethod 

find_best_output_device(pa, preferred_index=None)

Finds the most suitable working audio output device index (Speaker / Bluetooth).

Source code in edgevoiceai/audio/device.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
@classmethod
def find_best_output_device(cls, pa: Any, preferred_index: int | None = None) -> int | None:
    """
    Finds the most suitable working audio output device index (Speaker / Bluetooth).
    """
    count = pa.get_device_count()
    if count == 0:
        return None

    if preferred_index is not None:
        try:
            info = pa.get_device_info_by_index(preferred_index)
            if int(info.get("maxOutputChannels", 0)) > 0:
                return preferred_index
        except Exception:
            pass

    # 1. Try system default
    try:
        default_out = pa.get_default_output_device_info()
        if int(default_out.get("maxOutputChannels", 0)) > 0:
            return int(default_out["index"])
    except Exception:
        pass

    # 2. Search for preferred output devices (Pulse, bluez, default, headphone, speaker)
    first_output = None
    preferred_candidate = None
    for i in range(count):
        try:
            info = pa.get_device_info_by_index(i)
            if int(info.get("maxOutputChannels", 0)) > 0:
                if first_output is None:
                    first_output = i
                name_lower = str(info.get("name", "")).lower()
                if any(k in name_lower for k in ("pulse", "default", "bluez", "headphone", "speaker", "usb", "bcm2835", "jack", "audio", "alsa")):
                    preferred_candidate = i
                    break
        except Exception:
            continue

    return preferred_candidate if preferred_candidate is not None else first_output

find_device_index classmethod 

find_device_index(spec, is_input=True)

Resolves an audio device specification (integer index or fuzzy name string) to an actual PortAudio device index.

Supports regex and substring matching (e.g., 'USB', 'bcm2835', 'AirPods'). Returns None to use the system default if spec is None or device not found.

Source code in edgevoiceai/audio/device.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@classmethod
def find_device_index(
    cls,
    spec: int | str | None,
    is_input: bool = True,
) -> int | None:
    """
    Resolves an audio device specification (integer index or fuzzy name string)
    to an actual PortAudio device index.

    Supports regex and substring matching (e.g., 'USB', 'bcm2835', 'AirPods').
    Returns None to use the system default if spec is None or device not found.
    """
    if spec is None:
        return None

    # Direct integer index
    if isinstance(spec, int):
        return spec

    spec_str = str(spec).strip()
    if not spec_str:
        return None

    # Check if spec is an integer formatted as string
    if spec_str.isdigit():
        return int(spec_str)

    all_devices = cls.list_devices()
    candidates = [d for d in all_devices if (d.is_input if is_input else d.is_output)]

    # 1. Exact case-insensitive match
    for dev in candidates:
        if dev.name.lower() == spec_str.lower():
            return dev.index

    # 2. Substring match
    for dev in candidates:
        if spec_str.lower() in dev.name.lower():
            return dev.index

    # 3. Regex match
    try:
        pattern = re.compile(spec_str, re.IGNORECASE)
        for dev in candidates:
            if pattern.search(dev.name):
                return dev.index
    except re.error:
        pass

    return None

list_devices classmethod 

list_devices()

Returns all audio devices discovered on the host system.

Source code in edgevoiceai/audio/device.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def list_devices(cls) -> list[AudioDeviceInfo]:
    """Returns all audio devices discovered on the host system."""
    if not PYAUDIO_AVAILABLE or pyaudio is None:
        return []

    pa = pyaudio.PyAudio()
    devices: list[AudioDeviceInfo] = []
    try:
        count = pa.get_device_count()
        for i in range(count):
            try:
                info = pa.get_device_info_by_index(i)
                host_api_idx = int(info.get("hostApi", 0))
                host_api_info = pa.get_host_api_info_by_index(host_api_idx)
                api_name = host_api_info.get("name", "Unknown") if host_api_info else "Unknown"

                devices.append(
                    AudioDeviceInfo(
                        index=i,
                        name=str(info.get("name", f"Device {i}")),
                        max_input_channels=int(info.get("maxInputChannels", 0)),
                        max_output_channels=int(info.get("maxOutputChannels", 0)),
                        default_sample_rate=float(info.get("defaultSampleRate", 16000.0)),
                        host_api=str(api_name),
                    )
                )
            except Exception:
                continue
    finally:
        pa.terminate()

    return devices

print_diagnostics classmethod 

print_diagnostics()

Prints a user-friendly terminal table of available audio devices.

Source code in edgevoiceai/audio/device.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
@classmethod
def print_diagnostics(cls) -> None:
    """Prints a user-friendly terminal table of available audio devices."""
    devices = cls.list_devices()
    print("\n" + "=" * 65)
    print("🔊 [VOICE FRAMEWORK - AUDIO HARDWARE DIAGNOSTICS]")
    print("=" * 65)

    inputs = [d for d in devices if d.is_input]
    outputs = [d for d in devices if d.is_output]

    print("\n🎤 Microphone / Input Devices:")
    if not inputs:
        print("  (None detected)")
    for d in inputs:
        print(f"  • [{d.index}] {d.name} ({int(d.default_sample_rate)}Hz, {d.host_api})")

    print("\n🔈 Speaker / Output Devices:")
    if not outputs:
        print("  (None detected)")
    for d in outputs:
        print(f"  • [{d.index}] {d.name} ({int(d.default_sample_rate)}Hz, {d.host_api})")
    print("=" * 65 + "\n")

AudioDeviceInfo dataclass

Represents a discovered audio device with hardware characteristics.

Source code in edgevoiceai/audio/device.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass(slots=True)
class AudioDeviceInfo:
    """Represents a discovered audio device with hardware characteristics."""
    index: int
    name: str
    max_input_channels: int
    max_output_channels: int
    default_sample_rate: float
    host_api: str = "Unknown"

    @property
    def is_input(self) -> bool:
        return self.max_input_channels > 0

    @property
    def is_output(self) -> bool:
        return self.max_output_channels > 0

MicrophoneStream & AudioPlayer

edgevoiceai.audio.stream

Microphone Audio Input Stream. Captures continuous real-time audio from microphones on Raspberry Pi, macOS, Linux, and Windows. Automatically handles hardware rate negotiation and conversion to 16 kHz int16 mono PCM.

MicrophoneStream

Continuous microphone input capture stream. Opens PyAudio stream on the chosen hardware device and delivers standardized 16kHz mono PCM chunks to an async or threaded queue.

Source code in edgevoiceai/audio/stream.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class MicrophoneStream:
    """
    Continuous microphone input capture stream.
    Opens PyAudio stream on the chosen hardware device and delivers
    standardized 16kHz mono PCM chunks to an async or threaded queue.
    """

    def __init__(
        self,
        pa: Any | None = None,
        device_index: int | None = None,
        capture_rate: int = 16000,
        target_rate: int = 16000,
        channels: int = 1,
        frames_per_buffer: int = 512,
        max_queue_size: int = 300,
    ) -> None:
        self.device_index = device_index
        self.capture_rate = capture_rate
        self.target_rate = target_rate
        self.channels = channels
        self.frames_per_buffer = frames_per_buffer
        self.queue: queue.Queue[bytes | None] = queue.Queue(maxsize=max_queue_size)
        self.shutdown_event = threading.Event()
        self._pa_owner: bool = False
        self._stream: Any = None
        self._ratecv_state: Any = None

        if not PYAUDIO_AVAILABLE or pyaudio is None:
            return

        if pa is None:
            self._pa = pyaudio.PyAudio()
            self._pa_owner = True
        else:
            self._pa = pa

        # Hardware rate negotiation: test capture_rate, fallback to 48000 or 16000 if device complains
        actual_rate = self._negotiate_sample_rate()
        self.capture_rate = actual_rate

        try:
            self._stream = self._pa.open(
                format=pyaudio.paInt16,
                channels=channels,
                rate=actual_rate,
                input=True,
                input_device_index=device_index,
                frames_per_buffer=frames_per_buffer,
                stream_callback=self._audio_callback,
            )
            self._stream.start_stream()
        except Exception as exc:
            print(f"⚠️  [MicrophoneStream] Failed to open microphone input ({actual_rate}Hz, dev={device_index}): {exc}")

    def _negotiate_sample_rate(self) -> int:
        """Finds a compatible hardware capture rate supported by PortAudio."""
        candidate_rates = [16000, self.capture_rate, 48000, 44100]
        for rate in candidate_rates:
            try:
                supported = self._pa.is_format_supported(
                    rate=rate,
                    input_device=self.device_index,
                    input_channels=self.channels,
                    input_format=getattr(pyaudio, "paInt16", 2),
                )
                if supported:
                    return rate
            except Exception:
                continue
        return self.capture_rate

    def _audio_callback(self, in_data: Any, frame_count: int, time_info: Any, status: Any) -> tuple[None, int]:
        if self.shutdown_event.is_set():
            return None, getattr(pyaudio, "paComplete", 2)

        if not in_data:
            return None, getattr(pyaudio, "paContinue", 0)

        # Convert to 16kHz if needed
        if self.capture_rate != self.target_rate and audioop is not None:
            try:
                resampled, self._ratecv_state = audioop.ratecv(
                    in_data,
                    2,  # 2 bytes per sample (int16)
                    self.channels,
                    self.capture_rate,
                    self.target_rate,
                    self._ratecv_state,
                )
                chunk_to_queue = resampled
            except Exception:
                chunk_to_queue = in_data
        else:
            chunk_to_queue = in_data

        try:
            self.queue.put_nowait(chunk_to_queue)
        except queue.Full:
            pass  # Drop oldest or skip if consumer blocked

        return None, getattr(pyaudio, "paContinue", 0)

    def get_chunk(self, block: bool = True, timeout: float | None = None) -> bytes | None:
        """Retrieves the next 16kHz PCM chunk."""
        try:
            return self.queue.get(block=block, timeout=timeout)
        except queue.Empty:
            return None

    def close(self) -> None:
        """Closes the audio input stream."""
        self.shutdown_event.set()
        if self._stream is not None:
            try:
                self._stream.stop_stream()
                self._stream.close()
            except Exception:
                pass
            self._stream = None

        if self._pa_owner and self._pa is not None:
            try:
                self._pa.terminate()
            except Exception:
                pass

close 

close()

Closes the audio input stream.

Source code in edgevoiceai/audio/stream.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def close(self) -> None:
    """Closes the audio input stream."""
    self.shutdown_event.set()
    if self._stream is not None:
        try:
            self._stream.stop_stream()
            self._stream.close()
        except Exception:
            pass
        self._stream = None

    if self._pa_owner and self._pa is not None:
        try:
            self._pa.terminate()
        except Exception:
            pass

get_chunk 

get_chunk(block=True, timeout=None)

Retrieves the next 16kHz PCM chunk.

Source code in edgevoiceai/audio/stream.py 
133
134
135
136
137
138
def get_chunk(self, block: bool = True, timeout: float | None = None) -> bytes | None:
    """Retrieves the next 16kHz PCM chunk."""
    try:
        return self.queue.get(block=block, timeout=timeout)
    except queue.Empty:
        return None

MediaAudioPlayer

edgevoiceai.audio.media

Media Audio Player (MP3/WAV playback). Plays pre-recorded audio files on speakers with headless Linux/Raspberry Pi support, background thread monitoring, and instant interruption cancellation.

MediaAudioPlayer

Manages physical audio file playback (e.g. farmer experiences, music, prompts) with non-blocking execution and event hooks.

Source code in edgevoiceai/audio/media.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class MediaAudioPlayer:
    """
    Manages physical audio file playback (e.g. farmer experiences, music, prompts)
    with non-blocking execution and event hooks.
    """

    def __init__(self, media_dir: str = "audio_experiences") -> None:
        self.media_dir = media_dir
        self._playing_event = threading.Event()
        self._initialized = False

    def _init_pygame(self) -> None:
        if self._initialized:
            return
        if PYGAME_AVAILABLE and pygame is not None:
            try:
                if not pygame.mixer.get_init():
                    pygame.mixer.init()
                self._initialized = True
            except Exception as exc:
                print(f"⚠️  [MediaAudioPlayer] pygame mixer init warning: {exc}")

    def play(
        self,
        audio_url_or_path: str,
        on_complete: Callable[[], None] | None = None,
    ) -> bool:
        """
        Plays an MP3 or WAV file asynchronously on a background thread.
        Returns True if playback started successfully.
        """
        if not PYGAME_AVAILABLE or pygame is None:
            print("⚠️  [MediaAudioPlayer] pygame unavailable — skipping audio file playback")
            return False

        self._init_pygame()

        # Extract filename if URL format (e.g., /api/experiences/audio/valsad.mp3)
        clean_path = audio_url_or_path.replace("\\", "/")
        filename = clean_path.split("/")[-1]

        # Check both direct path and media_dir path
        filepath = clean_path if os.path.isfile(clean_path) else os.path.join(self.media_dir, filename)

        if not os.path.isfile(filepath):
            print(f"⚠️  [MediaAudioPlayer] File not found: {filepath}")
            return False

        try:
            self._init_pygame()
            pygame.mixer.music.load(filepath)
            pygame.mixer.music.play()
            self._playing_event.set()
            print(f"▶️  [MediaAudioPlayer] Playing audio file: {filename}")

            def _monitor() -> None:
                while pygame is not None and pygame.mixer.music.get_busy():
                    time.sleep(0.15)
                if self._playing_event.is_set():
                    self._playing_event.clear()
                    if on_complete:
                        try:
                            on_complete()
                        except Exception as cb_err:
                            print(f"⚠️  [MediaAudioPlayer] Completion callback error: {cb_err}")

            threading.Thread(target=_monitor, daemon=True).start()
            return True
        except Exception as exc:
            print(f"⚠️  [MediaAudioPlayer] Playback error: {exc}")
            self._playing_event.clear()
            return False

    def is_playing(self) -> bool:
        """Returns True if a media file is actively playing."""
        return self._playing_event.is_set()

    def stop(self) -> None:
        """Immediately stops media audio playback."""
        if not PYGAME_AVAILABLE or pygame is None:
            return
        self._playing_event.clear()
        try:
            if pygame.mixer.get_init() and pygame.mixer.music.get_busy():
                pygame.mixer.music.stop()
                print("⏹️  [MediaAudioPlayer] Audio playback stopped.")
        except Exception:
            pass

    def close(self) -> None:
        """Closes the audio subsystem."""
        self.stop()
        if PYGAME_AVAILABLE and pygame is not None:
            try:
                pygame.mixer.quit()
            except Exception:
                pass

close 

close()

Closes the audio subsystem.

Source code in edgevoiceai/audio/media.py 
115
116
117
118
119
120
121
122
def close(self) -> None:
    """Closes the audio subsystem."""
    self.stop()
    if PYGAME_AVAILABLE and pygame is not None:
        try:
            pygame.mixer.quit()
        except Exception:
            pass

is_playing 

is_playing()

Returns True if a media file is actively playing.

Source code in edgevoiceai/audio/media.py 
 99
100
101
def is_playing(self) -> bool:
    """Returns True if a media file is actively playing."""
    return self._playing_event.is_set()

play 

play(audio_url_or_path, on_complete=None)

Plays an MP3 or WAV file asynchronously on a background thread. Returns True if playback started successfully.

Source code in edgevoiceai/audio/media.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def play(
    self,
    audio_url_or_path: str,
    on_complete: Callable[[], None] | None = None,
) -> bool:
    """
    Plays an MP3 or WAV file asynchronously on a background thread.
    Returns True if playback started successfully.
    """
    if not PYGAME_AVAILABLE or pygame is None:
        print("⚠️  [MediaAudioPlayer] pygame unavailable — skipping audio file playback")
        return False

    self._init_pygame()

    # Extract filename if URL format (e.g., /api/experiences/audio/valsad.mp3)
    clean_path = audio_url_or_path.replace("\\", "/")
    filename = clean_path.split("/")[-1]

    # Check both direct path and media_dir path
    filepath = clean_path if os.path.isfile(clean_path) else os.path.join(self.media_dir, filename)

    if not os.path.isfile(filepath):
        print(f"⚠️  [MediaAudioPlayer] File not found: {filepath}")
        return False

    try:
        self._init_pygame()
        pygame.mixer.music.load(filepath)
        pygame.mixer.music.play()
        self._playing_event.set()
        print(f"▶️  [MediaAudioPlayer] Playing audio file: {filename}")

        def _monitor() -> None:
            while pygame is not None and pygame.mixer.music.get_busy():
                time.sleep(0.15)
            if self._playing_event.is_set():
                self._playing_event.clear()
                if on_complete:
                    try:
                        on_complete()
                    except Exception as cb_err:
                        print(f"⚠️  [MediaAudioPlayer] Completion callback error: {cb_err}")

        threading.Thread(target=_monitor, daemon=True).start()
        return True
    except Exception as exc:
        print(f"⚠️  [MediaAudioPlayer] Playback error: {exc}")
        self._playing_event.clear()
        return False

stop 

stop()

Immediately stops media audio playback.

Source code in edgevoiceai/audio/media.py 
103
104
105
106
107
108
109
110
111
112
113
def stop(self) -> None:
    """Immediately stops media audio playback."""
    if not PYGAME_AVAILABLE or pygame is None:
        return
    self._playing_event.clear()
    try:
        if pygame.mixer.get_init() and pygame.mixer.music.get_busy():
            pygame.mixer.music.stop()
            print("⏹️  [MediaAudioPlayer] Audio playback stopped.")
    except Exception:
        pass