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).
        """
        # Fail fast with no JACK server (see connect_player_to_outputs): the
        # retry loop is for the port-registration race, not for a host whose
        # jackd is absent/stubbed — there it just burned
        # len(audio_outputs) * max_retries * retry_delay seconds at startup.
        if self.conn_man.client is None:
            Logger.error(
                "No JACK server available - mixer outputs not wired; audio on"
                " this host will be SILENT. (jackd absent or stubbed -"
                " skipping the port wait.)"
            )
            return
        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} /"
                        f" {playback_port}"
                        f" (attempt {attempt + 1}/{max_retries})"
                    )
                    sleep(retry_delay)
            else:
                Logger.warning(
                    f"JACK ports not available after {max_retries} attempts:"
                    f"{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:"
                f"{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).
        # Gate ONLY on port_exists(); get_connections() returns [] (not None)
        # for a missing port, so the old guard broke the loop immediately and
        # connected a not-yet-registered port -> jackd Unknown source port.
        for attempt in range(max_retries):
            if 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"
                    f"{attempt + 1}/{max_retries})"
                )
                sleep(retry_delay)
        else:
            Logger.warning(
                f"JACK port {channel_0_output} not available after"
                f"{max_retries} attempts"
            )
            return False

        # 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,
        should_abort=None,
        ready=None,
    ):
        """
        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
            should_abort: optional predicate, checked on every attempt the
            port is still missing. True ends the wait at once with False. The
            caller passes "the player process has exited": its ports can
            never register then (869f9wqpn -- a STOP/load kills the player
            mid-wait). It does NOT shorten the wait for a live process.
            ready: optional threading.Event the player sets once it reports
            itself started (Player.ready, its "[OK] RUNNING!" line). When
            given, the wiring waits for it before looking at the ports: the
            ports exist before the player has finished starting, and RtAudio
            auto-connects them to system:playback only when its stream starts.
            Wiring earlier finds nothing to disconnect, and the auto-connect
            lands afterwards and stays - the cue then sounds through the mixer
            AND straight to the outputs (869fbyjzx / 869fcvz85). The ready wait
            and the port wait share one budget, today's port-wait ceiling.
            A player that never reports ready is wired on port presence, with
            a WARNING.

        KNOWN LIMIT (869fbyjzx): this makes an audio arm correct and as fast as the
        player allows, about 0.47 s cold (0.35 s warm), most of it the player's
        own start-up. A GO that reaches a cue less than that after its arm
        started still waits for the arm and fires LATE by the difference. On
        test2 (a node audio cue behind another node's pause, second GO after
        the hand-off): 0.3 s -> ~0.2 s late; 0.5 s and later -> on time.
        Only arming BEFORE the hand-off removes it. PLAN B, not built because
        no real show hits it today (sala1's node segments are video only; the
        cost is players and layers held for the whole pause): arm the first
        local segment at load even behind another node's pause, and the next
        one after each GO ("Part B", Plans/2026-10-01-engine-prearm-past-
        pauses-and-inventory.md; measured cost ~15 MB per held audio player,
        ~40 MB GPU per held 1080p video layer). A cheaper variant is to
        pre-launch players. Revisit if a venue reports a late node audio cue
        after a fast double GO.

        Returns:
            True if every required player→mixer connection was made, False if
            the player ports never registered, no mixer inputs resolved, or any
            connection failed (caller should treat False as a silent cue).
        """
        from time import monotonic, 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"

        # Fail fast when there is no JACK client at all. The retry loop below
        # exists for a TRANSIENT race (player ports registering a beat after
        # the process spawns); a missing JACK server is PERMANENT for this arm
        # attempt, and waiting max_retries * retry_delay (~15s) for it blocked
        # armed_ready — and therefore the cluster GO gate — on every load and
        # re-arm of a host whose jackd is deliberately stubbed (measured
        # load->GO 16s vs 1-2s on the test2 controller, regression window
        # 2026-07-23 -> 24 when this wait first deployed there). The client
        # property retries initialization once on this access, so a jackd
        # that just came up is still picked up.
        if self.conn_man.client is None:
            Logger.error(
                f"No JACK server available - cannot route {player_name} to the"
                f" mixer; cue will be SILENT despite showing armed status."
                f" (jackd absent or stubbed on this host - skipping the"
                f" {max_retries * retry_delay:.0f}s port wait.)"
            )
            return False

        # Wait for the player to report itself started (869fbyjzx). AFTER the
        # no-JACK fail-fast above: a JACK-stubbed host (the test2 controller)
        # must still return at once. Counted slices, not a monotonic deadline,
        # bound the wait, so a test can patch ready.wait. Whatever this phase
        # uses comes off the port wait below: one budget for both.
        port_attempts = max_retries
        if ready is not None:
            ready_slice = 0.05
            # Today's port wait sleeps between its checks, (max_retries - 1)
            # times: that total (14.5 s by default) is the whole budget.
            budget_s = max(0, max_retries - 1) * retry_delay
            ready_slices = max(1, int(round(budget_s / ready_slice)))
            ready_started = monotonic()
            slices_used = 0
            got_ready = False
            for _ in range(ready_slices):
                if ready.wait(ready_slice):
                    got_ready = True
                    break
                slices_used += 1
                if should_abort is not None and should_abort():
                    Logger.warning(
                        f"Player process for {player_name} exited before "
                        f"reporting RUNNING (waited "
                        f"{monotonic() - ready_started:.3f}s) - killed by a "
                        "STOP/load, or crashed. Not waiting any longer."
                    )
                    return False
            if got_ready:
                Logger.info(
                    f"Audio player {player_name} reported RUNNING after "
                    f"{monotonic() - ready_started:.3f}s"
                )
            else:
                Logger.warning(
                    f"Audio player {player_name} never reported RUNNING in "
                    f"{monotonic() - ready_started:.1f}s - wiring on port "
                    "presence (old or unusual player build)"
                )
            if retry_delay > 0:
                # Round UP (to 6 places first, against float noise): the ready
                # wait may never leave the port wait more than the budget.
                spent_attempts = math.ceil(
                    round(slices_used * ready_slice / retry_delay, 6)
                )
                port_attempts = max(1, max_retries - spent_attempts)
            else:
                port_attempts = 1

        # Wait for player JACK ports to be available.
        # NOTE: gate ONLY on port_exists(); get_connections() returns [] (not
        # None) for a missing port, so the old 'connections is not None' guard
        # made this loop break immediately and connect a not-yet-registered
        # port -> jackd 'Unknown source port' -> silent-but-green cue.
        #
        # 869f79ecc: only the FAILURE path left a trace (DEBUG per retry,
        # WARNING on giving up); the success path -- which is every normal
        # arm -- logged nothing, so nobody could tell whether this ~15s
        # ceiling (max_retries * retry_delay) reflects real registration
        # latency or is just an untested guess. It IS a guess -- CLAUDE.md:91
        # explains the pipeline (OSC bind + RtMidi + FFmpeg probe before
        # RtAudio registers the port) but gives no distribution, and the
        # 2026-07 isil multilingual incident shows underestimating it once
        # already cost a real show. Log the success path so the fleet's own
        # journals can answer that instead of another guess.
        wait_started = monotonic()
        for attempt in range(port_attempts):
            if self.conn_man.port_exists(channel_0_output):
                Logger.info(
                    f"JACK port {channel_0_output} registered after "
                    f"{monotonic() - wait_started:.3f}s (attempt {attempt + 1}"
                    f"/{port_attempts})"
                )
                break
            if should_abort is not None and should_abort():
                # WARNING, not ERROR: after a STOP or a load this is the
                # expected end of an arm that was in flight. A crash lands
                # here too, and the caller's "will be SILENT" error covers it.
                Logger.warning(
                    f"Player process for {player_name} exited before "
                    f"registering its JACK ports (waited "
                    f"{monotonic() - wait_started:.3f}s, attempt {attempt + 1}"
                    f"/{port_attempts}) - killed by a STOP/load, or crashed. "
                    "Not waiting any longer."
                )
                return False
            if attempt < port_attempts - 1:
                Logger.debug(
                    f"Waiting for JACK port {channel_0_output} (attempt"
                    f"{attempt + 1}/{port_attempts})"
                )
                sleep(retry_delay)
        else:
            Logger.warning(
                f"JACK port {channel_0_output} not available after"
                f"{port_attempts} attempts"
            )
            return False

        # A stereo player registers outport 1 alongside outport 0, but the two
        # can surface a beat apart. Give outport 1 a brief, bounded grace window
        # before deciding mono vs stereo, so a lagging outport 1 is not mis-read
        # as mono (which would fan outport 0 to both sides and drop the right
        # channel). A genuinely mono player just waits out this short window.
        for _ in range(6):
            if self.conn_man.port_exists(channel_1_output):
                break
            sleep(0.05)

        # 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)

        expected = self._expected_edges(
            player_name,
            player_output_prefix,
            selected_outputs,
            is_stereo,
            log_missing_inputs=True,
        )
        if not expected:
            Logger.error(f"No valid mixer inputs found for outputs: {selected_outputs}")
            return False

        Logger.info(
            f"Connecting {player_name} to outputs:"
            f"{selected_outputs} -> {[dst for _src, dst in expected]}"
        )

        all_connected = True
        for src, dst in expected:
            Logger.debug(f"{src} → {dst}")
            ok = self.conn_man.connect_by_name(src, dst)
            all_connected = all_connected and ok

        if not all_connected:
            Logger.error(
                f"One or more mixer connections failed for {player_name};"
                "cue may be silent"
            )
        return all_connected

    def _expected_edges(
        self,
        player_name: str,
        player_output_prefix: str,
        selected_outputs: list,
        is_stereo: bool,
        log_missing_inputs: bool = False,
    ) -> list:
        """The (player outport, mixer input) edges a player should have.

        The ONE copy of the routing, used by the arm-time wiring
        (connect_player_to_outputs), the GO-time check
        (player_connection_diff) and the tests, so they cannot drift apart
        (869fbyjzx). Selected outputs map to mixer inputs through
        audio_outputs; inputs that do not exist are dropped, which shifts the
        L/R pairing exactly as the wiring does. The targets are treated as
        alternating L/R pairs: even-indexed targets (0, 2, 4 ...) get outport
        0, odd-indexed ones outport 1, or outport 0 again on a mono player.
        This covers 1, 2 or any number of outputs uniformly.

        Returns [] when no mixer input resolves.
        """
        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"
        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)
                elif log_missing_inputs:
                    Logger.warning(f"Mixer input {mixer_input} does not exist")
        edges = []
        for i, mixer_input in enumerate(target_inputs):
            if i % 2 == 0 or not is_stereo:
                edges.append((channel_0_output, mixer_input))
            else:
                edges.append((channel_1_output, mixer_input))
        return edges

    def player_connection_diff(
        self,
        player_name: str,
        player_output_prefix: str = "outport",
        selected_outputs: list = None,
    ):
        """Compare a player's outport edges with the expected wiring.

        Returns (missing, stray), two lists of (source, destination) edges:
        missing = expected edges that are not connected; stray = edges on the
        player's outports that are not expected (e.g. RtAudio's own
        auto-connect to system:playback, 869fcvz85). An edge is never in
        both. Returns None when the graph cannot be wired at all: outport 0
        does not exist (the subprocess is gone) or no mixer input resolves —
        callers must treat None as a silent cue, never as "correct".

        One get_connections per outport (outport 1 only on a stereo player):
        this runs on the GO path before /offset.
        """
        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 None
        is_stereo = self.conn_man.port_exists(channel_1_output)
        expected = self._expected_edges(
            player_name, player_output_prefix, selected_outputs, is_stereo
        )
        if not expected:
            return None
        ports = [channel_0_output] + ([channel_1_output] if is_stereo else [])
        present = {port: list(self.conn_man.get_connections(port)) for port in ports}
        missing = [(src, dst) for src, dst in expected if dst not in present[src]]
        wanted = set(expected)
        stray = [
            (port, dst)
            for port in ports
            for dst in present[port]
            if (port, dst) not in wanted
        ]
        return missing, stray

    def player_connections_correct(
        self,
        player_name: str,
        player_output_prefix: str = "outport",
        selected_outputs: list = None,
    ) -> bool:
        """True when the player is wired exactly as connect_player_to_outputs
        wires it: every expected edge present and nothing else on its
        outports. False on a missing edge, a stray edge, a missing outport 0
        (subprocess gone) or no mixer input. See player_connection_diff.
        """
        diff = self.player_connection_diff(
            player_name, player_output_prefix, selected_outputs
        )
        return diff is not None and not diff[0] and not diff[1]

    def repair_player_connections(self, missing: list, stray: list):
        """Fix a player's wiring edge by edge, never by a full rewire.

        A GO can reach this while the cue is audible (a fresh GO or a 'play'
        action on a playing cue), so a working mixer edge is never
        disconnected: strays are disconnected one by one, missing edges
        connected one by one (869fbyjzx).

        Each stray is re-checked just before its disconnect: one already gone
        (a stale diff, a re-arm wiring the same player) is skipped, because a
        failed disconnect resets the SHARED JACK client under any arm wiring
        at the same moment (JackConnectionManager.disconnect_by_name).

        Returns (connected_ok, strays_left): whether every missing edge got
        connected, and the strays whose disconnect failed.
        """
        strays_left = []
        for src, dst in stray:
            if not self.conn_man.is_connected(src, dst):
                continue
            if not self.conn_man.disconnect_by_name(src, dst):
                # Another thread (a re-arm wiring the same player) may have
                # removed it between our check and our disconnect: then it is
                # gone, not a survivor.
                if self.conn_man.is_connected(src, dst):
                    strays_left.append((src, dst))
        connected_ok = True
        for src, dst in missing:
            ok = self.conn_man.connect_by_name(src, dst)
            connected_ok = connected_ok and ok
        return connected_ok, strays_left

    @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	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'
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:"
            f"{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).
    # Gate ONLY on port_exists(); get_connections() returns [] (not None)
    # for a missing port, so the old guard broke the loop immediately and
    # connected a not-yet-registered port -> jackd Unknown source port.
    for attempt in range(max_retries):
        if 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"
                f"{attempt + 1}/{max_retries})"
            )
            sleep(retry_delay)
    else:
        Logger.warning(
            f"JACK port {channel_0_output} not available after"
            f"{max_retries} attempts"
        )
        return False

    # 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, should_abort=None, ready=None)

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'
selected_outputs	list	List of output port names (e.g.,	None
['system		playback_1'])	required
max_retries	int	Maximum number of connection attempts	30
retry_delay	float	Delay between retries in seconds	0.5
should_abort		optional predicate, checked on every attempt the	None
caller passes "the player process has exited"		its ports can	required
ready		optional threading.Event the player sets once it reports	None
given, the wiring waits for it before looking at the ports		the	required
auto-connects them to system		playback only when its stream starts.	required

KNOWN LIMIT (869fbyjzx): this makes an audio arm correct and as fast as the player allows, about 0.47 s cold (0.35 s warm), most of it the player's own start-up. A GO that reaches a cue less than that after its arm started still waits for the arm and fires LATE by the difference. On test2 (a node audio cue behind another node's pause, second GO after the hand-off): 0.3 s -> ~0.2 s late; 0.5 s and later -> on time. Only arming BEFORE the hand-off removes it. PLAN B, not built because no real show hits it today (sala1's node segments are video only; the cost is players and layers held for the whole pause): arm the first local segment at load even behind another node's pause, and the next one after each GO ("Part B", Plans/2026-10-01-engine-prearm-past- pauses-and-inventory.md; measured cost ~15 MB per held audio player, ~40 MB GPU per held 1080p video layer). A cheaper variant is to pre-launch players. Revisit if a venue reports a late node audio cue after a fast double GO.

Returns:

Type	Description
	True if every required player→mixer connection was made, False if
	the player ports never registered, no mixer inputs resolved, or any
	connection failed (caller should treat False as a silent cue).

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,
    should_abort=None,
    ready=None,
):
    """
    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
        should_abort: optional predicate, checked on every attempt the
        port is still missing. True ends the wait at once with False. The
        caller passes "the player process has exited": its ports can
        never register then (869f9wqpn -- a STOP/load kills the player
        mid-wait). It does NOT shorten the wait for a live process.
        ready: optional threading.Event the player sets once it reports
        itself started (Player.ready, its "[OK] RUNNING!" line). When
        given, the wiring waits for it before looking at the ports: the
        ports exist before the player has finished starting, and RtAudio
        auto-connects them to system:playback only when its stream starts.
        Wiring earlier finds nothing to disconnect, and the auto-connect
        lands afterwards and stays - the cue then sounds through the mixer
        AND straight to the outputs (869fbyjzx / 869fcvz85). The ready wait
        and the port wait share one budget, today's port-wait ceiling.
        A player that never reports ready is wired on port presence, with
        a WARNING.

    KNOWN LIMIT (869fbyjzx): this makes an audio arm correct and as fast as the
    player allows, about 0.47 s cold (0.35 s warm), most of it the player's
    own start-up. A GO that reaches a cue less than that after its arm
    started still waits for the arm and fires LATE by the difference. On
    test2 (a node audio cue behind another node's pause, second GO after
    the hand-off): 0.3 s -> ~0.2 s late; 0.5 s and later -> on time.
    Only arming BEFORE the hand-off removes it. PLAN B, not built because
    no real show hits it today (sala1's node segments are video only; the
    cost is players and layers held for the whole pause): arm the first
    local segment at load even behind another node's pause, and the next
    one after each GO ("Part B", Plans/2026-10-01-engine-prearm-past-
    pauses-and-inventory.md; measured cost ~15 MB per held audio player,
    ~40 MB GPU per held 1080p video layer). A cheaper variant is to
    pre-launch players. Revisit if a venue reports a late node audio cue
    after a fast double GO.

    Returns:
        True if every required player→mixer connection was made, False if
        the player ports never registered, no mixer inputs resolved, or any
        connection failed (caller should treat False as a silent cue).
    """
    from time import monotonic, 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"

    # Fail fast when there is no JACK client at all. The retry loop below
    # exists for a TRANSIENT race (player ports registering a beat after
    # the process spawns); a missing JACK server is PERMANENT for this arm
    # attempt, and waiting max_retries * retry_delay (~15s) for it blocked
    # armed_ready — and therefore the cluster GO gate — on every load and
    # re-arm of a host whose jackd is deliberately stubbed (measured
    # load->GO 16s vs 1-2s on the test2 controller, regression window
    # 2026-07-23 -> 24 when this wait first deployed there). The client
    # property retries initialization once on this access, so a jackd
    # that just came up is still picked up.
    if self.conn_man.client is None:
        Logger.error(
            f"No JACK server available - cannot route {player_name} to the"
            f" mixer; cue will be SILENT despite showing armed status."
            f" (jackd absent or stubbed on this host - skipping the"
            f" {max_retries * retry_delay:.0f}s port wait.)"
        )
        return False

    # Wait for the player to report itself started (869fbyjzx). AFTER the
    # no-JACK fail-fast above: a JACK-stubbed host (the test2 controller)
    # must still return at once. Counted slices, not a monotonic deadline,
    # bound the wait, so a test can patch ready.wait. Whatever this phase
    # uses comes off the port wait below: one budget for both.
    port_attempts = max_retries
    if ready is not None:
        ready_slice = 0.05
        # Today's port wait sleeps between its checks, (max_retries - 1)
        # times: that total (14.5 s by default) is the whole budget.
        budget_s = max(0, max_retries - 1) * retry_delay
        ready_slices = max(1, int(round(budget_s / ready_slice)))
        ready_started = monotonic()
        slices_used = 0
        got_ready = False
        for _ in range(ready_slices):
            if ready.wait(ready_slice):
                got_ready = True
                break
            slices_used += 1
            if should_abort is not None and should_abort():
                Logger.warning(
                    f"Player process for {player_name} exited before "
                    f"reporting RUNNING (waited "
                    f"{monotonic() - ready_started:.3f}s) - killed by a "
                    "STOP/load, or crashed. Not waiting any longer."
                )
                return False
        if got_ready:
            Logger.info(
                f"Audio player {player_name} reported RUNNING after "
                f"{monotonic() - ready_started:.3f}s"
            )
        else:
            Logger.warning(
                f"Audio player {player_name} never reported RUNNING in "
                f"{monotonic() - ready_started:.1f}s - wiring on port "
                "presence (old or unusual player build)"
            )
        if retry_delay > 0:
            # Round UP (to 6 places first, against float noise): the ready
            # wait may never leave the port wait more than the budget.
            spent_attempts = math.ceil(
                round(slices_used * ready_slice / retry_delay, 6)
            )
            port_attempts = max(1, max_retries - spent_attempts)
        else:
            port_attempts = 1

    # Wait for player JACK ports to be available.
    # NOTE: gate ONLY on port_exists(); get_connections() returns [] (not
    # None) for a missing port, so the old 'connections is not None' guard
    # made this loop break immediately and connect a not-yet-registered
    # port -> jackd 'Unknown source port' -> silent-but-green cue.
    #
    # 869f79ecc: only the FAILURE path left a trace (DEBUG per retry,
    # WARNING on giving up); the success path -- which is every normal
    # arm -- logged nothing, so nobody could tell whether this ~15s
    # ceiling (max_retries * retry_delay) reflects real registration
    # latency or is just an untested guess. It IS a guess -- CLAUDE.md:91
    # explains the pipeline (OSC bind + RtMidi + FFmpeg probe before
    # RtAudio registers the port) but gives no distribution, and the
    # 2026-07 isil multilingual incident shows underestimating it once
    # already cost a real show. Log the success path so the fleet's own
    # journals can answer that instead of another guess.
    wait_started = monotonic()
    for attempt in range(port_attempts):
        if self.conn_man.port_exists(channel_0_output):
            Logger.info(
                f"JACK port {channel_0_output} registered after "
                f"{monotonic() - wait_started:.3f}s (attempt {attempt + 1}"
                f"/{port_attempts})"
            )
            break
        if should_abort is not None and should_abort():
            # WARNING, not ERROR: after a STOP or a load this is the
            # expected end of an arm that was in flight. A crash lands
            # here too, and the caller's "will be SILENT" error covers it.
            Logger.warning(
                f"Player process for {player_name} exited before "
                f"registering its JACK ports (waited "
                f"{monotonic() - wait_started:.3f}s, attempt {attempt + 1}"
                f"/{port_attempts}) - killed by a STOP/load, or crashed. "
                "Not waiting any longer."
            )
            return False
        if attempt < port_attempts - 1:
            Logger.debug(
                f"Waiting for JACK port {channel_0_output} (attempt"
                f"{attempt + 1}/{port_attempts})"
            )
            sleep(retry_delay)
    else:
        Logger.warning(
            f"JACK port {channel_0_output} not available after"
            f"{port_attempts} attempts"
        )
        return False

    # A stereo player registers outport 1 alongside outport 0, but the two
    # can surface a beat apart. Give outport 1 a brief, bounded grace window
    # before deciding mono vs stereo, so a lagging outport 1 is not mis-read
    # as mono (which would fan outport 0 to both sides and drop the right
    # channel). A genuinely mono player just waits out this short window.
    for _ in range(6):
        if self.conn_man.port_exists(channel_1_output):
            break
        sleep(0.05)

    # 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)

    expected = self._expected_edges(
        player_name,
        player_output_prefix,
        selected_outputs,
        is_stereo,
        log_missing_inputs=True,
    )
    if not expected:
        Logger.error(f"No valid mixer inputs found for outputs: {selected_outputs}")
        return False

    Logger.info(
        f"Connecting {player_name} to outputs:"
        f"{selected_outputs} -> {[dst for _src, dst in expected]}"
    )

    all_connected = True
    for src, dst in expected:
        Logger.debug(f"{src} → {dst}")
        ok = self.conn_man.connect_by_name(src, dst)
        all_connected = all_connected and ok

    if not all_connected:
        Logger.error(
            f"One or more mixer connections failed for {player_name};"
            "cue may be silent"
        )
    return all_connected

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).
    """
    # Fail fast with no JACK server (see connect_player_to_outputs): the
    # retry loop is for the port-registration race, not for a host whose
    # jackd is absent/stubbed — there it just burned
    # len(audio_outputs) * max_retries * retry_delay seconds at startup.
    if self.conn_man.client is None:
        Logger.error(
            "No JACK server available - mixer outputs not wired; audio on"
            " this host will be SILENT. (jackd absent or stubbed -"
            " skipping the port wait.)"
        )
        return
    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} /"
                    f" {playback_port}"
                    f" (attempt {attempt + 1}/{max_retries})"
                )
                sleep(retry_delay)
        else:
            Logger.warning(
                f"JACK ports not available after {max_retries} attempts:"
                f"{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_connection_diff(player_name, player_output_prefix='outport', selected_outputs=None)

Compare a player's outport edges with the expected wiring.

Returns (missing, stray), two lists of (source, destination) edges: missing = expected edges that are not connected; stray = edges on the player's outports that are not expected (e.g. RtAudio's own auto-connect to system:playback, 869fcvz85). An edge is never in both. Returns None when the graph cannot be wired at all: outport 0 does not exist (the subprocess is gone) or no mixer input resolves — callers must treat None as a silent cue, never as "correct".

