Players

start_audio_output(port, path, args, media, uuid, timeout=5.0)

Starts an audio output

Parameters:

Name	Type	Description	Default
port	int	The port to use for the audio output	required
path	str	The path to the audio player executable	required
args	list[str]	The arguments to pass to the audio player	required
media	str	The media to play	required
uuid	str	The uuid of the audio output	required
timeout	float	Maximum time to wait for player to start (seconds)	5.0

Returns:

Type	Description
tuple[AudioPlayer, AudioClient]	A tuple containing the audio player and client

Raises:

Type	Description
RuntimeError	If player fails to start within timeout or thread dies

Source code in src/cuemsengine/players/AudioPlayer.py

def start_audio_output(
    port: int,
    path: str,
    args: list[str],
    media: str,
    uuid: str,
    timeout: float = 5.0
) -> tuple[AudioPlayer, AudioClient]:
    """Starts an audio output

    Args:
        port: The port to use for the audio output
        path: The path to the audio player executable
        args: The arguments to pass to the audio player
        media: The media to play
        uuid: The uuid of the audio output
        timeout: Maximum time to wait for player to start (seconds)

    Returns:
        A tuple containing the audio player and client

    Raises:
        RuntimeError: If player fails to start within timeout or thread dies
    """
    player = AudioPlayer(
        port = port,
        path = path,
        args = args,
        media = media,
        uuid = uuid
    )
    player.start(timeout=timeout)

    try:
        client = AudioClient(
            player_port = port,
            name = f'audioplayer-{uuid}'
        )
    except Exception:
        # OSC client creation failed (e.g. port conflict); kill the subprocess so it doesn't linger
        try:
            player.kill()
        except Exception:
            pass
        raise

    return player, client

AudioMixer

Bases: Player

JACK audio mixer using jack-volume controlled via OSC.

This class manages a jack-volume process which provides volume control for multiple audio channels. It connects to JACK and exposes OSC control.

OSC address format: /audiomixer// where channel can be 'master' or '0', '1', '2', etc.

Source code in src/cuemsengine/players/AudioMixer.py

class AudioMixer(Player):
    """JACK audio mixer using jack-volume controlled via OSC.

    This class manages a jack-volume process which provides volume control
    for multiple audio channels. It connects to JACK and exposes OSC control.

    OSC address format: /audiomixer/<instance>/<channel>
    where channel can be 'master' or '0', '1', '2', etc.
    """

    def __init__(self, audio_outputs, port, mixer_id: str, path=None, args: str | None = None):
        """Initialize the AudioMixer.

        Args:
            audio_outputs: List of audio output configurations
            port: OSC port for jack-volume communication
            mixer_id: Unique identifier for this mixer
            path: Optional path to jack-volume binary (defaults to JACK_VOLUME_PATH)
        """
        super().__init__()
        self.conn_man = JackConnectionManager()
        self.port = port
        self.ports = self.conn_man.get_ports()
        self.path = path if path else JACK_VOLUME_PATH
        self.channel_number = len(audio_outputs)
        self.audio_outputs = audio_outputs
        self.client_name = get_mixer_client_name(mixer_id)
        self.extra_args = args

        # Build command line arguments for jack-volume
        self.args = [
            '-c', self.client_name,
            '-p', str(port),
            '-n', str(self.channel_number)
        ]

        # Note: start() will be called by start_audio_mixer() with timeout
        # self.connect_to_jack() will be called after start() in start_audio_mixer()

    @logged
    def run(self):
        """Start the jack-volume subprocess."""
        process_call_list = [self.path] + self.args
        if self.extra_args:
            for arg in self.extra_args.split():
                process_call_list.append(arg)
        Logger.info(f"Starting jack-volume with: {process_call_list}")
        self.call_subprocess(process_call_list)

    @logged
    def connect_to_jack(self, max_retries: int = 10, retry_delay: float = 0.5):
        """Connect mixer outputs to the configured playback ports.

        Retries if ports are not yet registered (race with jack-volume startup).
        """
        for i, playback_port in enumerate(self.audio_outputs):
            output_port = f"{self.client_name}:output_{i+1}"
            # Wait for both ports to be available
            for attempt in range(max_retries):
                if self.conn_man.port_exists(output_port) and self.conn_man.port_exists(playback_port):
                    break
                if attempt < max_retries - 1:
                    Logger.debug(f"Waiting for JACK ports {output_port} / {playback_port} (attempt {attempt + 1}/{max_retries})")
                    sleep(retry_delay)
            else:
                Logger.warning(f"JACK ports not available after {max_retries} attempts: {output_port} -> {playback_port}")
                continue
            Logger.debug(f"Connecting {output_port} to {playback_port}")
            self.conn_man.connect_by_name(output_port, playback_port)

    @logged
    def connect_player_to_mixer(self, player_name: str, player_output_prefix: str = 'output', mixer_channel: int = 0, max_retries: int = 30, retry_delay: float = 0.5):
        """Connect a player's output to a specific mixer input channel.

        First disconnects any existing connections from the player's outputs,
        then connects them to the mixer inputs. Will retry if ports are not
        immediately available (race condition with player startup).

        Handles both mono and stereo players:
        - Mono: output_0 → input_1 (single channel)
        - Stereo: output_0 → input_1, output_1 → input_2

        Args:
            player_name: Name of the player JACK client to connect
            player_output_prefix: Prefix for player's output ports (e.g., 'output')
            mixer_channel: Mixer input channel number (0-indexed)
            max_retries: Maximum number of connection attempts (default 10)
            retry_delay: Delay between retries in seconds (default 0.2)
        """
        from time import sleep

        if mixer_channel >= self.channel_number:
            Logger.error(f"Invalid mixer channel: {mixer_channel}. Max: {self.channel_number - 1}")
            return

        # Define player output ports
        # cuems-audioplayer uses space format: "outport 0", "outport 1"
        channel_0_output = f"{player_name}:{player_output_prefix} 0"
        channel_1_output = f"{player_name}:{player_output_prefix} 1"
        mixer_input_1 = f"{self.client_name}:input_{mixer_channel * 2 + 1}"
        mixer_input_2 = f"{self.client_name}:input_{mixer_channel * 2 + 2}"

        # Wait for player JACK ports to be available (retry mechanism)
        for attempt in range(max_retries):
            # Check if ports exist by trying to get connections
            connections = self.conn_man.get_connections(channel_0_output)
            if connections is not None or self.conn_man.port_exists(channel_0_output):
                break
            if attempt < max_retries - 1:
                Logger.debug(f"Waiting for JACK port {channel_0_output} (attempt {attempt + 1}/{max_retries})")
                sleep(retry_delay)
        else:
            Logger.warning(f"JACK port {channel_0_output} not available after {max_retries} attempts")

        # Check if player is stereo (has output_1) or mono (only output_0)
        is_stereo = self.conn_man.port_exists(channel_1_output)
        Logger.debug(f"Player {player_name} is {'stereo' if is_stereo else 'mono'}")

        # First, disconnect any existing connections from player outputs
        # Guard with port_exists to avoid sending disconnect requests for
        # ports that were destroyed by a concurrent /quit.
        if self.conn_man.port_exists(channel_0_output):
            Logger.debug(f"Disconnecting existing connections from {channel_0_output}")
            channel_0_connections = self.conn_man.get_connections(channel_0_output)
            for connection in channel_0_connections:
                Logger.debug(f"Disconnecting {channel_0_output} from {connection}")
                self.conn_man.disconnect_by_name(channel_0_output, connection)

        if is_stereo and self.conn_man.port_exists(channel_1_output):
            Logger.debug(f"Disconnecting existing connections from {channel_1_output}")
            channel_1_connections = self.conn_man.get_connections(channel_1_output)
            for connection in channel_1_connections:
                Logger.debug(f"Disconnecting {channel_1_output} from {connection}")
                self.conn_man.disconnect_by_name(channel_1_output, connection)

        # Connect to mixer inputs
        # For mono: connect output_0 to both input_1 and input_2 (if available)
        # For stereo: connect output_0 → input_1, output_1 → input_2

        # Connect first channel
        if self.conn_man.port_exists(mixer_input_1):
            Logger.debug(f"Connecting {channel_0_output} to {mixer_input_1}")
            self.conn_man.connect_by_name(channel_0_output, mixer_input_1)
        else:
            Logger.warning(f"Mixer input port {mixer_input_1} does not exist")

        # Connect second channel (if mixer has it)
        if self.conn_man.port_exists(mixer_input_2):
            if is_stereo:
                Logger.debug(f"Connecting {channel_1_output} to {mixer_input_2}")
                self.conn_man.connect_by_name(channel_1_output, mixer_input_2)
            else:
                # Mono player: connect output_0 to both mixer inputs for centered sound
                Logger.debug(f"Mono player: Connecting {channel_0_output} to {mixer_input_2}")
                self.conn_man.connect_by_name(channel_0_output, mixer_input_2)
        else:
            Logger.debug(f"Mixer input port {mixer_input_2} does not exist (mono mixer)")

    @logged
    def connect_player_to_outputs(self, player_name: str, player_output_prefix: str = 'outport', 
                                   selected_outputs: list = None, max_retries: int = 30, retry_delay: float = 0.5):
        """Connect a player to specific system outputs based on cue configuration.

        Maps selected output port names to mixer inputs:
        - system:playback_1 → mixer input_1
        - system:playback_2 → mixer input_2

        For stereo audio with a single output selected, both player channels
        are summed to that output. For both outputs, normal stereo routing.

        Args:
            player_name: Name of the player JACK client to connect
            player_output_prefix: Prefix for player's output ports (e.g., 'outport')
            selected_outputs: List of output port names (e.g., ['system:playback_1'])
            max_retries: Maximum number of connection attempts
            retry_delay: Delay between retries in seconds
        """
        from time import sleep

        # Default to stereo (both outputs) if none specified
        if not selected_outputs:
            selected_outputs = ['system:playback_1', 'system:playback_2']
            Logger.debug(f"No outputs specified, defaulting to stereo: {selected_outputs}")

        # Define player output ports - cuems-audioplayer uses "outport 0", "outport 1"
        channel_0_output = f"{player_name}:{player_output_prefix} 0"
        channel_1_output = f"{player_name}:{player_output_prefix} 1"

        # Build output→input mapping from the configured audio_outputs list
        output_to_input = {
            name: f"{self.client_name}:input_{i+1}"
            for i, name in enumerate(self.audio_outputs)
        }

        # Wait for player JACK ports to be available
        for attempt in range(max_retries):
            connections = self.conn_man.get_connections(channel_0_output)
            if connections is not None or self.conn_man.port_exists(channel_0_output):
                break
            if attempt < max_retries - 1:
                Logger.debug(f"Waiting for JACK port {channel_0_output} (attempt {attempt + 1}/{max_retries})")
                sleep(retry_delay)
        else:
            Logger.warning(f"JACK port {channel_0_output} not available after {max_retries} attempts")
            return

        # Check if player is stereo
        is_stereo = self.conn_man.port_exists(channel_1_output)
        Logger.debug(f"Player {player_name} is {'stereo' if is_stereo else 'mono'}")

        # First, disconnect any existing connections from player outputs
        # Guard with port_exists to avoid operating on destroyed ports.
        if self.conn_man.port_exists(channel_0_output):
            Logger.debug(f"Disconnecting existing connections from {channel_0_output}")
            channel_0_connections = self.conn_man.get_connections(channel_0_output)
            for connection in channel_0_connections:
                self.conn_man.disconnect_by_name(channel_0_output, connection)

        if is_stereo and self.conn_man.port_exists(channel_1_output):
            channel_1_connections = self.conn_man.get_connections(channel_1_output)
            for connection in channel_1_connections:
                self.conn_man.disconnect_by_name(channel_1_output, connection)

        # Determine which mixer inputs to connect to
        target_inputs = []
        for output in selected_outputs:
            if output in output_to_input:
                mixer_input = output_to_input[output]
                if self.conn_man.port_exists(mixer_input):
                    target_inputs.append(mixer_input)
                else:
                    Logger.warning(f"Mixer input {mixer_input} does not exist")

        if not target_inputs:
            Logger.error(f"No valid mixer inputs found for outputs: {selected_outputs}")
            return

        Logger.info(f"Connecting {player_name} to outputs: {selected_outputs} -> {target_inputs}")

        # Fan-out routing: treat target_inputs as alternating L/R pairs.
        # Even-indexed targets (0, 2, 4 …) receive outport 0 (L channel).
        # Odd-indexed targets  (1, 3, 5 …) receive outport 1 (R channel)
        #   or outport 0 again when the player is mono.
        # This covers 1, 2 or any number of outputs uniformly.
        for i, mixer_input in enumerate(target_inputs):
            if i % 2 == 0:
                Logger.debug(f"L → {mixer_input}")
                self.conn_man.connect_by_name(channel_0_output, mixer_input)
            else:
                if is_stereo:
                    Logger.debug(f"R → {mixer_input}")
                    self.conn_man.connect_by_name(channel_1_output, mixer_input)
                else:
                    Logger.debug(f"Mono → {mixer_input}")
                    self.conn_man.connect_by_name(channel_0_output, mixer_input)


    def player_connections_correct(self, player_name: str,
                                   player_output_prefix: str = 'outport',
                                   selected_outputs: list = None) -> bool:
        """Verify the player's outputs are wired exactly as connect_player_to_outputs would wire them.

        Mirrors the routing in connect_player_to_outputs: same output_to_input
        mapping (built from audio_outputs), same alternating L/R fan-out walk,
        same mono branch (outport 0 → both pair members when channel_1 absent).

        Returns False if any expected edge is missing, points elsewhere, or if
        outport 0 itself does not exist (subprocess gone). Caller decides
        whether to repair via connect_player_to_outputs or abort the cue.
        """
        if not selected_outputs:
            selected_outputs = ['system:playback_1', 'system:playback_2']

        channel_0_output = f"{player_name}:{player_output_prefix} 0"
        channel_1_output = f"{player_name}:{player_output_prefix} 1"

        if not self.conn_man.port_exists(channel_0_output):
            return False

        is_stereo = self.conn_man.port_exists(channel_1_output)

        output_to_input = {
            name: f"{self.client_name}:input_{i+1}"
            for i, name in enumerate(self.audio_outputs)
        }

        target_inputs = []
        for output in selected_outputs:
            if output in output_to_input:
                mixer_input = output_to_input[output]
                if self.conn_man.port_exists(mixer_input):
                    target_inputs.append(mixer_input)

        if not target_inputs:
            return False

        for i, mixer_input in enumerate(target_inputs):
            if i % 2 == 0 or not is_stereo:
                expected_src = channel_0_output
            else:
                expected_src = channel_1_output
            if not self.conn_man.is_connected(expected_src, mixer_input):
                return False

        return True

    @logged
    def disconnect_player(self, player_name: str, player_output_prefix: str = 'outport'):
        """Disconnect a player's outputs from the mixer.

        Must be called BEFORE the player's JACK client is destroyed (i.e. before
        sending /quit), otherwise JACK receives disconnect requests for ports
        that no longer exist, which can corrupt its shared memory registry.

        Args:
            player_name: Name of the player JACK client
            player_output_prefix: Prefix for player's output ports
        """
        channel_0_output = f"{player_name}:{player_output_prefix} 0"
        channel_1_output = f"{player_name}:{player_output_prefix} 1"

        for port_name in (channel_0_output, channel_1_output):
            if not self.conn_man.port_exists(port_name):
                continue
            connections = self.conn_man.get_connections(port_name)
            for connection in connections:
                Logger.debug(f"Disconnecting {port_name} from {connection}")
                self.conn_man.disconnect_by_name(port_name, connection)

__init__(audio_outputs, port, mixer_id, path=None, args=None)

Initialize the AudioMixer.

Parameters:

Name	Type	Description	Default
audio_outputs		List of audio output configurations	required
port		OSC port for jack-volume communication	required
mixer_id	str	Unique identifier for this mixer	required
path		Optional path to jack-volume binary (defaults to JACK_VOLUME_PATH)	None

Source code in src/cuemsengine/players/AudioMixer.py

def __init__(self, audio_outputs, port, mixer_id: str, path=None, args: str | None = None):
    """Initialize the AudioMixer.

    Args:
        audio_outputs: List of audio output configurations
        port: OSC port for jack-volume communication
        mixer_id: Unique identifier for this mixer
        path: Optional path to jack-volume binary (defaults to JACK_VOLUME_PATH)
    """
    super().__init__()
    self.conn_man = JackConnectionManager()
    self.port = port
    self.ports = self.conn_man.get_ports()
    self.path = path if path else JACK_VOLUME_PATH
    self.channel_number = len(audio_outputs)
    self.audio_outputs = audio_outputs
    self.client_name = get_mixer_client_name(mixer_id)
    self.extra_args = args

    # Build command line arguments for jack-volume
    self.args = [
        '-c', self.client_name,
        '-p', str(port),
        '-n', str(self.channel_number)
    ]

connect_player_to_mixer(player_name, player_output_prefix='output', mixer_channel=0, max_retries=30, retry_delay=0.5)

Connect a player's output to a specific mixer input channel.

First disconnects any existing connections from the player's outputs, then connects them to the mixer inputs. Will retry if ports are not immediately available (race condition with player startup).

Handles both mono and stereo players: - Mono: output_0 → input_1 (single channel) - Stereo: output_0 → input_1, output_1 → input_2

Parameters:

