From fd548d45f139f657f64b961a1893cae6adcd2a16 Mon Sep 17 00:00:00 2001 From: Thomas Farstrike Date: Tue, 13 Jan 2026 23:48:52 +0100 Subject: [PATCH] AudioFlinger framework: simplify import, use singleton class --- CHANGELOG.md | 4 +- .../assets/music_player.py | 2 +- .../assets/sound_recorder.py | 2 +- internal_filesystem/lib/mpos/__init__.py | 3 +- .../lib/mpos/audio/__init__.py | 51 +- .../lib/mpos/audio/audioflinger.py | 781 ++++++++++-------- .../lib/mpos/board/fri3d_2024.py | 2 +- internal_filesystem/lib/mpos/board/linux.py | 2 +- .../board/waveshare_esp32_s3_touch_lcd_2.py | 2 +- internal_filesystem/lib/mpos/info.py | 2 +- scripts/addr2line.sh | 0 tests/test_audioflinger.py | 13 +- 12 files changed, 503 insertions(+), 361 deletions(-) mode change 100644 => 100755 scripts/addr2line.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 93ba8690..6add2877 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,12 @@ -0.5.3 +0.6.0 ===== - AppStore app: add Settings screen to choose backend +- Camera app: fix aspect ratio for higher resolutions - WiFi app: check "hidden" in EditNetwork - Wifi app: add support for scanning wifi QR codes to "Add Network" - Make "Power Off" button on desktop exit completely - App framework: simplify MANIFEST.JSON +- AudioFlinger framework: simplify import, use singleton class - Create new SettingsActivity and SettingActivity framework so apps can easily add settings screens with just a few lines of code - Improve robustness by catching unhandled app exceptions - Improve robustness with custom exception that does not deinit() the TaskHandler diff --git a/internal_filesystem/apps/com.micropythonos.musicplayer/assets/music_player.py b/internal_filesystem/apps/com.micropythonos.musicplayer/assets/music_player.py index 1844d35d..c648a61a 100644 --- a/internal_filesystem/apps/com.micropythonos.musicplayer/assets/music_player.py +++ b/internal_filesystem/apps/com.micropythonos.musicplayer/assets/music_player.py @@ -2,7 +2,7 @@ import machine import os import time -from mpos import Activity, Intent, sdcard, get_event_name, audio as AudioFlinger +from mpos import Activity, Intent, sdcard, get_event_name, AudioFlinger class MusicPlayer(Activity): diff --git a/internal_filesystem/apps/com.micropythonos.soundrecorder/assets/sound_recorder.py b/internal_filesystem/apps/com.micropythonos.soundrecorder/assets/sound_recorder.py index 62a9822b..dd203498 100644 --- a/internal_filesystem/apps/com.micropythonos.soundrecorder/assets/sound_recorder.py +++ b/internal_filesystem/apps/com.micropythonos.soundrecorder/assets/sound_recorder.py @@ -2,7 +2,7 @@ import os import time -from mpos import Activity, ui, audio as AudioFlinger +from mpos import Activity, ui, AudioFlinger def _makedirs(path): diff --git a/internal_filesystem/lib/mpos/__init__.py b/internal_filesystem/lib/mpos/__init__.py index 49223fad..8ef39018 100644 --- a/internal_filesystem/lib/mpos/__init__.py +++ b/internal_filesystem/lib/mpos/__init__.py @@ -5,6 +5,7 @@ from .config import SharedPreferences from .net.connectivity_manager import ConnectivityManager from .net import download_manager as DownloadManager from .net.wifi_service import WifiService +from .audio.audioflinger import AudioFlinger from .content.intent import Intent from .activity_navigator import ActivityNavigator from .content.package_manager import PackageManager @@ -61,7 +62,7 @@ __all__ = [ "App", "Activity", "SharedPreferences", - "ConnectivityManager", "DownloadManager", "WifiService", "Intent", + "ConnectivityManager", "DownloadManager", "WifiService", "AudioFlinger", "Intent", "ActivityNavigator", "PackageManager", "TaskManager", # Common activities "ChooserActivity", "ViewActivity", "ShareActivity", diff --git a/internal_filesystem/lib/mpos/audio/__init__.py b/internal_filesystem/lib/mpos/audio/__init__.py index 37be5058..848fddc2 100644 --- a/internal_filesystem/lib/mpos/audio/__init__.py +++ b/internal_filesystem/lib/mpos/audio/__init__.py @@ -2,37 +2,36 @@ # Android-inspired audio routing with priority-based audio focus # Simple routing: play_wav() -> I2S, play_rtttl() -> buzzer, record_wav() -> I2S mic -from . import audioflinger +from .audioflinger import AudioFlinger -# Re-export main API -from .audioflinger import ( - # Stream types (for priority-based audio focus) - STREAM_MUSIC, - STREAM_NOTIFICATION, - STREAM_ALARM, +# Create singleton instance +_instance = AudioFlinger.get() - # Core playback functions - init, - play_wav, - play_rtttl, - stop, - pause, - resume, - set_volume, - get_volume, - is_playing, +# Re-export stream type constants for convenience +STREAM_MUSIC = AudioFlinger.STREAM_MUSIC +STREAM_NOTIFICATION = AudioFlinger.STREAM_NOTIFICATION +STREAM_ALARM = AudioFlinger.STREAM_ALARM - # Recording functions - record_wav, - is_recording, - - # Hardware availability checks - has_i2s, - has_buzzer, - has_microphone, -) +# Re-export main API from singleton instance for backward compatibility +init = _instance.init +play_wav = _instance.play_wav +play_rtttl = _instance.play_rtttl +stop = _instance.stop +pause = _instance.pause +resume = _instance.resume +set_volume = _instance.set_volume +get_volume = _instance.get_volume +is_playing = _instance.is_playing +record_wav = _instance.record_wav +is_recording = _instance.is_recording +has_i2s = _instance.has_i2s +has_buzzer = _instance.has_buzzer +has_microphone = _instance.has_microphone __all__ = [ + # Class + 'AudioFlinger', + # Stream types 'STREAM_MUSIC', 'STREAM_NOTIFICATION', diff --git a/internal_filesystem/lib/mpos/audio/audioflinger.py b/internal_filesystem/lib/mpos/audio/audioflinger.py index 031c3956..d49e3286 100644 --- a/internal_filesystem/lib/mpos/audio/audioflinger.py +++ b/internal_filesystem/lib/mpos/audio/audioflinger.py @@ -8,363 +8,500 @@ import _thread import mpos.apps -# Stream type constants (priority order: higher number = higher priority) -STREAM_MUSIC = 0 # Background music (lowest priority) -STREAM_NOTIFICATION = 1 # Notification sounds (medium priority) -STREAM_ALARM = 2 # Alarms/alerts (highest priority) -# Module-level state (singleton pattern, follows battery_voltage.py) -_i2s_pins = None # I2S pin configuration dict (created per-stream) -_buzzer_instance = None # PWM buzzer instance -_current_stream = None # Currently playing stream -_current_recording = None # Currently recording stream -_volume = 50 # System volume (0-100) - - -def init(i2s_pins=None, buzzer_instance=None): +class AudioFlinger: """ - Initialize AudioFlinger with hardware configuration. - - Args: - i2s_pins: Dict with 'sck', 'ws', 'sd' pin numbers (for I2S/WAV playback) - buzzer_instance: PWM instance for buzzer (for RTTTL playback) - """ - global _i2s_pins, _buzzer_instance - - _i2s_pins = i2s_pins - _buzzer_instance = buzzer_instance - - # Build status message - capabilities = [] - if i2s_pins: - capabilities.append("I2S (WAV)") - if buzzer_instance: - capabilities.append("Buzzer (RTTTL)") + Centralized audio management service with priority-based audio focus. + Implements singleton pattern for single audio service instance. - if capabilities: - print(f"AudioFlinger initialized: {', '.join(capabilities)}") - else: - print("AudioFlinger initialized: No audio hardware") - - -def has_i2s(): - """Check if I2S audio is available for WAV playback.""" - return _i2s_pins is not None - - -def has_buzzer(): - """Check if buzzer is available for RTTTL playback.""" - return _buzzer_instance is not None - - -def has_microphone(): - """Check if I2S microphone is available for recording.""" - return _i2s_pins is not None and 'sd_in' in _i2s_pins - - -def _check_audio_focus(stream_type): + Usage: + from mpos import AudioFlinger + + # Direct class method calls (no .get() needed) + AudioFlinger.init(i2s_pins=pins, buzzer_instance=buzzer) + AudioFlinger.play_wav("music.wav", stream_type=AudioFlinger.STREAM_MUSIC) + AudioFlinger.set_volume(80) + volume = AudioFlinger.get_volume() + AudioFlinger.stop() """ - Check if a stream with the given type can start playback. - Implements priority-based audio focus (Android-inspired). + + # Stream type constants (priority order: higher number = higher priority) + STREAM_MUSIC = 0 # Background music (lowest priority) + STREAM_NOTIFICATION = 1 # Notification sounds (medium priority) + STREAM_ALARM = 2 # Alarms/alerts (highest priority) + + _instance = None # Singleton instance + + def __init__(self): + """Initialize AudioFlinger instance.""" + if AudioFlinger._instance: + return + AudioFlinger._instance = self + + self._i2s_pins = None # I2S pin configuration dict (created per-stream) + self._buzzer_instance = None # PWM buzzer instance + self._current_stream = None # Currently playing stream + self._current_recording = None # Currently recording stream + self._volume = 50 # System volume (0-100) + + @classmethod + def get(cls): + """Get or create the singleton instance.""" + if cls._instance is None: + cls._instance = cls() + return cls._instance + + def init(self, i2s_pins=None, buzzer_instance=None): + """ + Initialize AudioFlinger with hardware configuration. - Args: - stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) + Args: + i2s_pins: Dict with 'sck', 'ws', 'sd' pin numbers (for I2S/WAV playback) + buzzer_instance: PWM instance for buzzer (for RTTTL playback) + """ + self._i2s_pins = i2s_pins + self._buzzer_instance = buzzer_instance - Returns: - bool: True if stream can start, False if rejected - """ - global _current_stream + # Build status message + capabilities = [] + if i2s_pins: + capabilities.append("I2S (WAV)") + if buzzer_instance: + capabilities.append("Buzzer (RTTTL)") + + if capabilities: + print(f"AudioFlinger initialized: {', '.join(capabilities)}") + else: + print("AudioFlinger initialized: No audio hardware") - if not _current_stream: - return True # No stream playing, OK to start + def has_i2s(self): + """Check if I2S audio is available for WAV playback.""" + return self._i2s_pins is not None - if not _current_stream.is_playing(): - return True # Current stream finished, OK to start + def has_buzzer(self): + """Check if buzzer is available for RTTTL playback.""" + return self._buzzer_instance is not None - # Check priority - if stream_type <= _current_stream.stream_type: - print(f"AudioFlinger: Stream rejected (priority {stream_type} <= current {_current_stream.stream_type})") - return False + def has_microphone(self): + """Check if I2S microphone is available for recording.""" + return self._i2s_pins is not None and 'sd_in' in self._i2s_pins - # Higher priority stream - interrupt current - print(f"AudioFlinger: Interrupting stream (priority {stream_type} > current {_current_stream.stream_type})") - _current_stream.stop() - return True + def _check_audio_focus(self, stream_type): + """ + Check if a stream with the given type can start playback. + Implements priority-based audio focus (Android-inspired). + Args: + stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) -def _playback_thread(stream): - """ - Thread function for audio playback. - Runs in a separate thread to avoid blocking the UI. + Returns: + bool: True if stream can start, False if rejected + """ + if not self._current_stream: + return True # No stream playing, OK to start - Args: - stream: Stream instance (WAVStream or RTTTLStream) - """ - global _current_stream + if not self._current_stream.is_playing(): + return True # Current stream finished, OK to start - _current_stream = stream + # Check priority + if stream_type <= self._current_stream.stream_type: + print(f"AudioFlinger: Stream rejected (priority {stream_type} <= current {self._current_stream.stream_type})") + return False - try: - # Run synchronous playback in this thread - stream.play() - except Exception as e: - print(f"AudioFlinger: Playback error: {e}") - finally: - # Clear current stream - if _current_stream == stream: - _current_stream = None - - -def play_wav(file_path, stream_type=STREAM_MUSIC, volume=None, on_complete=None): - """ - Play WAV file via I2S. - - Args: - file_path: Path to WAV file (e.g., "M:/sdcard/music/song.wav") - stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) - volume: Override volume (0-100), or None to use system volume - on_complete: Callback function(message) called when playback finishes - - Returns: - bool: True if playback started, False if rejected or unavailable - """ - if not _i2s_pins: - print("AudioFlinger: play_wav() failed - I2S not configured") - return False - - # Check audio focus - if not _check_audio_focus(stream_type): - return False - - # Create stream and start playback in separate thread - try: - from mpos.audio.stream_wav import WAVStream - - stream = WAVStream( - file_path=file_path, - stream_type=stream_type, - volume=volume if volume is not None else _volume, - i2s_pins=_i2s_pins, - on_complete=on_complete - ) - - _thread.stack_size(mpos.apps.good_stack_size()) - _thread.start_new_thread(_playback_thread, (stream,)) + # Higher priority stream - interrupt current + print(f"AudioFlinger: Interrupting stream (priority {stream_type} > current {self._current_stream.stream_type})") + self._current_stream.stop() return True - except Exception as e: - print(f"AudioFlinger: play_wav() failed: {e}") - return False + def _playback_thread(self, stream): + """ + Thread function for audio playback. + Runs in a separate thread to avoid blocking the UI. + + Args: + stream: Stream instance (WAVStream or RTTTLStream) + """ + self._current_stream = stream + + try: + # Run synchronous playback in this thread + stream.play() + except Exception as e: + print(f"AudioFlinger: Playback error: {e}") + finally: + # Clear current stream + if self._current_stream == stream: + self._current_stream = None + + def play_wav(self, file_path, stream_type=None, volume=None, on_complete=None): + """ + Play WAV file via I2S. + + Args: + file_path: Path to WAV file (e.g., "M:/sdcard/music/song.wav") + stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) + volume: Override volume (0-100), or None to use system volume + on_complete: Callback function(message) called when playback finishes + + Returns: + bool: True if playback started, False if rejected or unavailable + """ + if stream_type is None: + stream_type = self.STREAM_MUSIC + + if not self._i2s_pins: + print("AudioFlinger: play_wav() failed - I2S not configured") + return False + + # Check audio focus + if not self._check_audio_focus(stream_type): + return False + + # Create stream and start playback in separate thread + try: + from mpos.audio.stream_wav import WAVStream + + stream = WAVStream( + file_path=file_path, + stream_type=stream_type, + volume=volume if volume is not None else self._volume, + i2s_pins=self._i2s_pins, + on_complete=on_complete + ) + + _thread.stack_size(mpos.apps.good_stack_size()) + _thread.start_new_thread(self._playback_thread, (stream,)) + return True + + except Exception as e: + print(f"AudioFlinger: play_wav() failed: {e}") + return False + + def play_rtttl(self, rtttl_string, stream_type=None, volume=None, on_complete=None): + """ + Play RTTTL ringtone via buzzer. + + Args: + rtttl_string: RTTTL format string (e.g., "Nokia:d=4,o=5,b=225:8e6,8d6...") + stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) + volume: Override volume (0-100), or None to use system volume + on_complete: Callback function(message) called when playback finishes + + Returns: + bool: True if playback started, False if rejected or unavailable + """ + if stream_type is None: + stream_type = self.STREAM_NOTIFICATION + + if not self._buzzer_instance: + print("AudioFlinger: play_rtttl() failed - buzzer not configured") + return False + + # Check audio focus + if not self._check_audio_focus(stream_type): + return False + + # Create stream and start playback in separate thread + try: + from mpos.audio.stream_rtttl import RTTTLStream + + stream = RTTTLStream( + rtttl_string=rtttl_string, + stream_type=stream_type, + volume=volume if volume is not None else self._volume, + buzzer_instance=self._buzzer_instance, + on_complete=on_complete + ) + + _thread.stack_size(mpos.apps.good_stack_size()) + _thread.start_new_thread(self._playback_thread, (stream,)) + return True + + except Exception as e: + print(f"AudioFlinger: play_rtttl() failed: {e}") + return False + + def _recording_thread(self, stream): + """ + Thread function for audio recording. + Runs in a separate thread to avoid blocking the UI. + + Args: + stream: RecordStream instance + """ + self._current_recording = stream + + try: + # Run synchronous recording in this thread + stream.record() + except Exception as e: + print(f"AudioFlinger: Recording error: {e}") + finally: + # Clear current recording + if self._current_recording == stream: + self._current_recording = None + + def record_wav(self, file_path, duration_ms=None, on_complete=None, sample_rate=16000): + """ + Record audio from I2S microphone to WAV file. + + Args: + file_path: Path to save WAV file (e.g., "data/recording.wav") + duration_ms: Recording duration in milliseconds (None = 60 seconds default) + on_complete: Callback function(message) when recording finishes + sample_rate: Sample rate in Hz (default 16000 for voice) + + Returns: + bool: True if recording started, False if rejected or unavailable + """ + print(f"AudioFlinger.record_wav() called") + print(f" file_path: {file_path}") + print(f" duration_ms: {duration_ms}") + print(f" sample_rate: {sample_rate}") + print(f" _i2s_pins: {self._i2s_pins}") + print(f" has_microphone(): {self.has_microphone()}") + + if not self.has_microphone(): + print("AudioFlinger: record_wav() failed - microphone not configured") + return False + + # Cannot record while playing (I2S can only be TX or RX, not both) + if self.is_playing(): + print("AudioFlinger: Cannot record while playing") + return False + + # Cannot start new recording while already recording + if self.is_recording(): + print("AudioFlinger: Already recording") + return False + + # Create stream and start recording in separate thread + try: + print("AudioFlinger: Importing RecordStream...") + from mpos.audio.stream_record import RecordStream + + print("AudioFlinger: Creating RecordStream instance...") + stream = RecordStream( + file_path=file_path, + duration_ms=duration_ms, + sample_rate=sample_rate, + i2s_pins=self._i2s_pins, + on_complete=on_complete + ) + + print("AudioFlinger: Starting recording thread...") + _thread.stack_size(mpos.apps.good_stack_size()) + _thread.start_new_thread(self._recording_thread, (stream,)) + print("AudioFlinger: Recording thread started successfully") + return True + + except Exception as e: + import sys + print(f"AudioFlinger: record_wav() failed: {e}") + sys.print_exception(e) + return False + + def stop(self): + """Stop current audio playback or recording.""" + stopped = False + + if self._current_stream: + self._current_stream.stop() + print("AudioFlinger: Playback stopped") + stopped = True + + if self._current_recording: + self._current_recording.stop() + print("AudioFlinger: Recording stopped") + stopped = True + + if not stopped: + print("AudioFlinger: No playback or recording to stop") + + def pause(self): + """ + Pause current audio playback (if supported by stream). + Note: Most streams don't support pause, only stop. + """ + if self._current_stream and hasattr(self._current_stream, 'pause'): + self._current_stream.pause() + print("AudioFlinger: Playback paused") + else: + print("AudioFlinger: Pause not supported or no playback active") + + def resume(self): + """ + Resume paused audio playback (if supported by stream). + Note: Most streams don't support resume, only play. + """ + if self._current_stream and hasattr(self._current_stream, 'resume'): + self._current_stream.resume() + print("AudioFlinger: Playback resumed") + else: + print("AudioFlinger: Resume not supported or no playback active") + + def set_volume(self, volume): + """ + Set system volume (affects new streams, not current playback). + + Args: + volume: Volume level (0-100) + """ + self._volume = max(0, min(100, volume)) + if self._current_stream: + self._current_stream.set_volume(self._volume) + + def get_volume(self): + """ + Get system volume. + + Returns: + int: Current system volume (0-100) + """ + return self._volume + + def is_playing(self): + """ + Check if audio is currently playing. + + Returns: + bool: True if playback active, False otherwise + """ + return self._current_stream is not None and self._current_stream.is_playing() + + def is_recording(self): + """ + Check if audio is currently being recorded. + + Returns: + bool: True if recording active, False otherwise + """ + return self._current_recording is not None and self._current_recording.is_recording() -def play_rtttl(rtttl_string, stream_type=STREAM_NOTIFICATION, volume=None, on_complete=None): - """ - Play RTTTL ringtone via buzzer. +# ============================================================================ +# Class methods that delegate to singleton instance (like DownloadManager) +# ============================================================================ - Args: - rtttl_string: RTTTL format string (e.g., "Nokia:d=4,o=5,b=225:8e6,8d6...") - stream_type: Stream type (STREAM_MUSIC, STREAM_NOTIFICATION, STREAM_ALARM) - volume: Override volume (0-100), or None to use system volume - on_complete: Callback function(message) called when playback finishes - - Returns: - bool: True if playback started, False if rejected or unavailable - """ - if not _buzzer_instance: - print("AudioFlinger: play_rtttl() failed - buzzer not configured") - return False - - # Check audio focus - if not _check_audio_focus(stream_type): - return False - - # Create stream and start playback in separate thread - try: - from mpos.audio.stream_rtttl import RTTTLStream - - stream = RTTTLStream( - rtttl_string=rtttl_string, - stream_type=stream_type, - volume=volume if volume is not None else _volume, - buzzer_instance=_buzzer_instance, - on_complete=on_complete - ) - - _thread.stack_size(mpos.apps.good_stack_size()) - _thread.start_new_thread(_playback_thread, (stream,)) - return True - - except Exception as e: - print(f"AudioFlinger: play_rtttl() failed: {e}") - return False +# Store original instance methods before creating class methods +_init_impl = AudioFlinger.init +_play_wav_impl = AudioFlinger.play_wav +_play_rtttl_impl = AudioFlinger.play_rtttl +_record_wav_impl = AudioFlinger.record_wav +_stop_impl = AudioFlinger.stop +_pause_impl = AudioFlinger.pause +_resume_impl = AudioFlinger.resume +_set_volume_impl = AudioFlinger.set_volume +_get_volume_impl = AudioFlinger.get_volume +_is_playing_impl = AudioFlinger.is_playing +_is_recording_impl = AudioFlinger.is_recording +_has_i2s_impl = AudioFlinger.has_i2s +_has_buzzer_impl = AudioFlinger.has_buzzer +_has_microphone_impl = AudioFlinger.has_microphone -def _recording_thread(stream): - """ - Thread function for audio recording. - Runs in a separate thread to avoid blocking the UI. +# Create class methods that delegate to singleton +@classmethod +def init(cls, i2s_pins=None, buzzer_instance=None): + """Initialize AudioFlinger with hardware configuration.""" + return cls.get()._init_impl(i2s_pins=i2s_pins, buzzer_instance=buzzer_instance) - Args: - stream: RecordStream instance - """ - global _current_recording +@classmethod +def play_wav(cls, file_path, stream_type=None, volume=None, on_complete=None): + """Play WAV file via I2S.""" + return cls.get()._play_wav_impl(file_path=file_path, stream_type=stream_type, + volume=volume, on_complete=on_complete) - _current_recording = stream +@classmethod +def play_rtttl(cls, rtttl_string, stream_type=None, volume=None, on_complete=None): + """Play RTTTL ringtone via buzzer.""" + return cls.get()._play_rtttl_impl(rtttl_string=rtttl_string, stream_type=stream_type, + volume=volume, on_complete=on_complete) - try: - # Run synchronous recording in this thread - stream.record() - except Exception as e: - print(f"AudioFlinger: Recording error: {e}") - finally: - # Clear current recording - if _current_recording == stream: - _current_recording = None +@classmethod +def record_wav(cls, file_path, duration_ms=None, on_complete=None, sample_rate=16000): + """Record audio from I2S microphone to WAV file.""" + return cls.get()._record_wav_impl(file_path=file_path, duration_ms=duration_ms, + on_complete=on_complete, sample_rate=sample_rate) - -def record_wav(file_path, duration_ms=None, on_complete=None, sample_rate=16000): - """ - Record audio from I2S microphone to WAV file. - - Args: - file_path: Path to save WAV file (e.g., "data/recording.wav") - duration_ms: Recording duration in milliseconds (None = 60 seconds default) - on_complete: Callback function(message) when recording finishes - sample_rate: Sample rate in Hz (default 16000 for voice) - - Returns: - bool: True if recording started, False if rejected or unavailable - """ - print(f"AudioFlinger.record_wav() called") - print(f" file_path: {file_path}") - print(f" duration_ms: {duration_ms}") - print(f" sample_rate: {sample_rate}") - print(f" _i2s_pins: {_i2s_pins}") - print(f" has_microphone(): {has_microphone()}") - - if not has_microphone(): - print("AudioFlinger: record_wav() failed - microphone not configured") - return False - - # Cannot record while playing (I2S can only be TX or RX, not both) - if is_playing(): - print("AudioFlinger: Cannot record while playing") - return False - - # Cannot start new recording while already recording - if is_recording(): - print("AudioFlinger: Already recording") - return False - - # Create stream and start recording in separate thread - try: - print("AudioFlinger: Importing RecordStream...") - from mpos.audio.stream_record import RecordStream - - print("AudioFlinger: Creating RecordStream instance...") - stream = RecordStream( - file_path=file_path, - duration_ms=duration_ms, - sample_rate=sample_rate, - i2s_pins=_i2s_pins, - on_complete=on_complete - ) - - print("AudioFlinger: Starting recording thread...") - _thread.stack_size(mpos.apps.good_stack_size()) - _thread.start_new_thread(_recording_thread, (stream,)) - print("AudioFlinger: Recording thread started successfully") - return True - - except Exception as e: - import sys - print(f"AudioFlinger: record_wav() failed: {e}") - sys.print_exception(e) - return False - - -def stop(): +@classmethod +def stop(cls): """Stop current audio playback or recording.""" - global _current_stream, _current_recording + return cls.get()._stop_impl() - stopped = False +@classmethod +def pause(cls): + """Pause current audio playback.""" + return cls.get()._pause_impl() - if _current_stream: - _current_stream.stop() - print("AudioFlinger: Playback stopped") - stopped = True +@classmethod +def resume(cls): + """Resume paused audio playback.""" + return cls.get()._resume_impl() - if _current_recording: - _current_recording.stop() - print("AudioFlinger: Recording stopped") - stopped = True +@classmethod +def set_volume(cls, volume): + """Set system volume.""" + return cls.get()._set_volume_impl(volume) - if not stopped: - print("AudioFlinger: No playback or recording to stop") +@classmethod +def get_volume(cls): + """Get system volume.""" + return cls.get()._get_volume_impl() +@classmethod +def is_playing(cls): + """Check if audio is currently playing.""" + return cls.get()._is_playing_impl() -def pause(): - """ - Pause current audio playback (if supported by stream). - Note: Most streams don't support pause, only stop. - """ - if _current_stream and hasattr(_current_stream, 'pause'): - _current_stream.pause() - print("AudioFlinger: Playback paused") - else: - print("AudioFlinger: Pause not supported or no playback active") +@classmethod +def is_recording(cls): + """Check if audio is currently being recorded.""" + return cls.get()._is_recording_impl() +@classmethod +def has_i2s(cls): + """Check if I2S audio is available.""" + return cls.get()._has_i2s_impl() -def resume(): - """ - Resume paused audio playback (if supported by stream). - Note: Most streams don't support resume, only play. - """ - if _current_stream and hasattr(_current_stream, 'resume'): - _current_stream.resume() - print("AudioFlinger: Playback resumed") - else: - print("AudioFlinger: Resume not supported or no playback active") +@classmethod +def has_buzzer(cls): + """Check if buzzer is available.""" + return cls.get()._has_buzzer_impl() +@classmethod +def has_microphone(cls): + """Check if I2S microphone is available.""" + return cls.get()._has_microphone_impl() -def set_volume(volume): - """ - Set system volume (affects new streams, not current playback). +# Attach class methods to AudioFlinger class +AudioFlinger.init = init +AudioFlinger.play_wav = play_wav +AudioFlinger.play_rtttl = play_rtttl +AudioFlinger.record_wav = record_wav +AudioFlinger.stop = stop +AudioFlinger.pause = pause +AudioFlinger.resume = resume +AudioFlinger.set_volume = set_volume +AudioFlinger.get_volume = get_volume +AudioFlinger.is_playing = is_playing +AudioFlinger.is_recording = is_recording +AudioFlinger.has_i2s = has_i2s +AudioFlinger.has_buzzer = has_buzzer +AudioFlinger.has_microphone = has_microphone - Args: - volume: Volume level (0-100) - """ - global _volume - _volume = max(0, min(100, volume)) - if _current_stream: - _current_stream.set_volume(_volume) - - -def get_volume(): - """ - Get system volume. - - Returns: - int: Current system volume (0-100) - """ - return _volume - - -def is_playing(): - """ - Check if audio is currently playing. - - Returns: - bool: True if playback active, False otherwise - """ - return _current_stream is not None and _current_stream.is_playing() - - -def is_recording(): - """ - Check if audio is currently being recorded. - - Returns: - bool: True if recording active, False otherwise - """ - return _current_recording is not None and _current_recording.is_recording() +# Rename instance methods to avoid conflicts +AudioFlinger._init_impl = _init_impl +AudioFlinger._play_wav_impl = _play_wav_impl +AudioFlinger._play_rtttl_impl = _play_rtttl_impl +AudioFlinger._record_wav_impl = _record_wav_impl +AudioFlinger._stop_impl = _stop_impl +AudioFlinger._pause_impl = _pause_impl +AudioFlinger._resume_impl = _resume_impl +AudioFlinger._set_volume_impl = _set_volume_impl +AudioFlinger._get_volume_impl = _get_volume_impl +AudioFlinger._is_playing_impl = _is_playing_impl +AudioFlinger._is_recording_impl = _is_recording_impl +AudioFlinger._has_i2s_impl = _has_i2s_impl +AudioFlinger._has_buzzer_impl = _has_buzzer_impl +AudioFlinger._has_microphone_impl = _has_microphone_impl diff --git a/internal_filesystem/lib/mpos/board/fri3d_2024.py b/internal_filesystem/lib/mpos/board/fri3d_2024.py index 86c8b6fe..3391dd60 100644 --- a/internal_filesystem/lib/mpos/board/fri3d_2024.py +++ b/internal_filesystem/lib/mpos/board/fri3d_2024.py @@ -291,7 +291,7 @@ mpos.sdcard.init(spi_bus, cs_pin=14) # === AUDIO HARDWARE === from machine import PWM, Pin -import mpos.audio.audioflinger as AudioFlinger +from mpos import AudioFlinger # Initialize buzzer (GPIO 46) buzzer = PWM(Pin(46), freq=550, duty=0) diff --git a/internal_filesystem/lib/mpos/board/linux.py b/internal_filesystem/lib/mpos/board/linux.py index 9522344c..a85c58da 100644 --- a/internal_filesystem/lib/mpos/board/linux.py +++ b/internal_filesystem/lib/mpos/board/linux.py @@ -96,7 +96,7 @@ def adc_to_voltage(adc_value): mpos.battery_voltage.init_adc(999, adc_to_voltage) # === AUDIO HARDWARE === -import mpos.audio.audioflinger as AudioFlinger +from mpos import AudioFlinger # Desktop builds have no real audio hardware, but we simulate microphone # recording with a 440Hz sine wave for testing WAV file generation diff --git a/internal_filesystem/lib/mpos/board/waveshare_esp32_s3_touch_lcd_2.py b/internal_filesystem/lib/mpos/board/waveshare_esp32_s3_touch_lcd_2.py index 15642eec..047540e4 100644 --- a/internal_filesystem/lib/mpos/board/waveshare_esp32_s3_touch_lcd_2.py +++ b/internal_filesystem/lib/mpos/board/waveshare_esp32_s3_touch_lcd_2.py @@ -111,7 +111,7 @@ except Exception as e: print(f"Warning: powering off camera got exception: {e}") # === AUDIO HARDWARE === -import mpos.audio.audioflinger as AudioFlinger +from mpos import AudioFlinger # Note: Waveshare board has no buzzer or I2S audio AudioFlinger.init() diff --git a/internal_filesystem/lib/mpos/info.py b/internal_filesystem/lib/mpos/info.py index 956d86ed..9afcf9d4 100644 --- a/internal_filesystem/lib/mpos/info.py +++ b/internal_filesystem/lib/mpos/info.py @@ -1,4 +1,4 @@ -CURRENT_OS_VERSION = "0.5.3" +CURRENT_OS_VERSION = "0.6.0" # Unique string that defines the hardware, used by OSUpdate and the About app _hardware_id = "missing-hardware-info" diff --git a/scripts/addr2line.sh b/scripts/addr2line.sh old mode 100644 new mode 100755 diff --git a/tests/test_audioflinger.py b/tests/test_audioflinger.py index da9414ee..1e7f8dd2 100644 --- a/tests/test_audioflinger.py +++ b/tests/test_audioflinger.py @@ -20,7 +20,7 @@ inject_mocks({ }) # Now import the module to test -import mpos.audio.audioflinger as AudioFlinger +from mpos.audio.audioflinger import AudioFlinger class TestAudioFlinger(unittest.TestCase): @@ -45,8 +45,9 @@ class TestAudioFlinger(unittest.TestCase): def test_initialization(self): """Test that AudioFlinger initializes correctly.""" - self.assertEqual(AudioFlinger._i2s_pins, self.i2s_pins) - self.assertEqual(AudioFlinger._buzzer_instance, self.buzzer) + af = AudioFlinger.get() + self.assertEqual(af._i2s_pins, self.i2s_pins) + self.assertEqual(af._buzzer_instance, self.buzzer) def test_has_i2s(self): """Test has_i2s() returns correct value.""" @@ -134,7 +135,8 @@ class TestAudioFlinger(unittest.TestCase): def test_audio_focus_check_no_current_stream(self): """Test audio focus allows playback when no stream is active.""" - result = AudioFlinger._check_audio_focus(AudioFlinger.STREAM_MUSIC) + af = AudioFlinger.get() + result = af._check_audio_focus(AudioFlinger.STREAM_MUSIC) self.assertTrue(result) def test_volume_default_value(self): @@ -156,7 +158,8 @@ class TestAudioFlingerRecording(unittest.TestCase): self.i2s_pins_no_mic = {'sck': 2, 'ws': 47, 'sd': 16} # Reset state - AudioFlinger._current_recording = None + af = AudioFlinger.get() + af._current_recording = None AudioFlinger.set_volume(70) AudioFlinger.init(