One get_connections per outport (outport 1 only on a stereo player): this runs on the GO path before /offset.

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

def player_connection_diff(
    self,
    player_name: str,
    player_output_prefix: str = "outport",
    selected_outputs: list = None,
):
    """Compare a player's outport edges with the expected wiring.

    Returns (missing, stray), two lists of (source, destination) edges:
    missing = expected edges that are not connected; stray = edges on the
    player's outports that are not expected (e.g. RtAudio's own
    auto-connect to system:playback, 869fcvz85). An edge is never in
    both. Returns None when the graph cannot be wired at all: outport 0
    does not exist (the subprocess is gone) or no mixer input resolves —
    callers must treat None as a silent cue, never as "correct".

    One get_connections per outport (outport 1 only on a stereo player):
    this runs on the GO path before /offset.
    """
    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 None
    is_stereo = self.conn_man.port_exists(channel_1_output)
    expected = self._expected_edges(
        player_name, player_output_prefix, selected_outputs, is_stereo
    )
    if not expected:
        return None
    ports = [channel_0_output] + ([channel_1_output] if is_stereo else [])
    present = {port: list(self.conn_man.get_connections(port)) for port in ports}
    missing = [(src, dst) for src, dst in expected if dst not in present[src]]
    wanted = set(expected)
    stray = [
        (port, dst)
        for port in ports
        for dst in present[port]
        if (port, dst) not in wanted
    ]
    return missing, stray

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

