CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-mopidy

Mopidy is an extensible music server written in Python

—

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

audio-system.mddocs/

Audio System

Mopidy's audio system provides a GStreamer-based audio pipeline with playback control, volume management, and support for various audio formats and output methods. The system handles audio streaming, format conversion, and hardware integration through a flexible architecture.

Capabilities

Audio Actor

The main audio system actor that manages the GStreamer pipeline and audio playback operations.

class Audio(pykka.ThreadingActor):
    """
    Main audio actor managing GStreamer-based audio pipeline.
    
    Parameters:
    - config: Audio configuration
    - mixer: Audio mixer instance
    """
    def __init__(self, config, mixer): ...
    
    def set_uri(self, uri):
        """
        Set URI for audio playback.
        
        Parameters:
        - uri (str): Audio URI to play
        
        Returns:
        - bool: True if URI was set successfully
        """
        ...
    
    def get_uri(self):
        """
        Get current audio URI.
        
        Returns:
        - str or None: Current audio URI
        """
        ...
    
    def start_playback(self):
        """
        Start audio playback.
        
        Returns:
        - bool: True if playback started successfully
        """
        ...
    
    def pause_playback(self):
        """
        Pause audio playback.
        
        Returns:
        - bool: True if playback was paused successfully
        """
        ...
    
    def prepare_change(self):
        """
        Prepare for URI change by pausing.
        
        Returns:
        - bool: True if prepared successfully
        """
        ...
    
    def stop_playback(self):
        """
        Stop audio playback and reset pipeline.
        
        Returns:
        - bool: True if playback was stopped successfully
        """
        ...
    
    def get_position(self):
        """
        Get current playback position.
        
        Returns:
        - int: Position in milliseconds
        """
        ...
    
    def set_position(self, position):
        """
        Seek to specific position.
        
        Parameters:
        - position (int): Position in milliseconds
        
        Returns:
        - bool: True if seek was successful
        """
        ...
    
    def get_volume(self):
        """
        Get current volume level.
        
        Returns:
        - int or None: Volume level (0-100)
        """
        ...
    
    def set_volume(self, volume):
        """
        Set volume level.
        
        Parameters:
        - volume (int): Volume level (0-100)
        
        Returns:
        - bool: True if volume was set successfully
        """
        ...
    
    def get_mute(self):
        """
        Get mute state.
        
        Returns:
        - bool or None: True if muted
        """
        ...
    
    def set_mute(self, mute):
        """
        Set mute state.
        
        Parameters:
        - mute (bool): True to mute, False to unmute
        
        Returns:
        - bool: True if mute state was changed successfully
        """
        ...
    
    def set_metadata(self, track):
        """
        Set track metadata for current playback.
        
        Parameters:
        - track (Track): Track metadata
        """
        ...
    
    def get_appsrc(self):
        """
        Get GStreamer appsrc element for custom audio sources.
        
        Returns:
        - GstElement or None: appsrc element
        """
        ...

Usage example:

# Audio actor is typically managed by core system
audio_actor = Audio.start(config=audio_config, mixer=mixer_instance)
audio_ref = audio_actor.actor_ref

# Control playback
audio_ref.tell({"method": "set_uri", "uri": "file:///path/to/song.mp3"})
audio_ref.tell({"method": "start_playback"})
audio_ref.tell({"method": "set_volume", "volume": 75})

Playback State Management

Constants and utilities for managing audio playback states.

class PlaybackState:
    """Constants for audio playback states."""
    STOPPED = "stopped"
    PLAYING = "playing"
    PAUSED = "paused"
    
    @classmethod
    def all_states(cls):
        """
        Get all valid playback states.
        
        Returns:
        - list[str]: All playback state constants
        """
        ...

Audio Utility Functions

Helper functions for audio processing, format detection, and stream management.

def supported_uri_schemes():
    """
    Get list of supported URI schemes for audio playback.
    
    Returns:
    - list[str]: Supported URI schemes (file, http, https, etc.)
    """
    ...

