You've already forked MicroPythonOS
mirror of
https://github.com/m5stack/MicroPythonOS.git
synced 2026-05-20 11:51:27 -07:00
Rename AudioFlinger to AudioManager framework
This commit is contained in:
Thomas Farstrike
@@ -2,7 +2,7 @@ import machine
|
||||
import os
|
||||
import time
|
||||
|
||||
from mpos import Activity, Intent, sdcard, get_event_name, AudioFlinger
|
||||
from mpos import Activity, Intent, sdcard, get_event_name, AudioManager
|
||||
|
||||
class MusicPlayer(Activity):
|
||||
|
||||
@@ -63,17 +63,17 @@ class FullscreenPlayer(Activity):
|
||||
self._filename = self.getIntent().extras.get("filename")
|
||||
qr_screen = lv.obj()
|
||||
self._slider_label=lv.label(qr_screen)
|
||||
self._slider_label.set_text(f"Volume: {AudioFlinger.get_volume()}%")
|
||||
self._slider_label.set_text(f"Volume: {AudioManager.get_volume()}%")
|
||||
self._slider_label.align(lv.ALIGN.TOP_MID,0,lv.pct(4))
|
||||
self._slider=lv.slider(qr_screen)
|
||||
self._slider.set_range(0,16)
|
||||
self._slider.set_value(int(AudioFlinger.get_volume()/6.25), False)
|
||||
self._slider.set_value(int(AudioManager.get_volume()/6.25), False)
|
||||
self._slider.set_width(lv.pct(90))
|
||||
self._slider.align_to(self._slider_label,lv.ALIGN.OUT_BOTTOM_MID,0,10)
|
||||
def volume_slider_changed(e):
|
||||
volume_int = self._slider.get_value()*6.25
|
||||
self._slider_label.set_text(f"Volume: {volume_int}%")
|
||||
AudioFlinger.set_volume(volume_int)
|
||||
AudioManager.set_volume(volume_int)
|
||||
self._slider.add_event_cb(volume_slider_changed,lv.EVENT.VALUE_CHANGED,None)
|
||||
self._filename_label = lv.label(qr_screen)
|
||||
self._filename_label.align(lv.ALIGN.CENTER,0,0)
|
||||
@@ -100,12 +100,12 @@ class FullscreenPlayer(Activity):
|
||||
print("Not playing any file...")
|
||||
else:
|
||||
print(f"Playing file {self._filename}")
|
||||
AudioFlinger.stop()
|
||||
AudioManager.stop()
|
||||
time.sleep(0.1)
|
||||
|
||||
success = AudioFlinger.play_wav(
|
||||
success = AudioManager.play_wav(
|
||||
self._filename,
|
||||
stream_type=AudioFlinger.STREAM_MUSIC,
|
||||
stream_type=AudioManager.STREAM_MUSIC,
|
||||
on_complete=self.player_finished
|
||||
)
|
||||
|
||||
@@ -125,7 +125,7 @@ class FullscreenPlayer(Activity):
|
||||
obj.set_style_border_width(0, lv.PART.MAIN)
|
||||
|
||||
def stop_button_clicked(self, event):
|
||||
AudioFlinger.stop()
|
||||
AudioManager.stop()
|
||||
self.finish()
|
||||
|
||||
def player_finished(self, result=None):
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
import os
|
||||
import time
|
||||
|
||||
from mpos import Activity, ui, AudioFlinger
|
||||
from mpos import Activity, ui, AudioManager
|
||||
|
||||
|
||||
def _makedirs(path):
|
||||
@@ -136,7 +136,7 @@ class SoundRecorder(Activity):
|
||||
|
||||
def _update_status(self):
|
||||
"""Update status label based on microphone availability."""
|
||||
if AudioFlinger.has_microphone():
|
||||
if AudioManager.has_microphone():
|
||||
self._status_label.set_text("Microphone ready")
|
||||
self._status_label.set_style_text_color(lv.color_hex(0x00AA00), lv.PART.MAIN)
|
||||
self._record_button.remove_flag(lv.obj.FLAG.HIDDEN)
|
||||
@@ -243,9 +243,9 @@ class SoundRecorder(Activity):
|
||||
def _start_recording(self):
|
||||
"""Start recording audio."""
|
||||
print("SoundRecorder: _start_recording called")
|
||||
print(f"SoundRecorder: has_microphone() = {AudioFlinger.has_microphone()}")
|
||||
print(f"SoundRecorder: has_microphone() = {AudioManager.has_microphone()}")
|
||||
|
||||
if not AudioFlinger.has_microphone():
|
||||
if not AudioManager.has_microphone():
|
||||
print("SoundRecorder: No microphone available - aborting")
|
||||
return
|
||||
|
||||
@@ -263,12 +263,12 @@ class SoundRecorder(Activity):
|
||||
return
|
||||
|
||||
# Start recording
|
||||
print(f"SoundRecorder: Calling AudioFlinger.record_wav()")
|
||||
print(f"SoundRecorder: Calling AudioManager.record_wav()")
|
||||
print(f" file_path: {file_path}")
|
||||
print(f" duration_ms: {self._current_max_duration_ms}")
|
||||
print(f" sample_rate: {self.SAMPLE_RATE}")
|
||||
|
||||
success = AudioFlinger.record_wav(
|
||||
success = AudioManager.record_wav(
|
||||
file_path=file_path,
|
||||
duration_ms=self._current_max_duration_ms,
|
||||
on_complete=self._on_recording_complete,
|
||||
@@ -302,7 +302,7 @@ class SoundRecorder(Activity):
|
||||
|
||||
def _stop_recording(self):
|
||||
"""Stop recording audio."""
|
||||
AudioFlinger.stop()
|
||||
AudioManager.stop()
|
||||
self._is_recording = False
|
||||
|
||||
# Show "Saving..." status immediately (file finalization takes time on SD card)
|
||||
@@ -364,13 +364,13 @@ class SoundRecorder(Activity):
|
||||
"""Handle play button click."""
|
||||
if self._last_recording and not self._is_recording:
|
||||
# Stop any current playback
|
||||
AudioFlinger.stop()
|
||||
AudioManager.stop()
|
||||
time.sleep_ms(100)
|
||||
|
||||
# Play the recording
|
||||
success = AudioFlinger.play_wav(
|
||||
success = AudioManager.play_wav(
|
||||
self._last_recording,
|
||||
stream_type=AudioFlinger.STREAM_MUSIC,
|
||||
stream_type=AudioManager.STREAM_MUSIC,
|
||||
on_complete=self._on_playback_complete,
|
||||
volume=100
|
||||
)
|
||||
|
||||
@@ -8,7 +8,7 @@ from .content.app_manager import AppManager
|
||||
from .config import SharedPreferences
|
||||
from .net.connectivity_manager import ConnectivityManager
|
||||
from .net.wifi_service import WifiService
|
||||
from .audio.audioflinger import AudioFlinger
|
||||
from .audio.audiomanager import AudioManager
|
||||
from .net.download_manager import DownloadManager
|
||||
from .task_manager import TaskManager
|
||||
from .camera_manager import CameraManager
|
||||
@@ -66,7 +66,7 @@ __all__ = [
|
||||
"App",
|
||||
"Activity",
|
||||
"SharedPreferences",
|
||||
"ConnectivityManager", "DownloadManager", "WifiService", "AudioFlinger", "Intent",
|
||||
"ConnectivityManager", "DownloadManager", "WifiService", "AudioManager", "Intent",
|
||||
"ActivityNavigator", "AppManager", "TaskManager", "CameraManager", "BatteryManager",
|
||||
# Device and build info
|
||||
"DeviceInfo", "BuildInfo",
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
# AudioFlinger - Centralized Audio Management Service for MicroPythonOS
|
||||
# AudioManager - Centralized Audio Management Service for MicroPythonOS
|
||||
# Android-inspired audio routing with priority-based audio focus
|
||||
# Simple routing: play_wav() -> I2S, play_rtttl() -> buzzer, record_wav() -> I2S mic
|
||||
|
||||
from .audioflinger import AudioFlinger
|
||||
from .audiomanager import AudioManager
|
||||
|
||||
+40
-40
@@ -1,4 +1,4 @@
|
||||
# AudioFlinger - Core Audio Management Service
|
||||
# AudioManager - Core Audio Management Service
|
||||
# Centralized audio routing with priority-based audio focus (Android-inspired)
|
||||
# Supports I2S (digital audio) and PWM buzzer (tones/ringtones)
|
||||
#
|
||||
@@ -9,20 +9,20 @@ import _thread
|
||||
from ..task_manager import TaskManager
|
||||
|
||||
|
||||
class AudioFlinger:
|
||||
class AudioManager:
|
||||
"""
|
||||
Centralized audio management service with priority-based audio focus.
|
||||
Implements singleton pattern for single audio service instance.
|
||||
|
||||
Usage:
|
||||
from mpos import AudioFlinger
|
||||
from mpos import AudioManager
|
||||
|
||||
# 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()
|
||||
AudioManager.init(i2s_pins=pins, buzzer_instance=buzzer)
|
||||
AudioManager.play_wav("music.wav", stream_type=AudioManager.STREAM_MUSIC)
|
||||
AudioManager.set_volume(80)
|
||||
volume = AudioManager.get_volume()
|
||||
AudioManager.stop()
|
||||
"""
|
||||
|
||||
# Stream type constants (priority order: higher number = higher priority)
|
||||
@@ -34,15 +34,15 @@ class AudioFlinger:
|
||||
|
||||
def __init__(self, i2s_pins=None, buzzer_instance=None):
|
||||
"""
|
||||
Initialize AudioFlinger instance with optional hardware configuration.
|
||||
Initialize AudioManager instance with optional 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)
|
||||
"""
|
||||
if AudioFlinger._instance:
|
||||
if AudioManager._instance:
|
||||
return
|
||||
AudioFlinger._instance = self
|
||||
AudioManager._instance = self
|
||||
|
||||
self._i2s_pins = i2s_pins # I2S pin configuration dict (created per-stream)
|
||||
self._buzzer_instance = buzzer_instance # PWM buzzer instance
|
||||
@@ -58,9 +58,9 @@ class AudioFlinger:
|
||||
capabilities.append("Buzzer (RTTTL)")
|
||||
|
||||
if capabilities:
|
||||
print(f"AudioFlinger initialized: {', '.join(capabilities)}")
|
||||
print(f"AudioManager initialized: {', '.join(capabilities)}")
|
||||
else:
|
||||
print("AudioFlinger initialized: No audio hardware")
|
||||
print("AudioManager initialized: No audio hardware")
|
||||
|
||||
@classmethod
|
||||
def get(cls):
|
||||
@@ -100,11 +100,11 @@ class AudioFlinger:
|
||||
|
||||
# Check priority
|
||||
if stream_type <= self._current_stream.stream_type:
|
||||
print(f"AudioFlinger: Stream rejected (priority {stream_type} <= current {self._current_stream.stream_type})")
|
||||
print(f"AudioManager: Stream rejected (priority {stream_type} <= current {self._current_stream.stream_type})")
|
||||
return False
|
||||
|
||||
# Higher priority stream - interrupt current
|
||||
print(f"AudioFlinger: Interrupting stream (priority {stream_type} > current {self._current_stream.stream_type})")
|
||||
print(f"AudioManager: Interrupting stream (priority {stream_type} > current {self._current_stream.stream_type})")
|
||||
self._current_stream.stop()
|
||||
return True
|
||||
|
||||
@@ -122,7 +122,7 @@ class AudioFlinger:
|
||||
# Run synchronous playback in this thread
|
||||
stream.play()
|
||||
except Exception as e:
|
||||
print(f"AudioFlinger: Playback error: {e}")
|
||||
print(f"AudioManager: Playback error: {e}")
|
||||
finally:
|
||||
# Clear current stream
|
||||
if self._current_stream == stream:
|
||||
@@ -145,7 +145,7 @@ class AudioFlinger:
|
||||
stream_type = self.STREAM_MUSIC
|
||||
|
||||
if not self._i2s_pins:
|
||||
print("AudioFlinger: play_wav() failed - I2S not configured")
|
||||
print("AudioManager: play_wav() failed - I2S not configured")
|
||||
return False
|
||||
|
||||
# Check audio focus
|
||||
@@ -169,7 +169,7 @@ class AudioFlinger:
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"AudioFlinger: play_wav() failed: {e}")
|
||||
print(f"AudioManager: play_wav() failed: {e}")
|
||||
return False
|
||||
|
||||
def play_rtttl(self, rtttl_string, stream_type=None, volume=None, on_complete=None):
|
||||
@@ -189,7 +189,7 @@ class AudioFlinger:
|
||||
stream_type = self.STREAM_NOTIFICATION
|
||||
|
||||
if not self._buzzer_instance:
|
||||
print("AudioFlinger: play_rtttl() failed - buzzer not configured")
|
||||
print("AudioManager: play_rtttl() failed - buzzer not configured")
|
||||
return False
|
||||
|
||||
# Check audio focus
|
||||
@@ -213,7 +213,7 @@ class AudioFlinger:
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"AudioFlinger: play_rtttl() failed: {e}")
|
||||
print(f"AudioManager: play_rtttl() failed: {e}")
|
||||
return False
|
||||
|
||||
def _recording_thread(self, stream):
|
||||
@@ -230,7 +230,7 @@ class AudioFlinger:
|
||||
# Run synchronous recording in this thread
|
||||
stream.record()
|
||||
except Exception as e:
|
||||
print(f"AudioFlinger: Recording error: {e}")
|
||||
print(f"AudioManager: Recording error: {e}")
|
||||
finally:
|
||||
# Clear current recording
|
||||
if self._current_recording == stream:
|
||||
@@ -249,7 +249,7 @@ class AudioFlinger:
|
||||
Returns:
|
||||
bool: True if recording started, False if rejected or unavailable
|
||||
"""
|
||||
print(f"AudioFlinger.record_wav() called")
|
||||
print(f"AudioManager.record_wav() called")
|
||||
print(f" file_path: {file_path}")
|
||||
print(f" duration_ms: {duration_ms}")
|
||||
print(f" sample_rate: {sample_rate}")
|
||||
@@ -257,25 +257,25 @@ class AudioFlinger:
|
||||
print(f" has_microphone(): {self.has_microphone()}")
|
||||
|
||||
if not self.has_microphone():
|
||||
print("AudioFlinger: record_wav() failed - microphone not configured")
|
||||
print("AudioManager: 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")
|
||||
print("AudioManager: Cannot record while playing")
|
||||
return False
|
||||
|
||||
# Cannot start new recording while already recording
|
||||
if self.is_recording():
|
||||
print("AudioFlinger: Already recording")
|
||||
print("AudioManager: Already recording")
|
||||
return False
|
||||
|
||||
# Create stream and start recording in separate thread
|
||||
try:
|
||||
print("AudioFlinger: Importing RecordStream...")
|
||||
print("AudioManager: Importing RecordStream...")
|
||||
from mpos.audio.stream_record import RecordStream
|
||||
|
||||
print("AudioFlinger: Creating RecordStream instance...")
|
||||
print("AudioManager: Creating RecordStream instance...")
|
||||
stream = RecordStream(
|
||||
file_path=file_path,
|
||||
duration_ms=duration_ms,
|
||||
@@ -284,15 +284,15 @@ class AudioFlinger:
|
||||
on_complete=on_complete
|
||||
)
|
||||
|
||||
print("AudioFlinger: Starting recording thread...")
|
||||
print("AudioManager: Starting recording thread...")
|
||||
_thread.stack_size(TaskManager.good_stack_size())
|
||||
_thread.start_new_thread(self._recording_thread, (stream,))
|
||||
print("AudioFlinger: Recording thread started successfully")
|
||||
print("AudioManager: Recording thread started successfully")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
import sys
|
||||
print(f"AudioFlinger: record_wav() failed: {e}")
|
||||
print(f"AudioManager: record_wav() failed: {e}")
|
||||
sys.print_exception(e)
|
||||
return False
|
||||
|
||||
@@ -302,16 +302,16 @@ class AudioFlinger:
|
||||
|
||||
if self._current_stream:
|
||||
self._current_stream.stop()
|
||||
print("AudioFlinger: Playback stopped")
|
||||
print("AudioManager: Playback stopped")
|
||||
stopped = True
|
||||
|
||||
if self._current_recording:
|
||||
self._current_recording.stop()
|
||||
print("AudioFlinger: Recording stopped")
|
||||
print("AudioManager: Recording stopped")
|
||||
stopped = True
|
||||
|
||||
if not stopped:
|
||||
print("AudioFlinger: No playback or recording to stop")
|
||||
print("AudioManager: No playback or recording to stop")
|
||||
|
||||
def pause(self):
|
||||
"""
|
||||
@@ -320,9 +320,9 @@ class AudioFlinger:
|
||||
"""
|
||||
if self._current_stream and hasattr(self._current_stream, 'pause'):
|
||||
self._current_stream.pause()
|
||||
print("AudioFlinger: Playback paused")
|
||||
print("AudioManager: Playback paused")
|
||||
else:
|
||||
print("AudioFlinger: Pause not supported or no playback active")
|
||||
print("AudioManager: Pause not supported or no playback active")
|
||||
|
||||
def resume(self):
|
||||
"""
|
||||
@@ -331,9 +331,9 @@ class AudioFlinger:
|
||||
"""
|
||||
if self._current_stream and hasattr(self._current_stream, 'resume'):
|
||||
self._current_stream.resume()
|
||||
print("AudioFlinger: Playback resumed")
|
||||
print("AudioManager: Playback resumed")
|
||||
else:
|
||||
print("AudioFlinger: Resume not supported or no playback active")
|
||||
print("AudioManager: Resume not supported or no playback active")
|
||||
|
||||
def set_volume(self, volume):
|
||||
"""
|
||||
@@ -396,7 +396,7 @@ _methods_to_delegate = [
|
||||
]
|
||||
|
||||
for method_name in _methods_to_delegate:
|
||||
_original_methods[method_name] = getattr(AudioFlinger, method_name)
|
||||
_original_methods[method_name] = getattr(AudioManager, method_name)
|
||||
|
||||
# Helper to create delegating class methods
|
||||
def _make_class_method(method_name):
|
||||
@@ -410,6 +410,6 @@ def _make_class_method(method_name):
|
||||
|
||||
return class_method
|
||||
|
||||
# Attach class methods to AudioFlinger
|
||||
# Attach class methods to AudioManager
|
||||
for method_name in _methods_to_delegate:
|
||||
setattr(AudioFlinger, method_name, _make_class_method(method_name))
|
||||
setattr(AudioManager, method_name, _make_class_method(method_name))
|
||||
@@ -1,4 +1,4 @@
|
||||
# RecordStream - WAV File Recording Stream for AudioFlinger
|
||||
# RecordStream - WAV File Recording Stream for AudioManager
|
||||
# Records 16-bit mono PCM audio from I2S microphone to WAV file
|
||||
# Uses synchronous recording in a separate thread for non-blocking operation
|
||||
# On desktop (no I2S hardware), generates a 440Hz sine wave for testing
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# RTTTLStream - RTTTL Ringtone Playback Stream for AudioFlinger
|
||||
# RTTTLStream - RTTTL Ringtone Playback Stream for AudioManager
|
||||
# Ring Tone Text Transfer Language parser and player
|
||||
# Uses synchronous playback in a separate thread for non-blocking operation
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# WAVStream - WAV File Playback Stream for AudioFlinger
|
||||
# WAVStream - WAV File Playback Stream for AudioManager
|
||||
# Supports 8/16/24/32-bit PCM, mono+stereo, auto-upsampling, volume control
|
||||
# Uses synchronous playback in a separate thread for non-blocking operation
|
||||
|
||||
|
||||
@@ -296,7 +296,7 @@ mpos.sdcard.init(spi_bus, cs_pin=14)
|
||||
|
||||
# === AUDIO HARDWARE ===
|
||||
from machine import PWM, Pin
|
||||
from mpos import AudioFlinger
|
||||
from mpos import AudioManager
|
||||
|
||||
# Initialize buzzer (GPIO 46)
|
||||
buzzer = PWM(Pin(46), freq=550, duty=0)
|
||||
@@ -315,8 +315,8 @@ i2s_pins = {
|
||||
'sd_in': 15, # DIN - Serial Data IN (microphone)
|
||||
}
|
||||
|
||||
# Initialize AudioFlinger with I2S and buzzer
|
||||
AudioFlinger(i2s_pins=i2s_pins, buzzer_instance=buzzer)
|
||||
# Initialize AudioManager with I2S and buzzer
|
||||
AudioManager(i2s_pins=i2s_pins, buzzer_instance=buzzer)
|
||||
|
||||
# === LED HARDWARE ===
|
||||
import mpos.lights as LightsManager
|
||||
@@ -349,9 +349,9 @@ def startup_wow_effect():
|
||||
#startup_jingle = "ShortBeeps:d=32,o=5,b=320:c6,c7"
|
||||
|
||||
# Start the jingle
|
||||
AudioFlinger.play_rtttl(
|
||||
AudioManager.play_rtttl(
|
||||
startup_jingle,
|
||||
stream_type=AudioFlinger.STREAM_NOTIFICATION,
|
||||
stream_type=AudioManager.STREAM_NOTIFICATION,
|
||||
volume=60
|
||||
)
|
||||
|
||||
|
||||
@@ -200,7 +200,7 @@ mpos.sdcard.init(spi_bus, cs_pin=14)
|
||||
|
||||
# === AUDIO HARDWARE ===
|
||||
from machine import PWM, Pin
|
||||
from mpos import AudioFlinger
|
||||
from mpos import AudioManager
|
||||
|
||||
# Initialize buzzer: now sits on PC14/CC1 of the CH32X035GxUx so needs custom code
|
||||
#buzzer = PWM(Pin(46), freq=550, duty=0)
|
||||
@@ -219,8 +219,8 @@ i2s_pins = {
|
||||
'sd_in': 15, # DIN - Serial Data IN (microphone)
|
||||
}
|
||||
|
||||
# Initialize AudioFlinger with I2S (buzzer TODO)
|
||||
AudioFlinger(i2s_pins=i2s_pins)
|
||||
# Initialize AudioManager with I2S (buzzer TODO)
|
||||
AudioManager(i2s_pins=i2s_pins)
|
||||
|
||||
# === LED HARDWARE ===
|
||||
import mpos.lights as LightsManager
|
||||
|
||||
@@ -98,7 +98,7 @@ def adc_to_voltage(adc_value):
|
||||
BatteryManager.init_adc(999, adc_to_voltage)
|
||||
|
||||
# === AUDIO HARDWARE ===
|
||||
from mpos import AudioFlinger
|
||||
from mpos import AudioManager
|
||||
|
||||
# Desktop builds have no real audio hardware, but we simulate microphone
|
||||
# recording with a 440Hz sine wave for testing WAV file generation
|
||||
@@ -110,7 +110,7 @@ i2s_pins = {
|
||||
'sck_in': 0, # Simulated - not used on desktop
|
||||
'sd_in': 0, # Simulated - enables microphone simulation
|
||||
}
|
||||
AudioFlinger(i2s_pins=i2s_pins)
|
||||
AudioManager(i2s_pins=i2s_pins)
|
||||
|
||||
# === LED HARDWARE ===
|
||||
# Note: Desktop builds have no LED hardware
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
# Fri3d Camp 2024 Badge Hardware Drivers
|
||||
# These are simple wrappers that can be used by services like AudioFlinger
|
||||
# These are simple wrappers that can be used by services like AudioManager
|
||||
|
||||
from .buzzer import BuzzerConfig
|
||||
from .leds import LEDConfig
|
||||
|
||||
@@ -1,225 +0,0 @@
|
||||
# Unit tests for AudioFlinger service
|
||||
import unittest
|
||||
import sys
|
||||
|
||||
# Import centralized mocks
|
||||
from mpos.testing import (
|
||||
MockMachine,
|
||||
MockPWM,
|
||||
MockPin,
|
||||
MockThread,
|
||||
inject_mocks,
|
||||
)
|
||||
|
||||
# Inject mocks before importing AudioFlinger
|
||||
inject_mocks({
|
||||
'machine': MockMachine(),
|
||||
'_thread': MockThread,
|
||||
})
|
||||
|
||||
# Now import the module to test
|
||||
from mpos.audio.audioflinger import AudioFlinger
|
||||
|
||||
|
||||
class TestAudioFlinger(unittest.TestCase):
|
||||
"""Test cases for AudioFlinger service."""
|
||||
|
||||
def setUp(self):
|
||||
"""Initialize AudioFlinger before each test."""
|
||||
self.buzzer = MockPWM(MockPin(46))
|
||||
self.i2s_pins = {'sck': 2, 'ws': 47, 'sd': 16}
|
||||
|
||||
# Reset singleton instance for each test
|
||||
AudioFlinger._instance = None
|
||||
|
||||
AudioFlinger(
|
||||
i2s_pins=self.i2s_pins,
|
||||
buzzer_instance=self.buzzer
|
||||
)
|
||||
|
||||
# Reset volume to default after creating instance
|
||||
AudioFlinger.set_volume(70)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after each test."""
|
||||
AudioFlinger.stop()
|
||||
|
||||
def test_initialization(self):
|
||||
"""Test that AudioFlinger initializes correctly."""
|
||||
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."""
|
||||
# With I2S configured
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins, buzzer_instance=None)
|
||||
self.assertTrue(AudioFlinger.has_i2s())
|
||||
|
||||
# Without I2S configured
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=self.buzzer)
|
||||
self.assertFalse(AudioFlinger.has_i2s())
|
||||
|
||||
def test_has_buzzer(self):
|
||||
"""Test has_buzzer() returns correct value."""
|
||||
# With buzzer configured
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=self.buzzer)
|
||||
self.assertTrue(AudioFlinger.has_buzzer())
|
||||
|
||||
# Without buzzer configured
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins, buzzer_instance=None)
|
||||
self.assertFalse(AudioFlinger.has_buzzer())
|
||||
|
||||
def test_stream_types(self):
|
||||
"""Test stream type constants and priority order."""
|
||||
self.assertEqual(AudioFlinger.STREAM_MUSIC, 0)
|
||||
self.assertEqual(AudioFlinger.STREAM_NOTIFICATION, 1)
|
||||
self.assertEqual(AudioFlinger.STREAM_ALARM, 2)
|
||||
|
||||
# Higher number = higher priority
|
||||
self.assertTrue(AudioFlinger.STREAM_MUSIC < AudioFlinger.STREAM_NOTIFICATION)
|
||||
self.assertTrue(AudioFlinger.STREAM_NOTIFICATION < AudioFlinger.STREAM_ALARM)
|
||||
|
||||
def test_volume_control(self):
|
||||
"""Test volume get/set operations."""
|
||||
# Set volume
|
||||
AudioFlinger.set_volume(50)
|
||||
self.assertEqual(AudioFlinger.get_volume(), 50)
|
||||
|
||||
# Test clamping to 0-100 range
|
||||
AudioFlinger.set_volume(150)
|
||||
self.assertEqual(AudioFlinger.get_volume(), 100)
|
||||
|
||||
AudioFlinger.set_volume(-10)
|
||||
self.assertEqual(AudioFlinger.get_volume(), 0)
|
||||
|
||||
def test_no_hardware_rejects_playback(self):
|
||||
"""Test that no hardware rejects all playback requests."""
|
||||
# Re-initialize with no hardware
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=None)
|
||||
|
||||
# WAV should be rejected (no I2S)
|
||||
result = AudioFlinger.play_wav("test.wav")
|
||||
self.assertFalse(result)
|
||||
|
||||
# RTTTL should be rejected (no buzzer)
|
||||
result = AudioFlinger.play_rtttl("Test:d=4,o=5,b=120:c")
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_i2s_only_rejects_rtttl(self):
|
||||
"""Test that I2S-only config rejects buzzer playback."""
|
||||
# Re-initialize with I2S only
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins, buzzer_instance=None)
|
||||
|
||||
# RTTTL should be rejected (no buzzer)
|
||||
result = AudioFlinger.play_rtttl("Test:d=4,o=5,b=120:c")
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_buzzer_only_rejects_wav(self):
|
||||
"""Test that buzzer-only config rejects I2S playback."""
|
||||
# Re-initialize with buzzer only
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=self.buzzer)
|
||||
|
||||
# WAV should be rejected (no I2S)
|
||||
result = AudioFlinger.play_wav("test.wav")
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_is_playing_initially_false(self):
|
||||
"""Test that is_playing() returns False initially."""
|
||||
# Reset to ensure clean state
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins, buzzer_instance=self.buzzer)
|
||||
self.assertFalse(AudioFlinger.is_playing())
|
||||
|
||||
def test_stop_with_no_playback(self):
|
||||
"""Test that stop() can be called when nothing is playing."""
|
||||
# Should not raise exception
|
||||
AudioFlinger.stop()
|
||||
self.assertFalse(AudioFlinger.is_playing())
|
||||
|
||||
def test_audio_focus_check_no_current_stream(self):
|
||||
"""Test audio focus allows playback when no stream is active."""
|
||||
af = AudioFlinger.get()
|
||||
result = af._check_audio_focus(AudioFlinger.STREAM_MUSIC)
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_volume_default_value(self):
|
||||
"""Test that default volume is reasonable."""
|
||||
# After init, volume should be at default (70)
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=None)
|
||||
self.assertEqual(AudioFlinger.get_volume(), 70)
|
||||
|
||||
|
||||
class TestAudioFlingerRecording(unittest.TestCase):
|
||||
"""Test cases for AudioFlinger recording functionality."""
|
||||
|
||||
def setUp(self):
|
||||
"""Initialize AudioFlinger with microphone before each test."""
|
||||
self.buzzer = MockPWM(MockPin(46))
|
||||
# I2S pins with microphone input
|
||||
self.i2s_pins_with_mic = {'sck': 2, 'ws': 47, 'sd': 16, 'sd_in': 15}
|
||||
# I2S pins without microphone input
|
||||
self.i2s_pins_no_mic = {'sck': 2, 'ws': 47, 'sd': 16}
|
||||
|
||||
# Reset singleton instance for each test
|
||||
AudioFlinger._instance = None
|
||||
|
||||
AudioFlinger(
|
||||
i2s_pins=self.i2s_pins_with_mic,
|
||||
buzzer_instance=self.buzzer
|
||||
)
|
||||
|
||||
# Reset volume to default after creating instance
|
||||
AudioFlinger.set_volume(70)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up after each test."""
|
||||
AudioFlinger.stop()
|
||||
|
||||
def test_has_microphone_with_sd_in(self):
|
||||
"""Test has_microphone() returns True when sd_in pin is configured."""
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins_with_mic, buzzer_instance=None)
|
||||
self.assertTrue(AudioFlinger.has_microphone())
|
||||
|
||||
def test_has_microphone_without_sd_in(self):
|
||||
"""Test has_microphone() returns False when sd_in pin is not configured."""
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins_no_mic, buzzer_instance=None)
|
||||
self.assertFalse(AudioFlinger.has_microphone())
|
||||
|
||||
def test_has_microphone_no_i2s(self):
|
||||
"""Test has_microphone() returns False when no I2S is configured."""
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=self.buzzer)
|
||||
self.assertFalse(AudioFlinger.has_microphone())
|
||||
|
||||
def test_is_recording_initially_false(self):
|
||||
"""Test that is_recording() returns False initially."""
|
||||
self.assertFalse(AudioFlinger.is_recording())
|
||||
|
||||
def test_record_wav_no_microphone(self):
|
||||
"""Test that record_wav() fails when no microphone is configured."""
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=self.i2s_pins_no_mic, buzzer_instance=None)
|
||||
result = AudioFlinger.record_wav("test.wav")
|
||||
self.assertFalse(result, "record_wav() fails when no microphone is configured")
|
||||
|
||||
def test_record_wav_no_i2s(self):
|
||||
AudioFlinger._instance = None
|
||||
AudioFlinger(i2s_pins=None, buzzer_instance=self.buzzer)
|
||||
result = AudioFlinger.record_wav("test.wav")
|
||||
self.assertFalse(result, "record_wav() should fail when no I2S is configured")
|
||||
|
||||
def test_stop_with_no_recording(self):
|
||||
"""Test that stop() can be called when nothing is recording."""
|
||||
# Should not raise exception
|
||||
AudioFlinger.stop()
|
||||
self.assertFalse(AudioFlinger.is_recording())
|
||||
Reference in New Issue
Block a user