True when the player is wired exactly as connect_player_to_outputs wires it: every expected edge present and nothing else on its outports. False on a missing edge, a stray edge, a missing outport 0 (subprocess gone) or no mixer input. See player_connection_diff.

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:
    """True when the player is wired exactly as connect_player_to_outputs
    wires it: every expected edge present and nothing else on its
    outports. False on a missing edge, a stray edge, a missing outport 0
    (subprocess gone) or no mixer input. See player_connection_diff.
    """
    diff = self.player_connection_diff(
        player_name, player_output_prefix, selected_outputs
    )
    return diff is not None and not diff[0] and not diff[1]

repair_player_connections(missing, stray)

Fix a player's wiring edge by edge, never by a full rewire.

A GO can reach this while the cue is audible (a fresh GO or a 'play' action on a playing cue), so a working mixer edge is never disconnected: strays are disconnected one by one, missing edges connected one by one (869fbyjzx).

Each stray is re-checked just before its disconnect: one already gone (a stale diff, a re-arm wiring the same player) is skipped, because a failed disconnect resets the SHARED JACK client under any arm wiring at the same moment (JackConnectionManager.disconnect_by_name).

Returns (connected_ok, strays_left): whether every missing edge got connected, and the strays whose disconnect failed.

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

def repair_player_connections(self, missing: list, stray: list):
    """Fix a player's wiring edge by edge, never by a full rewire.

    A GO can reach this while the cue is audible (a fresh GO or a 'play'
    action on a playing cue), so a working mixer edge is never
    disconnected: strays are disconnected one by one, missing edges
    connected one by one (869fbyjzx).

    Each stray is re-checked just before its disconnect: one already gone
    (a stale diff, a re-arm wiring the same player) is skipped, because a
    failed disconnect resets the SHARED JACK client under any arm wiring
    at the same moment (JackConnectionManager.disconnect_by_name).

    Returns (connected_ok, strays_left): whether every missing edge got
    connected, and the strays whose disconnect failed.
    """
    strays_left = []
    for src, dst in stray:
        if not self.conn_man.is_connected(src, dst):
            continue
        if not self.conn_man.disconnect_by_name(src, dst):
            # Another thread (a re-arm wiring the same player) may have
            # removed it between our check and our disconnect: then it is
            # gone, not a survivor.
            if self.conn_man.is_connected(src, dst):
                strays_left.append((src, dst))
    connected_ok = True
    for src, dst in missing:
        ok = self.conn_man.connect_by_name(src, dst)
        connected_ok = connected_ok and ok
    return connected_ok, strays_left

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

        # Shadow of the gain commanded to jack-volume via the typed setters
        # (jack-volume is write-only — it cannot be read back). Keys are
        # "master" and channel indices as strings ("0", "1", …); values are
        # linear gains. reset_volumes populates it (through apply_volume); the
        # node snapshots it to the controller after each reset so the UI can
        # read back the true (post-reset) mixer volume.
        self._gain_state: dict[str, float] = {}

        # 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 apply_volume(self, channel, gain: float) -> bool:
        """Send a volume to jack-volume AND record it in `_gain_state`.

        The typed setters (set_master_volume / set_channel_volume, hence
        reset_volumes) route through here, so a reset populates _gain_state with
        the values it commanded. The node snapshots that state to the controller
        right after each reset, letting the UI read back the true (post-reset)
        mixer gain — jack-volume itself is write-only.

        (Live per-cue UI writes stay on the raw set_value path and are shadowed
        controller-side; they are not recorded here because the node snapshots
        only on reset, never per write — a per-write report would race the
        controller's optimistic populate.)

        Args:
            channel: "master" or a 0-indexed channel (int or numeric string)
            gain: Volume gain (0.0 to 1.0)

        Returns:
            True if applied; False if gain/channel was invalid (no OSC sent,
            no state change) — same reject-and-no-op semantics the typed
            setters have always had.
        """
        if not 0.0 <= gain <= 1.0:
            Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
            return False

        ch = str(channel)
        if ch == "master":
            path = f"/audiomixer/{self.client_name}/master"
        else:
            try:
                ch_i = int(ch)
            except (TypeError, ValueError):
                Logger.error(f"Invalid mixer channel: {channel!r}")
                return False
            if not 0 <= ch_i < self.channel_number:
                Logger.error(f"Invalid channel: {ch_i}. Max: {self.channel_number - 1}")
                return False
            ch = str(ch_i)
            path = f"/audiomixer/{self.client_name}/{ch_i}"

        Logger.debug(f"Setting mixer {ch} volume to {gain}")
        self.set_value(path, gain)
        self._gain_state[ch] = gain
        return True

    def snapshot(self) -> dict:
        """Return a copy of the authoritative gain state for reporting.

        Shape: {"master": gain, "0": gain, "1": gain, …}.
        """
        return dict(self._gain_state)

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

        Args:
            gain: Volume gain (0.0 to 1.0)
        """
        self.apply_volume("master", 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)
        """
        self.apply_volume(channel, 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"
            f"{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

    # Shadow of the gain commanded to jack-volume via the typed setters
    # (jack-volume is write-only — it cannot be read back). Keys are
    # "master" and channel indices as strings ("0", "1", …); values are
    # linear gains. reset_volumes populates it (through apply_volume); the
    # node snapshots it to the controller after each reset so the UI can
    # read back the true (post-reset) mixer volume.
    self._gain_state: dict[str, float] = {}

    # 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"
        f"{len(endpoints)} endpoints"
    )