def calculate_duration(num_samples, sample_rate):
    """
    Calculate duration from audio samples and sample rate.
    
    Parameters:
    - num_samples (int): Number of audio samples
    - sample_rate (int): Sample rate in Hz
    
    Returns:
    - int: Duration in milliseconds
    """
    ...

def millisecond_to_clocktime(milliseconds):
    """
    Convert milliseconds to GStreamer clocktime format.
    
    Parameters:
    - milliseconds (int): Time in milliseconds
    
    Returns:
    - int: Time in GStreamer clocktime format (nanoseconds)
    """
    ...

def create_buffer(data, timestamp=None, duration=None):
    """
    Create GStreamer buffer from audio data.
    
    Parameters:
    - data (bytes): Audio data
    - timestamp (int, optional): Buffer timestamp in nanoseconds
    - duration (int, optional): Buffer duration in nanoseconds
    
    Returns:
    - GstBuffer: GStreamer buffer object
    """
    ...

Usage example:

from mopidy.audio import calculate_duration, supported_uri_schemes

# Check supported formats
schemes = supported_uri_schemes()
print(f"Supported schemes: {schemes}")

# Calculate duration for audio data
samples = 44100 * 180  # 3 minutes at 44.1kHz
duration_ms = calculate_duration(samples, 44100)
print(f"Duration: {duration_ms}ms")

Audio Event Listeners

Event notification system for audio state changes and pipeline events.

class AudioListener:
    """Base class for audio event listeners."""
    
    def reached_end_of_stream(self):
        """Called when audio stream reaches end."""
        ...
    
    def position_changed(self, position):
        """
        Called when playback position changes.
        
        Parameters:
        - position (int): New position in milliseconds
        """
        ...
    
    def state_changed(self, old_state, new_state):
        """
        Called when playback state changes.
        
        Parameters:
        - old_state (str): Previous playback state
        - new_state (str): New playback state
        """
        ...
    
    def stream_changed(self, uri):
        """
        Called when audio stream URI changes.
        
        Parameters:
        - uri (str): New stream URI
        """
        ...
    
    def tags_changed(self, tags):
        """
        Called when stream tags/metadata change.
        
        Parameters:
        - tags (dict): Stream tags and metadata
        """
        ...

Usage example:

class MyAudioListener(AudioListener):
    def reached_end_of_stream(self):
        print("Track finished playing")
    
    def position_changed(self, position):
        print(f"Playback position: {position}ms")
    
    def state_changed(self, old_state, new_state):
        print(f"State changed: {old_state} -> {new_state}")

# Register listener with audio system
listener = MyAudioListener()
# Registration depends on specific audio system setup

Audio Configuration

Configuration schema and options for the audio system.

# Audio configuration schema
_audio_schema = ConfigSchema("audio")
_audio_schema["mixer"] = String()  # Mixer implementation to use
_audio_schema["mixer_volume"] = Integer(optional=True, minimum=0, maximum=100)
_audio_schema["output"] = String()  # Audio output configuration
_audio_schema["buffer_time"] = Integer(optional=True, minimum=1)  # Buffer time in milliseconds

# Example audio configuration
"""
[audio]
mixer = software
mixer_volume = 80
output = autoaudiosink
buffer_time = 200
"""

GStreamer Pipeline Integration

Low-level GStreamer pipeline management and element handling.

class GstElement:
    """Type hint for GStreamer elements."""
    pass

def get_audio_element():
    """
    Get main audio pipeline element.
    
    Returns:
    - GstElement: Main audio pipeline element
    """
    ...

def setup_output_element(config):
    """
    Set up audio output element based on configuration.
    
    Parameters:
    - config (dict): Audio configuration
    
    Returns:
    - GstElement: Configured output element
    """
    ...