Name	Type	Description	Default
player_name	str	Name of the player JACK client to connect	required
player_output_prefix	str	Prefix for player's output ports (e.g., 'output')	'output'
mixer_channel	int	Mixer input channel number (0-indexed)	0
max_retries	int	Maximum number of connection attempts (default 10)	30
retry_delay	float	Delay between retries in seconds (default 0.2)	0.5

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def connect_player_to_mixer(self, player_name: str, player_output_prefix: str = 'output', mixer_channel: int = 0, max_retries: int = 30, retry_delay: float = 0.5):
    """Connect a player's output to a specific mixer input channel.

    First disconnects any existing connections from the player's outputs,
    then connects them to the mixer inputs. Will retry if ports are not
    immediately available (race condition with player startup).

    Handles both mono and stereo players:
    - Mono: output_0 → input_1 (single channel)
    - Stereo: output_0 → input_1, output_1 → input_2

    Args:
        player_name: Name of the player JACK client to connect
        player_output_prefix: Prefix for player's output ports (e.g., 'output')
        mixer_channel: Mixer input channel number (0-indexed)
        max_retries: Maximum number of connection attempts (default 10)
        retry_delay: Delay between retries in seconds (default 0.2)
    """
    from time import sleep

    if mixer_channel >= self.channel_number:
        Logger.error(f"Invalid mixer channel: {mixer_channel}. Max: {self.channel_number - 1}")
        return

    # Define player output ports
    # cuems-audioplayer uses space format: "outport 0", "outport 1"
    channel_0_output = f"{player_name}:{player_output_prefix} 0"
    channel_1_output = f"{player_name}:{player_output_prefix} 1"
    mixer_input_1 = f"{self.client_name}:input_{mixer_channel * 2 + 1}"
    mixer_input_2 = f"{self.client_name}:input_{mixer_channel * 2 + 2}"

    # Wait for player JACK ports to be available (retry mechanism)
    for attempt in range(max_retries):
        # Check if ports exist by trying to get connections
        connections = self.conn_man.get_connections(channel_0_output)
        if connections is not None or self.conn_man.port_exists(channel_0_output):
            break
        if attempt < max_retries - 1:
            Logger.debug(f"Waiting for JACK port {channel_0_output} (attempt {attempt + 1}/{max_retries})")
            sleep(retry_delay)
    else:
        Logger.warning(f"JACK port {channel_0_output} not available after {max_retries} attempts")

    # Check if player is stereo (has output_1) or mono (only output_0)
    is_stereo = self.conn_man.port_exists(channel_1_output)
    Logger.debug(f"Player {player_name} is {'stereo' if is_stereo else 'mono'}")

    # First, disconnect any existing connections from player outputs
    # Guard with port_exists to avoid sending disconnect requests for
    # ports that were destroyed by a concurrent /quit.
    if self.conn_man.port_exists(channel_0_output):
        Logger.debug(f"Disconnecting existing connections from {channel_0_output}")
        channel_0_connections = self.conn_man.get_connections(channel_0_output)
        for connection in channel_0_connections:
            Logger.debug(f"Disconnecting {channel_0_output} from {connection}")
            self.conn_man.disconnect_by_name(channel_0_output, connection)

    if is_stereo and self.conn_man.port_exists(channel_1_output):
        Logger.debug(f"Disconnecting existing connections from {channel_1_output}")
        channel_1_connections = self.conn_man.get_connections(channel_1_output)
        for connection in channel_1_connections:
            Logger.debug(f"Disconnecting {channel_1_output} from {connection}")
            self.conn_man.disconnect_by_name(channel_1_output, connection)

    # Connect to mixer inputs
    # For mono: connect output_0 to both input_1 and input_2 (if available)
    # For stereo: connect output_0 → input_1, output_1 → input_2

    # Connect first channel
    if self.conn_man.port_exists(mixer_input_1):
        Logger.debug(f"Connecting {channel_0_output} to {mixer_input_1}")
        self.conn_man.connect_by_name(channel_0_output, mixer_input_1)
    else:
        Logger.warning(f"Mixer input port {mixer_input_1} does not exist")

    # Connect second channel (if mixer has it)
    if self.conn_man.port_exists(mixer_input_2):
        if is_stereo:
            Logger.debug(f"Connecting {channel_1_output} to {mixer_input_2}")
            self.conn_man.connect_by_name(channel_1_output, mixer_input_2)
        else:
            # Mono player: connect output_0 to both mixer inputs for centered sound
            Logger.debug(f"Mono player: Connecting {channel_0_output} to {mixer_input_2}")
            self.conn_man.connect_by_name(channel_0_output, mixer_input_2)
    else:
        Logger.debug(f"Mixer input port {mixer_input_2} does not exist (mono mixer)")

connect_player_to_outputs(player_name, player_output_prefix='outport', selected_outputs=None, max_retries=30, retry_delay=0.5)

Connect a player to specific system outputs based on cue configuration.

Maps selected output port names to mixer inputs: - system:playback_1 → mixer input_1 - system:playback_2 → mixer input_2

For stereo audio with a single output selected, both player channels are summed to that output. For both outputs, normal stereo routing.

Parameters:

Name	Type	Description	Default
player_name	str	Name of the player JACK client to connect	required
player_output_prefix	str	Prefix for player's output ports (e.g., 'outport')	'outport'
selected_outputs	list	List of output port names (e.g., ['system:playback_1'])	None
max_retries	int	Maximum number of connection attempts	30
retry_delay	float	Delay between retries in seconds	0.5

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def connect_player_to_outputs(self, player_name: str, player_output_prefix: str = 'outport', 
                               selected_outputs: list = None, max_retries: int = 30, retry_delay: float = 0.5):
    """Connect a player to specific system outputs based on cue configuration.

    Maps selected output port names to mixer inputs:
    - system:playback_1 → mixer input_1
    - system:playback_2 → mixer input_2

    For stereo audio with a single output selected, both player channels
    are summed to that output. For both outputs, normal stereo routing.

    Args:
        player_name: Name of the player JACK client to connect
        player_output_prefix: Prefix for player's output ports (e.g., 'outport')
        selected_outputs: List of output port names (e.g., ['system:playback_1'])
        max_retries: Maximum number of connection attempts
        retry_delay: Delay between retries in seconds
    """
    from time import sleep

    # Default to stereo (both outputs) if none specified
    if not selected_outputs:
        selected_outputs = ['system:playback_1', 'system:playback_2']
        Logger.debug(f"No outputs specified, defaulting to stereo: {selected_outputs}")

    # Define player output ports - cuems-audioplayer uses "outport 0", "outport 1"
    channel_0_output = f"{player_name}:{player_output_prefix} 0"
    channel_1_output = f"{player_name}:{player_output_prefix} 1"

    # Build output→input mapping from the configured audio_outputs list
    output_to_input = {
        name: f"{self.client_name}:input_{i+1}"
        for i, name in enumerate(self.audio_outputs)
    }

    # Wait for player JACK ports to be available
    for attempt in range(max_retries):
        connections = self.conn_man.get_connections(channel_0_output)
        if connections is not None or self.conn_man.port_exists(channel_0_output):
            break
        if attempt < max_retries - 1:
            Logger.debug(f"Waiting for JACK port {channel_0_output} (attempt {attempt + 1}/{max_retries})")
            sleep(retry_delay)
    else:
        Logger.warning(f"JACK port {channel_0_output} not available after {max_retries} attempts")
        return

    # Check if player is stereo
    is_stereo = self.conn_man.port_exists(channel_1_output)
    Logger.debug(f"Player {player_name} is {'stereo' if is_stereo else 'mono'}")

    # First, disconnect any existing connections from player outputs
    # Guard with port_exists to avoid operating on destroyed ports.
    if self.conn_man.port_exists(channel_0_output):
        Logger.debug(f"Disconnecting existing connections from {channel_0_output}")
        channel_0_connections = self.conn_man.get_connections(channel_0_output)
        for connection in channel_0_connections:
            self.conn_man.disconnect_by_name(channel_0_output, connection)

    if is_stereo and self.conn_man.port_exists(channel_1_output):
        channel_1_connections = self.conn_man.get_connections(channel_1_output)
        for connection in channel_1_connections:
            self.conn_man.disconnect_by_name(channel_1_output, connection)

    # Determine which mixer inputs to connect to
    target_inputs = []
    for output in selected_outputs:
        if output in output_to_input:
            mixer_input = output_to_input[output]
            if self.conn_man.port_exists(mixer_input):
                target_inputs.append(mixer_input)
            else:
                Logger.warning(f"Mixer input {mixer_input} does not exist")

    if not target_inputs:
        Logger.error(f"No valid mixer inputs found for outputs: {selected_outputs}")
        return

    Logger.info(f"Connecting {player_name} to outputs: {selected_outputs} -> {target_inputs}")

    # Fan-out routing: treat target_inputs as alternating L/R pairs.
    # Even-indexed targets (0, 2, 4 …) receive outport 0 (L channel).
    # Odd-indexed targets  (1, 3, 5 …) receive outport 1 (R channel)
    #   or outport 0 again when the player is mono.
    # This covers 1, 2 or any number of outputs uniformly.
    for i, mixer_input in enumerate(target_inputs):
        if i % 2 == 0:
            Logger.debug(f"L → {mixer_input}")
            self.conn_man.connect_by_name(channel_0_output, mixer_input)
        else:
            if is_stereo:
                Logger.debug(f"R → {mixer_input}")
                self.conn_man.connect_by_name(channel_1_output, mixer_input)
            else:
                Logger.debug(f"Mono → {mixer_input}")
                self.conn_man.connect_by_name(channel_0_output, mixer_input)

connect_to_jack(max_retries=10, retry_delay=0.5)

Connect mixer outputs to the configured playback ports.

Retries if ports are not yet registered (race with jack-volume startup).

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def connect_to_jack(self, max_retries: int = 10, retry_delay: float = 0.5):
    """Connect mixer outputs to the configured playback ports.

    Retries if ports are not yet registered (race with jack-volume startup).
    """
    for i, playback_port in enumerate(self.audio_outputs):
        output_port = f"{self.client_name}:output_{i+1}"
        # Wait for both ports to be available
        for attempt in range(max_retries):
            if self.conn_man.port_exists(output_port) and self.conn_man.port_exists(playback_port):
                break
            if attempt < max_retries - 1:
                Logger.debug(f"Waiting for JACK ports {output_port} / {playback_port} (attempt {attempt + 1}/{max_retries})")
                sleep(retry_delay)
        else:
            Logger.warning(f"JACK ports not available after {max_retries} attempts: {output_port} -> {playback_port}")
            continue
        Logger.debug(f"Connecting {output_port} to {playback_port}")
        self.conn_man.connect_by_name(output_port, playback_port)

disconnect_player(player_name, player_output_prefix='outport')

Disconnect a player's outputs from the mixer.

Must be called BEFORE the player's JACK client is destroyed (i.e. before sending /quit), otherwise JACK receives disconnect requests for ports that no longer exist, which can corrupt its shared memory registry.

Parameters:

Name	Type	Description	Default
player_name	str	Name of the player JACK client	required
player_output_prefix	str	Prefix for player's output ports	'outport'

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def disconnect_player(self, player_name: str, player_output_prefix: str = 'outport'):
    """Disconnect a player's outputs from the mixer.

    Must be called BEFORE the player's JACK client is destroyed (i.e. before
    sending /quit), otherwise JACK receives disconnect requests for ports
    that no longer exist, which can corrupt its shared memory registry.

    Args:
        player_name: Name of the player JACK client
        player_output_prefix: Prefix for player's output ports
    """
    channel_0_output = f"{player_name}:{player_output_prefix} 0"
    channel_1_output = f"{player_name}:{player_output_prefix} 1"

    for port_name in (channel_0_output, channel_1_output):
        if not self.conn_man.port_exists(port_name):
            continue
        connections = self.conn_man.get_connections(port_name)
        for connection in connections:
            Logger.debug(f"Disconnecting {port_name} from {connection}")
            self.conn_man.disconnect_by_name(port_name, connection)

player_connections_correct(player_name, player_output_prefix='outport', selected_outputs=None)

Verify the player's outputs are wired exactly as connect_player_to_outputs would wire them.

Mirrors the routing in connect_player_to_outputs: same output_to_input mapping (built from audio_outputs), same alternating L/R fan-out walk, same mono branch (outport 0 → both pair members when channel_1 absent).

Returns False if any expected edge is missing, points elsewhere, or if outport 0 itself does not exist (subprocess gone). Caller decides whether to repair via connect_player_to_outputs or abort the cue.

Source code in src/cuemsengine/players/AudioMixer.py

def player_connections_correct(self, player_name: str,
                               player_output_prefix: str = 'outport',
                               selected_outputs: list = None) -> bool:
    """Verify the player's outputs are wired exactly as connect_player_to_outputs would wire them.

    Mirrors the routing in connect_player_to_outputs: same output_to_input
    mapping (built from audio_outputs), same alternating L/R fan-out walk,
    same mono branch (outport 0 → both pair members when channel_1 absent).

    Returns False if any expected edge is missing, points elsewhere, or if
    outport 0 itself does not exist (subprocess gone). Caller decides
    whether to repair via connect_player_to_outputs or abort the cue.
    """
    if not selected_outputs:
        selected_outputs = ['system:playback_1', 'system:playback_2']

    channel_0_output = f"{player_name}:{player_output_prefix} 0"
    channel_1_output = f"{player_name}:{player_output_prefix} 1"

    if not self.conn_man.port_exists(channel_0_output):
        return False

    is_stereo = self.conn_man.port_exists(channel_1_output)

    output_to_input = {
        name: f"{self.client_name}:input_{i+1}"
        for i, name in enumerate(self.audio_outputs)
    }

    target_inputs = []
    for output in selected_outputs:
        if output in output_to_input:
            mixer_input = output_to_input[output]
            if self.conn_man.port_exists(mixer_input):
                target_inputs.append(mixer_input)

    if not target_inputs:
        return False

    for i, mixer_input in enumerate(target_inputs):
        if i % 2 == 0 or not is_stereo:
            expected_src = channel_0_output
        else:
            expected_src = channel_1_output
        if not self.conn_man.is_connected(expected_src, mixer_input):
            return False

    return True

run()

Start the jack-volume subprocess.

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def run(self):
    """Start the jack-volume subprocess."""
    process_call_list = [self.path] + self.args
    if self.extra_args:
        for arg in self.extra_args.split():
            process_call_list.append(arg)
    Logger.info(f"Starting jack-volume with: {process_call_list}")
    self.call_subprocess(process_call_list)

MixerClient

Bases: PlayerClient

OSC Client for controlling the AudioMixer via jack-volume.

Provides methods to control volume for individual channels and master volume. Uses OSC addresses: /audiomixer// where channel can be 'master' or '0', '1', '2', etc.

Source code in src/cuemsengine/players/AudioMixer.py

class MixerClient(PlayerClient):
    """OSC Client for controlling the AudioMixer via jack-volume.

    Provides methods to control volume for individual channels and master volume.
    Uses OSC addresses: /audiomixer/<instance>/<channel>
    where channel can be 'master' or '0', '1', '2', etc.
    """

    def __init__(self, player_port: int, channel_number: int, mixer_id: str):
        """Initialize the MixerClient.

        Args:
            player_port: OSC port where jack-volume is listening
            channel_number: Number of audio channels in the mixer
            mixer_id: Unique identifier for this mixer
        """
        self.client_name = get_mixer_client_name(mixer_id)
        self.channel_number = channel_number

        # Build OSC endpoint configuration for jack-volume
        endpoints = build_mixer_osc_endpoints(self.client_name, channel_number)

        super().__init__(
            player_port=player_port,
            endpoints=endpoints,
            name=f'mixer-{mixer_id}'
        )

    @logged
    def set_master_volume(self, gain: float):
        """Set the master volume gain.

        Args:
            gain: Volume gain (0.0 to 1.0)
        """
        if not 0.0 <= gain <= 1.0:
            Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
            return

        path = f'/audiomixer/{self.client_name}/master'
        Logger.debug(f"Setting master volume to {gain}")
        self.set_value(path, gain)

    @logged
    def set_channel_volume(self, channel: int, gain: float):
        """Set volume for a specific channel.

        Args:
            channel: Channel number (0-indexed)
            gain: Volume gain (0.0 to 1.0)
        """
        if not 0.0 <= gain <= 1.0:
            Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
            return

        if channel >= self.channel_number:
            Logger.error(f"Invalid channel: {channel}. Max: {self.channel_number - 1}")
            return

        path = f'/audiomixer/{self.client_name}/{channel}'
        Logger.debug(f"Setting channel {channel} volume to {gain}")
        self.set_value(path, gain)

    @logged
    def set_all_channels_volume(self, gain: float):
        """Set volume for all channels (excluding master).

        Args:
            gain: Volume gain (0.0 to 1.0)
        """
        for i in range(self.channel_number):
            self.set_channel_volume(i, gain)

    @logged
    def reset_volumes(self):
        """Reset all volumes to maximum (1.0).

        Call this when loading a project or starting playback to ensure
        consistent volume levels.
        """
        Logger.info("Resetting mixer volumes to default (1.0)")
        self.set_master_volume(1.0)
        self.set_all_channels_volume(1.0)

    @logged
    def mute_channel(self, channel: int):
        """Mute a specific channel by setting its volume to 0.0.

        Args:
            channel: Channel number (0-indexed)
        """
        self.set_channel_volume(channel, 0.0)

    @logged
    def unmute_channel(self, channel: int, gain: float = 1.0):
        """Unmute a specific channel by setting its volume.

        Args:
            channel: Channel number (0-indexed)
            gain: Volume gain to restore (0.0 to 1.0), defaults to 1.0
        """
        self.set_channel_volume(channel, gain)

    @logged
    def mute_master(self):
        """Mute master volume."""
        self.set_master_volume(0.0)

    @logged
    def unmute_master(self, gain: float = 1.0):
        """Unmute master volume.

        Args:
            gain: Volume gain to restore (0.0 to 1.0), defaults to 1.0
        """
        self.set_master_volume(gain)

    @logged
    def add_to_oscquery_server(self, oscquery_server):
        """Add this mixer's OSC routes to a local OSCQuery server.

        This allows the mixer controls to be visible and controllable
        through the OSCQuery server interface.

        Args:
            oscquery_server: OssiaServer instance to add endpoints to
        """
        Logger.info(f"Adding mixer {self.client_name} to OSCQuery server")

        # Get endpoints from this client
        endpoints = self.get_endpoints()
        Logger.debug(f"Mixer endpoints: {list(endpoints.keys())}")

        # Create callback that forwards values from server to this client
        def server_to_client_callback(value):
            """Forward OSC values from server to mixer client."""
            Logger.debug(f"Forwarding value to mixer: {value}")
            # The value will be automatically sent to jack-volume via the OSC client

        # Add callback to all endpoints
        endpoints_with_callbacks = add_callback_to_all(endpoints, server_to_client_callback)

        # Add endpoints to the OSCQuery server
        oscquery_server.add_endpoints(endpoints_with_callbacks)

        Logger.info(f"Mixer {self.client_name} added to OSCQuery server with {len(endpoints)} endpoints")

__init__(player_port, channel_number, mixer_id)

Initialize the MixerClient.

Parameters:

Name	Type	Description	Default
player_port	int	OSC port where jack-volume is listening	required
channel_number	int	Number of audio channels in the mixer	required
mixer_id	str	Unique identifier for this mixer	required

Source code in src/cuemsengine/players/AudioMixer.py

def __init__(self, player_port: int, channel_number: int, mixer_id: str):
    """Initialize the MixerClient.

    Args:
        player_port: OSC port where jack-volume is listening
        channel_number: Number of audio channels in the mixer
        mixer_id: Unique identifier for this mixer
    """
    self.client_name = get_mixer_client_name(mixer_id)
    self.channel_number = channel_number

    # Build OSC endpoint configuration for jack-volume
    endpoints = build_mixer_osc_endpoints(self.client_name, channel_number)

    super().__init__(
        player_port=player_port,
        endpoints=endpoints,
        name=f'mixer-{mixer_id}'
    )