apply_volume(channel, gain)

Send a volume to jack-volume AND record it in _gain_state.

The typed setters (set_master_volume / set_channel_volume, hence reset_volumes) route through here, so a reset populates _gain_state with the values it commanded. The node snapshots that state to the controller right after each reset, letting the UI read back the true (post-reset) mixer gain — jack-volume itself is write-only.

(Live per-cue UI writes stay on the raw set_value path and are shadowed controller-side; they are not recorded here because the node snapshots only on reset, never per write — a per-write report would race the controller's optimistic populate.)

Parameters:

Name	Type	Description	Default
channel		"master" or a 0-indexed channel (int or numeric string)	required
gain	float	Volume gain (0.0 to 1.0)	required

Returns:

Type	Description
bool	True if applied; False if gain/channel was invalid (no OSC sent,
bool	no state change) — same reject-and-no-op semantics the typed
bool	setters have always had.

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

@logged
def apply_volume(self, channel, gain: float) -> bool:
    """Send a volume to jack-volume AND record it in `_gain_state`.

    The typed setters (set_master_volume / set_channel_volume, hence
    reset_volumes) route through here, so a reset populates _gain_state with
    the values it commanded. The node snapshots that state to the controller
    right after each reset, letting the UI read back the true (post-reset)
    mixer gain — jack-volume itself is write-only.

    (Live per-cue UI writes stay on the raw set_value path and are shadowed
    controller-side; they are not recorded here because the node snapshots
    only on reset, never per write — a per-write report would race the
    controller's optimistic populate.)

    Args:
        channel: "master" or a 0-indexed channel (int or numeric string)
        gain: Volume gain (0.0 to 1.0)

    Returns:
        True if applied; False if gain/channel was invalid (no OSC sent,
        no state change) — same reject-and-no-op semantics the typed
        setters have always had.
    """
    if not 0.0 <= gain <= 1.0:
        Logger.error(f"Invalid gain value: {gain}. Must be between 0.0 and 1.0")
        return False

    ch = str(channel)
    if ch == "master":
        path = f"/audiomixer/{self.client_name}/master"
    else:
        try:
            ch_i = int(ch)
        except (TypeError, ValueError):
            Logger.error(f"Invalid mixer channel: {channel!r}")
            return False
        if not 0 <= ch_i < self.channel_number:
            Logger.error(f"Invalid channel: {ch_i}. Max: {self.channel_number - 1}")
            return False
        ch = str(ch_i)
        path = f"/audiomixer/{self.client_name}/{ch_i}"

    Logger.debug(f"Setting mixer {ch} volume to {gain}")
    self.set_value(path, gain)
    self._gain_state[ch] = gain
    return True

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)
    """
    self.apply_volume(channel, 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)
    """
    self.apply_volume("master", gain)

snapshot()

Return a copy of the authoritative gain state for reporting.

Shape: {"master": gain, "0": gain, "1": gain, …}.

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