def setup_mixer_element(config):
    """
    Set up audio mixer element based on configuration.
    
    Parameters:
    - config (dict): Audio configuration
    
    Returns:
    - GstElement: Configured mixer element
    """
    ...

Audio Format Support

Information about supported audio formats, codecs, and containers.

def get_supported_formats():
    """
    Get list of supported audio formats.
    
    Returns:
    - list[str]: Supported audio format extensions
    """
    ...

def get_supported_codecs():
    """
    Get list of supported audio codecs.
    
    Returns:
    - list[str]: Supported audio codec names
    """
    ...

def detect_format(uri):
    """
    Detect audio format from URI or file.
    
    Parameters:
    - uri (str): Audio URI or file path
    
    Returns:
    - str or None: Detected format or None if unknown
    """
    ...

Stream Management

Utilities for handling audio streams, including network streams and local files.

def prepare_stream(uri, headers=None, timeout=None):
    """
    Prepare audio stream for playback.
    
    Parameters:
    - uri (str): Stream URI
    - headers (dict, optional): HTTP headers for network streams
    - timeout (int, optional): Connection timeout in seconds
    
    Returns:
    - bool: True if stream was prepared successfully
    """
    ...

def get_stream_metadata(uri):
    """
    Get metadata from audio stream.
    
    Parameters:
    - uri (str): Stream URI
    
    Returns:
    - dict: Stream metadata (title, artist, album, etc.)
    """
    ...

def validate_stream_uri(uri):
    """
    Validate that URI can be played by audio system.
    
    Parameters:
    - uri (str): URI to validate
    
    Returns:
    - bool: True if URI is valid and playable
    """
    ...

Usage example:

# Prepare and validate streams
uri = "http://stream.example.com/radio.mp3"
if validate_stream_uri(uri):
    headers = {"User-Agent": "Mopidy/3.4.2"}
    if prepare_stream(uri, headers=headers, timeout=30):
        metadata = get_stream_metadata(uri)
        print(f"Stream title: {metadata.get('title', 'Unknown')}")

Audio Buffer Management

Low-level buffer management for custom audio sources and processing.

def create_audio_buffer(data, sample_rate=44100, channels=2, format="S16LE"):
    """
    Create audio buffer from raw audio data.
    
    Parameters:
    - data (bytes): Raw audio data
    - sample_rate (int): Sample rate in Hz
    - channels (int): Number of audio channels  
    - format (str): Audio format (S16LE, F32LE, etc.)
    
    Returns:
    - GstBuffer: GStreamer audio buffer
    """
    ...

def buffer_to_samples(buffer):
    """
    Extract audio samples from GStreamer buffer.
    
    Parameters:
    - buffer (GstBuffer): GStreamer buffer
    
    Returns:
    - bytes: Raw audio sample data
    """
    ...

def get_buffer_info(buffer):
    """
    Get information about audio buffer.
    
    Parameters:
    - buffer (GstBuffer): GStreamer buffer
    
    Returns:
    - dict: Buffer info (duration, timestamp, size, etc.)
    """
    ...

Audio Device Management

Interface for discovering and managing audio devices.

def get_audio_devices():
    """
    Get list of available audio devices.
    
    Returns:
    - list[dict]: Available audio devices with properties
    """
    ...

def get_default_audio_device():
    """
    Get default audio output device.
    
    Returns:
    - dict: Default device information
    """
    ...

def set_audio_device(device_id):
    """
    Set audio output device.
    
    Parameters:
    - device_id (str): Device identifier
    
    Returns:
    - bool: True if device was set successfully
    """
    ...

Audio System Architecture

Pipeline Structure

The audio system uses GStreamer pipelines with the following general structure:

uridecodebin -> audioconvert -> audioresample -> volume -> output

Key components:

  • uridecodebin: Automatically decodes audio from various URI schemes
  • audioconvert: Converts between audio formats as needed
  • audioresample: Resamples audio to match output requirements
  • volume: Provides software volume control
  • output: Routes audio to speakers, files, or other destinations