add_to_oscquery_server(oscquery_server)

Add this mixer's OSC routes to a local OSCQuery server.

This allows the mixer controls to be visible and controllable through the OSCQuery server interface.

Parameters:

Name	Type	Description	Default
oscquery_server		OssiaServer instance to add endpoints to	required

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def add_to_oscquery_server(self, oscquery_server):
    """Add this mixer's OSC routes to a local OSCQuery server.

    This allows the mixer controls to be visible and controllable
    through the OSCQuery server interface.

    Args:
        oscquery_server: OssiaServer instance to add endpoints to
    """
    Logger.info(f"Adding mixer {self.client_name} to OSCQuery server")

    # Get endpoints from this client
    endpoints = self.get_endpoints()
    Logger.debug(f"Mixer endpoints: {list(endpoints.keys())}")

    # Create callback that forwards values from server to this client
    def server_to_client_callback(value):
        """Forward OSC values from server to mixer client."""
        Logger.debug(f"Forwarding value to mixer: {value}")
        # The value will be automatically sent to jack-volume via the OSC client

    # Add callback to all endpoints
    endpoints_with_callbacks = add_callback_to_all(endpoints, server_to_client_callback)

    # Add endpoints to the OSCQuery server
    oscquery_server.add_endpoints(endpoints_with_callbacks)

    Logger.info(f"Mixer {self.client_name} added to OSCQuery server with {len(endpoints)} endpoints")

mute_channel(channel)

Mute a specific channel by setting its volume to 0.0.

Parameters:

Name	Type	Description	Default
channel	int	Channel number (0-indexed)	required

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def mute_channel(self, channel: int):
    """Mute a specific channel by setting its volume to 0.0.

    Args:
        channel: Channel number (0-indexed)
    """
    self.set_channel_volume(channel, 0.0)

mute_master()

Mute master volume.

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def mute_master(self):
    """Mute master volume."""
    self.set_master_volume(0.0)

reset_volumes()

Reset all volumes to maximum (1.0).

Call this when loading a project or starting playback to ensure consistent volume levels.

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def reset_volumes(self):
    """Reset all volumes to maximum (1.0).

    Call this when loading a project or starting playback to ensure
    consistent volume levels.
    """
    Logger.info("Resetting mixer volumes to default (1.0)")
    self.set_master_volume(1.0)
    self.set_all_channels_volume(1.0)

set_all_channels_volume(gain)

Set volume for all channels (excluding master).

Parameters:

Name	Type	Description	Default
gain	float	Volume gain (0.0 to 1.0)	required

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def set_all_channels_volume(self, gain: float):
    """Set volume for all channels (excluding master).

    Args:
        gain: Volume gain (0.0 to 1.0)
    """
    for i in range(self.channel_number):
        self.set_channel_volume(i, gain)

set_channel_volume(channel, gain)

Set volume for a specific channel.

Parameters:

Name	Type	Description	Default
channel	int	Channel number (0-indexed)	required
gain	float	Volume gain (0.0 to 1.0)	required

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def set_channel_volume(self, channel: int, gain: float):
    """Set volume for a specific channel.

    Args:
        channel: Channel number (0-indexed)
        gain: Volume gain (0.0 to 1.0)
    """
    if not 0.0 <= gain <= 1.0:
        Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
        return

    if channel >= self.channel_number:
        Logger.error(f"Invalid channel: {channel}. Max: {self.channel_number - 1}")
        return

    path = f'/audiomixer/{self.client_name}/{channel}'
    Logger.debug(f"Setting channel {channel} volume to {gain}")
    self.set_value(path, gain)

set_master_volume(gain)

Set the master volume gain.

Parameters:

Name	Type	Description	Default
gain	float	Volume gain (0.0 to 1.0)	required

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def set_master_volume(self, gain: float):
    """Set the master volume gain.

    Args:
        gain: Volume gain (0.0 to 1.0)
    """
    if not 0.0 <= gain <= 1.0:
        Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
        return

    path = f'/audiomixer/{self.client_name}/master'
    Logger.debug(f"Setting master volume to {gain}")
    self.set_value(path, gain)

unmute_channel(channel, gain=1.0)

Unmute a specific channel by setting its volume.

Parameters:

Name	Type	Description	Default
channel	int	Channel number (0-indexed)	required
gain	float	Volume gain to restore (0.0 to 1.0), defaults to 1.0	1.0

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def unmute_channel(self, channel: int, gain: float = 1.0):
    """Unmute a specific channel by setting its volume.

    Args:
        channel: Channel number (0-indexed)
        gain: Volume gain to restore (0.0 to 1.0), defaults to 1.0
    """
    self.set_channel_volume(channel, gain)

unmute_master(gain=1.0)

Unmute master volume.

Parameters:

Name	Type	Description	Default
gain	float	Volume gain to restore (0.0 to 1.0), defaults to 1.0	1.0

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def unmute_master(self, gain: float = 1.0):
    """Unmute master volume.

    Args:
        gain: Volume gain to restore (0.0 to 1.0), defaults to 1.0
    """
    self.set_master_volume(gain)

build_mixer_osc_endpoints(client_name, channel_number)

Build OSC endpoint configuration for audio mixer.

Creates OSC addresses in the format expected by jack-volume (audiomixer_routes branch): /audiomixer/{client_name}/master /audiomixer/{client_name}/0 /audiomixer/{client_name}/1 etc.

Parameters:

Name	Type	Description	Default
client_name	str	Name of the mixer client instance (JACK client name)	required
channel_number	int	Number of audio channels in the mixer	required

Returns:

Type	Description
dict	Dictionary of OSC endpoints with their configuration

Source code in src/cuemsengine/players/AudioMixer.py

def build_mixer_osc_endpoints(client_name: str, channel_number: int) -> dict:
    """Build OSC endpoint configuration for audio mixer.

    Creates OSC addresses in the format expected by jack-volume (audiomixer_routes branch):
    /audiomixer/{client_name}/master
    /audiomixer/{client_name}/0
    /audiomixer/{client_name}/1
    etc.

    Args:
        client_name: Name of the mixer client instance (JACK client name)
        channel_number: Number of audio channels in the mixer

    Returns:
        Dictionary of OSC endpoints with their configuration
    """
    endpoints = {}
    base_path = f'/audiomixer/{client_name}'

    # Master volume control
    endpoints[f'{base_path}/master'] = [ValueType.Float, None, 1.0]

    # Individual channel volume controls
    for i in range(channel_number):
        endpoints[f'{base_path}/{i}'] = [ValueType.Float, None, 1.0]

    return endpoints

get_mixer_client_name(mixer_id)

Get the client name for the mixer.

Parameters:

Name	Type	Description	Default
mixer_id	str	Unique identifier for this mixer	required

Returns:

Type	Description
str	Client name for the mixer

Source code in src/cuemsengine/players/AudioMixer.py

def get_mixer_client_name(mixer_id: str) -> str:
    """Get the client name for the mixer.

    Args:
        mixer_id: Unique identifier for this mixer

    Returns:
        Client name for the mixer
    """
    return f'{mixer_id}_mixer'

start_audio_mixer(audio_outputs, port, mixer_id, path=None, args=None, timeout=5.0)

Start an audio mixer and its OSC client.

This function creates and starts a jack-volume mixer process and sets up an OSC client to control it.

Parameters:

Name	Type	Description	Default
audio_outputs	list	List of audio output configurations	required
port	int	OSC port for jack-volume communication	required
mixer_id	str	Unique identifier for this mixer	required
path	str	Optional path to jack-volume binary	None
args	str | None	Additional arguments for jack-volume	None
timeout	float	Maximum time to wait for mixer to start (seconds)	5.0

Returns:

Type	Description
tuple[AudioMixer, MixerClient]	Tuple containing the AudioMixer and MixerClient instances

Raises:

Type	Description
RuntimeError	If mixer fails to start within timeout or thread dies

Source code in src/cuemsengine/players/AudioMixer.py

@logged
def start_audio_mixer(
    audio_outputs: list,
    port: int,
    mixer_id: str,
    path: str = None,
    args: str | None = None,
    timeout: float = 5.0
) -> tuple[AudioMixer, MixerClient]:
    """Start an audio mixer and its OSC client.

    This function creates and starts a jack-volume mixer process and
    sets up an OSC client to control it.

    Args:
        audio_outputs: List of audio output configurations
        port: OSC port for jack-volume communication
        mixer_id: Unique identifier for this mixer
        path: Optional path to jack-volume binary
        args: Additional arguments for jack-volume
        timeout: Maximum time to wait for mixer to start (seconds)

    Returns:
        Tuple containing the AudioMixer and MixerClient instances

    Raises:
        RuntimeError: If mixer fails to start within timeout or thread dies
    """
    # Create the mixer
    mixer = AudioMixer(
        audio_outputs=audio_outputs,
        port=port,
        mixer_id=mixer_id,
        path=path,
        args=args
    )

    # Start with timeout handling
    mixer.start(timeout=timeout)

    # Wait for jack-volume to fully initialize before connecting
    sleep(2)

    # Connect JACK ports
    mixer.connect_to_jack()

    # Create OSC client for controlling the mixer
    client = MixerClient(
        player_port=port,
        channel_number=len(audio_outputs),
        mixer_id=mixer_id
    )

    Logger.info(f"Audio mixer {mixer_id} started on port {port}")
    return mixer, client

DmxClient

Bases: PlayerClient

Source code in src/cuemsengine/players/DmxPlayer.py

class DmxClient(PlayerClient):
    def __init__(self, player_port: int, client_name: str, host: str = "127.0.0.1"):
        """Initialize the DMX client.

        Args:
            player_port: OSC port for communication
            client_name: Name for this client instance
            host: Host IP address of the dmxplayer
        """
        super().__init__(
            player_port = player_port,
            endpoints = OSC_DMXPLAYER_CONF,
            name = client_name
        )
        self.host = host
        self.player_port = player_port

        # Bundle parameters for building OSC bundles (pyossia now sends proper #bundle wire format)
        self._create_bundle_parameters()
        Logger.debug(f"DMX bundle parameters created for {self.name}")

    def _create_bundle_parameters(self) -> None:
        """Create parameters on the OSC device for bundle construction."""
        root = self.device.root_node
        self._frame_param = root.add_node("/frame").create_parameter(ossia.ValueType.List)
        self._mtc_time_param = root.add_node("/mtc_time").create_parameter(ossia.ValueType.String)
        self._start_offset_param = root.add_node("/start_offset").create_parameter(ossia.ValueType.Int)
        self._fade_time_param = root.add_node("/fade_time").create_parameter(ossia.ValueType.Float)
        self._mtcfollow_param = root.add_node("/mtcfollow").create_parameter(ossia.ValueType.Int)

    def enable_mtcfollow(self) -> None:
        """Enable MTC following so the dmxplayer tracks timecode."""
        self._mtcfollow_param.push_value(1)
        Logger.debug("DMX mtcfollow enabled")

    def disable_mtcfollow(self) -> None:
        """Disable MTC following so the dmxplayer stops advancing its playhead."""
        self._mtcfollow_param.push_value(0)
        Logger.debug("DMX mtcfollow disabled")

    @logged
    def send_dmx_scene(
        self,
        universe_frames: dict[int, dict[int, int]],
        mtc_time: str | int,
        fade_time: float = 0.0
    ) -> None:
        """Send a complete DMX scene as an OSC bundle via pyossia.

        Constructs an OSC bundle containing:
        - /frame messages: universe_id followed by channel/value pairs
        - /mtc_time or /start_offset: timing information
        - /fade_time: fade duration
        """
        try:
            bundle = ossia.Bundle()

            for universe_id, channels in universe_frames.items():
                if channels:
                    frame_data = [int(universe_id)]
                    for channel, value in sorted(channels.items()):
                        frame_data.append(int(channel))
                        frame_data.append(int(value))
                    bundle.append(self._frame_param, frame_data)
                    Logger.debug(f"Added frame for universe {universe_id} with {len(channels)} channels")

            if isinstance(mtc_time, int):
                bundle.append(self._start_offset_param, int(mtc_time))
                Logger.debug(f"Added start_offset: {mtc_time}ms")
            else:
                bundle.append(self._mtc_time_param, str(mtc_time))
                Logger.debug(f"Added mtc_time: {mtc_time}")

            bundle.append(self._fade_time_param, float(fade_time))
            Logger.debug(f"Added fade_time: {fade_time}s")

            self.device.push_bundle(bundle)

            Logger.info(
                f"Sent DMX scene bundle: {len(universe_frames)} universe(s), "
                f"mtc={mtc_time}, fade={fade_time}s"
            )

        except Exception as e:
            Logger.error(f"Error sending DMX scene bundle: {e}")
            Logger.exception(e)
            raise

    @logged
    def send_blackout(self, universe_ids: int | tuple[int, ...] = (0, 1)) -> None:
        """Send blackout: clear dmxplayer fades + direct OLA backup.

        Sends /blackout to the dmxplayer which clears all queued scenes,
        active fades, and writes zeros to OLA. The direct ola_set_dmx
        backup covers the case where the dmxplayer hasn't processed
        the command yet.

        Args:
            universe_ids: DMX universe(s) to black out.
        """
        import subprocess

        if isinstance(universe_ids, int):
            universe_ids = (universe_ids,)

        # Tell the dmxplayer to clear all scenes/fades and send zeros to OLA.
        try:
            self.set_value('/blackout', None)
        except Exception as e:
            Logger.warning(f'Blackout command to dmxplayer failed: {e}')

        # Backup: write zeros directly to OLA.
        zeros = ','.join(['0'] * 512)
        for uid in universe_ids:
            try:
                subprocess.run(
                    ['ola_set_dmx', '-u', str(uid), '-d', zeros],
                    timeout=2, check=True,
                    capture_output=True,
                )
            except Exception as e:
                Logger.error(f"Blackout ola_set_dmx failed for universe {uid}: {e}")

        Logger.info(f"Sent DMX blackout for universe(s) {universe_ids}")

__init__(player_port, client_name, host='127.0.0.1')

Initialize the DMX client.

Parameters:

Name	Type	Description	Default
player_port	int	OSC port for communication	required
client_name	str	Name for this client instance	required
host	str	Host IP address of the dmxplayer	'127.0.0.1'

Source code in src/cuemsengine/players/DmxPlayer.py

def __init__(self, player_port: int, client_name: str, host: str = "127.0.0.1"):
    """Initialize the DMX client.

    Args:
        player_port: OSC port for communication
        client_name: Name for this client instance
        host: Host IP address of the dmxplayer
    """
    super().__init__(
        player_port = player_port,
        endpoints = OSC_DMXPLAYER_CONF,
        name = client_name
    )
    self.host = host
    self.player_port = player_port

    # Bundle parameters for building OSC bundles (pyossia now sends proper #bundle wire format)
    self._create_bundle_parameters()
    Logger.debug(f"DMX bundle parameters created for {self.name}")

disable_mtcfollow()

Disable MTC following so the dmxplayer stops advancing its playhead.

Source code in src/cuemsengine/players/DmxPlayer.py

def disable_mtcfollow(self) -> None:
    """Disable MTC following so the dmxplayer stops advancing its playhead."""
    self._mtcfollow_param.push_value(0)
    Logger.debug("DMX mtcfollow disabled")

enable_mtcfollow()

Enable MTC following so the dmxplayer tracks timecode.

Source code in src/cuemsengine/players/DmxPlayer.py

def enable_mtcfollow(self) -> None:
    """Enable MTC following so the dmxplayer tracks timecode."""
    self._mtcfollow_param.push_value(1)
    Logger.debug("DMX mtcfollow enabled")

send_blackout(universe_ids=(0, 1))

Send blackout: clear dmxplayer fades + direct OLA backup.

Sends /blackout to the dmxplayer which clears all queued scenes, active fades, and writes zeros to OLA. The direct ola_set_dmx backup covers the case where the dmxplayer hasn't processed the command yet.

Parameters:

Name	Type	Description	Default
universe_ids	int | tuple[int, ...]	DMX universe(s) to black out.	(0, 1)

Source code in src/cuemsengine/players/DmxPlayer.py

@logged
def send_blackout(self, universe_ids: int | tuple[int, ...] = (0, 1)) -> None:
    """Send blackout: clear dmxplayer fades + direct OLA backup.

    Sends /blackout to the dmxplayer which clears all queued scenes,
    active fades, and writes zeros to OLA. The direct ola_set_dmx
    backup covers the case where the dmxplayer hasn't processed
    the command yet.

    Args:
        universe_ids: DMX universe(s) to black out.
    """
    import subprocess

    if isinstance(universe_ids, int):
        universe_ids = (universe_ids,)

    # Tell the dmxplayer to clear all scenes/fades and send zeros to OLA.
    try:
        self.set_value('/blackout', None)
    except Exception as e:
        Logger.warning(f'Blackout command to dmxplayer failed: {e}')

    # Backup: write zeros directly to OLA.
    zeros = ','.join(['0'] * 512)
    for uid in universe_ids:
        try:
            subprocess.run(
                ['ola_set_dmx', '-u', str(uid), '-d', zeros],
                timeout=2, check=True,
                capture_output=True,
            )
        except Exception as e:
            Logger.error(f"Blackout ola_set_dmx failed for universe {uid}: {e}")

    Logger.info(f"Sent DMX blackout for universe(s) {universe_ids}")

send_dmx_scene(universe_frames, mtc_time, fade_time=0.0)

Send a complete DMX scene as an OSC bundle via pyossia.

Constructs an OSC bundle containing: - /frame messages: universe_id followed by channel/value pairs - /mtc_time or /start_offset: timing information - /fade_time: fade duration

Source code in src/cuemsengine/players/DmxPlayer.py