def snapshot(self) -> dict:
    """Return a copy of the authoritative gain state for reporting.

    Shape: {"master": gain, "0": gain, "1": gain, …}.
    """
    return dict(self._gain_state)

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"
                        f"{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"
                    f"{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, media_w: int | None = None, media_h: int | None = None
    ) -> tuple[float, float]:
        """Returns a uniform (s, s) that renders the media at its NATIVE pixel
        size, clamped so the layer never exceeds this output's region.

        The videocomposer base-renders each layer letterboxed against the whole
        canvas (fills canvas height) and applies scale as quad geometry sampling
        the source texture, so s = media_h/canvas_h reproduces the media 1:1 (no
        resample, no stretch). The two clamp terms keep an oversized / wider-
        aspect media inside this region instead of bleeding onto neighbouring
        outputs:
          - (region_w * media_h) / (media_w * canvas_h): clamp width to region
          - region_h / canvas_h: clamp height to region
        The min() is ESSENTIAL — for oversized media the media_h/canvas_h term
        alone could be > 1 and scale the layer UP, causing bleed. Do not remove it.

        Falls back to the legacy region-height ratio when media dims are unknown
        (ffprobe failed / non-video media). NOTE: that fallback reproduces the
        original overflow for an output whose aspect != media aspect, so it is
        logged at WARNING to make a silent re-bleed visible.
        """
        ch = self.canvas_height
        if not ch:
            return (1.0, 1.0)
        region_w = self.canvas_region["width"]
        region_h = self.canvas_region["height"]
        if media_w and media_h:
            s = min(
                media_h / ch,
                (region_w * media_h) / (media_w * ch),
                region_h / ch,
            )
        else:
            s = region_h / ch
            Logger.warning(
                f"VideoOutput {self.mapped_to}: media dims unknown, using legacy "
                f"region-height scale {s:.4f} (may overflow if media aspect != region)"
            )
        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}"
            f"{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}"
        f"{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(media_w=None, media_h=None)

Returns a uniform (s, s) that renders the media at its NATIVE pixel size, clamped so the layer never exceeds this output's region.

The videocomposer base-renders each layer letterboxed against the whole canvas (fills canvas height) and applies scale as quad geometry sampling the source texture, so s = media_h/canvas_h reproduces the media 1:1 (no resample, no stretch). The two clamp terms keep an oversized / wider- aspect media inside this region instead of bleeding onto neighbouring outputs: - (region_w * media_h) / (media_w * canvas_h): clamp width to region - region_h / canvas_h: clamp height to region The min() is ESSENTIAL — for oversized media the media_h/canvas_h term alone could be > 1 and scale the layer UP, causing bleed. Do not remove it.

Falls back to the legacy region-height ratio when media dims are unknown (ffprobe failed / non-video media). NOTE: that fallback reproduces the original overflow for an output whose aspect != media aspect, so it is logged at WARNING to make a silent re-bleed visible.

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