Buffer Management

Audio data flows through the pipeline in discrete buffers, each containing:

  • Raw audio samples
  • Timestamp information
  • Duration metadata
  • Format specifications

Event Handling

The audio system generates events for:

  • Playback state changes (play, pause, stop)
  • End of stream notifications
  • Position updates
  • Error conditions
  • Metadata changes

Thread Safety

The audio system uses Pykka actors to ensure thread-safe operation across multiple components while maintaining real-time audio performance requirements.

Mixer System

Mopidy's mixer system provides audio volume and mute control across different audio backends and hardware configurations.

Mixer Base Class

Base mixer implementation that backends can extend to provide volume control.

class Mixer:
    """
    Audio mixer API for volume and mute control.
    
    Parameters:
    - config (dict): Mopidy configuration
    """
    
    name: str  # Name of the mixer (matches extension name)
    
    def __init__(self, config): ...
    
    def get_volume(self):
        """
        Get volume level on a linear scale from 0 to 100.
        
        Returns:
        - int or None: Volume level (0-100), None if unknown
        """
        ...
    
    def set_volume(self, volume):
        """
        Set volume level of the mixer.
        
        Parameters:
        - volume (int): Volume level (0-100)
        
        Returns:
        - bool: True if successful
        """
        ...
    
    def get_mute(self):
        """
        Get mute state of the mixer.
        
        Returns:
        - bool or None: True if muted, False if not, None if unknown
        """
        ...
    
    def set_mute(self, mute):
        """
        Mute or unmute the mixer.
        
        Parameters:
        - mute (bool): True to mute, False to unmute
        
        Returns:
        - bool: True if successful
        """
        ...
    
    def trigger_volume_changed(self, volume):
        """
        Send volume_changed event to all mixer listeners.
        
        Should be called by subclasses when volume changes.
        
        Parameters:
        - volume (int): New volume level
        """
        ...
    
    def trigger_mute_changed(self, mute):
        """
        Send mute_changed event to all mixer listeners.
        
        Should be called by subclasses when mute state changes.
        
        Parameters:
        - mute (bool): New mute state
        """
        ...
    
    def ping(self):
        """
        Check if the mixer actor is still alive.
        
        Returns:
        - bool: True if alive
        """
        ...

Mixer Listener

Event listener interface for receiving mixer state change notifications.

class MixerListener:
    """Listener interface for mixer events."""
    
    def volume_changed(self, volume):
        """
        Called after the volume has changed.
        
        Parameters:
        - volume (int): New volume level (0-100)
        """
        ...
    
    def mute_changed(self, mute):
        """
        Called after the mute state has changed.
        
        Parameters:
        - mute (bool): New mute state
        """
        ...

def send(event, **kwargs):
    """
    Helper to send mixer listener events.
    
    Parameters:
    - event (str): Event name ('volume_changed' or 'mute_changed')
    - **kwargs: Event parameters
    """
    ...

Usage Examples

# Custom mixer implementation
class MyMixer(Mixer):
    name = 'mymixer'
    
    def __init__(self, config):
        super().__init__(config)
        self._volume = 50
        self._muted = False
    
    def get_volume(self):
        return self._volume
    
    def set_volume(self, volume):
        if 0 <= volume <= 100:
            self._volume = volume
            self.trigger_volume_changed(volume)
            return True
        return False
    
    def get_mute(self):
        return self._muted
    
    def set_mute(self, mute):
        self._muted = bool(mute)
        self.trigger_mute_changed(self._muted)
        return True

# Using mixer events
class MyMixerListener(MixerListener):
    def volume_changed(self, volume):
        print(f"Volume changed to {volume}%")
    
    def mute_changed(self, mute):
        print(f"Mute {'enabled' if mute else 'disabled'}")

docs

audio-system.md

backend-system.md

configuration.md

core-controllers.md

extension-system.md

index.md

models.md

tile.json