@logged
def send_dmx_scene(
    self,
    universe_frames: dict[int, dict[int, int]],
    mtc_time: str | int,
    fade_time: float = 0.0
) -> None:
    """Send a complete DMX scene as an OSC bundle via pyossia.

    Constructs an OSC bundle containing:
    - /frame messages: universe_id followed by channel/value pairs
    - /mtc_time or /start_offset: timing information
    - /fade_time: fade duration
    """
    try:
        bundle = ossia.Bundle()

        for universe_id, channels in universe_frames.items():
            if channels:
                frame_data = [int(universe_id)]
                for channel, value in sorted(channels.items()):
                    frame_data.append(int(channel))
                    frame_data.append(int(value))
                bundle.append(self._frame_param, frame_data)
                Logger.debug(f"Added frame for universe {universe_id} with {len(channels)} channels")

        if isinstance(mtc_time, int):
            bundle.append(self._start_offset_param, int(mtc_time))
            Logger.debug(f"Added start_offset: {mtc_time}ms")
        else:
            bundle.append(self._mtc_time_param, str(mtc_time))
            Logger.debug(f"Added mtc_time: {mtc_time}")

        bundle.append(self._fade_time_param, float(fade_time))
        Logger.debug(f"Added fade_time: {fade_time}s")

        self.device.push_bundle(bundle)

        Logger.info(
            f"Sent DMX scene bundle: {len(universe_frames)} universe(s), "
            f"mtc={mtc_time}, fade={fade_time}s"
        )

    except Exception as e:
        Logger.error(f"Error sending DMX scene bundle: {e}")
        Logger.exception(e)
        raise

DmxPlayer

Bases: Player

DMX player process wrapper.

Manages a single cuems-dmxplayer process per node and exposes OSC control.

Source code in src/cuemsengine/players/DmxPlayer.py

class DmxPlayer(Player):
    """DMX player process wrapper.

    Manages a single cuems-dmxplayer process per node and exposes OSC control.
    """

    def __init__(self, port, node_uuid, path=None, args: str | None = None):
        """Initialize the DmxPlayer.

        Args:
            port: OSC port for dmxplayer communication
            node_uuid: Unique identifier for this player node
            path: Path to cuems-dmxplayer binary
        """
        super().__init__()
        self.node_uuid = node_uuid
        self.port = port
        self.path = path
        self.client_name = f'{self.node_uuid}_dmxplayer'
        self.args = args
        self.stdout = None
        self.stderr = None

    @logged
    def run(self):
        """Call cuems-dmxplayer in a subprocess"""
        process_call_list = [self.path]
        if self.args:
            for arg in self.args.split():
                process_call_list.append(arg)
        process_call_list.extend(['--port', str(self.port)])
        process_call_list.extend(['--uuid', str(self.node_uuid)])
        Logger.info(f"Starting dmxplayer with: {process_call_list}")
        self.call_subprocess(process_call_list)

__init__(port, node_uuid, path=None, args=None)

Initialize the DmxPlayer.

Parameters:

Name	Type	Description	Default
port		OSC port for dmxplayer communication	required
node_uuid		Unique identifier for this player node	required
path		Path to cuems-dmxplayer binary	None

Source code in src/cuemsengine/players/DmxPlayer.py

def __init__(self, port, node_uuid, path=None, args: str | None = None):
    """Initialize the DmxPlayer.

    Args:
        port: OSC port for dmxplayer communication
        node_uuid: Unique identifier for this player node
        path: Path to cuems-dmxplayer binary
    """
    super().__init__()
    self.node_uuid = node_uuid
    self.port = port
    self.path = path
    self.client_name = f'{self.node_uuid}_dmxplayer'
    self.args = args
    self.stdout = None
    self.stderr = None

run()

Call cuems-dmxplayer in a subprocess

Source code in src/cuemsengine/players/DmxPlayer.py

@logged
def run(self):
    """Call cuems-dmxplayer in a subprocess"""
    process_call_list = [self.path]
    if self.args:
        for arg in self.args.split():
            process_call_list.append(arg)
    process_call_list.extend(['--port', str(self.port)])
    process_call_list.extend(['--uuid', str(self.node_uuid)])
    Logger.info(f"Starting dmxplayer with: {process_call_list}")
    self.call_subprocess(process_call_list)

start_dmx_player(port, node_uuid, path, args=None, timeout=5.0)

Start a DMX player and its OSC client.

This function creates and starts a cuems-dmxplayer process and sets up an OSC client to control it.

Parameters:

Name	Type	Description	Default
port	int	OSC port for dmxplayer communication	required
node_uuid	str	Unique identifier for this player node	required
path	str	Path to cuems-dmxplayer binary	required
args	str | None	Additional arguments for cuems-dmxplayer	None
timeout	float	Maximum time to wait for player to start (seconds)	5.0

Returns:

Type	Description
tuple[DmxPlayer, DmxClient]	Tuple containing the DmxPlayer and DmxClient instances

Raises:

Type	Description
RuntimeError	If player fails to start within timeout or thread dies

Source code in src/cuemsengine/players/DmxPlayer.py

@logged
def start_dmx_player(
    port: int,
    node_uuid: str,
    path: str,
    args: str | None = None,
    timeout: float = 5.0
) -> tuple[DmxPlayer, DmxClient]:
    """Start a DMX player and its OSC client.

    This function creates and starts a cuems-dmxplayer process and
    sets up an OSC client to control it.

    Args:
        port: OSC port for dmxplayer communication
        node_uuid: Unique identifier for this player node
        path: Path to cuems-dmxplayer binary
        args: Additional arguments for cuems-dmxplayer
        timeout: Maximum time to wait for player to start (seconds)

    Returns:
        Tuple containing the DmxPlayer and DmxClient instances

    Raises:
        RuntimeError: If player fails to start within timeout or thread dies
    """
    # Create and start the player with timeout handling
    player = DmxPlayer(
        port=port,
        node_uuid=node_uuid,
        path=path,
        args=args
    )
    player.start(timeout=timeout)

    # Create OSC client for controlling the player
    client = DmxClient(
        player_port=port,
        client_name=f'{node_uuid}_dmxplayer'
    )

    Logger.info(f"DMX player started: {node_uuid}_dmxplayer on port {port}")
    return player, client

VideoClient

Bases: PlayerClient

Source code in src/cuemsengine/players/VideoPlayer.py

class VideoClient(PlayerClient):
    def __init__(self, player_port: int, name: str = "videocomposer"):
        super().__init__(
            player_port = player_port,
            name = name,
            endpoints = OSC_VIDEOPLAYER_CONF
        )

    def create_layer_endpoints(self, layer_id: str) -> None:
        """Register per-layer OSC endpoints for the given layer_id."""
        layer_endpoints = {
            k.format(layer_id): v
            for k, v in OSC_VIDEOPLAYER_LAYER_CONF.items()
        }
        self.create_endpoints(layer_endpoints)

    def remove_layer_endpoints(self, layer_id: str) -> None:
        """Remove per-layer OSC endpoints for the given layer_id."""
        for template_path in OSC_VIDEOPLAYER_LAYER_CONF:
            path = template_path.format(layer_id)
            try:
                self.remove_node(path)
            except Exception as e:
                Logger.debug(f'Could not remove endpoint {path}: {e}')

create_layer_endpoints(layer_id)

Register per-layer OSC endpoints for the given layer_id.

Source code in src/cuemsengine/players/VideoPlayer.py

def create_layer_endpoints(self, layer_id: str) -> None:
    """Register per-layer OSC endpoints for the given layer_id."""
    layer_endpoints = {
        k.format(layer_id): v
        for k, v in OSC_VIDEOPLAYER_LAYER_CONF.items()
    }
    self.create_endpoints(layer_endpoints)

remove_layer_endpoints(layer_id)

Remove per-layer OSC endpoints for the given layer_id.

Source code in src/cuemsengine/players/VideoPlayer.py

def remove_layer_endpoints(self, layer_id: str) -> None:
    """Remove per-layer OSC endpoints for the given layer_id."""
    for template_path in OSC_VIDEOPLAYER_LAYER_CONF:
        path = template_path.format(layer_id)
        try:
            self.remove_node(path)
        except Exception as e:
            Logger.debug(f'Could not remove endpoint {path}: {e}')

VideoOutput

Source code in src/cuemsengine/players/VideoPlayer.py

class VideoOutput:
    def __init__(self, **kwargs):
        self.name = kwargs.get('name')
        self.mapped_to = kwargs.get('mapped_to', self.name)
        self.x = kwargs.get('x', 0)
        self.y = kwargs.get('y', 0)
        self.width = kwargs.get('width', 1920)
        self.height = kwargs.get('height', 1080)
        self.resolution = kwargs.get('resolution', "1080p")
        self.canvas_region = kwargs.get('canvas_region', {
            'x': self.x, 'y': self.y,
            'width': self.width, 'height': self.height,
        })
        self.canvas_width = kwargs.get('canvas_width', self.width)
        self.canvas_height = kwargs.get('canvas_height', self.height)

    def get_layer_placement(self) -> tuple[int, int]:
        """Returns (x, y) offset from canvas center to this output's center.

        The videocomposer uses center-relative coordinates: (0, 0) = canvas center.
        The renderer negates Y (glTranslatef(x, -y, 0)) because OpenGL Y points
        up while screen Y points down.  The canvas FBO also has Y=0 at the
        bottom, so we negate Y here to compensate — positive Y in the returned
        value means "below canvas center" in screen coords, which maps to the
        correct FBO position after the renderer's negation.
        """
        output_cx = self.canvas_region['x'] + self.canvas_region['width'] // 2
        output_cy = self.canvas_region['y'] + self.canvas_region['height'] // 2
        canvas_cx = self.canvas_width // 2
        canvas_cy = self.canvas_height // 2
        return (output_cx - canvas_cx, canvas_cy - output_cy)

    def get_layer_scale(self) -> tuple[float, float]:
        """Returns (scaleX, scaleY) to fit the video layer within this output's region.

        The videocomposer renders layers at full canvas size with letterboxing.
        For typical setups (ultra-wide canvas, 16:9 video), the video fills the
        canvas height and is letterboxed horizontally.  The height ratio therefore
        determines the correct uniform scale to fit the output region.
        """
        s = self.canvas_region['height'] / self.canvas_height if self.canvas_height else 1.0
        return (s, s)

    def apply_config(self, video_client: VideoClient) -> None:
        """No-op: videocomposer reads display config from display.conf at startup.

        /run/cuems/display.conf is the shared contract between engine and
        videocomposer for canvas geometry. cuems-generate-display-conf
        (videocomposer's ExecStartPre) writes it from default_mappings.xml;
        both VC and the engine (via cuemsengine.display_conf.read_display_conf)
        read it independently. The engine must NOT send /display/region or
        resolution_mode here because that caused the MultiOutputRenderer to
        reconfigure (and sometimes switch to native 4K resolution, corrupting
        the canvas layout). Phase 2 will gate runtime tweaks behind explicit
        edit-mode OSC handlers.
        """
        Logger.info(f'VideoOutput {self.mapped_to}: region ({self.x},{self.y} {self.width}x{self.height})')

apply_config(video_client)

No-op: videocomposer reads display config from display.conf at startup.