def get_layer_scale(
    self, media_w: int | None = None, media_h: int | None = None
) -> tuple[float, float]:
    """Returns a uniform (s, s) that renders the media at its NATIVE pixel
    size, clamped so the layer never exceeds this output's region.

    The videocomposer base-renders each layer letterboxed against the whole
    canvas (fills canvas height) and applies scale as quad geometry sampling
    the source texture, so s = media_h/canvas_h reproduces the media 1:1 (no
    resample, no stretch). The two clamp terms keep an oversized / wider-
    aspect media inside this region instead of bleeding onto neighbouring
    outputs:
      - (region_w * media_h) / (media_w * canvas_h): clamp width to region
      - region_h / canvas_h: clamp height to region
    The min() is ESSENTIAL — for oversized media the media_h/canvas_h term
    alone could be > 1 and scale the layer UP, causing bleed. Do not remove it.

    Falls back to the legacy region-height ratio when media dims are unknown
    (ffprobe failed / non-video media). NOTE: that fallback reproduces the
    original overflow for an output whose aspect != media aspect, so it is
    logged at WARNING to make a silent re-bleed visible.
    """
    ch = self.canvas_height
    if not ch:
        return (1.0, 1.0)
    region_w = self.canvas_region["width"]
    region_h = self.canvas_region["height"]
    if media_w and media_h:
        s = min(
            media_h / ch,
            (region_w * media_h) / (media_w * ch),
            region_h / ch,
        )
    else:
        s = region_h / ch
        Logger.warning(
            f"VideoOutput {self.mapped_to}: media dims unknown, using legacy "
            f"region-height scale {s:.4f} (may overflow if media aspect != region)"
        )
    return (s, s)

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
        # Set when the subprocess prints READY_LINE. An audio arm waits on it
        # before wiring the player to the mixer: the player's JACK ports exist
        # before it has finished starting, and wiring them that early leaves
        # RtAudio's own auto-connect to system:playback in place (869fbyjzx).
        # Known limit: waiting for it makes an audio arm ~0.47 s cold, so a GO
        # that comes sooner after the arm began is still late. Plan B (arm
        # before the hand-off) is described in AudioMixer.connect_player_to_
        # outputs and the 869fbyjzx plan; it is not built.
        self.ready = Event()

    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:
            # This run starts not ready: an Event left set by an earlier run of
            # the same Player object would pass the audio wiring's gate at once.
            self.ready.clear()
            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}")
                    if READY_LINE in line:
                        self.ready.set()
                # 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
        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 = "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	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
    # Set when the subprocess prints READY_LINE. An audio arm waits on it
    # before wiring the player to the mixer: the player's JACK ports exist
    # before it has finished starting, and wiring them that early leaves
    # RtAudio's own auto-connect to system:playback in place (869fbyjzx).
    # Known limit: waiting for it makes an audio arm ~0.47 s cold, so a GO
    # that comes sooner after the arm began is still late. Plan B (arm
    # before the hand-off) is described in AudioMixer.connect_player_to_
    # outputs and the 869fbyjzx plan; it is not built.
    self.ready = Event()

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:
        # This run starts not ready: an Event left set by an earlier run of
        # the same Player object would pass the audio wiring's gate at once.
        self.ready.clear()
        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}")
                if READY_LINE in line:
                    self.ready.set()
            # 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

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
    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 = "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
    _media_dims_cache: dict
    _unrouted_warned: set

    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
            cls._instance._media_dims_cache = {}
            # Files whose stored dimensions were missing or stale, already
            # warned about once (cue_media_dimensions).
            cls._instance._media_dims_warned = set()
            # (cue id, output name) pairs whose "output not driven" error was
            # already logged in this load (869fgq3xv); cleared at each load.
            cls._instance._unrouted_warned = set()
        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()
            self._media_dims_cache = {}
            self._media_dims_warned = set()
            self._unrouted_warned = set()

    # ---------------------------
    # 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[tuple[AudioPlayer, AudioClient]](
            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}")
        # Clean up any surviving jack-volume from a prior engine lifetime so the
        # new mixer binds "0_mixer" cleanly (avoids the JackNameNotUnique rename
        # that would leave the engine addressing a disconnected zombie).
        self.kill_orphaned_mixer_processes()
        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"
                f"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. Like
        the orphan sweeps it is hygiene, so it never raises into the load
        (869evtdf7): a JACK error skips the cleanup with a warning, and audio
        cues then fail loudly at their own arm.

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

        try:
            all_ports = self._audio_mixer.conn_man.get_ports(
                pattern="Audio_Player-.*", is_audio=True, is_output=True
            )
        except Exception as e:
            Logger.warning(
                f"JACK not reachable, zombie JACK client cleanup skipped: {e!r}"
            )
            return 0
        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.

        Runs in the middle of a project load, so it never raises (869evtdf7).
        """
        pids = _orphan_pids(AUDIOPLAYER_ORPHAN_PATTERN, "audioplayer")
        if not pids:
            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 in pids:
            if pid not in tracked_pids:
                _kill_orphan(pid, "audioplayer")

    def kill_orphaned_mixer_processes(self):
        """Kill jack-volume mixer processes not tracked by this engine.

        Mirrors kill_orphaned_audio_processes for the mixer. On a node-engine
        restart the previous jack-volume survives as an independent subprocess
        and keeps the "0_mixer" JACK client name; a freshly spawned jack-volume
        then hits JackNameNotUnique and is renamed by JACK, ending up a
        disconnected zombie while the engine still addresses "0_mixer" (the
        orphan) — every restart also leaks another zombie. Kill any survivor
        first so the new mixer binds "0_mixer" cleanly.

        Match "jack-volume -c" (the binary invoked with its client-name flag)
        rather than the bare word, so an operator's `tail -f .../jack-volume.log`
        or a grep isn't caught and killed. Covers both the plain and
        "cuems-jack-volume" install names.

        Never raises (869evtdf7): an orphan it cannot kill used to stop the
        mixer from starting, leaving every audio cue on the node silent.
        """
        pids = _orphan_pids(MIXER_ORPHAN_PATTERN, "jack-volume mixer")
        if not pids:
            return

        tracked_pid = None
        if self._audio_mixer is not None and getattr(self._audio_mixer, "p", None):
            tracked_pid = self._audio_mixer.p.pid

        for pid in pids:
            if pid != tracked_pid:
                _kill_orphan(pid, "jack-volume mixer")

    # ---------------------------
    # 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'
                            f"a JACK port"
                        )

            if not selected_outputs:
                Logger.warning(
                    f"No valid audio outputs resolved for cue {cue.id},"
                    f"skipping mixer connection"
                )
            else:
                Logger.info(f"Connecting {player_name} to outputs: {selected_outputs}")
                ready = getattr(player, "ready", None)
                if ready is None:
                    # Every Player has one. Without it the wiring would fall
                    # back to port presence and the doubled-audio race (869fcvz85)
                    # would come back unnoticed.
                    Logger.warning(
                        f"Audio player {player_name} has no ready event - wiring "
                        "on port presence (the auto-connect race is possible)"
                    )
                connected = self._audio_mixer.connect_player_to_outputs(
                    player_name=player_name,
                    player_output_prefix="outport",
                    selected_outputs=selected_outputs,
                    # A STOP/load may kill this player while we wait for its
                    # ports; they will never register then. Stop waiting as
                    # soon as the process is gone instead of holding this
                    # cue's arm for the full ~15 s (869f9wqpn).
                    should_abort=lambda: _process_exited(player),
                    # Wire only once the player reports itself started: its
                    # ports exist earlier, before RtAudio's auto-connect,
                    # which an earlier wiring would leave in place (869fbyjzx).
                    ready=ready,
                )
                if connected is False:
                    # Route to the mixer failed: the cue would show armed/green
                    # in the UI but produce no sound. Surface it loudly rather
                    # than proceeding silently. (Not raised on purpose: aborting
                    # the arm here could disrupt an in-progress GO chain.)
                    Logger.error(
                        f"Audio cue {cue.id} failed to route {player_name} to the mixer"
                        " - it will be SILENT despite showing armed status "
                        "(player ports may not have registered in time, or "
                        "mixer inputs are missing)."
                    )

    # ---------------------------
    # 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_name: str) -> None:
        """Construct (or replace) the GradientClient for this node.

        node_name must match gradient-motiond's own --node-name (defaults to
        the OS hostname; see node-identity-contract.md in cuems-common) —
        NOT the node's UUID, which the daemon's node_name filter will never
        match, silently dropping every message.

        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.
        """
        self._gradient_client = GradientClient(
            host="127.0.0.1",
            port=port,
            node_name=node_name,
        )
        Logger.info(f"GradientClient: bound to 127.0.0.1:{port} node_name={node_name}")

    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"
                    f"per-output "
                    f"bounding box {bbox_w}x{bbox_h}; monitors would be"
                    f"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 partition_cue_outputs(
        self, cue, output_names: list
    ) -> tuple[list[tuple[int, str, VideoOutput]], list[tuple[str, str]]]:
        """Split a video cue's outputs into the ones this node drives and the
        ones it doesn't (869fgq3xv).

        Returns ``(routed, unrouted)``: ``routed`` holds ``(index, output_name,
        VideoOutput)`` with the output's position in ``output_names`` (layer
        ids are built from it), ``unrouted`` holds ``(output_name, reason)``.

        An output is unrouted when it does not resolve: an alias this node
        dropped at start (no region in display.conf, DISPLAY_MISMATCH), a
        custom region with no VideoCueOutput, no canvas at all. Anything else
        the resolver raises (a malformed custom region, cue.outputs None) is
        unrouted too, with its traceback: with no resolved output there is no
        right place to paint the layer, and the videocomposer's default
        placement is the whole canvas, i.e. the other screens.

        The one function behind both the arm and the load-time preflight, so
        the two can never disagree. Never raises.
        """
        routed, unrouted = [], []
        for index, output_name in enumerate(output_names):
            try:
                output = self.resolve_video_output_for_cue(cue, output_name)
            except (KeyError, RuntimeError) as e:
                unrouted.append((output_name, f"{type(e).__name__}: {e}"))
                continue
            except Exception as e:
                Logger.exception(
                    f'Unexpected error resolving output "{output_name}" of video'
                    f" cue {cue.id}; treated as not driven by this node"
                )
                unrouted.append((output_name, f"unexpected {type(e).__name__}: {e}"))
                continue
            routed.append((index, output_name, output))
        return routed, unrouted

    def driven_video_outputs(self) -> list[str]:
        """The outputs this node drives, as ``id=connector``, for log lines."""
        return [
            f"{name}={getattr(output, 'mapped_to', '?')}"
            for name, output in sorted(self._video_outputs.items())
        ]

    def first_unrouted_report(self, cue_id, output_name) -> bool:
        """True the first time a (cue, output) is reported unrouted in this
        load, so the arm logs it at ERROR once and every re-arm (STOP, loop,
        enable, go() fallback) at DEBUG."""
        key = (str(cue_id), output_name)
        with self._lock:
            if key in self._unrouted_warned:
                return False
            self._unrouted_warned.add(key)
            return True

    def reset_unrouted_warnings(self) -> None:
        """Forget which unrouted outputs were reported (called at each load)."""
        with self._lock:
            self._unrouted_warned = set()

    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 is_layer_registered(self, layer_id: str) -> bool:
        """True while the layer is tracked as loaded in the videocomposer.

        reset_videocomposer / reset_video_layers / quit_videocomposer clear
        the tracking together with the layers' endpoints.
        """
        with self._lock:
            return layer_id in self._loaded_layer_ids

    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 cue_media_dimensions(self, cue) -> tuple[int | None, int | None]:
        """(width, height) px of a video cue's media, as stored in the project.

        The editor stores ``pixel_width`` / ``pixel_height`` and the file's
        ``file_size`` in the cue's ``Media`` when the file is uploaded, as it
        stores the duration, so arming a video cue needs no ``ffprobe``: a
        cue selection arms the selected chain under the command lock, and a
        GO sent right after it waited for the probes (869fat84r).

        Falls back to :meth:`media_dimensions` (the probe), warning once per
        file, when the values are missing or invalid (a project saved before
        they existed, or by an older editor), or when ``file_size`` differs
        from this node's copy of the file (replaced under the same name).
        Read with ``.get()``: projects are parsed without ``Media``'s
        setters, and an older cuemsutils has none for these keys.
        """
        media = cue.media
        file_name = media["file_name"]
        width, height = media.get("pixel_width"), media.get("pixel_height")
        if not (_positive_int(width) and _positive_int(height)):
            self._warn_media_dims_once(
                ("missing", file_name),
                f"No stored dimensions for {file_name} in the project; probing it, "
                "which delays the first arm of this file. A current editor stores "
                "them when the project is saved.",
            )
            return self.media_dimensions(file_name)
        stored_size = media.get("file_size")
        if _positive_int(stored_size):
            try:
                actual_size = os.stat(self.media_path(file_name)).st_size
            except OSError:
                # A missing file is reported by the layer load, not here.
                actual_size = None
            if actual_size is not None and actual_size != stored_size:
                self._warn_media_dims_once(
                    ("size", file_name, stored_size, actual_size),
                    f"Stored dimensions of {file_name} do not match the file on this "
                    f"node (file_size {stored_size} in the project, {actual_size} on "
                    "disk): it was replaced or not synced; probing it.",
                )
                return self.media_dimensions(file_name)
        return (width, height)

    def _warn_media_dims_once(self, key: tuple, message: str) -> None:
        with self._lock:
            if key in self._media_dims_warned:
                return
            self._media_dims_warned.add(key)
        Logger.warning(message)

    def media_dimensions(self, file_name: str) -> tuple[int | None, int | None]:
        """Return (width, height) px of the media's first video stream via
        ffprobe, cached by (path, st_mtime). Returns (None, None) on any
        failure (ffprobe missing/error/timeout, non-video media, file gone),
        in which case callers fall back to the legacy region-ratio scale.
        """
        file_path = self.media_path(file_name)
        try:
            key = (file_path, os.stat(file_path).st_mtime)
        except OSError:
            return (None, None)
        with self._lock:
            if key in self._media_dims_cache:
                return self._media_dims_cache[key]
        # Probe OUTSIDE the lock so a slow ffprobe never blocks other arms.
        # Two cues with the same file may both miss and both probe: a benign
        # double-probe (NOT a real race) — leave it, do not "fix" with a lock
        # held across the subprocess.
        dims: tuple[int | None, int | None] = (None, None)
        ffprobe = shutil.which("ffprobe")
        if ffprobe:
            t0 = monotonic()
            try:
                out = subprocess.run(
                    [
                        ffprobe,
                        "-v",
                        "error",
                        "-select_streams",
                        "v:0",
                        "-show_entries",
                        "stream=width,height",
                        "-of",
                        "csv=p=0",
                        file_path,
                    ],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    timeout=5,
                    text=True,
                )
                parts = out.stdout.strip().split(",")
                if out.returncode == 0 and len(parts) == 2:
                    w, h = int(parts[0]), int(parts[1])
                    if w > 0 and h > 0:
                        dims = (w, h)
            except (subprocess.TimeoutExpired, ValueError, OSError) as e:
                Logger.warning(f"ffprobe failed for {file_path}: {e}")
            Logger.debug(f"ffprobe {file_path} -> {dims} in {monotonic() - t0:.3f}s")
        else:
            Logger.warning("ffprobe not found; video layers use legacy region scale")
        with self._lock:
            self._media_dims_cache[key] = dims
        return dims

    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
        cls._instance._media_dims_cache = {}
        # Files whose stored dimensions were missing or stale, already
        # warned about once (cue_media_dimensions).
        cls._instance._media_dims_warned = set()
        # (cue id, output name) pairs whose "output not driven" error was
        # already logged in this load (869fgq3xv); cleared at each load.
        cls._instance._unrouted_warned = set()
    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. Like the orphan sweeps it is hygiene, so it never raises into the load (869evtdf7): a JACK error skips the cleanup with a warning, and audio cues then fail loudly at their own arm.

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. Like
    the orphan sweeps it is hygiene, so it never raises into the load
    (869evtdf7): a JACK error skips the cleanup with a warning, and audio
    cues then fail loudly at their own arm.

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

    try:
        all_ports = self._audio_mixer.conn_man.get_ports(
            pattern="Audio_Player-.*", is_audio=True, is_output=True
        )
    except Exception as e:
        Logger.warning(
            f"JACK not reachable, zombie JACK client cleanup skipped: {e!r}"
        )
        return 0
    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)

cue_media_dimensions(cue)

(width, height) px of a video cue's media, as stored in the project.

The editor stores pixel_width / pixel_height and the file's file_size in the cue's Media when the file is uploaded, as it stores the duration, so arming a video cue needs no ffprobe: a cue selection arms the selected chain under the command lock, and a GO sent right after it waited for the probes (869fat84r).

Falls back to :meth:media_dimensions (the probe), warning once per file, when the values are missing or invalid (a project saved before they existed, or by an older editor), or when file_size differs from this node's copy of the file (replaced under the same name). Read with .get(): projects are parsed without Media's setters, and an older cuemsutils has none for these keys.

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

def cue_media_dimensions(self, cue) -> tuple[int | None, int | None]:
    """(width, height) px of a video cue's media, as stored in the project.

    The editor stores ``pixel_width`` / ``pixel_height`` and the file's
    ``file_size`` in the cue's ``Media`` when the file is uploaded, as it
    stores the duration, so arming a video cue needs no ``ffprobe``: a
    cue selection arms the selected chain under the command lock, and a
    GO sent right after it waited for the probes (869fat84r).

    Falls back to :meth:`media_dimensions` (the probe), warning once per
    file, when the values are missing or invalid (a project saved before
    they existed, or by an older editor), or when ``file_size`` differs
    from this node's copy of the file (replaced under the same name).
    Read with ``.get()``: projects are parsed without ``Media``'s
    setters, and an older cuemsutils has none for these keys.
    """
    media = cue.media
    file_name = media["file_name"]
    width, height = media.get("pixel_width"), media.get("pixel_height")
    if not (_positive_int(width) and _positive_int(height)):
        self._warn_media_dims_once(
            ("missing", file_name),
            f"No stored dimensions for {file_name} in the project; probing it, "
            "which delays the first arm of this file. A current editor stores "
            "them when the project is saved.",
        )
        return self.media_dimensions(file_name)
    stored_size = media.get("file_size")
    if _positive_int(stored_size):
        try:
            actual_size = os.stat(self.media_path(file_name)).st_size
        except OSError:
            # A missing file is reported by the layer load, not here.
            actual_size = None
        if actual_size is not None and actual_size != stored_size:
            self._warn_media_dims_once(
                ("size", file_name, stored_size, actual_size),
                f"Stored dimensions of {file_name} do not match the file on this "
                f"node (file_size {stored_size} in the project, {actual_size} on "
                "disk): it was replaced or not synced; probing it.",
            )
            return self.media_dimensions(file_name)
    return (width, height)

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)

driven_video_outputs()

The outputs this node drives, as id=connector, for log lines.

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

def driven_video_outputs(self) -> list[str]:
    """The outputs this node drives, as ``id=connector``, for log lines."""
    return [
        f"{name}={getattr(output, 'mapped_to', '?')}"
        for name, output in sorted(self._video_outputs.items())
    ]

first_unrouted_report(cue_id, output_name)

True the first time a (cue, output) is reported unrouted in this load, so the arm logs it at ERROR once and every re-arm (STOP, loop, enable, go() fallback) at DEBUG.

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

def first_unrouted_report(self, cue_id, output_name) -> bool:
    """True the first time a (cue, output) is reported unrouted in this
    load, so the arm logs it at ERROR once and every re-arm (STOP, loop,
    enable, go() fallback) at DEBUG."""
    key = (str(cue_id), output_name)
    with self._lock:
        if key in self._unrouted_warned:
            return False
        self._unrouted_warned.add(key)
        return True

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
str | None	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]

is_layer_registered(layer_id)

True while the layer is tracked as loaded in the videocomposer.

reset_videocomposer / reset_video_layers / quit_videocomposer clear the tracking together with the layers' endpoints.

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

def is_layer_registered(self, layer_id: str) -> bool:
    """True while the layer is tracked as loaded in the videocomposer.

    reset_videocomposer / reset_video_layers / quit_videocomposer clear
    the tracking together with the layers' endpoints.
    """
    with self._lock:
        return layer_id in self._loaded_layer_ids

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.

Runs in the middle of a project load, so it never raises (869evtdf7).

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.

    Runs in the middle of a project load, so it never raises (869evtdf7).
    """
    pids = _orphan_pids(AUDIOPLAYER_ORPHAN_PATTERN, "audioplayer")
    if not pids:
        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 in pids:
        if pid not in tracked_pids:
            _kill_orphan(pid, "audioplayer")

kill_orphaned_mixer_processes()

Kill jack-volume mixer processes not tracked by this engine.

Mirrors kill_orphaned_audio_processes for the mixer. On a node-engine restart the previous jack-volume survives as an independent subprocess and keeps the "0_mixer" JACK client name; a freshly spawned jack-volume then hits JackNameNotUnique and is renamed by JACK, ending up a disconnected zombie while the engine still addresses "0_mixer" (the orphan) — every restart also leaks another zombie. Kill any survivor first so the new mixer binds "0_mixer" cleanly.

Match "jack-volume -c" (the binary invoked with its client-name flag) rather than the bare word, so an operator's tail -f .../jack-volume.log or a grep isn't caught and killed. Covers both the plain and "cuems-jack-volume" install names.

Never raises (869evtdf7): an orphan it cannot kill used to stop the mixer from starting, leaving every audio cue on the node silent.

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

def kill_orphaned_mixer_processes(self):
    """Kill jack-volume mixer processes not tracked by this engine.

    Mirrors kill_orphaned_audio_processes for the mixer. On a node-engine
    restart the previous jack-volume survives as an independent subprocess
    and keeps the "0_mixer" JACK client name; a freshly spawned jack-volume
    then hits JackNameNotUnique and is renamed by JACK, ending up a
    disconnected zombie while the engine still addresses "0_mixer" (the
    orphan) — every restart also leaks another zombie. Kill any survivor
    first so the new mixer binds "0_mixer" cleanly.

    Match "jack-volume -c" (the binary invoked with its client-name flag)
    rather than the bare word, so an operator's `tail -f .../jack-volume.log`
    or a grep isn't caught and killed. Covers both the plain and
    "cuems-jack-volume" install names.

    Never raises (869evtdf7): an orphan it cannot kill used to stop the
    mixer from starting, leaving every audio cue on the node silent.
    """
    pids = _orphan_pids(MIXER_ORPHAN_PATTERN, "jack-volume mixer")
    if not pids:
        return

    tracked_pid = None
    if self._audio_mixer is not None and getattr(self._audio_mixer, "p", None):
        tracked_pid = self._audio_mixer.p.pid

    for pid in pids:
        if pid != tracked_pid:
            _kill_orphan(pid, "jack-volume mixer")

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_dimensions(file_name)

Return (width, height) px of the media's first video stream via ffprobe, cached by (path, st_mtime). Returns (None, None) on any failure (ffprobe missing/error/timeout, non-video media, file gone), in which case callers fall back to the legacy region-ratio scale.

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