/run/cuems/display.conf is the shared contract between engine and videocomposer for canvas geometry. cuems-generate-display-conf (videocomposer's ExecStartPre) writes it from default_mappings.xml; both VC and the engine (via cuemsengine.display_conf.read_display_conf) read it independently. The engine must NOT send /display/region or resolution_mode here because that caused the MultiOutputRenderer to reconfigure (and sometimes switch to native 4K resolution, corrupting the canvas layout). Phase 2 will gate runtime tweaks behind explicit edit-mode OSC handlers.

Source code in src/cuemsengine/players/VideoPlayer.py

def apply_config(self, video_client: VideoClient) -> None:
    """No-op: videocomposer reads display config from display.conf at startup.

    /run/cuems/display.conf is the shared contract between engine and
    videocomposer for canvas geometry. cuems-generate-display-conf
    (videocomposer's ExecStartPre) writes it from default_mappings.xml;
    both VC and the engine (via cuemsengine.display_conf.read_display_conf)
    read it independently. The engine must NOT send /display/region or
    resolution_mode here because that caused the MultiOutputRenderer to
    reconfigure (and sometimes switch to native 4K resolution, corrupting
    the canvas layout). Phase 2 will gate runtime tweaks behind explicit
    edit-mode OSC handlers.
    """
    Logger.info(f'VideoOutput {self.mapped_to}: region ({self.x},{self.y} {self.width}x{self.height})')

get_layer_placement()

Returns (x, y) offset from canvas center to this output's center.

The videocomposer uses center-relative coordinates: (0, 0) = canvas center. The renderer negates Y (glTranslatef(x, -y, 0)) because OpenGL Y points up while screen Y points down. The canvas FBO also has Y=0 at the bottom, so we negate Y here to compensate — positive Y in the returned value means "below canvas center" in screen coords, which maps to the correct FBO position after the renderer's negation.

Source code in src/cuemsengine/players/VideoPlayer.py

def get_layer_placement(self) -> tuple[int, int]:
    """Returns (x, y) offset from canvas center to this output's center.

    The videocomposer uses center-relative coordinates: (0, 0) = canvas center.
    The renderer negates Y (glTranslatef(x, -y, 0)) because OpenGL Y points
    up while screen Y points down.  The canvas FBO also has Y=0 at the
    bottom, so we negate Y here to compensate — positive Y in the returned
    value means "below canvas center" in screen coords, which maps to the
    correct FBO position after the renderer's negation.
    """
    output_cx = self.canvas_region['x'] + self.canvas_region['width'] // 2
    output_cy = self.canvas_region['y'] + self.canvas_region['height'] // 2
    canvas_cx = self.canvas_width // 2
    canvas_cy = self.canvas_height // 2
    return (output_cx - canvas_cx, canvas_cy - output_cy)

get_layer_scale()

Returns (scaleX, scaleY) to fit the video layer within this output's region.

The videocomposer renders layers at full canvas size with letterboxing. For typical setups (ultra-wide canvas, 16:9 video), the video fills the canvas height and is letterboxed horizontally. The height ratio therefore determines the correct uniform scale to fit the output region.

Source code in src/cuemsengine/players/VideoPlayer.py

def get_layer_scale(self) -> tuple[float, float]:
    """Returns (scaleX, scaleY) to fit the video layer within this output's region.

    The videocomposer renders layers at full canvas size with letterboxing.
    For typical setups (ultra-wide canvas, 16:9 video), the video fills the
    canvas height and is letterboxed horizontally.  The height ratio therefore
    determines the correct uniform scale to fit the output region.
    """
    s = self.canvas_region['height'] / self.canvas_height if self.canvas_height else 1.0
    return (s, s)

VideoPlayer

Bases: Player

Video player systemd service wrapper.

This class restarts the videocomposer service.

IMPORTANT: This class should not be used, since videocomposer is a systemd service and not a subprocess.

Source code in src/cuemsengine/players/VideoPlayer.py

class VideoPlayer(Player):
    """Video player systemd service wrapper.

    This class restarts the videocomposer service.

    IMPORTANT: This class should not be used, since videocomposer is a systemd service and not a subprocess.
    """
    def __init__(self):
        super().__init__()
        Logger.warning('Restarting the videocomposer service. Use VideoClient only to control videocomposer.')

    @logged
    def run(self):
        process_call_list = [
            'systemctl',
            'restart',
            'videocomposer.service'
        ]
        Logger.info(f'Restarting videocomposer service: {process_call_list}')
        self.call_subprocess(process_call_list)

Player

Bases: Thread

Base class for all players in the system. Holds the common methods and attributes for all players. Extends the Thread class. Can call a subprocess, kill it and start the Thread.

IMPORTANT: The run method must be implemented in the child classes.

Source code in src/cuemsengine/players/Player.py

class Player(Thread):
    """Base class for all players in the system.
        Holds the common methods and attributes for all players.
        Extends the Thread class.
        Can call a subprocess, kill it and start the Thread.

        IMPORTANT: The run method must be implemented in the child classes.

    """
    def __init__(self, daemon: bool = True):
        """Initializes the Player object and a Thread object with the daemon attribute set to True.

        Args:
            daemon (bool, optional): Sets the daemon attribute of the Thread object. Defaults to True.
        """
        super().__init__(daemon = daemon)
        self.p = None
        self.pid = None
        self.firstrun = True
        self.started = False
        self.status = 'starting'  # 'starting', 'running', 'failed'
        self.error = None

    def run(self):
        raise NotImplementedError

    @logged
    def call_subprocess(self, call_args):
        """Calls a subprocess with the given arguments.

        Automatically handles exceptions and updates status/error attributes.
        Sets status to 'running' on success, 'failed' on error.
        """
        try:
            my_env= os.environ.copy()
            my_env["DISPLAY"] = ":0"
            self.p = Popen(call_args, stdout=PIPE, stderr=STDOUT, env=my_env)
            self.pid = self.p.pid

            stdout_lines_iterator = iter(self.p.stdout.readline, b'')
            while self.p.poll() is None:
                for line in stdout_lines_iterator:
                    Logger.debug(f"Subprocess output: {line}")
                # Prevent CPU spinning when subprocess has no output
                sleep(0.01)

            self.status = 'running'
        except Exception as e:
            self.status = 'failed'
            self.error = e
            Logger.error(f"Failed to start player subprocess: {e}")
            Logger.exception(e)
            raise

    @logged
    def kill(self):
        """Kills the subprocess."""
        if self.p:
            self.p.kill()
            self.started = False

    @logged    
    def start(self, timeout: float = 5.0):
        """Starts the player and waits for it to initialize.

        Args:
            timeout: Maximum time to wait for player to start (seconds)

        Raises:
            RuntimeError: If player fails to start within timeout or thread dies
        """
        # Start the thread
        if self.firstrun:
            super().start()
            self.firstrun = False
        elif not self.is_alive():
            super().start()
        self.started = True

        # Wait for player process to start with timeout
        from time import sleep
        elapsed = 0.0
        interval = 0.01
        while self.pid is None and elapsed < timeout:
            # Check if the thread is still alive
            if not self.is_alive():
                error_msg = f"Player thread died during startup"
                if self.error:
                    error_msg += f": {self.error}"
                Logger.error(error_msg)
                raise RuntimeError(error_msg)

            # Check if player failed
            if self.status == 'failed':
                error_msg = f"Player failed to start: {self.error}"
                Logger.error(error_msg)
                raise RuntimeError(error_msg)

            sleep(interval)
            elapsed += interval

        # Timeout check
        if self.pid is None:
            error_msg = f"Player failed to start within {timeout}s timeout"
            Logger.error(error_msg)
            self.kill()
            raise RuntimeError(error_msg)

__init__(daemon=True)

Initializes the Player object and a Thread object with the daemon attribute set to True.

Parameters:

Name	Type	Description	Default
daemon	bool	Sets the daemon attribute of the Thread object. Defaults to True.	True

Source code in src/cuemsengine/players/Player.py

def __init__(self, daemon: bool = True):
    """Initializes the Player object and a Thread object with the daemon attribute set to True.

    Args:
        daemon (bool, optional): Sets the daemon attribute of the Thread object. Defaults to True.
    """
    super().__init__(daemon = daemon)
    self.p = None
    self.pid = None
    self.firstrun = True
    self.started = False
    self.status = 'starting'  # 'starting', 'running', 'failed'
    self.error = None

call_subprocess(call_args)

Calls a subprocess with the given arguments.

Automatically handles exceptions and updates status/error attributes. Sets status to 'running' on success, 'failed' on error.

Source code in src/cuemsengine/players/Player.py

@logged
def call_subprocess(self, call_args):
    """Calls a subprocess with the given arguments.

    Automatically handles exceptions and updates status/error attributes.
    Sets status to 'running' on success, 'failed' on error.
    """
    try:
        my_env= os.environ.copy()
        my_env["DISPLAY"] = ":0"
        self.p = Popen(call_args, stdout=PIPE, stderr=STDOUT, env=my_env)
        self.pid = self.p.pid

        stdout_lines_iterator = iter(self.p.stdout.readline, b'')
        while self.p.poll() is None:
            for line in stdout_lines_iterator:
                Logger.debug(f"Subprocess output: {line}")
            # Prevent CPU spinning when subprocess has no output
            sleep(0.01)

        self.status = 'running'
    except Exception as e:
        self.status = 'failed'
        self.error = e
        Logger.error(f"Failed to start player subprocess: {e}")
        Logger.exception(e)
        raise

kill()

Kills the subprocess.

Source code in src/cuemsengine/players/Player.py

@logged
def kill(self):
    """Kills the subprocess."""
    if self.p:
        self.p.kill()
        self.started = False

start(timeout=5.0)

Starts the player and waits for it to initialize.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum time to wait for player to start (seconds)	5.0

Raises:

Type	Description
RuntimeError	If player fails to start within timeout or thread dies

Source code in src/cuemsengine/players/Player.py

@logged    
def start(self, timeout: float = 5.0):
    """Starts the player and waits for it to initialize.

    Args:
        timeout: Maximum time to wait for player to start (seconds)

    Raises:
        RuntimeError: If player fails to start within timeout or thread dies
    """
    # Start the thread
    if self.firstrun:
        super().start()
        self.firstrun = False
    elif not self.is_alive():
        super().start()
    self.started = True

    # Wait for player process to start with timeout
    from time import sleep
    elapsed = 0.0
    interval = 0.01
    while self.pid is None and elapsed < timeout:
        # Check if the thread is still alive
        if not self.is_alive():
            error_msg = f"Player thread died during startup"
            if self.error:
                error_msg += f": {self.error}"
            Logger.error(error_msg)
            raise RuntimeError(error_msg)

        # Check if player failed
        if self.status == 'failed':
            error_msg = f"Player failed to start: {self.error}"
            Logger.error(error_msg)
            raise RuntimeError(error_msg)

        sleep(interval)
        elapsed += interval

    # Timeout check
    if self.pid is None:
        error_msg = f"Player failed to start within {timeout}s timeout"
        Logger.error(error_msg)
        self.kill()
        raise RuntimeError(error_msg)

PlayerHandler

This class is responsible for handling and generating player objects.

It is a singleton class, so it will only be instantiated once.

Holds a list of armed cues and provides methods to use them.

Source code in src/cuemsengine/players/PlayerHandler.py

class PlayerHandler:
    """
    This class is responsible for handling and generating player objects.

    It is a singleton class, so it will
    only be instantiated once.

    Holds a list of armed cues and provides methods to use them.
    """
    _instance: 'PlayerHandler | None' = None

    # Instance attributes (declared for IDE/type checker support)
    _audio_output_generator: partial | None
    _audio_mixer: AudioMixer | None
    _audio_mixer_client: MixerClient | None
    _cue_players: dict[Cue, Player]
    _audio_players_by_id: dict[str, AudioPlayer]
    _dmx_player: DmxPlayer | None
    _dmx_player_client: DmxClient | None
    _player_endpoints_generator: partial | None
    _video_client: VideoClient | None
    _gradient_client: 'GradientClient | None'
    _video_outputs: dict[str, VideoOutput]
    _audio_outputs: dict[str, dict]
    _loaded_layer_ids: set[str]
    _outputs_map: dict | None
    _lock: RLock
    _media_folder: str
    _node_uuid: str | None

    def __new__(cls, *args, **kwargs):
        """Singleton pattern: Ensure only one instance is created"""
        if not cls._instance:
            cls._instance = super(PlayerHandler, cls).__new__(cls)

            cls._instance._audio_output_generator = None
            cls._instance._audio_mixer = None
            cls._instance._audio_mixer_client = None
            cls._instance._cue_players = {}
            cls._instance._audio_players_by_id = {}
            cls._instance._dmx_player = None
            cls._instance._dmx_player_client = None
            cls._instance._player_endpoints_generator = None
            cls._instance._video_client = None
            cls._instance._gradient_client = None
            cls._instance._video_outputs = {}
            cls._instance._audio_outputs = {}
            cls._instance._loaded_layer_ids = set()
            cls._instance._outputs_map = None
            cls._instance._lock = RLock()
            cls._instance._media_folder = DEFAULT_MEDIA_FOLDER
            cls._instance._node_uuid = None
        return cls._instance


    # ---------------------------
    # Players List Management
    # ---------------------------

    def store_cue_player(self, cue: Cue, player: Player):
        """Stores a cue player"""
        with self._lock:
            self._cue_players[cue] = player

    def get_cue_player(self, cue: Cue) -> Player:
        """Gets a cue player"""
        with self._lock:
            return self._cue_players[cue]

    def remove_cue_player(self, cue: Cue):
            """Removes a cue player"""
            osc_client = None
            cue_id = str(cue.id)
            with self._lock:
                try:
                    player = self._cue_players.pop(cue)
                except KeyError:
                    # Try to find by ID in _audio_players_by_id
                    player = self._audio_players_by_id.pop(cue_id, None)
                    if player is None:
                        Logger.debug(f'Cue player not found for cue {cue.id}')
                        return

                # Also remove from ID-based tracking
                self._audio_players_by_id.pop(cue_id, None)

                # Save OSC client reference before clearing
                osc_client = getattr(cue, '_osc', None)
                cue._osc = None
            if isinstance(player, AudioPlayer):
                killed = self._kill_audio_player(player, osc_client, cue_id)
                # Free port AFTER process is dead to prevent concurrent arm
                # from getting a port the OS still has bound (Bug 2 fix).
                # Skip if kill failed — process still holds the port.
                if killed:
                    PORT_HANDLER.remove_ports(cue)

    def reset_all(self):
        """Complete reset of PlayerHandler for testing"""
        Logger.debug('Performing complete PlayerHandler reset')
        self.reset_video_layers()
        self._video_outputs = {}
        self._cue_players = {}
        self._outputs_map = None
        with self._lock:
            self._loaded_layer_ids.clear()


    # ---------------------------
    # Audio Player Management
    # ---------------------------

    def set_audio_output_generator(self, path: str, args: str):
        """Sets the audio player generator"""
        Logger.info(f'Setting audio output generator to {path} {args}')
        self._audio_output_generator = partial(start_audio_output, path=path, args=args)

    def set_audio_outputs(self, audio_outputs: dict[str, dict]) -> None:
        """Store audio output configs keyed by <id>."""
        self._audio_outputs = audio_outputs

    def resolve_audio_port(self, output_id: str) -> str | None:
        """Resolve an output <id> to its JACK port name (mapped_to)."""
        output = self._audio_outputs.get(output_id)
        if output:
            return output.get('mapped_to')
        return None

    def start_audio_mixer(self, audio_outputs: list, port: int, mixer_id: str, path: str = None, args: str | None = None) -> tuple[AudioMixer, MixerClient]:
        """Starts the audio mixer for this node.

        Args:
            audio_outputs: List of audio output configurations
            port: OSC port for jack-volume communication
            node_uuid: Unique identifier for this mixer node
            path: Optional path to jack-volume binary

        Returns:
            Tuple containing the AudioMixer and MixerClient instances
        """
        Logger.info(f'Starting audio mixer {mixer_id}')
        self._audio_mixer, self._audio_mixer_client = start_audio_mixer(
            audio_outputs=audio_outputs,
            port=port,
            mixer_id=mixer_id,
            path=path,
            args=args
        )
        return self._audio_mixer, self._audio_mixer_client

    def get_audio_mixer(self) -> AudioMixer:
        """Returns the audio mixer instance."""
        return self._audio_mixer

    def get_audio_mixer_client(self) -> MixerClient:
        """Returns the audio mixer client instance."""
        return self._audio_mixer_client

    def _kill_audio_player(self, player: AudioPlayer, osc_client: AudioClient, cue_id: str) -> bool:
        """Helper method to kill an audio player process.

        The order is critical: disconnect JACK ports first, THEN send /quit.
        If /quit is sent first the player destroys its JACK client immediately,
        and subsequent disconnect calls hit non-existent ports which can corrupt
        JACK's shared-memory semaphore registry.

        Returns:
            True if the process was successfully killed (or was already dead),
            False if the process could not be killed (still alive after timeout).
        """
        if player is None:
            return True

        # 1. Disconnect player from the mixer BEFORE destroying its JACK client
        if self._audio_mixer is not None:
            try:
                uuid_slug = ''.join(cue_id.split('-'))
                player_name = f'Audio_Player-{uuid_slug}'
                self._audio_mixer.disconnect_player(player_name)
                Logger.debug(f'Disconnected {player_name} from mixer')
            except Exception as e:
                Logger.warning(f'Failed to disconnect audio player from mixer: {e}')

        # 2. Send /quit OSC command to gracefully stop the player
        if osc_client is not None:
            try:
                osc_client.set_value('/quit', True)
                Logger.debug(f'Sent /quit command to audio player for cue {cue_id}')
            except Exception as e:
                Logger.warning(f'Failed to send /quit to audio player: {e}')

            # Free the random OSC local port back into the pool
            local_port = getattr(osc_client, 'local_port', None)
            if local_port is not None:
                PORT_HANDLER.remove_random_port(local_port)

        # 3. Kill the subprocess and wait for the OS to release its resources.
        #    SIGKILL is near-instant; 1s timeout handles edge cases (D state).
        process_dead = True
        try:
            if player.p is not None:
                player.p.kill()
                player.p.wait(timeout=1.0)
                Logger.debug(f'Killed audio player subprocess for cue {cue_id}')
        except subprocess.TimeoutExpired:
            Logger.error(f'Audio player process for cue {cue_id} did not die after SIGKILL — port may still be bound')
            process_dead = False
        except Exception as e:
            Logger.warning(f'Failed to kill audio player subprocess: {e}')

        # Wait for thread to finish
        try:
            player.join(timeout=0.5)
        except Exception as e:
            Logger.warning(f'Failed to join audio player thread: {e}')

        # 4. Verify JACK has removed the dead client's ports.
        #    wait() reaps the process, which triggers JACK to unregister the
        #    client. Poll briefly to confirm ports are gone before returning.
        if process_dead and self._audio_mixer is not None:
            uuid_slug = ''.join(cue_id.split('-'))
            player_name = f'Audio_Player-{uuid_slug}'
            for _ in range(10):
                if not self._audio_mixer.conn_man.port_exists(f'{player_name}:outport 0'):
                    break
                sleep(0.1)
            else:
                Logger.warning(f'JACK client {player_name} still has ports after kill')

        return process_dead

    def kill_all_audio_players(self):
        """Kill ALL tracked audio players - used during project cleanup"""
        with self._lock:
            players_to_kill = list(self._audio_players_by_id.items())
            self._audio_players_by_id.clear()

            # Also clear audio players from _cue_players, saving the OSC
            # client so _kill_audio_player can free the random port.
            cue_players_to_remove = []
            for cue, player in self._cue_players.items():
                if isinstance(player, AudioPlayer):
                    osc_client = getattr(cue, '_osc', None)
                    cue._osc = None
                    cue_players_to_remove.append((cue, player, osc_client))
            for cue, player, osc_client in cue_players_to_remove:
                self._cue_players.pop(cue, None)
                players_to_kill.append((str(cue.id), player, osc_client))

        Logger.info(f'Killing {len(players_to_kill)} audio players during cleanup')
        for entry in players_to_kill:
            if len(entry) == 3:
                cue_id, player, osc_client = entry
            else:
                cue_id, player = entry
                osc_client = None
            self._kill_audio_player(player, osc_client, cue_id)

    def cleanup_zombie_jack_clients(self) -> int:
        """Scan for JACK Audio_Player clients whose processes have died.

        Enumerates all JACK ports matching Audio_Player-* and cross-references
        with tracked players in _audio_players_by_id. Unmatched ports are
        zombies left by crashed processes — disconnect them from the mixer.

        Called on project load to clear stale state from previous runs.

        Returns:
            Number of zombie clients found and cleaned up.
        """
        if self._audio_mixer is None:
            return 0

        all_ports = self._audio_mixer.conn_man.get_ports(
            pattern='Audio_Player-.*', is_audio=True, is_output=True
        )
        if not all_ports:
            return 0

        # Extract unique client names from port names (e.g. "Audio_Player-abc123:outport 0" → "Audio_Player-abc123")
        jack_clients = set()
        for port_name in all_ports:
            client_name = port_name.split(':')[0]
            jack_clients.add(client_name)

        # Build set of tracked player client names
        with self._lock:
            tracked_slugs = set()
            for cue_id in self._audio_players_by_id:
                slug = ''.join(cue_id.split('-'))
                tracked_slugs.add(f'Audio_Player-{slug}')

        zombies = jack_clients - tracked_slugs
        if not zombies:
            return 0

        Logger.warning(f'Found {len(zombies)} zombie JACK audio clients: {zombies}')
        for client_name in zombies:
            try:
                self._audio_mixer.disconnect_player(client_name)
                Logger.info(f'Disconnected zombie JACK client {client_name}')
            except Exception as e:
                Logger.warning(f'Failed to disconnect zombie {client_name}: {e}')

        return len(zombies)

    def kill_orphaned_audio_processes(self):
        """Kill cuems-audioplayer OS processes not tracked by this engine.

        On engine restart, previously spawned audioplayer processes survive
        because they are independent subprocesses. The new engine has no
        reference to them, so they steal JACK client names and cause silence.
        """
        import os
        import signal
        result = subprocess.run(
            ['pgrep', '-f', 'cuems-audioplayer'],
            capture_output=True, text=True
        )
        if result.returncode != 0:
            return

        tracked_pids = set()
        with self._lock:
            for player in self._audio_players_by_id.values():
                if player and player.p:
                    tracked_pids.add(player.p.pid)

        for pid_str in result.stdout.strip().split('\n'):
            if not pid_str:
                continue
            pid = int(pid_str)
            if pid not in tracked_pids:
                Logger.warning(f'Killing orphaned audioplayer process {pid}')
                try:
                    os.kill(pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass

    # ---------------------------
    # Audio Cue Management
    # ---------------------------

    def new_audio_output(self, cue: AudioCue) -> None:
        """Creates a new audio output for the given cue

        The player is stored in the player handler and the osc client is assigned to the cue.
        After creating the player, it will be automatically connected to the audio mixer if one exists.

        Args:
            cue: The cue to create the audio output for

        Returns:
            None
        """
        Logger.debug(f'Creating new audio output for cue {cue.id}')
        if self._audio_output_generator is None:
            raise ValueError("Audio output generator not set")

        # Kill any existing player for this cue before spawning a new one.
        # This prevents orphaned audioplayer processes when a cue is re-armed
        # without being disarmed first (the old process would keep running,
        # holding its JACK client and OSC port, while its reference is silently
        # overwritten in _audio_players_by_id).
        cue_id = str(cue.id)
        with self._lock:
            existing_player = self._audio_players_by_id.pop(cue_id, None)
            self._cue_players.pop(cue, None)
        if existing_player is not None:
            Logger.warning(f'Killing existing audio player for cue {cue_id} before re-arm')
            # Save and clear OSC client so loop_audioCue stops sending to the
            # dying player (it will hit AttributeError, caught by its blanket
            # except AttributeError handler and exit silently).
            existing_osc = getattr(cue, '_osc', None)
            cue._osc = None
            killed = self._kill_audio_player(existing_player, existing_osc, cue_id)
            # Free assigned port AFTER process is dead to avoid Bug 2's race.
            # Skip if kill failed — process still holds the port.
            if killed:
                PORT_HANDLER.remove_ports(cue)

        ports = PORT_HANDLER.assign_ports(['audio_output'], cue)
        player, client = self._audio_output_generator(
            port=ports['audio_output'],
            media=self.media_path(cue.media['file_name']),
            uuid=str(cue.id)
        )
        cue._osc = client
        self.set_player_endpoints(cue)
        self.store_cue_player(cue, player)

        # Also track by cue ID string for cleanup when cue object is lost
        with self._lock:
            self._audio_players_by_id[str(cue.id)] = player

        # Connect the player to the audio mixer if available
        if self._audio_mixer is not None:
            uuid_slug = ''.join(str(cue.id).split('-'))
            player_name = f'Audio_Player-{uuid_slug}'

            # Resolve each output_name to its JACK port via the ID in the mappings.
            # output_name format: "{node_uuid}_{output_id}"  (e.g. "a3811d78-..._6")
            # resolve_audio_port maps the numeric ID → JACK port name (e.g. "usb_audio:playback_1")
            selected_outputs = []
            for output in getattr(cue, 'outputs', []):
                raw = output.get('output_name', '')
                output_id = raw[37:] if len(raw) > 37 else None  # strip "{uuid}_"
                if output_id is not None:
                    jack_port = self.resolve_audio_port(output_id)
                    if jack_port:
                        selected_outputs.append(jack_port)
                    else:
                        Logger.warning(f'Cannot resolve audio output ID "{output_id}" to a JACK port')

            if not selected_outputs:
                Logger.warning(f'No valid audio outputs resolved for cue {cue.id}, skipping mixer connection')
            else:
                Logger.info(f'Connecting {player_name} to outputs: {selected_outputs}')
                self._audio_mixer.connect_player_to_outputs(
                    player_name=player_name,
                    player_output_prefix='outport',
                    selected_outputs=selected_outputs
                )


    # ---------------------------
    # DMX Player Management
    # ---------------------------

    def start_dmx_player(self, port: int, node_uuid: str, path: str, args: str | None = None) -> tuple[DmxPlayer, DmxClient]:
        """Starts the DMX player for this node.

        Args:
            port: OSC port for dmxplayer communication
            node_uuid: Unique identifier for this player node
            path: Path to cuems-dmxplayer binary

        Returns:
            Tuple containing the DmxPlayer and DmxClient instances
        """
        Logger.info(f'Starting DMX player for node {node_uuid}')
        self._dmx_player, self._dmx_player_client = start_dmx_player(
            port=port,
            node_uuid=node_uuid,
            path=path,
            args=args
        )
        return self._dmx_player, self._dmx_player_client

    def get_dmx_player(self) -> DmxPlayer:
        """Returns the DMX player instance."""
        return self._dmx_player

    def get_dmx_player_client(self) -> DmxClient:
        """Returns the DMX player client instance."""
        return self._dmx_player_client

    # def set_dmx_output_generator(cls, path: str, args: str):
    #     """Sets the dmx player generator"""
    #     cls._dmx_output_generator = partial(start_dmx_output, path, args)

    # def new_dmx_output(cls, cue: DmxCue) -> None:
    #     """Creates a new audio output for the given cue

    #     The player is stored in the player handler and the osc client is assigned to the cue.

    #     Args:
    #         cue: The cue to create the dmx output for

    #     Returns:
    #         None
    #     """
    #     if cls._dmx_output_generator is None:
    #         raise ValueError("Audio output generator not set")
    #     ports = PORT_HANDLER.assign_ports(['dmx_output'], cue)
    #     player, client = cls._dmx_output_generator(
    #         ports['dmx_output'],
    #         cue.media['file_name']
    #     )
    #     cue._osc = client
    #     cls.store_cue_player(cue, player)


    # ---------------------------
    # Video Player Management
    # ---------------------------

    def get_video_client(self) -> VideoClient:
        """Returns the video client instance."""
        return self._video_client

    def set_video_client(self, port: int) -> None:
        """Sets the video client for this node."""
        Logger.info(f'Setting video client for node {self._node_uuid}')
        self._video_client = VideoClient(player_port=port)

    def get_gradient_client(self) -> 'GradientClient | None':
        """Returns the GradientClient instance, or None if not yet initialised."""
        return self._gradient_client

    def set_gradient_client(self, port: int, node_uuid: str) -> None:
        """Construct (or replace) the GradientClient for this node.

        Safe to call multiple times: any new call replaces the prior client.
        PyOscClient is fire-and-forget UDP with no held resources, so no
        teardown of the prior client is needed.
        """
        from .GradientClient import GradientClient
        self._gradient_client = GradientClient(
            host='127.0.0.1', port=port, node_uuid=node_uuid,
        )
        Logger.info(
            f'GradientClient: bound to 127.0.0.1:{port} node_uuid={node_uuid}'
        )

    def start_video_outputs(
        self,
        output_names: dict[str, dict[str, any]],
        canvas_override: tuple[int, int] | None = None,
    ) -> None:
        """Ensures that the all the required video output exist.

        ``canvas_override`` is an optional ``(width, height)`` carrying the
        engine reader's authoritative canvas size — set when display.conf
        has a ``canvas_size=`` global key. When provided, it must be >=
        the per-region bounding box (we validate as defense in depth — the
        reader already validates, but a stale caller could pass garbage).
        When ``None``, fall back to bbox computed from the output regions.
        """
        Logger.info(f'Checking & starting video outputs for {output_names} ')
        bbox_w, bbox_h = 0, 0
        for cfg in output_names.values():
            region = cfg.get('canvas_region') or {}
            right = region.get('x', 0) + region.get('width', 1920)
            bottom = region.get('y', 0) + region.get('height', 1080)
            bbox_w = max(bbox_w, right)
            bbox_h = max(bbox_h, bottom)
        if canvas_override is not None:
            cw, ch = canvas_override
            if cw < bbox_w or ch < bbox_h:
                raise ValueError(
                    f"canvas_override {cw}x{ch} is smaller than the per-output "
                    f"bounding box {bbox_w}x{bbox_h}; monitors would be cropped"
                )
            canvas_w, canvas_h = cw, ch
        else:
            canvas_w, canvas_h = bbox_w, bbox_h
        Logger.info(f'Canvas: {canvas_w}x{canvas_h} (bbox={bbox_w}x{bbox_h})')
        for output_name, output_config in output_names.items():
            output_config['canvas_width'] = canvas_w
            output_config['canvas_height'] = canvas_h
            video_output = VideoOutput(**output_config)
            video_output.apply_config(self._video_client)
            self._video_outputs[output_name] = video_output

    def get_video_output(self, output_name: str) -> VideoOutput:
        """Returns the VideoOutput object for a given output name."""
        return self._video_outputs[output_name]

    def _resolve_canvas_dimensions(self) -> tuple[int, int]:
        """Return the node's canvas (width, height) in pixels.

        All alias VideoOutputs on a node share the same canvas totals,
        written by start_video_outputs. Raises if no aliases exist yet —
        custom outputs have no independent canvas dimensions.
        """
        for vo in self._video_outputs.values():
            return vo.canvas_width, vo.canvas_height
        raise RuntimeError(
            "Cannot resolve canvas dimensions: no named video outputs "
            "are registered. Custom outputs require at least one alias "
            "on the same node."
        )

    def make_custom_video_output(self, cue_output) -> VideoOutput:
        """Build a VideoOutput for a per-cue custom region.

        cue_output is a dict-like VideoCueOutput with a canvas_region
        holding normalized floats in [0, 1]. Converts to pixel integers
        so VideoOutput.get_layer_placement / get_layer_scale work the
        same way they do for alias outputs.
        """
        region_norm = cue_output["canvas_region"]
        canvas_w, canvas_h = self._resolve_canvas_dimensions()
        region_px = {
            "x": int(region_norm["x"] * canvas_w),
            "y": int(region_norm["y"] * canvas_h),
            "width": int(region_norm["width"] * canvas_w),
            "height": int(region_norm["height"] * canvas_h),
        }
        return VideoOutput(
            name=cue_output.get("output_name", "custom"),
            canvas_region=region_px,
            canvas_width=canvas_w,
            canvas_height=canvas_h,
            width=region_px["width"],
            height=region_px["height"],
        )

    def resolve_video_output_for_cue(self, cue, output_name: str) -> VideoOutput:
        """Resolve an output_name suffix to a VideoOutput.

        For alias suffixes (<int>) looks up the cached VideoOutput.
        For custom suffixes (custom_<n>) synthesizes a VideoOutput from
        the matching VideoCueOutput's inline canvas_region.
        """
        if output_name.startswith("custom_"):
            full = f"{self._node_uuid}_{output_name}"
            cue_output = next(
                (o for o in cue.outputs if o.get("output_name") == full),
                None,
            )
            if cue_output is None:
                raise KeyError(f"No VideoCueOutput match for {full}")
            return self.make_custom_video_output(cue_output)
        return self._video_outputs[output_name]

    def register_layer(self, layer_id: str) -> None:
        """Track a layer as active in the videocomposer."""
        with self._lock:
            self._loaded_layer_ids.add(layer_id)

    def deregister_layer(self, layer_id: str) -> None:
        """Remove a layer from active tracking."""
        with self._lock:
            self._loaded_layer_ids.discard(layer_id)

    def reset_videocomposer(self):
        """Send atomic reset to videocomposer (removes all layers + resets master)."""
        Logger.debug('Sending atomic reset to videocomposer')
        if self._video_client is not None:
            try:
                self._video_client.set_value('/videocomposer/reset', None)
            except Exception as e:
                Logger.warning(f'Error sending reset to videocomposer: {e}')
            # Remove all layer endpoints from the OSC client
            with self._lock:
                for layer_id in list(self._loaded_layer_ids):
                    try:
                        self._video_client.remove_layer_endpoints(layer_id)
                    except Exception as e:
                        Logger.debug(f'Error removing layer endpoints {layer_id}: {e}')
        with self._lock:
            self._loaded_layer_ids.clear()

    def reset_video_layers(self):
        """Unload all tracked video layers (video blackout). Legacy per-layer method."""
        Logger.debug('Resetting video layers')
        with self._lock:
            if self._video_client is None:
                self._loaded_layer_ids.clear()
                return
            for layer_id in list(self._loaded_layer_ids):
                try:
                    self._video_client.set_value('/videocomposer/layer/unload', layer_id)
                    self._video_client.remove_layer_endpoints(layer_id)
                except Exception as e:
                    Logger.debug(f'Error unloading layer {layer_id}: {e}')
            self._loaded_layer_ids.clear()

    def quit_videocomposer(self):
        """Quits the videocomposer process."""
        Logger.debug('Quitting videocomposer')
        if self._video_client is not None:
            try:
                self._video_client.set_value('/videocomposer/quit', None)
            except Exception as e:
                Logger.debug(f'Error sending quit to videocomposer: {e}')
        self._video_client = None
        self._video_outputs = {}
        with self._lock:
            self._loaded_layer_ids.clear()


    # ---------------------------
    # Helper functions
    # ---------------------------

    def set_player_endpoints_generator(self, func: Callable, *args, **kwargs):
        """Sets the player endpoints generator"""
        Logger.info(f'Setting player endpoints generator to {func}')
        self._player_endpoints_generator = partial(func, *args, **kwargs)

    def set_player_endpoints(self, cue: Cue) -> None:
        """Sets the player endpoints for a given cue"""
        if self._player_endpoints_generator is None:
            raise ValueError("Player endpoints generator not set")
        try:
            self._player_endpoints_generator(cue)
        except Exception as e:
            Logger.error(f'Error setting player endpoints for cue {cue.id}: {e}')

    def set_outputs_map(self, outputs_map: dict):
        """Set the outputs map for the player handler"""
        self._outputs_map = outputs_map

    def get_cue_output_name(self, cue: Cue) -> str | None:
        """Get the output name for a given cue from the outputs map.

        Args:
            cue: The cue to get the output name for

        Returns:
            The output name for the given cue or None if the cue is not found in the outputs map

        Raises:
            AttributeError: If the outputs map is not set
        """
        if self._outputs_map is None:
            Logger.error('Outputs map not set')
            raise AttributeError('Outputs map not set')
        outputs = self._outputs_map.get(cue.id, None)
        # outputs_map stores lists, but callers expect a single string
        if isinstance(outputs, list) and len(outputs) > 0:
            return outputs[0]
        return outputs

    def get_all_cue_output_names(self, cue: Cue) -> list:
        """Get all output names for a given cue from the outputs map.

        Args:
            cue: The cue to get the output names for

        Returns:
            List of output names for the given cue, or empty list if not found

        Raises:
            AttributeError: If the outputs map is not set
        """
        if self._outputs_map is None:
            Logger.error('Outputs map not set')
            raise AttributeError('Outputs map not set')
        outputs = self._outputs_map.get(cue.id, None)
        if isinstance(outputs, list):
            return outputs
        elif outputs:
            return [outputs]
        return []

    def add_media_folder(self, path: str):
        """Adds a media folder to the player handler"""
        path = path.split('/')
        if path[-1] != 'media':
            path.append('media')
        self._media_folder = '/' + '/'.join(path)
        if self._media_folder[0:2] == "//":
            self._media_folder = self._media_folder[1:]

    def media_path(self, file_name: str) -> str:
        """Returns the media path for a given file name"""
        return self._media_folder + '/' + file_name

    def add_node_uuid(self, uuid: str):
        """Adds a node uuid to the player handler"""
        self._node_uuid = uuid

    @property
    def node_uuid(self) -> str | None:
        """Public read-only accessor for the node uuid."""
        return self._node_uuid

node_uuid property

Public read-only accessor for the node uuid.

__new__(*args, **kwargs)

Singleton pattern: Ensure only one instance is created

Source code in src/cuemsengine/players/PlayerHandler.py

def __new__(cls, *args, **kwargs):
    """Singleton pattern: Ensure only one instance is created"""
    if not cls._instance:
        cls._instance = super(PlayerHandler, cls).__new__(cls)

        cls._instance._audio_output_generator = None
        cls._instance._audio_mixer = None
        cls._instance._audio_mixer_client = None
        cls._instance._cue_players = {}
        cls._instance._audio_players_by_id = {}
        cls._instance._dmx_player = None
        cls._instance._dmx_player_client = None
        cls._instance._player_endpoints_generator = None
        cls._instance._video_client = None
        cls._instance._gradient_client = None
        cls._instance._video_outputs = {}
        cls._instance._audio_outputs = {}
        cls._instance._loaded_layer_ids = set()
        cls._instance._outputs_map = None
        cls._instance._lock = RLock()
        cls._instance._media_folder = DEFAULT_MEDIA_FOLDER
        cls._instance._node_uuid = None
    return cls._instance

add_media_folder(path)

Adds a media folder to the player handler

Source code in src/cuemsengine/players/PlayerHandler.py

def add_media_folder(self, path: str):
    """Adds a media folder to the player handler"""
    path = path.split('/')
    if path[-1] != 'media':
        path.append('media')
    self._media_folder = '/' + '/'.join(path)
    if self._media_folder[0:2] == "//":
        self._media_folder = self._media_folder[1:]

add_node_uuid(uuid)

Adds a node uuid to the player handler

Source code in src/cuemsengine/players/PlayerHandler.py

def add_node_uuid(self, uuid: str):
    """Adds a node uuid to the player handler"""
    self._node_uuid = uuid

cleanup_zombie_jack_clients()

Scan for JACK Audio_Player clients whose processes have died.

Enumerates all JACK ports matching Audio_Player-* and cross-references with tracked players in _audio_players_by_id. Unmatched ports are zombies left by crashed processes — disconnect them from the mixer.

Called on project load to clear stale state from previous runs.

Returns:

Type	Description
int	Number of zombie clients found and cleaned up.

Source code in src/cuemsengine/players/PlayerHandler.py

def cleanup_zombie_jack_clients(self) -> int:
    """Scan for JACK Audio_Player clients whose processes have died.

    Enumerates all JACK ports matching Audio_Player-* and cross-references
    with tracked players in _audio_players_by_id. Unmatched ports are
    zombies left by crashed processes — disconnect them from the mixer.

    Called on project load to clear stale state from previous runs.

    Returns:
        Number of zombie clients found and cleaned up.
    """
    if self._audio_mixer is None:
        return 0

    all_ports = self._audio_mixer.conn_man.get_ports(
        pattern='Audio_Player-.*', is_audio=True, is_output=True
    )
    if not all_ports:
        return 0

    # Extract unique client names from port names (e.g. "Audio_Player-abc123:outport 0" → "Audio_Player-abc123")
    jack_clients = set()
    for port_name in all_ports:
        client_name = port_name.split(':')[0]
        jack_clients.add(client_name)

    # Build set of tracked player client names
    with self._lock:
        tracked_slugs = set()
        for cue_id in self._audio_players_by_id:
            slug = ''.join(cue_id.split('-'))
            tracked_slugs.add(f'Audio_Player-{slug}')

    zombies = jack_clients - tracked_slugs
    if not zombies:
        return 0

    Logger.warning(f'Found {len(zombies)} zombie JACK audio clients: {zombies}')
    for client_name in zombies:
        try:
            self._audio_mixer.disconnect_player(client_name)
            Logger.info(f'Disconnected zombie JACK client {client_name}')
        except Exception as e:
            Logger.warning(f'Failed to disconnect zombie {client_name}: {e}')

    return len(zombies)

deregister_layer(layer_id)

Remove a layer from active tracking.

Source code in src/cuemsengine/players/PlayerHandler.py

def deregister_layer(self, layer_id: str) -> None:
    """Remove a layer from active tracking."""
    with self._lock:
        self._loaded_layer_ids.discard(layer_id)

get_all_cue_output_names(cue)

Get all output names for a given cue from the outputs map.

Parameters:

Name	Type	Description	Default
cue	Cue	The cue to get the output names for	required

Returns:

Type	Description
list	List of output names for the given cue, or empty list if not found

Raises:

Type	Description
AttributeError	If the outputs map is not set

Source code in src/cuemsengine/players/PlayerHandler.py

def get_all_cue_output_names(self, cue: Cue) -> list:
    """Get all output names for a given cue from the outputs map.

    Args:
        cue: The cue to get the output names for

    Returns:
        List of output names for the given cue, or empty list if not found

    Raises:
        AttributeError: If the outputs map is not set
    """
    if self._outputs_map is None:
        Logger.error('Outputs map not set')
        raise AttributeError('Outputs map not set')
    outputs = self._outputs_map.get(cue.id, None)
    if isinstance(outputs, list):
        return outputs
    elif outputs:
        return [outputs]
    return []

get_audio_mixer()

Returns the audio mixer instance.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_audio_mixer(self) -> AudioMixer:
    """Returns the audio mixer instance."""
    return self._audio_mixer

get_audio_mixer_client()

Returns the audio mixer client instance.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_audio_mixer_client(self) -> MixerClient:
    """Returns the audio mixer client instance."""
    return self._audio_mixer_client

get_cue_output_name(cue)

Get the output name for a given cue from the outputs map.

Parameters:

Name	Type	Description	Default
cue	Cue	The cue to get the output name for	required

Returns:

Type	Description
str | None	The output name for the given cue or None if the cue is not found in the outputs map

Raises:

Type	Description
AttributeError	If the outputs map is not set

Source code in src/cuemsengine/players/PlayerHandler.py

def get_cue_output_name(self, cue: Cue) -> str | None:
    """Get the output name for a given cue from the outputs map.

    Args:
        cue: The cue to get the output name for

    Returns:
        The output name for the given cue or None if the cue is not found in the outputs map

    Raises:
        AttributeError: If the outputs map is not set
    """
    if self._outputs_map is None:
        Logger.error('Outputs map not set')
        raise AttributeError('Outputs map not set')
    outputs = self._outputs_map.get(cue.id, None)
    # outputs_map stores lists, but callers expect a single string
    if isinstance(outputs, list) and len(outputs) > 0:
        return outputs[0]
    return outputs

get_cue_player(cue)

Gets a cue player

Source code in src/cuemsengine/players/PlayerHandler.py

def get_cue_player(self, cue: Cue) -> Player:
    """Gets a cue player"""
    with self._lock:
        return self._cue_players[cue]

get_dmx_player()

Returns the DMX player instance.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_dmx_player(self) -> DmxPlayer:
    """Returns the DMX player instance."""
    return self._dmx_player

get_dmx_player_client()

Returns the DMX player client instance.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_dmx_player_client(self) -> DmxClient:
    """Returns the DMX player client instance."""
    return self._dmx_player_client

get_gradient_client()

Returns the GradientClient instance, or None if not yet initialised.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_gradient_client(self) -> 'GradientClient | None':
    """Returns the GradientClient instance, or None if not yet initialised."""
    return self._gradient_client

get_video_client()

Returns the video client instance.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_video_client(self) -> VideoClient:
    """Returns the video client instance."""
    return self._video_client

get_video_output(output_name)

Returns the VideoOutput object for a given output name.

Source code in src/cuemsengine/players/PlayerHandler.py

def get_video_output(self, output_name: str) -> VideoOutput:
    """Returns the VideoOutput object for a given output name."""
    return self._video_outputs[output_name]

kill_all_audio_players()

Kill ALL tracked audio players - used during project cleanup

Source code in src/cuemsengine/players/PlayerHandler.py

def kill_all_audio_players(self):
    """Kill ALL tracked audio players - used during project cleanup"""
    with self._lock:
        players_to_kill = list(self._audio_players_by_id.items())
        self._audio_players_by_id.clear()

        # Also clear audio players from _cue_players, saving the OSC
        # client so _kill_audio_player can free the random port.
        cue_players_to_remove = []
        for cue, player in self._cue_players.items():
            if isinstance(player, AudioPlayer):
                osc_client = getattr(cue, '_osc', None)
                cue._osc = None
                cue_players_to_remove.append((cue, player, osc_client))
        for cue, player, osc_client in cue_players_to_remove:
            self._cue_players.pop(cue, None)
            players_to_kill.append((str(cue.id), player, osc_client))

    Logger.info(f'Killing {len(players_to_kill)} audio players during cleanup')
    for entry in players_to_kill:
        if len(entry) == 3:
            cue_id, player, osc_client = entry
        else:
            cue_id, player = entry
            osc_client = None
        self._kill_audio_player(player, osc_client, cue_id)

kill_orphaned_audio_processes()

Kill cuems-audioplayer OS processes not tracked by this engine.

On engine restart, previously spawned audioplayer processes survive because they are independent subprocesses. The new engine has no reference to them, so they steal JACK client names and cause silence.

Source code in src/cuemsengine/players/PlayerHandler.py

def kill_orphaned_audio_processes(self):
    """Kill cuems-audioplayer OS processes not tracked by this engine.

    On engine restart, previously spawned audioplayer processes survive
    because they are independent subprocesses. The new engine has no
    reference to them, so they steal JACK client names and cause silence.
    """
    import os
    import signal
    result = subprocess.run(
        ['pgrep', '-f', 'cuems-audioplayer'],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        return

    tracked_pids = set()
    with self._lock:
        for player in self._audio_players_by_id.values():
            if player and player.p:
                tracked_pids.add(player.p.pid)

    for pid_str in result.stdout.strip().split('\n'):
        if not pid_str:
            continue
        pid = int(pid_str)
        if pid not in tracked_pids:
            Logger.warning(f'Killing orphaned audioplayer process {pid}')
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                pass

make_custom_video_output(cue_output)

Build a VideoOutput for a per-cue custom region.

cue_output is a dict-like VideoCueOutput with a canvas_region holding normalized floats in [0, 1]. Converts to pixel integers so VideoOutput.get_layer_placement / get_layer_scale work the same way they do for alias outputs.

Source code in src/cuemsengine/players/PlayerHandler.py

def make_custom_video_output(self, cue_output) -> VideoOutput:
    """Build a VideoOutput for a per-cue custom region.

    cue_output is a dict-like VideoCueOutput with a canvas_region
    holding normalized floats in [0, 1]. Converts to pixel integers
    so VideoOutput.get_layer_placement / get_layer_scale work the
    same way they do for alias outputs.
    """
    region_norm = cue_output["canvas_region"]
    canvas_w, canvas_h = self._resolve_canvas_dimensions()
    region_px = {
        "x": int(region_norm["x"] * canvas_w),
        "y": int(region_norm["y"] * canvas_h),
        "width": int(region_norm["width"] * canvas_w),
        "height": int(region_norm["height"] * canvas_h),
    }
    return VideoOutput(
        name=cue_output.get("output_name", "custom"),
        canvas_region=region_px,
        canvas_width=canvas_w,
        canvas_height=canvas_h,
        width=region_px["width"],
        height=region_px["height"],
    )

media_path(file_name)

Returns the media path for a given file name

Source code in src/cuemsengine/players/PlayerHandler.py

def media_path(self, file_name: str) -> str:
    """Returns the media path for a given file name"""
    return self._media_folder + '/' + file_name

new_audio_output(cue)

Creates a new audio output for the given cue

The player is stored in the player handler and the osc client is assigned to the cue. After creating the player, it will be automatically connected to the audio mixer if one exists.

Parameters:

Name	Type	Description	Default
cue	AudioCue	The cue to create the audio output for	required

Returns:

Type	Description
None	None

Source code in src/cuemsengine/players/PlayerHandler.py

def new_audio_output(self, cue: AudioCue) -> None:
    """Creates a new audio output for the given cue

    The player is stored in the player handler and the osc client is assigned to the cue.
    After creating the player, it will be automatically connected to the audio mixer if one exists.

    Args:
        cue: The cue to create the audio output for

    Returns:
        None
    """
    Logger.debug(f'Creating new audio output for cue {cue.id}')
    if self._audio_output_generator is None:
        raise ValueError("Audio output generator not set")

    # Kill any existing player for this cue before spawning a new one.
    # This prevents orphaned audioplayer processes when a cue is re-armed
    # without being disarmed first (the old process would keep running,
    # holding its JACK client and OSC port, while its reference is silently
    # overwritten in _audio_players_by_id).
    cue_id = str(cue.id)
    with self._lock:
        existing_player = self._audio_players_by_id.pop(cue_id, None)
        self._cue_players.pop(cue, None)
    if existing_player is not None:
        Logger.warning(f'Killing existing audio player for cue {cue_id} before re-arm')
        # Save and clear OSC client so loop_audioCue stops sending to the
        # dying player (it will hit AttributeError, caught by its blanket
        # except AttributeError handler and exit silently).
        existing_osc = getattr(cue, '_osc', None)
        cue._osc = None
        killed = self._kill_audio_player(existing_player, existing_osc, cue_id)
        # Free assigned port AFTER process is dead to avoid Bug 2's race.
        # Skip if kill failed — process still holds the port.
        if killed:
            PORT_HANDLER.remove_ports(cue)

    ports = PORT_HANDLER.assign_ports(['audio_output'], cue)
    player, client = self._audio_output_generator(
        port=ports['audio_output'],
        media=self.media_path(cue.media['file_name']),
        uuid=str(cue.id)
    )
    cue._osc = client
    self.set_player_endpoints(cue)
    self.store_cue_player(cue, player)

    # Also track by cue ID string for cleanup when cue object is lost
    with self._lock:
        self._audio_players_by_id[str(cue.id)] = player

    # Connect the player to the audio mixer if available
    if self._audio_mixer is not None:
        uuid_slug = ''.join(str(cue.id).split('-'))
        player_name = f'Audio_Player-{uuid_slug}'

        # Resolve each output_name to its JACK port via the ID in the mappings.
        # output_name format: "{node_uuid}_{output_id}"  (e.g. "a3811d78-..._6")
        # resolve_audio_port maps the numeric ID → JACK port name (e.g. "usb_audio:playback_1")
        selected_outputs = []
        for output in getattr(cue, 'outputs', []):
            raw = output.get('output_name', '')
            output_id = raw[37:] if len(raw) > 37 else None  # strip "{uuid}_"
            if output_id is not None:
                jack_port = self.resolve_audio_port(output_id)
                if jack_port:
                    selected_outputs.append(jack_port)
                else:
                    Logger.warning(f'Cannot resolve audio output ID "{output_id}" to a JACK port')

        if not selected_outputs:
            Logger.warning(f'No valid audio outputs resolved for cue {cue.id}, skipping mixer connection')
        else:
            Logger.info(f'Connecting {player_name} to outputs: {selected_outputs}')
            self._audio_mixer.connect_player_to_outputs(
                player_name=player_name,
                player_output_prefix='outport',
                selected_outputs=selected_outputs
            )

quit_videocomposer()

Quits the videocomposer process.

Source code in src/cuemsengine/players/PlayerHandler.py

def quit_videocomposer(self):
    """Quits the videocomposer process."""
    Logger.debug('Quitting videocomposer')
    if self._video_client is not None:
        try:
            self._video_client.set_value('/videocomposer/quit', None)
        except Exception as e:
            Logger.debug(f'Error sending quit to videocomposer: {e}')
    self._video_client = None
    self._video_outputs = {}
    with self._lock:
        self._loaded_layer_ids.clear()

register_layer(layer_id)

Track a layer as active in the videocomposer.

Source code in src/cuemsengine/players/PlayerHandler.py

def register_layer(self, layer_id: str) -> None:
    """Track a layer as active in the videocomposer."""
    with self._lock:
        self._loaded_layer_ids.add(layer_id)

remove_cue_player(cue)

Removes a cue player

Source code in src/cuemsengine/players/PlayerHandler.py

def remove_cue_player(self, cue: Cue):
        """Removes a cue player"""
        osc_client = None
        cue_id = str(cue.id)
        with self._lock:
            try:
                player = self._cue_players.pop(cue)
            except KeyError:
                # Try to find by ID in _audio_players_by_id
                player = self._audio_players_by_id.pop(cue_id, None)
                if player is None:
                    Logger.debug(f'Cue player not found for cue {cue.id}')
                    return

            # Also remove from ID-based tracking
            self._audio_players_by_id.pop(cue_id, None)

            # Save OSC client reference before clearing
            osc_client = getattr(cue, '_osc', None)
            cue._osc = None
        if isinstance(player, AudioPlayer):
            killed = self._kill_audio_player(player, osc_client, cue_id)
            # Free port AFTER process is dead to prevent concurrent arm
            # from getting a port the OS still has bound (Bug 2 fix).
            # Skip if kill failed — process still holds the port.
            if killed:
                PORT_HANDLER.remove_ports(cue)

reset_all()

Complete reset of PlayerHandler for testing

Source code in src/cuemsengine/players/PlayerHandler.py

def reset_all(self):
    """Complete reset of PlayerHandler for testing"""
    Logger.debug('Performing complete PlayerHandler reset')
    self.reset_video_layers()
    self._video_outputs = {}
    self._cue_players = {}
    self._outputs_map = None
    with self._lock:
        self._loaded_layer_ids.clear()

reset_video_layers()

Unload all tracked video layers (video blackout). Legacy per-layer method.

Source code in src/cuemsengine/players/PlayerHandler.py

def reset_video_layers(self):
    """Unload all tracked video layers (video blackout). Legacy per-layer method."""
    Logger.debug('Resetting video layers')
    with self._lock:
        if self._video_client is None:
            self._loaded_layer_ids.clear()
            return
        for layer_id in list(self._loaded_layer_ids):
            try:
                self._video_client.set_value('/videocomposer/layer/unload', layer_id)
                self._video_client.remove_layer_endpoints(layer_id)
            except Exception as e:
                Logger.debug(f'Error unloading layer {layer_id}: {e}')
        self._loaded_layer_ids.clear()

reset_videocomposer()

Send atomic reset to videocomposer (removes all layers + resets master).

Source code in src/cuemsengine/players/PlayerHandler.py

def reset_videocomposer(self):
    """Send atomic reset to videocomposer (removes all layers + resets master)."""
    Logger.debug('Sending atomic reset to videocomposer')
    if self._video_client is not None:
        try:
            self._video_client.set_value('/videocomposer/reset', None)
        except Exception as e:
            Logger.warning(f'Error sending reset to videocomposer: {e}')
        # Remove all layer endpoints from the OSC client
        with self._lock:
            for layer_id in list(self._loaded_layer_ids):
                try:
                    self._video_client.remove_layer_endpoints(layer_id)
                except Exception as e:
                    Logger.debug(f'Error removing layer endpoints {layer_id}: {e}')
    with self._lock:
        self._loaded_layer_ids.clear()

resolve_audio_port(output_id)

Resolve an output to its JACK port name (mapped_to).

Source code in src/cuemsengine/players/PlayerHandler.py

def resolve_audio_port(self, output_id: str) -> str | None:
    """Resolve an output <id> to its JACK port name (mapped_to)."""
    output = self._audio_outputs.get(output_id)
    if output:
        return output.get('mapped_to')
    return None

resolve_video_output_for_cue(cue, output_name)

Resolve an output_name suffix to a VideoOutput.

For alias suffixes () looks up the cached VideoOutput. For custom suffixes (custom_) synthesizes a VideoOutput from the matching VideoCueOutput's inline canvas_region.

Source code in src/cuemsengine/players/PlayerHandler.py

def resolve_video_output_for_cue(self, cue, output_name: str) -> VideoOutput:
    """Resolve an output_name suffix to a VideoOutput.

    For alias suffixes (<int>) looks up the cached VideoOutput.
    For custom suffixes (custom_<n>) synthesizes a VideoOutput from
    the matching VideoCueOutput's inline canvas_region.
    """
    if output_name.startswith("custom_"):
        full = f"{self._node_uuid}_{output_name}"
        cue_output = next(
            (o for o in cue.outputs if o.get("output_name") == full),
            None,
        )
        if cue_output is None:
            raise KeyError(f"No VideoCueOutput match for {full}")
        return self.make_custom_video_output(cue_output)
    return self._video_outputs[output_name]

set_audio_output_generator(path, args)

Sets the audio player generator

Source code in src/cuemsengine/players/PlayerHandler.py

def set_audio_output_generator(self, path: str, args: str):
    """Sets the audio player generator"""
    Logger.info(f'Setting audio output generator to {path} {args}')
    self._audio_output_generator = partial(start_audio_output, path=path, args=args)

set_audio_outputs(audio_outputs)

Store audio output configs keyed by .

Source code in src/cuemsengine/players/PlayerHandler.py

def set_audio_outputs(self, audio_outputs: dict[str, dict]) -> None:
    """Store audio output configs keyed by <id>."""
    self._audio_outputs = audio_outputs

set_gradient_client(port, node_uuid)

Construct (or replace) the GradientClient for this node.

Safe to call multiple times: any new call replaces the prior client. PyOscClient is fire-and-forget UDP with no held resources, so no teardown of the prior client is needed.

Source code in src/cuemsengine/players/PlayerHandler.py

def set_gradient_client(self, port: int, node_uuid: str) -> None:
    """Construct (or replace) the GradientClient for this node.

    Safe to call multiple times: any new call replaces the prior client.
    PyOscClient is fire-and-forget UDP with no held resources, so no
    teardown of the prior client is needed.
    """
    from .GradientClient import GradientClient
    self._gradient_client = GradientClient(
        host='127.0.0.1', port=port, node_uuid=node_uuid,
    )
    Logger.info(
        f'GradientClient: bound to 127.0.0.1:{port} node_uuid={node_uuid}'
    )

set_outputs_map(outputs_map)

Set the outputs map for the player handler

Source code in src/cuemsengine/players/PlayerHandler.py

def set_outputs_map(self, outputs_map: dict):
    """Set the outputs map for the player handler"""
    self._outputs_map = outputs_map

set_player_endpoints(cue)

Sets the player endpoints for a given cue

Source code in src/cuemsengine/players/PlayerHandler.py

def set_player_endpoints(self, cue: Cue) -> None:
    """Sets the player endpoints for a given cue"""
    if self._player_endpoints_generator is None:
        raise ValueError("Player endpoints generator not set")
    try:
        self._player_endpoints_generator(cue)
    except Exception as e:
        Logger.error(f'Error setting player endpoints for cue {cue.id}: {e}')

set_player_endpoints_generator(func, *args, **kwargs)

Sets the player endpoints generator

Source code in src/cuemsengine/players/PlayerHandler.py

def set_player_endpoints_generator(self, func: Callable, *args, **kwargs):
    """Sets the player endpoints generator"""
    Logger.info(f'Setting player endpoints generator to {func}')
    self._player_endpoints_generator = partial(func, *args, **kwargs)

set_video_client(port)

Sets the video client for this node.

Source code in src/cuemsengine/players/PlayerHandler.py

def set_video_client(self, port: int) -> None:
    """Sets the video client for this node."""
    Logger.info(f'Setting video client for node {self._node_uuid}')
    self._video_client = VideoClient(player_port=port)

start_audio_mixer(audio_outputs, port, mixer_id, path=None, args=None)

Starts the audio mixer for this node.

Parameters:

Name	Type	Description	Default
audio_outputs	list	List of audio output configurations	required
port	int	OSC port for jack-volume communication	required
node_uuid		Unique identifier for this mixer node	required
path	str	Optional path to jack-volume binary	None

Returns:

Type	Description
tuple[AudioMixer, MixerClient]	Tuple containing the AudioMixer and MixerClient instances

Source code in src/cuemsengine/players/PlayerHandler.py

def start_audio_mixer(self, audio_outputs: list, port: int, mixer_id: str, path: str = None, args: str | None = None) -> tuple[AudioMixer, MixerClient]:
    """Starts the audio mixer for this node.

    Args:
        audio_outputs: List of audio output configurations
        port: OSC port for jack-volume communication
        node_uuid: Unique identifier for this mixer node
        path: Optional path to jack-volume binary

    Returns:
        Tuple containing the AudioMixer and MixerClient instances
    """
    Logger.info(f'Starting audio mixer {mixer_id}')
    self._audio_mixer, self._audio_mixer_client = start_audio_mixer(
        audio_outputs=audio_outputs,
        port=port,
        mixer_id=mixer_id,
        path=path,
        args=args
    )
    return self._audio_mixer, self._audio_mixer_client

start_dmx_player(port, node_uuid, path, args=None)

Starts the DMX player for this node.

Parameters:

Name	Type	Description	Default
port	int	OSC port for dmxplayer communication	required
node_uuid	str	Unique identifier for this player node	required
path	str	Path to cuems-dmxplayer binary	required

Returns:

Type	Description
tuple[DmxPlayer, DmxClient]	Tuple containing the DmxPlayer and DmxClient instances

Source code in src/cuemsengine/players/PlayerHandler.py

def start_dmx_player(self, port: int, node_uuid: str, path: str, args: str | None = None) -> tuple[DmxPlayer, DmxClient]:
    """Starts the DMX player for this node.

    Args:
        port: OSC port for dmxplayer communication
        node_uuid: Unique identifier for this player node
        path: Path to cuems-dmxplayer binary

    Returns:
        Tuple containing the DmxPlayer and DmxClient instances
    """
    Logger.info(f'Starting DMX player for node {node_uuid}')
    self._dmx_player, self._dmx_player_client = start_dmx_player(
        port=port,
        node_uuid=node_uuid,
        path=path,
        args=args
    )
    return self._dmx_player, self._dmx_player_client

start_video_outputs(output_names, canvas_override=None)

Ensures that the all the required video output exist.

canvas_override is an optional (width, height) carrying the engine reader's authoritative canvas size — set when display.conf has a canvas_size= global key. When provided, it must be >= the per-region bounding box (we validate as defense in depth — the reader already validates, but a stale caller could pass garbage). When None, fall back to bbox computed from the output regions.

Source code in src/cuemsengine/players/PlayerHandler.py

def start_video_outputs(
    self,
    output_names: dict[str, dict[str, any]],
    canvas_override: tuple[int, int] | None = None,
) -> None:
    """Ensures that the all the required video output exist.

    ``canvas_override`` is an optional ``(width, height)`` carrying the
    engine reader's authoritative canvas size — set when display.conf
    has a ``canvas_size=`` global key. When provided, it must be >=
    the per-region bounding box (we validate as defense in depth — the
    reader already validates, but a stale caller could pass garbage).
    When ``None``, fall back to bbox computed from the output regions.
    """
    Logger.info(f'Checking & starting video outputs for {output_names} ')
    bbox_w, bbox_h = 0, 0
    for cfg in output_names.values():
        region = cfg.get('canvas_region') or {}
        right = region.get('x', 0) + region.get('width', 1920)
        bottom = region.get('y', 0) + region.get('height', 1080)
        bbox_w = max(bbox_w, right)
        bbox_h = max(bbox_h, bottom)
    if canvas_override is not None:
        cw, ch = canvas_override
        if cw < bbox_w or ch < bbox_h:
            raise ValueError(
                f"canvas_override {cw}x{ch} is smaller than the per-output "
                f"bounding box {bbox_w}x{bbox_h}; monitors would be cropped"
            )
        canvas_w, canvas_h = cw, ch
    else:
        canvas_w, canvas_h = bbox_w, bbox_h
    Logger.info(f'Canvas: {canvas_w}x{canvas_h} (bbox={bbox_w}x{bbox_h})')
    for output_name, output_config in output_names.items():
        output_config['canvas_width'] = canvas_w
        output_config['canvas_height'] = canvas_h
        video_output = VideoOutput(**output_config)
        video_output.apply_config(self._video_client)
        self._video_outputs[output_name] = video_output

store_cue_player(cue, player)

Stores a cue player

Source code in src/cuemsengine/players/PlayerHandler.py

def store_cue_player(self, cue: Cue, player: Player):
    """Stores a cue player"""
    with self._lock:
        self._cue_players[cue] = player

Fire-and-forget UDP OSC client for gradient-motiond v0.3.0.

GradientClient

Fire-and-forget UDP OSC client for gradient-motiond v0.3.0.

Holds node_uuid at construction and injects it as node_name on every send_fade — callers do not pass it. Safe to construct multiple times; each new instance replaces the prior one in PlayerHandler.

Source code in src/cuemsengine/players/GradientClient.py

class GradientClient:
    """Fire-and-forget UDP OSC client for gradient-motiond v0.3.0.

    Holds node_uuid at construction and injects it as node_name on every
    send_fade — callers do not pass it. Safe to construct multiple times;
    each new instance replaces the prior one in PlayerHandler.
    """

    def __init__(self, host: str = '127.0.0.1', port: int = 7100,
                 node_uuid: str = '') -> None:
        self._host = host
        self._port = port
        self._node_uuid = node_uuid
        self._osc = PyOscClient(host=host, port=port)

    def send_fade(
        self,
        motion_id: str,
        osc_host: str,
        osc_port: int,
        osc_path: str,
        start_value: float,
        end_value: float,
        start_mtc_ms: int,
        duration_ms: int,
        curve_type: str,
        curve_params_json: str = '{}',
    ) -> None:
        builder = OscMessageBuilder(address='/gradient/start_fade')
        builder.add_arg(motion_id,              arg_type='s')
        builder.add_arg(self._node_uuid,        arg_type='s')  # node_name — self-injected
        builder.add_arg(osc_host,               arg_type='s')
        builder.add_arg(int(osc_port),          arg_type='i')
        builder.add_arg(osc_path,               arg_type='s')
        builder.add_arg(float(start_value),     arg_type='f')
        builder.add_arg(float(end_value),       arg_type='f')
        builder.add_arg(int(start_mtc_ms),      arg_type='h')  # int64 — REQUIRED
        builder.add_arg(int(duration_ms),       arg_type='i')
        builder.add_arg(curve_type,             arg_type='s')
        builder.add_arg(curve_params_json,      arg_type='s')
        try:
            self._osc.client.send(builder.build())
        except Exception as exc:
            Logger.error(f'GradientClient.send_fade failed: {exc}')
            raise

    def send_cancel_motion(self, motion_id: str) -> None:
        try:
            self._osc.client.send_message('/gradient/cancel_motion', motion_id)
        except Exception as exc:
            Logger.error(f'GradientClient.send_cancel_motion failed: {exc}')
            raise

    def send_cancel_all(self) -> None:
        try:
            self._osc.client.send_message('/gradient/cancel_all', [])
        except Exception as exc:
            Logger.error(f'GradientClient.send_cancel_all failed: {exc}')
            raise

JACK Connection Manager

This module provides a simple interface for managing JACK audio connections using the python-jack (JACK-Client) library.

JackConnectionManager

Manager for JACK audio connections.

Uses the python-jack (JACK-Client) library to manage JACK port connections. Creates a lightweight client just for querying and connection management.

Source code in src/cuemsengine/players/JackConnectionManager.py

class JackConnectionManager:
    """Manager for JACK audio connections.

    Uses the python-jack (JACK-Client) library to manage JACK port connections.
    Creates a lightweight client just for querying and connection management.
    """

    def __init__(self, client_name: str = 'cuems_connection_manager'):
        """Initialize the JACK connection manager.

        Args:
            client_name: Name for the JACK client (default: 'cuems_connection_manager')
        """
        self.client_name = client_name
        self._client = None
        self._initialize_client()

    def _initialize_client(self):
        """Initialize the JACK client."""
        if jack is None:
            Logger.warning("JACK library not available -- JackConnectionManager running in no-op mode")
            self._client = None
            return
        try:
            # Create a client without ports, just for connection management
            self._client = jack.Client(self.client_name, no_start_server=True)
            Logger.debug(f"JACK connection manager client '{self.client_name}' initialized")
        except jack.JackError as e:
            Logger.error(f"Failed to initialize JACK client: {e}")
            self._client = None

    @property
    def client(self):
        """Get the JACK client, reinitializing if necessary."""
        if self._client is None:
            self._initialize_client()
        return self._client

    def _reset_client(self):
        """Discard the current client so the next access reinitializes.

        Needed because jackd can deregister our client after a process-graph
        error (e.g. an audioplayer XRun that trips ProcessGraphAsyncMaster).
        Python still holds a reference to a dead handle; every subsequent
        connect/disconnect silently returns -1. Calling this on a JackError
        lets the retry path get a fresh, registered client.
        """
        if self._client is not None:
            try:
                self._client.close()
            except Exception:
                pass
            self._client = None

    @logged
    def get_ports(self, pattern: str = None, is_audio: bool = True, 
                  is_output: bool = None, is_input: bool = None) -> list[str]:
        """Get list of JACK ports.

        Args:
            pattern: Optional regex pattern to filter port names
            is_audio: Filter for audio ports (default: True)
            is_output: Filter for output ports (default: None = all)
            is_input: Filter for input ports (default: None = all)

        Returns:
            List of port names
        """
        if self.client is None:
            Logger.error("JACK client not initialized")
            return []

        try:
            ports = self.client.get_ports(
                name_pattern=pattern if pattern else '',
                is_audio=is_audio,
                is_output=is_output,
                is_input=is_input
            )
            port_names = [p.name for p in ports]
            Logger.debug(f"Found {len(port_names)} JACK ports")
            return port_names

        except jack.JackError as e:
            Logger.error(f"Error getting JACK ports: {e}")
            return []
        except Exception as e:
            Logger.error(f"Unexpected error getting JACK ports: {e}")
            return []

    def port_exists(self, port_name: str) -> bool:
        """Check if a JACK port exists.

        Args:
            port_name: Full name of the port (e.g., 'client_name:port_name')

        Returns:
            True if the port exists, False otherwise
        """
        if self.client is None:
            return False

        try:
            ports = self.client.get_ports(name_pattern=f'^{port_name}$')
            return len(ports) > 0
        except Exception:
            return False

    @logged
    def connect_by_name(self, source_port: str, destination_port: str) -> bool:
        """Connect two JACK ports by name.

        Args:
            source_port: Name of the source port (output)
            destination_port: Name of the destination port (input)

        Returns:
            True if connection successful, False otherwise
        """
        for attempt in (0, 1):
            if self.client is None:
                Logger.error("JACK client not initialized")
                return False
            try:
                if self.is_connected(source_port, destination_port):
                    Logger.debug(f"Ports already connected: {source_port} -> {destination_port}")
                    return True
                self.client.connect(source_port, destination_port)
                Logger.info(f"Connected {source_port} -> {destination_port}")
                return True
            except jack.JackError as e:
                if attempt == 0:
                    Logger.warning(f"connect failed, retrying with fresh client: {source_port} -> {destination_port}: {e}")
                    self._reset_client()
                    continue
                Logger.warning(f"Failed to connect {source_port} -> {destination_port}: {e}")
                return False
            except Exception as e:
                Logger.error(f"Unexpected error connecting JACK ports: {e}")
                return False
        return False

    @logged
    def disconnect_by_name(self, source_port: str, destination_port: str) -> bool:
        """Disconnect two JACK ports by name.

        Args:
            source_port: Name of the source port (output)
            destination_port: Name of the destination port (input)

        Returns:
            True if disconnection successful, False otherwise
        """
        for attempt in (0, 1):
            if self.client is None:
                Logger.error("JACK client not initialized")
                return False
            try:
                self.client.disconnect(source_port, destination_port)
                Logger.info(f"Disconnected {source_port} -> {destination_port}")
                return True
            except jack.JackError as e:
                if attempt == 0:
                    Logger.warning(f"disconnect failed, retrying with fresh client: {source_port} -> {destination_port}: {e}")
                    self._reset_client()
                    continue
                Logger.warning(f"Failed to disconnect {source_port} -> {destination_port}: {e}")
                return False
            except Exception as e:
                Logger.error(f"Unexpected error disconnecting JACK ports: {e}")
                return False
        return False

    @logged
    def get_connections(self, port_name: str) -> list[str]:
        """Get all connections for a given port.

        Args:
            port_name: Name of the port to query

        Returns:
            List of connected port names
        """
        if self.client is None:
            Logger.error("JACK client not initialized")
            return []

        try:
            # Get the port object
            ports = self.client.get_ports(name_pattern=f'^{port_name}$')
            if not ports:
                Logger.warning(f"Port not found: {port_name}")
                return []

            port = ports[0]

            # Get connections
            connections = self.client.get_all_connections(port)
            connection_names = [conn.name for conn in connections]

            return connection_names

        except jack.JackError as e:
            Logger.error(f"Error getting connections for port {port_name}: {e}")
            return []
        except Exception as e:
            Logger.error(f"Unexpected error getting connections: {e}")
            return []

    @logged
    def is_connected(self, source_port: str, destination_port: str) -> bool:
        """Check if two ports are connected.

        Args:
            source_port: Name of the source port
            destination_port: Name of the destination port

        Returns:
            True if connected, False otherwise
        """
        connections = self.get_connections(source_port)
        return destination_port in connections

    def __del__(self):
        """Cleanup JACK client on deletion."""
        if self._client is not None:
            try:
                self._client.close()
                Logger.debug(f"JACK connection manager client '{self.client_name}' closed")
            except Exception as e:
                Logger.debug(f"Error closing JACK client: {e}")

client property

Get the JACK client, reinitializing if necessary.

__del__()

Cleanup JACK client on deletion.

Source code in src/cuemsengine/players/JackConnectionManager.py

def __del__(self):
    """Cleanup JACK client on deletion."""
    if self._client is not None:
        try:
            self._client.close()
            Logger.debug(f"JACK connection manager client '{self.client_name}' closed")
        except Exception as e:
            Logger.debug(f"Error closing JACK client: {e}")

__init__(client_name='cuems_connection_manager')

Initialize the JACK connection manager.

Parameters:

Name	Type	Description	Default
client_name	str	Name for the JACK client (default: 'cuems_connection_manager')	'cuems_connection_manager'

Source code in src/cuemsengine/players/JackConnectionManager.py

def __init__(self, client_name: str = 'cuems_connection_manager'):
    """Initialize the JACK connection manager.

    Args:
        client_name: Name for the JACK client (default: 'cuems_connection_manager')
    """
    self.client_name = client_name
    self._client = None
    self._initialize_client()

connect_by_name(source_port, destination_port)

Connect two JACK ports by name.

Parameters:

Name	Type	Description	Default
source_port	str	Name of the source port (output)	required
destination_port	str	Name of the destination port (input)	required

Returns:

Type	Description
bool	True if connection successful, False otherwise

Source code in src/cuemsengine/players/JackConnectionManager.py

@logged
def connect_by_name(self, source_port: str, destination_port: str) -> bool:
    """Connect two JACK ports by name.

    Args:
        source_port: Name of the source port (output)
        destination_port: Name of the destination port (input)

    Returns:
        True if connection successful, False otherwise
    """
    for attempt in (0, 1):
        if self.client is None:
            Logger.error("JACK client not initialized")
            return False
        try:
            if self.is_connected(source_port, destination_port):
                Logger.debug(f"Ports already connected: {source_port} -> {destination_port}")
                return True
            self.client.connect(source_port, destination_port)
            Logger.info(f"Connected {source_port} -> {destination_port}")
            return True
        except jack.JackError as e:
            if attempt == 0:
                Logger.warning(f"connect failed, retrying with fresh client: {source_port} -> {destination_port}: {e}")
                self._reset_client()
                continue
            Logger.warning(f"Failed to connect {source_port} -> {destination_port}: {e}")
            return False
        except Exception as e:
            Logger.error(f"Unexpected error connecting JACK ports: {e}")
            return False
    return False

disconnect_by_name(source_port, destination_port)

Disconnect two JACK ports by name.

Parameters:

Name	Type	Description	Default
source_port	str	Name of the source port (output)	required
destination_port	str	Name of the destination port (input)	required

Returns:

Type	Description
bool	True if disconnection successful, False otherwise

Source code in src/cuemsengine/players/JackConnectionManager.py

@logged
def disconnect_by_name(self, source_port: str, destination_port: str) -> bool:
    """Disconnect two JACK ports by name.

    Args:
        source_port: Name of the source port (output)
        destination_port: Name of the destination port (input)

    Returns:
        True if disconnection successful, False otherwise
    """
    for attempt in (0, 1):
        if self.client is None:
            Logger.error("JACK client not initialized")
            return False
        try:
            self.client.disconnect(source_port, destination_port)
            Logger.info(f"Disconnected {source_port} -> {destination_port}")
            return True
        except jack.JackError as e:
            if attempt == 0:
                Logger.warning(f"disconnect failed, retrying with fresh client: {source_port} -> {destination_port}: {e}")
                self._reset_client()
                continue
            Logger.warning(f"Failed to disconnect {source_port} -> {destination_port}: {e}")
            return False
        except Exception as e:
            Logger.error(f"Unexpected error disconnecting JACK ports: {e}")
            return False
    return False

get_connections(port_name)

Get all connections for a given port.

Parameters:

Name	Type	Description	Default
port_name	str	Name of the port to query	required

Returns:

Type	Description
list[str]	List of connected port names

Source code in src/cuemsengine/players/JackConnectionManager.py

@logged
def get_connections(self, port_name: str) -> list[str]:
    """Get all connections for a given port.

    Args:
        port_name: Name of the port to query

    Returns:
        List of connected port names
    """
    if self.client is None:
        Logger.error("JACK client not initialized")
        return []

    try:
        # Get the port object
        ports = self.client.get_ports(name_pattern=f'^{port_name}$')
        if not ports:
            Logger.warning(f"Port not found: {port_name}")
            return []

        port = ports[0]

        # Get connections
        connections = self.client.get_all_connections(port)
        connection_names = [conn.name for conn in connections]

        return connection_names

    except jack.JackError as e:
        Logger.error(f"Error getting connections for port {port_name}: {e}")
        return []
    except Exception as e:
        Logger.error(f"Unexpected error getting connections: {e}")
        return []

get_ports(pattern=None, is_audio=True, is_output=None, is_input=None)

Get list of JACK ports.

Parameters:

Name	Type	Description	Default
pattern	str	Optional regex pattern to filter port names	None
is_audio	bool	Filter for audio ports (default: True)	True
is_output	bool	Filter for output ports (default: None = all)	None
is_input	bool	Filter for input ports (default: None = all)	None

Returns:

Type	Description
list[str]	List of port names

Source code in src/cuemsengine/players/JackConnectionManager.py

@logged
def get_ports(self, pattern: str = None, is_audio: bool = True, 
              is_output: bool = None, is_input: bool = None) -> list[str]:
    """Get list of JACK ports.

    Args:
        pattern: Optional regex pattern to filter port names
        is_audio: Filter for audio ports (default: True)
        is_output: Filter for output ports (default: None = all)
        is_input: Filter for input ports (default: None = all)

    Returns:
        List of port names
    """
    if self.client is None:
        Logger.error("JACK client not initialized")
        return []

    try:
        ports = self.client.get_ports(
            name_pattern=pattern if pattern else '',
            is_audio=is_audio,
            is_output=is_output,
            is_input=is_input
        )
        port_names = [p.name for p in ports]
        Logger.debug(f"Found {len(port_names)} JACK ports")
        return port_names

    except jack.JackError as e:
        Logger.error(f"Error getting JACK ports: {e}")
        return []
    except Exception as e:
        Logger.error(f"Unexpected error getting JACK ports: {e}")
        return []

is_connected(source_port, destination_port)

Check if two ports are connected.

Parameters:

Name	Type	Description	Default
source_port	str	Name of the source port	required
destination_port	str	Name of the destination port	required

Returns:

Type	Description
bool	True if connected, False otherwise

Source code in src/cuemsengine/players/JackConnectionManager.py

@logged
def is_connected(self, source_port: str, destination_port: str) -> bool:
    """Check if two ports are connected.

    Args:
        source_port: Name of the source port
        destination_port: Name of the destination port

    Returns:
        True if connected, False otherwise
    """
    connections = self.get_connections(source_port)
    return destination_port in connections

port_exists(port_name)

Check if a JACK port exists.

Parameters:

Name	Type	Description	Default
port_name	str	Full name of the port (e.g., 'client_name:port_name')	required

Returns:

Type	Description
bool	True if the port exists, False otherwise

Source code in src/cuemsengine/players/JackConnectionManager.py

def port_exists(self, port_name: str) -> bool:
    """Check if a JACK port exists.

    Args:
        port_name: Full name of the port (e.g., 'client_name:port_name')

    Returns:
        True if the port exists, False otherwise
    """
    if self.client is None:
        return False

    try:
        ports = self.client.get_ports(name_pattern=f'^{port_name}$')
        return len(ports) > 0
    except Exception:
        return False