def media_dimensions(self, file_name: str) -> tuple[int | None, int | None]:
    """Return (width, height) px of the media's first video stream via
    ffprobe, cached by (path, st_mtime). Returns (None, None) on any
    failure (ffprobe missing/error/timeout, non-video media, file gone),
    in which case callers fall back to the legacy region-ratio scale.
    """
    file_path = self.media_path(file_name)
    try:
        key = (file_path, os.stat(file_path).st_mtime)
    except OSError:
        return (None, None)
    with self._lock:
        if key in self._media_dims_cache:
            return self._media_dims_cache[key]
    # Probe OUTSIDE the lock so a slow ffprobe never blocks other arms.
    # Two cues with the same file may both miss and both probe: a benign
    # double-probe (NOT a real race) — leave it, do not "fix" with a lock
    # held across the subprocess.
    dims: tuple[int | None, int | None] = (None, None)
    ffprobe = shutil.which("ffprobe")
    if ffprobe:
        t0 = monotonic()
        try:
            out = subprocess.run(
                [
                    ffprobe,
                    "-v",
                    "error",
                    "-select_streams",
                    "v:0",
                    "-show_entries",
                    "stream=width,height",
                    "-of",
                    "csv=p=0",
                    file_path,
                ],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=5,
                text=True,
            )
            parts = out.stdout.strip().split(",")
            if out.returncode == 0 and len(parts) == 2:
                w, h = int(parts[0]), int(parts[1])
                if w > 0 and h > 0:
                    dims = (w, h)
        except (subprocess.TimeoutExpired, ValueError, OSError) as e:
            Logger.warning(f"ffprobe failed for {file_path}: {e}")
        Logger.debug(f"ffprobe {file_path} -> {dims} in {monotonic() - t0:.3f}s")
    else:
        Logger.warning("ffprobe not found; video layers use legacy region scale")
    with self._lock:
        self._media_dims_cache[key] = dims
    return dims

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'
                        f"a JACK port"
                    )

        if not selected_outputs:
            Logger.warning(
                f"No valid audio outputs resolved for cue {cue.id},"
                f"skipping mixer connection"
            )
        else:
            Logger.info(f"Connecting {player_name} to outputs: {selected_outputs}")
            ready = getattr(player, "ready", None)
            if ready is None:
                # Every Player has one. Without it the wiring would fall
                # back to port presence and the doubled-audio race (869fcvz85)
                # would come back unnoticed.
                Logger.warning(
                    f"Audio player {player_name} has no ready event - wiring "
                    "on port presence (the auto-connect race is possible)"
                )
            connected = self._audio_mixer.connect_player_to_outputs(
                player_name=player_name,
                player_output_prefix="outport",
                selected_outputs=selected_outputs,
                # A STOP/load may kill this player while we wait for its
                # ports; they will never register then. Stop waiting as
                # soon as the process is gone instead of holding this
                # cue's arm for the full ~15 s (869f9wqpn).
                should_abort=lambda: _process_exited(player),
                # Wire only once the player reports itself started: its
                # ports exist earlier, before RtAudio's auto-connect,
                # which an earlier wiring would leave in place (869fbyjzx).
                ready=ready,
            )
            if connected is False:
                # Route to the mixer failed: the cue would show armed/green
                # in the UI but produce no sound. Surface it loudly rather
                # than proceeding silently. (Not raised on purpose: aborting
                # the arm here could disrupt an in-progress GO chain.)
                Logger.error(
                    f"Audio cue {cue.id} failed to route {player_name} to the mixer"
                    " - it will be SILENT despite showing armed status "
                    "(player ports may not have registered in time, or "
                    "mixer inputs are missing)."
                )

partition_cue_outputs(cue, output_names)

Split a video cue's outputs into the ones this node drives and the ones it doesn't (869fgq3xv).

Returns (routed, unrouted): routed holds (index, output_name, VideoOutput) with the output's position in output_names (layer ids are built from it), unrouted holds (output_name, reason).

An output is unrouted when it does not resolve: an alias this node dropped at start (no region in display.conf, DISPLAY_MISMATCH), a custom region with no VideoCueOutput, no canvas at all. Anything else the resolver raises (a malformed custom region, cue.outputs None) is unrouted too, with its traceback: with no resolved output there is no right place to paint the layer, and the videocomposer's default placement is the whole canvas, i.e. the other screens.

The one function behind both the arm and the load-time preflight, so the two can never disagree. Never raises.

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

def partition_cue_outputs(
    self, cue, output_names: list
) -> tuple[list[tuple[int, str, VideoOutput]], list[tuple[str, str]]]:
    """Split a video cue's outputs into the ones this node drives and the
    ones it doesn't (869fgq3xv).

    Returns ``(routed, unrouted)``: ``routed`` holds ``(index, output_name,
    VideoOutput)`` with the output's position in ``output_names`` (layer
    ids are built from it), ``unrouted`` holds ``(output_name, reason)``.

    An output is unrouted when it does not resolve: an alias this node
    dropped at start (no region in display.conf, DISPLAY_MISMATCH), a
    custom region with no VideoCueOutput, no canvas at all. Anything else
    the resolver raises (a malformed custom region, cue.outputs None) is
    unrouted too, with its traceback: with no resolved output there is no
    right place to paint the layer, and the videocomposer's default
    placement is the whole canvas, i.e. the other screens.

    The one function behind both the arm and the load-time preflight, so
    the two can never disagree. Never raises.
    """
    routed, unrouted = [], []
    for index, output_name in enumerate(output_names):
        try:
            output = self.resolve_video_output_for_cue(cue, output_name)
        except (KeyError, RuntimeError) as e:
            unrouted.append((output_name, f"{type(e).__name__}: {e}"))
            continue
        except Exception as e:
            Logger.exception(
                f'Unexpected error resolving output "{output_name}" of video'
                f" cue {cue.id}; treated as not driven by this node"
            )
            unrouted.append((output_name, f"unexpected {type(e).__name__}: {e}"))
            continue
        routed.append((index, output_name, output))
    return routed, unrouted

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()
        self._media_dims_cache = {}
        self._media_dims_warned = set()
        self._unrouted_warned = set()

reset_unrouted_warnings()

Forget which unrouted outputs were reported (called at each load).

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

def reset_unrouted_warnings(self) -> None:
    """Forget which unrouted outputs were reported (called at each load)."""
    with self._lock:
        self._unrouted_warned = set()

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[tuple[AudioPlayer, AudioClient]](
        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_name)

Construct (or replace) the GradientClient for this node.

node_name must match gradient-motiond's own --node-name (defaults to the OS hostname; see node-identity-contract.md in cuems-common) — NOT the node's UUID, which the daemon's node_name filter will never match, silently dropping every message.

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_name: str) -> None:
    """Construct (or replace) the GradientClient for this node.

    node_name must match gradient-motiond's own --node-name (defaults to
    the OS hostname; see node-identity-contract.md in cuems-common) —
    NOT the node's UUID, which the daemon's node_name filter will never
    match, silently dropping every message.

    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.
    """
    self._gradient_client = GradientClient(
        host="127.0.0.1",
        port=port,
        node_name=node_name,
    )
    Logger.info(f"GradientClient: bound to 127.0.0.1:{port} node_name={node_name}")

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}")
    # Clean up any surviving jack-volume from a prior engine lifetime so the
    # new mixer binds "0_mixer" cleanly (avoids the JackNameNotUnique rename
    # that would leave the engine addressing a disconnected zombie).
    self.kill_orphaned_mixer_processes()
    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"
                f"per-output "
                f"bounding box {bbox_w}x{bbox_h}; monitors would be"
                f"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

check_orphan_binary_names(node_conf)

Warn about configured player binaries the orphan cleanups cannot see.

Returns the settings keys whose path basename is not a known name. A missing key is skipped: not every node configures every player.

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

def check_orphan_binary_names(node_conf: dict) -> list[str]:
    """Warn about configured player binaries the orphan cleanups cannot see.

    Returns the settings keys whose path basename is not a known name. A
    missing key is skipped: not every node configures every player.
    """
    unknown = []
    for key, names in ORPHAN_BINARY_NAMES.items():
        entry = node_conf.get(key) if isinstance(node_conf, dict) else None
        path = entry.get("path") if isinstance(entry, dict) else None
        if not path:
            continue
        base = os.path.basename(str(path))
        if base not in names:
            unknown.append(key)
            Logger.warning(
                f"{key}.path basename '{base}' is not one the orphan cleanup "
                f"knows ({', '.join(names)}): orphaned {key} processes from an "
                f"earlier engine run will not be cleaned up"
            )
    return unknown

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_name 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.

node_name MUST match the value gradient-motiond filters on (its --node-name CLI flag, defaulting to the OS hostname — see node-identity-contract.md in cuems-common). It is NOT the node's UUID: the daemon's OscServer drops any message whose node_name field doesn't match byte-for-byte, silently (DEBUG-only log), so a UUID never matches and every send is a no-op.

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

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

    Holds node_name 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.

    node_name MUST match the value gradient-motiond filters on (its
    --node-name CLI flag, defaulting to the OS hostname — see
    node-identity-contract.md in cuems-common). It is NOT the node's UUID:
    the daemon's OscServer drops any message whose node_name field doesn't
    match byte-for-byte, silently (DEBUG-only log), so a UUID never
    matches and every send is a no-op.
    """

    def __init__(
        self, host: str = "127.0.0.1", port: int = 7100, node_name: str = ""
    ) -> None:
        self._host = host
        self._port = port
        self._node_name = node_name
        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")
        # node_name — self-injected
        builder.add_arg(self._node_name, arg_type="s")
        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, self._node_name)
            )
        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", self._node_name)
        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} ->"
                        f"{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:"
                        f"{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:"
                        f"{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'

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} ->"
                    f"{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:"
                    f"{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:"
                    f"{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