API Documentation

This API is still in development and may change without notice.

ControllerEngine

Bases: BaseEngine

The main engine class for the CUEMS system.

An object of this class runs all the inner logical part of communications with

• The WebSocket system
• The Ossia System
• The MTC System
• The NodeEngine local and remote instances
• The NNG communication system

It is responsible for

• Monitoring the NodeEngine local and remote instances
• Restarting the NodeEngine local and remote instances
• Updating the NodeEngine local and remote instances
• Handling the NodeEngine local and remote instances failures
• Handling the NNG communication system
• Handling the WebSocket system
• Handling the Ossia System
• Handling the MTC master system
• Handling the NodeConf system

Source code in src/cuemsengine/ControllerEngine.py

class ControllerEngine(BaseEngine):
    '''
    The main engine class for the CUEMS system.

    An object of this class runs all the inner logical part of communications with:
      - The WebSocket system
      - The Ossia System
      - The MTC System
      - The NodeEngine local and remote instances
      - The NNG communication system

    It is responsible for:
      - Monitoring the NodeEngine local and remote instances
      - Restarting the NodeEngine local and remote instances
      - Updating the NodeEngine local and remote instances
      - Handling the NodeEngine local and remote instances failures
      - Handling the NNG communication system
      - Handling the WebSocket system
      - Handling the Ossia System
      - Handling the MTC master system
      - Handling the NodeConf system
    '''
    # Controller→UI WebSocket throttle for cue percentage updates.
    # State transitions (0, 1, 100) always bypass this and broadcast immediately.
    # Only in-progress percentage values (2-99) are throttled.
    # Two-tier throttle: Tier 1 is node-side (CUE_STATUS_UPDATE_HZ in loop_cue.py);
    # Tier 2 is here, capping WS broadcasts even when multiple nodes send updates
    # in quick succession.
    CUE_BROADCAST_MIN_INTERVAL = 0.25  # seconds — max 4 Hz to UI per cue

    def __init__(self, **kwargs):
        # Must be set before super().__init__() because BaseEngine sets
        # self.timecode = None which triggers on_timecode_change() via the
        # property setter, and that method reads these attributes.
        self._last_timecode_second: int = -1  # last whole-second value broadcast to UI
        # Per-cue status dict: maps cue uuid → int status value.
        # Values: 0=unplayed, 1-99=playing (1 until percentage enabled), 100=played, -1=error
        self.cue_status: dict[str, int] = {}
        # Per-cue enabled status: maps cue uuid → bool.
        # Initialised from XML on load_project, updated by show-time toggles.
        # Resets to XML values on reload; persists across stop/go.
        self.cue_enabled_status: dict[str, bool] = {}
        # Per-cue last-broadcast timestamps for WS throttle (Tier 2).
        self._cue_broadcast_timestamps: dict[str, float] = {}
        # Per-mixer-channel volume state: maps "{node_uuid}/{output_index}/{channel}"
        # to float 0.0-1.0. Channel is "master" or a stringified index. Persists
        # across project loads; resets only on engine restart. All access is on
        # the asyncio event loop (WS receive handler + _on_ws_client_connect),
        # so no lock is needed — keep it that way.
        self.mixer_status: dict[str, float] = {}

        # Cluster-state tracking. Populated at load time by _resolve_cluster_state
        # (chunk 3); used to gate the armed=yes flip on all required nodes having
        # reported armed_ready. _pong_responses is populated by the comms thread
        # via status_operation_callback as pong replies arrive. All set mutations
        # and the read in _probe_cluster_liveness happen under _cluster_lock.
        self._adopted_nodes: set[str] = set()
        self._required_nodes: set[str] = set()
        self._armed_nodes: set[str] = set()
        self._finished_nodes: set[str] = set()
        self._pong_responses: set[str] = set()
        self._pong_expected: set[str] = set()
        self._pong_event = threading.Event()
        self._cluster_lock = threading.Lock()
        # One-shot watchdog: fires N seconds after _resolve_cluster_state if
        # _armed_nodes still doesn't cover _required_nodes (e.g. a node ponged
        # alive but then died mid-rsync). Logs an error listing the pending
        # nodes. Does NOT force armed=yes — operator decides.
        self._arm_watchdog: threading.Timer | None = None

        super().__init__(**kwargs)
        self.set_editor_request('')
        self.set_node_operation_callback()

    def start(self):
        self.create_timecode()
        self.set_comms()
        # Always re-detect after create_timecode(): the MtcMaster sender port
        # ("MtcMaster:MTCPort") only appears in the ALSA port list AFTER the
        # sender is created.  Connecting the listener directly to that port is
        # the most reliable loopback path; any earlier detection would have
        # picked a wrong/fallback port (e.g. rtpmidid:Announcements).
        Logger.info('Re-detecting MIDI port after MTC sender creation...')
        self.mtc_listener._MtcListener__open_port(None)
        self.mtc_listener.start()
        super().start()

    def set_status(self, property: str, value: str, strict: bool = False) -> None:
        """Set status and push to UI via WebSocket when running, armed, or load."""
        super().set_status(property, value, strict)
        if property in ('running', 'armed', 'load', 'nextcue'):
            self._broadcast_status(property, value)

    @logged
    def set_comms(self):
        # Start communicators with WebSocket handler on port 9190
        self.set_communicators()

    def set_communicators(self):
        Logger.info('Setting up Communicators')

        # Get OSC hub host from ConfigManager or use default
        if hasattr(self, 'cm') and self.cm:
            osc_hub_host = self.cm.controller_url
        else:
            osc_hub_host = CONTROLLER_HOST

        # Get NNG hub port from config (must match NodeEngine)
        if hasattr(self, 'cm') and self.cm and hasattr(self.cm, 'node_conf'):
            nng_hub_port = self.cm.node_conf.get('nng_hub_port', 9093)
            # Use port 9190 for WebSocket OSC - we start BEFORE pyossia to claim this port
            # This allows UI to send commands via Apache's /realtime proxy to ws://127.0.0.1:9190
            websocket_osc_port = self.cm.node_conf.get('oscquery_ws_port', 9190)
            node_id = self.cm.node_conf.get('uuid', 'controller')
        else:
            nng_hub_port = 9093
            websocket_osc_port = 9190  # Take port 9190 for WebSocket OSC
            node_id = 'controller'

        # LISTENER binds to all interfaces (0.0.0.0) so it does not depend on the
        # avahi link-local address (169.254.x.x) being assigned before startup.
        # NodeEngine (DIALER) still targets the specific controller_url IP.
        nng_hub_address = f"tcp://0.0.0.0:{nng_hub_port}"

        Logger.info(f'NNG Hub address: {nng_hub_address}')

        # WebSocket OSC configuration for receiving commands from UI
        # Uses port 9190 (same as Apache /realtime proxy target) to receive
        # OSC commands directly. Started BEFORE pyossia to claim the port.
        websocket_osc_config = {
            'host': '0.0.0.0',
            'port': websocket_osc_port,
            'node_id': node_id
        }
        Logger.info(f'WebSocket OSC port: {websocket_osc_port}')

        self.communications_thread = ControllerCommunications(
            nng_hub_address=nng_hub_address,
            editor_callback=self.editor_command_callback,
            node_operation_callback=self.node_operation_callback,
            websocket_osc_config=websocket_osc_config
        )

        # Register command handlers for WebSocket OSC
        self._register_osc_command_handlers()
        self.communications_thread.set_on_client_connect(self._on_ws_client_connect)

        self.communications_thread.start()

        # Wait for NNG thread to initialize (prevents race condition in nni_random)
        from time import sleep
        max_wait = 5.0  # seconds
        wait_interval = 0.1
        waited = 0.0
        while waited < max_wait:
            if (self.communications_thread.is_alive() and 
                self.communications_thread.event_loop is not None):
                Logger.info(f"NNG communications thread ready after {waited:.1f}s")
                break
            sleep(wait_interval)
            waited += wait_interval
        else:
            Logger.warning(f"NNG communications thread not ready after {max_wait}s")

    def _register_osc_command_handlers(self):
        """Register OSC command handlers for WebSocket OSC receiving.

        These handlers are called when commands are received from the UI via
        WebSocket OSC. Commands are also forwarded to NodeEngine via NNG.
        """
        # Command handlers - same as used in _command_poll_loop
        self.communications_thread.register_command_handler(
            '/engine/command/go', self.go_script, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            '/engine/command/load', self.deploy_project, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            '/engine/command/stop', self.stop_script, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            '/engine/command/setnextcue', self._setnextcue_handler, forward_to_nodes=False
        )
        self.communications_thread.register_command_handler(
            '/engine/command/cue_enabled', self._cue_enabled_handler, forward_to_nodes=False
        )

        # Register wildcard handler for player messages (engine format)
        self.communications_thread.register_osc_handler(
            '/engine/players/*', self._handle_player_osc_message
        )

        # Register direct player handler for every adopted node in the network map.
        # UI sends /{node_uuid}/<type>/... for both controller and worker nodes;
        # without per-node registration the WS dispatcher silently drops the
        # message and the NNG forward never happens.
        # The set deduplicates so the controller's own UUID isn't registered
        # twice (it appears in both node_conf and network_map['node_list']).
        node_uuids: set[str] = set()
        own_uuid = self.cm.node_conf.get('uuid', '') if hasattr(self, 'cm') and self.cm else ''
        if own_uuid:
            node_uuids.add(own_uuid)
        try:
            if self.cm and self.cm.network_map:
                adopted, _new = NetworkMap.get_nodes_by_adoption(self.cm.network_map)
                for entry in adopted:
                    nuuid = (entry.get('node') or {}).get('uuid')
                    if nuuid:
                        node_uuids.add(nuuid)
        except Exception as e:
            Logger.warning(f"Could not enumerate node UUIDs from network_map: {e}")

        for nuuid in node_uuids:
            self.communications_thread.register_osc_handler(
                f'/{nuuid}/*', self._handle_direct_player_osc_message
            )
            Logger.info(f"Registered direct player OSC handler for /{nuuid}/*")

        Logger.info("OSC command handlers registered for WebSocket receiving")

    def _handle_direct_player_osc_message(self, address: str, args: list):
        """Handle direct player OSC messages from UI (/<node_uuid>/<type>/...).

        These are forwarded directly to the local node's player handlers.
        """
        value = args[0] if args else None

        # Parse: /<node_uuid>/<type>/<...>
        parts = address.strip('/').split('/')
        if len(parts) < 2:
            Logger.warning(f"Invalid direct player OSC address: {address}")
            return

        # parts[0] is node_uuid, parts[1] is type (audiomixer, jadeo, etc.)
        player_type = parts[1]

        Logger.debug(f"Direct player OSC: {address} = {repr(value)}")

        # Shadow audio mixer volume so /realtime clients can recover state on
        # reconnect or page reload. Address shape:
        #   /{node_uuid}/audio/mixer/{output_index}/{channel}/volume
        # NaN/Inf are dropped to avoid polluting the dict.
        if (len(parts) >= 6
                and parts[1] == 'audio' and parts[2] == 'mixer'
                and parts[-1] == 'volume'
                and value is not None):
            try:
                vol = float(value)
            except (TypeError, ValueError):
                vol = None
            if vol is not None and math.isfinite(vol):
                node_uuid_part, output_index, channel = parts[0], parts[3], parts[4]
                key = f'{node_uuid_part}/{output_index}/{channel}'
                self.mixer_status[key] = vol
                self._broadcast_status(
                    f'audio/mixer/{node_uuid_part}/{output_index}/{channel}/volume',
                    vol,
                )

        # Forward to NodeEngine via NNG as player_control
        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=self.cm.node_conf.get('uuid', 'controller') if hasattr(self, 'cm') and self.cm else 'controller',
            target='player_control',
            data={'address': address, 'value': value}
        )

        try:
            import asyncio
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop
            )
            Logger.debug(f"Forwarded direct player OSC to nodes: {address} = {repr(value)}")
        except Exception as e:
            Logger.error(f"Error forwarding direct player OSC to nodes: {e}")

    def _handle_player_osc_message(self, address: str, args: list):
        """Handle player-related OSC messages from UI.

        These are forwarded to NodeEngine via NNG for player control
        (video, audio mixer, DMX, etc.)
        """
        # Forward to NodeEngine via NNG
        value = args[0] if args else None

        # Create a COMMAND operation for player control
        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=self.cm.node_conf.get('uuid', 'controller') if hasattr(self, 'cm') and self.cm else 'controller',
            target='player_control',
            data={'address': address, 'value': value}
        )

        try:
            import asyncio
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop
            )
            Logger.debug(f"Forwarded player OSC to nodes: {address} = {repr(value)}")
        except Exception as e:
            Logger.error(f"Error forwarding player OSC to nodes: {e}")

    def _forward_load_to_nodes(self, project_name: str) -> None:
        """Forward a load command to NodeEngine via NNG."""
        self._forward_command_to_nodes('/engine/command/load', project_name)

    def stop(self):
        self.stop_comms()
        super().stop()

    @logged
    def stop_comms(self):
        if self.with_mtc:
            self.stop_timecode()
        if hasattr(self, 'communications_thread'):
            self.communications_thread.stop()

    #########################
    # Timecode
    #########################
    def create_timecode(self):
        if self.with_mtc:
            self.mtcmaster = libmtcmaster.MTCSender_create()
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")

    def start_timecode(self):
        if self.with_mtc:
            libmtcmaster.MTCSender_play(self.mtcmaster)
            Logger.info("Midi TimeCode started.")
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")

    def stop_timecode(self):
        if self.with_mtc:
            libmtcmaster.MTCSender_stop(self.mtcmaster)
            Logger.info("Midi TimeCode stopped.")
        else:
            Logger.info("Midi TimeCode requires with_mtc to be True.")


    #########################
    # Operation callbacks
    #########################
    def set_node_operation_callback(self):
        self.node_operation_callback = {
            OperationType.PLAYER: self.player_operation_callback,
            OperationType.CUE: self.cue_operation_callback,
            OperationType.STATUS: self.status_operation_callback
        }

    def player_operation_callback(self, operation: NodeOperation):
        """
        Callback invoked when players are received from nodes.

        Parameters:
        - operation: NodeOperation with sender, target (player_id), and action
        """
        Logger.info(f'Player operation received: {operation}')

    def cue_operation_callback(self, operation: NodeOperation):
        """Callback invoked when cues are received from nodes.

        Handles three action types:
        - ADD:    cue started playing on a node → status 1, broadcast immediately
        - REMOVE: cue finished playing on a node → status 100, broadcast immediately
        - UPDATE: percentage progress from a node (future) → throttled broadcast
        """
        Logger.info(f'Cue operation received: {operation}')
        cue_id = operation.data.get('id') if operation.data else None

        # Drop operations for cues not belonging to the current project.
        # This prevents stale REMOVE/ADD notifications from the NodeEngine
        # (sent when it disarms the previous project) from being broadcast
        # to the UI as unknown UUIDs.
        if cue_id and cue_id not in self.cue_status:
            Logger.debug(f'Ignoring cue operation for unknown/stale cue_id {cue_id} (action={operation.action})')
            return

        if operation.action == ActionType.ADD:
            # Cue started playing: mark as playing (1) and broadcast immediately.
            if cue_id:
                self.cue_status[cue_id] = 1
                self._broadcast_cue_status(cue_id, 1, force=True)
            try:
                self.status.currentcue = [operation.data['id'], operation.data['offset']]
                Logger.debug(f"Current cue updated: {self.status.currentcue}")
            except Exception as e:
                Logger.error(f'Error updating currentcue: {e}')

        elif operation.action == ActionType.REMOVE:
            # Cue finished playing: mark as played (100) and broadcast immediately.
            # Only transition to 100 if the cue was actually playing (status == 1).
            # REMOVEs that arrive while status is 0 (e.g. NodeEngine disarming the
            # previous project after a reload) are stale and must be silently dropped.
            if cue_id:
                if self.cue_status.get(cue_id) == 1:
                    self.cue_status[cue_id] = 100
                    self._broadcast_cue_status(cue_id, 100, force=True)
                else:
                    Logger.debug(f'Ignoring stale REMOVE for cue {cue_id} (status={self.cue_status.get(cue_id)}, expected 1)')
            self.status.remove_currentcue(operation.data['id'])
            Logger.debug(f"Cue removed from currentcue: {operation.data['id']}")

        elif operation.action == ActionType.UPDATE:
            # Future: percentage progress updates from loop_cue() during playback.
            # Throttled by _broadcast_cue_status (Tier 2 / controller-side).
            # The node-side Tier 1 throttle (CUE_STATUS_UPDATE_HZ) limits NNG traffic.
            if cue_id:
                pct = operation.data.get('percentage', 1)
                self.cue_status[cue_id] = pct
                self._broadcast_cue_status(cue_id, pct)  # throttled
            Logger.debug(f"Cue percentage update: {cue_id} = {operation.data.get('percentage')}")

        else:
            Logger.warning(f'Unknown cue action: {operation.action}')

    def status_operation_callback(self, operation: NodeOperation):
        """Callback invoked when status updates are received from nodes.

        Handles script_finished, armed_ready, nextcue, cue_enabled, and pong
        notifications.
        """
        Logger.info(f'Status operation received: {operation}')
        if operation.target == 'pong':
            # Cluster liveness probe reply. Runs on the AsyncCommsThread event
            # loop; _probe_cluster_liveness on the main thread waits on the event.
            with self._cluster_lock:
                self._pong_responses.add(operation.sender)
                if self._pong_responses >= self._pong_expected:
                    self._pong_event.set()
            Logger.debug(f'Pong from {operation.sender}')
            return
        if operation.target == 'script_finished':
            if operation.data and operation.data.get('running') == 'no':
                # Aggregate per-node: only flip running=no when all required
                # nodes have reported. Filter foreign senders to keep the
                # tracker bounded.
                sender = operation.sender
                if sender not in self._adopted_nodes:
                    Logger.debug(
                        f'Ignoring script_finished from non-adopted node {sender}'
                    )
                    return
                with self._cluster_lock:
                    self._finished_nodes.add(sender)
                    finished_now = set(self._finished_nodes)
                    required = set(self._required_nodes)
                Logger.info(
                    f'Node {sender} script_finished '
                    f'({len(finished_now)}/{len(required)})'
                )
                if finished_now >= required and self.get_status('running') == 'yes':
                    Logger.info('All required nodes finished — updating running status')
                    self.set_status('running', 'no')
        elif operation.target == 'armed_ready':
            if operation.data and operation.data.get('armed') == 'yes':
                # Aggregate per-node: only flip armed=yes when all required
                # nodes have reported. Filter foreign senders to keep the
                # tracker bounded.
                sender = operation.sender
                if sender not in self._adopted_nodes:
                    Logger.debug(
                        f'Ignoring armed_ready from non-adopted node {sender}'
                    )
                    return
                with self._cluster_lock:
                    self._armed_nodes.add(sender)
                    armed_now = set(self._armed_nodes)
                    required = set(self._required_nodes)
                Logger.info(
                    f'Node {sender} armed ({len(armed_now)}/{len(required)})'
                )
                if armed_now >= required and self.get_status('armed') != 'yes':
                    if self.go_offset is None:
                        Logger.info(
                            'Re-arm after stop complete — restarting timecode and enabling GO'
                        )
                        self.start_timecode()
                        self.go_offset = 0
                    else:
                        Logger.info('All required nodes armed — enabling GO')
                    self.set_status('armed', 'yes')
                    self._cancel_arm_watchdog()
        elif operation.target == 'nextcue':
            nextcue_id = operation.data.get('nextcue', '') if operation.data else ''
            self.set_status('nextcue', nextcue_id)
            Logger.info(f'Next cue updated: {nextcue_id or "(none)"}')
        elif operation.target == 'cue_enabled':
            cue_id = operation.data.get('cue_id') if operation.data else None
            enabled = operation.data.get('enabled', True) if operation.data else True
            if cue_id and cue_id in self.cue_enabled_status:
                self.cue_enabled_status[cue_id] = enabled
                self._broadcast_cue_enabled(cue_id, enabled)
                Logger.info(f'Cue {cue_id} enabled status updated from node: {enabled}')
        else:
            Logger.debug(f'Unknown status target: {operation.target}')

    #########################
    # Editor commands
    #########################

    def editor_command_callback(self, item: dict, context):
        Logger.debug(f'Received editor command: {item}, with context: {context}')
        _item_keys = item.keys()
        if 'value' not in _item_keys:
            item['value'] = ''
        if 'action_uuid' not in _item_keys:
            self.error_to_editor(context, "No action uuid submitted")
        self.set_editor_request(item['action_uuid'])

        if 'type' in _item_keys:
            if item['type'] not in ['error', 'initial_settings']:

                self.set_editor_request('')
            self.error_to_editor(context, "Response not recognized")

        try:
            self.handle_editor_command(
                action = item['action'],
                value = item['value'], 
                context = context
            ) 
        except Exception as e:
            Logger.error(f'{type(e)} handling editor command: {e}')

            request_uuid = self.get_editor_request()
            self.set_editor_request('')
            self.error_to_editor(context, value=f"Command {type(e)}: {e}", request_uuid=request_uuid)

    def handle_editor_command(self, action, value, context=None):
        command_dict = {
            'project_deploy': partial(self.load_project, deploy_only=True),
            'project_ready': self.load_project,
            'hw_discovery': self.hwdiscovery,
            'nodeconf': self.nodeconf,
            'go_script': self.go_script,
            'project_status': self.get_project_status,
            'project_unload': self.unload_project,
        }
        if action in command_dict.keys():
            result = command_dict[action](value, context)
            if result:
                reply_value = result if isinstance(result, dict) else 'OK'
                self.confirm_to_editor(
                    context, type=action, value=reply_value
                )
                # Clear the editor request after successful confirmation
                self.set_editor_request('')

        else:
            raise ValueError(f'Command {action} not recognized')

    def confirm_to_editor(self, context, type=None, value=None):
        return_message={
            'type': type,
            'value': value,
            'action_uuid': self.get_editor_request()
        }
        Logger.debug(f'Sending confirm to editor: {return_message}')

        try:
            self.communications_thread.reply_to_editor(return_message, context)
        except Exception as e:
            Logger.error(f'{type(e)} confirming to editor: {e}')

    def error_to_editor(self, context, value=None, request_uuid = None, action = None):
        if not request_uuid:
            request_uuid = self.get_editor_request()
        return_message={
            'type': 'error',
            'value': value,
            'action_uuid': request_uuid
        }
        if action:
            return_message['action'] = action
        Logger.debug(f'Sending error to editor: {return_message}')
        try:
            self.communications_thread.reply_to_editor(return_message, context)
        except Exception as e:
            Logger.error(f'{type(e)} sending error to editor: {e}')


    def set_editor_request(self, value):
        self._editor_request_uuid = value

    def get_editor_request(self):
        return self._editor_request_uuid


    #########################
    # External services
    #########################

    def hwdiscovery(self, message: dict, context=None) -> bool:
        Logger.debug(f'sending HW discovery request: {message}')
        try:
            reply = self.communications_thread.request_to_hwdiscovery(message)
            Logger.debug(f'Received HW discovery reply: {reply}')
            if 'OK' in reply.values():
                return True
            else:
                return False            
        except Exception as e:
            Logger.error(f'{type(e)} sending HW discovery request: {e}')
            return False

    def nodeconf(self, message: dict, context=None) -> bool:
        Logger.debug(f'sending nodeconf request: {message}')
        try:
            reply = self.communications_thread.request_to_nodeconf(message)
            Logger.debug(f'Received nodeconf reply: {reply}')
            if 'OK' in reply.values():
                return True
            else:
                return False            
        except Exception as e:
            Logger.error(f'{type(e)} sending nodeconf request: {e}')
            return False


    #########################
    # Status Updates (stub - OSCQuery removed)
    #########################

    def set_oscquery_values(self, values: dict):
        """Stub for OSCQuery value setting - OSCQuery server has been removed.

        Status updates are now handled via internal state tracking.
        TODO: Implement WebSocket status push if UI needs real-time status.
        """
        for key, value in values.items():
            Logger.debug(f"Status update (no-op): {key} = {repr(value)}")

    def _collect_project_nodes(self, cuelist) -> set[str]:
        """Walk a cuelist and return the set of node UUIDs referenced by any
        cue's output_name.

        output_name format depends on cue type:
          - video / audio: "<36-char UUID>_<index>" (e.g. "07131798-...-...d18f_0")
          - DMX:           "<UUID>" (no suffix)
        In both cases, output_name[:36] is the node UUID. We validate with a
        cheap UUID-shape check (hyphens at positions 8/13/18/23) so garbage
        output names don't leak non-UUID strings into the set.
        """
        from cuemsutils.cues import CueList
        nodes: set[str] = set()
        if not (hasattr(cuelist, 'contents') and cuelist.contents):
            return nodes
        for item in cuelist.contents:
            if item is None:
                continue
            outputs = getattr(item, 'outputs', None) or []
            for out in outputs:
                if not isinstance(out, dict):
                    continue
                name = out.get('output_name') or ''
                if len(name) >= 36:
                    head = name[:36]
                    if (
                        head[8] == '-' and head[13] == '-'
                        and head[18] == '-' and head[23] == '-'
                    ):
                        nodes.add(head)
            if isinstance(item, CueList):
                nodes.update(self._collect_project_nodes(item))
        return nodes

    def _collect_cue_ids(self, cuelist) -> list[str]:
        """Recursively collect all cue IDs from a cuelist (including nested CueLists)."""
        from cuemsutils.cues import CueList
        ids = []
        if hasattr(cuelist, 'contents') and cuelist.contents:
            for item in cuelist.contents:
                if item is None:
                    continue
                ids.append(item.id)
                if isinstance(item, CueList):
                    ids.extend(self._collect_cue_ids(item))
        return ids

    def _collect_cue_enabled(self, cuelist) -> dict[str, bool]:
        """Recursively collect cue enabled states from a cuelist."""
        from cuemsutils.cues import CueList
        result = {}
        if hasattr(cuelist, 'contents') and cuelist.contents:
            for item in cuelist.contents:
                if item is None:
                    continue
                result[item.id] = item.enabled
                if isinstance(item, CueList):
                    result.update(self._collect_cue_enabled(item))
        return result

    def _broadcast_cue_enabled(self, cue_id: str, enabled: bool) -> None:
        """Broadcast per-cue enabled status to UI at /engine/status/cue_enabled/{uuid}."""
        if hasattr(self, 'communications_thread') and self.communications_thread \
                and hasattr(self.communications_thread, 'broadcast_osc'):
            self.communications_thread.broadcast_osc(
                f'/engine/status/cue_enabled/{cue_id}', 1 if enabled else 0)

    def _broadcast_cue_status(self, cue_id: str, value: int, force: bool = False) -> None:
        """Broadcast per-cue status to UI via WebSocket OSC at /engine/status/cue/{uuid}.

        Values: 0=unplayed, 1-99=playing (1 until percentage is enabled), 100=played, -1=error.

        State transitions (force=True: values 0, 1, 100) bypass throttle and broadcast
        immediately. In-progress percentage updates (2-99) are throttled per-cue to
        CUE_BROADCAST_MIN_INTERVAL to limit WS traffic even when multiple remote nodes
        send updates in quick succession (Tier 2 of the two-tier throttle strategy).
        """
        if not force:
            now = time.monotonic()
            last = self._cue_broadcast_timestamps.get(cue_id, 0)
            if now - last < self.CUE_BROADCAST_MIN_INTERVAL:
                return
            self._cue_broadcast_timestamps[cue_id] = now
        if hasattr(self, 'communications_thread') and self.communications_thread \
                and hasattr(self.communications_thread, 'broadcast_osc'):
            self.communications_thread.broadcast_osc(f'/engine/status/cue/{cue_id}', value)

    def _broadcast_status(self, key: str, value) -> None:
        """Push status to UI via WebSocket OSC (realtime)."""
        if hasattr(self, 'communications_thread') and self.communications_thread and hasattr(self.communications_thread, 'broadcast_osc'):
            self.communications_thread.broadcast_osc(f'/engine/status/{key}', value)

    async def _on_ws_client_connect(self, websocket) -> None:
        """Send full state dump to a newly connected WebSocket client."""
        from .osc.WebSocketOscHandler import build_osc_message

        # Engine status
        for key in ('running', 'armed', 'load', 'nextcue'):
            val = self.get_status(key)
            if val is not None:
                data = build_osc_message(f'/engine/status/{key}', val)
                if data:
                    await websocket.send(data)

        # Per-cue playback status
        for cid, status in self.cue_status.items():
            data = build_osc_message(f'/engine/status/cue/{cid}', status)
            if data:
                await websocket.send(data)

        # Per-cue enabled status
        for cid, enabled in self.cue_enabled_status.items():
            data = build_osc_message(
                f'/engine/status/cue_enabled/{cid}', 1 if enabled else 0)
            if data:
                await websocket.send(data)

        # Per-mixer-channel volume status.
        # Scale note: each entry is one ~80-byte WS message. Acceptable up to
        # ~500 entries (~40 KB / ~500 ms over LAN). If a deployment ever
        # exceeds that (e.g. 8 nodes × 64 channels), switch to a single OSC
        # bundle via build_osc_bundle() instead of per-message sends.
        for key, vol in self.mixer_status.items():
            data = build_osc_message(f'/engine/status/audio/mixer/{key}/volume', vol)
            if data:
                await websocket.send(data)

        Logger.info(f'Late-join state dump sent to new WebSocket client')

    def on_timecode_change(self, value) -> None:
        """Broadcast timecode to UI as integer ms (whole seconds only), once per second."""
        try:
            ms = int(value) if value is not None else 0
        except (TypeError, ValueError):
            return
        current_second = ms // 1000
        if current_second != self._last_timecode_second:
            self._last_timecode_second = current_second
            self._broadcast_status('timecode', current_second * 1000)
            Logger.debug(f'Timecode broadcast {current_second}s')

    def _clear_playback_state(self):
        """Clear runtime playback tracking: timestamps, timecode, armed, nextcue.

        Also clears the per-node armed/finished accumulators. The
        _required_nodes / _adopted_nodes snapshots stay — they belong to
        the loaded project, not playback state, and get recomputed on the
        next load_project. unload_project clears them explicitly.
        """
        self._cue_broadcast_timestamps.clear()
        self._last_timecode_second = -1
        self._broadcast_status('timecode', 0)
        self.set_status('armed', 'no')
        self.set_status('nextcue', '')
        self.stop_timecode()
        with self._cluster_lock:
            self._armed_nodes.clear()
            self._finished_nodes.clear()
        self._cancel_arm_watchdog()

    #########################
    # Project management
    #########################

    def load_project(self, project_name, context=None, deploy_only=False):
        # Don't allow loading while script is running
        if self.get_status('running') == "yes":
            Logger.warning(f'Cannot load project {project_name} while script is running. Stop first.')
            return False

        Logger.info(f'Loading project {project_name}')
        self._clear_playback_state()
        self.reset_script()

        if deploy_only:
            Logger.info(f"Deploy only requested for {project_name}")
            return True

        try:
            self.cm.load_project_config(project_name)
        except Exception as e:
            Logger.error(f'Error loading project config: {e}')

            request_uuid = self.get_editor_request()
            self.set_editor_request('')
            self.error_to_editor(context, 
                f"Project config error: {e}",
                request_uuid=request_uuid,
                action='project_ready'
            )
            return False

        try:
            self.read_script(project_name)
        except Exception as e:
            Logger.error(f'Error loading project script: {e}')

            request_uuid = self.get_editor_request()
            self.set_editor_request('')
            self.error_to_editor(context, 
                f"Project script error: {e}",
                request_uuid=request_uuid,
                action='project_ready'
            )
            return False

        Logger.info(f'Script from {project_name} loaded')
        self.script.unix_name = project_name

        # Initialise per-cue status: every cue starts as unplayed (0).
        # Broadcasts one WS message per cue so the UI can populate its cue list.
        self.cue_status = {cid: 0 for cid in self._collect_cue_ids(self.script.cuelist)}
        for cid in self.cue_status:
            self._broadcast_cue_status(cid, 0, force=True)
        Logger.info(f'Cue status initialised for {len(self.cue_status)} cues')

        # Initialise per-cue enabled status from XML (resets show-time overrides).
        self.cue_enabled_status = self._collect_cue_enabled(self.script.cuelist)
        for cid, enabled in self.cue_enabled_status.items():
            self._broadcast_cue_enabled(cid, enabled)
        Logger.info(f'Cue enabled status initialised for {len(self.cue_enabled_status)} cues')

        # Update internal status
        # TODO: send project UUID instead of name for robustness (would break UI contract)
        self.set_status('load', project_name)

        # Probe cluster, derive _required_nodes for GO gating, refresh <online>
        # in network_map. Done BEFORE _forward_load_to_nodes so the gating set
        # is in place by the time nodes start sending armed_ready.
        self._resolve_cluster_state()

        # Forward load command to NodeEngine via NNG (nodes will arm cues)
        self._forward_load_to_nodes(project_name)

        # Timecode starts on load; runs until next load or engine shutdown
        self.start_timecode()
        self.go_offset = 0  # Enable mtc_callback → on_timecode_change → broadcast
        # armed=yes is NOT set here -- it's set when NodeEngine reports armed_ready
        # via status_operation_callback, ensuring cues are actually armed before
        # the UI shows GO as available

        # Confirm the project is loaded
        self.set_show_lock_file()
        Logger.info(f'Project {project_name} loaded')
        # Note: Don't clear editor_request here - handle_editor_command will clear it after confirmation
        return True

    def deploy_project(self, project_name):
        self.load_project(project_name)

    def go_script(self, value, context=None):
        if self.get_status('armed') != "yes":
            Logger.warning('Cues not armed. GO not available.')
            return

        if not self.script:
            Logger.warning('No script loaded, cannot process GO command.')
            return

        self.set_status('running', "yes")

        # Forward GO to NodeEngine via NNG (needed when called from editor;
        # when called from WebSocket the comms layer also forwards, but the
        # NodeEngine's run_command is idempotent so a double-call is harmless)
        self._forward_command_to_nodes('/engine/command/go', value)

        Logger.info(f'GO command processed')
        return True

    def _setnextcue_handler(self, value):
        """Handle setnextcue from UI — forward to NodeEngine which owns the pointer."""
        self._forward_command_to_nodes('/engine/command/setnextcue', value)

    def _cue_enabled_handler(self, value):
        """Handle cue_enabled toggle from UI.

        Value format: "<cue_id> <0|1>" (space-separated UUID and enabled flag).
        """
        if not value or not isinstance(value, str):
            Logger.warning(f'Invalid cue_enabled value: {repr(value)}')
            return

        parts = value.split(' ', 1)
        if len(parts) != 2 or parts[1] not in ('0', '1'):
            Logger.warning(f'Invalid cue_enabled format (expected "uuid 0|1"): {repr(value)}')
            return

        cue_id, enabled_str = parts
        enabled = enabled_str == '1'

        if cue_id not in self.cue_enabled_status:
            Logger.warning(f'cue_enabled: unknown cue_id {cue_id}')
            return

        self.cue_enabled_status[cue_id] = enabled
        self._broadcast_cue_enabled(cue_id, enabled)
        self._forward_command_to_nodes('/engine/command/cue_enabled', value)
        Logger.info(f'Cue {cue_id} {"enabled" if enabled else "disabled"}')

    def _forward_command_to_nodes(self, address: str, value) -> None:
        """Forward a generic command to NodeEngine via NNG."""
        if not hasattr(self, 'communications_thread') or not self.communications_thread:
            Logger.warning("Cannot forward command to nodes: communications thread not available")
            return

        parts = address.strip('/').split('/')
        command_name = parts[-1] if parts else address

        operation = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=self.cm.node_conf.get('uuid', 'controller') if hasattr(self, 'cm') and self.cm else 'controller',
            target=command_name,
            data={'value': value, 'address': address}
        )

        try:
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(operation),
                self.communications_thread.event_loop
            )
            Logger.debug(f"Forwarded command to nodes: {command_name} = {repr(value)}")
        except Exception as e:
            Logger.error(f"Error forwarding command to nodes: {e}")

    def _controller_uuid(self) -> str:
        if hasattr(self, 'cm') and self.cm:
            return self.cm.node_conf.get('uuid', 'controller')
        return 'controller'

    def _adopted_uuids_from_network_map(self) -> set[str]:
        """Read adopted node UUIDs from the in-memory network_map.

        We avoid NetworkMap.get_nodes_by_adoption() because it mutates the dict
        via strtobool — only works on the first pass; subsequent calls raise
        because `adopted`/`online` are no longer string-typed.
        """
        out: set[str] = set()
        try:
            node_list = (self.cm.network_map or {}).get('node_list', [])
            for entry in node_list:
                if not isinstance(entry, dict):
                    continue
                node = entry.get('node') or {}
                uuid = node.get('uuid')
                adopted = node.get('adopted', False)
                if isinstance(adopted, str):
                    adopted = adopted.strip().lower() in ('true', '1', 'yes')
                if adopted and uuid:
                    out.add(uuid)
        except Exception as e:
            Logger.warning(f'Could not read network_map: {e}')
        return out

    def _probe_cluster_liveness(self, timeout: float = 1.5) -> set[str]:
        """Broadcast a ping to all nodes and collect pong replies.

        The set of senders that respond within `timeout` is the authoritative
        "alive right now" view of the cluster. The controller's own UUID is
        always included (it never needs to ping itself).

        Returns the set of UUIDs that are alive.
        """
        controller_uuid = self._controller_uuid()
        adopted_uuids = self._adopted_uuids_from_network_map()

        expected = adopted_uuids - {controller_uuid}

        with self._cluster_lock:
            self._pong_responses.clear()
            self._pong_expected = set(expected)
        self._pong_event.clear()

        if not expected:
            Logger.debug('Cluster probe: no remote nodes to ping')
            return {controller_uuid}

        if not hasattr(self, 'communications_thread') or not self.communications_thread:
            Logger.warning('Cluster probe skipped: communications thread not available')
            return {controller_uuid}

        ping_op = NodeOperation(
            type=OperationType.COMMAND,
            action=ActionType.UPDATE,
            sender=controller_uuid,
            target='ping',
            data={'value': None, 'address': '/engine/cluster/ping'},
        )
        try:
            asyncio.run_coroutine_threadsafe(
                self.communications_thread.nng_hub.send_operation(ping_op),
                self.communications_thread.event_loop,
            )
        except Exception as e:
            Logger.warning(f'Could not broadcast cluster ping: {e}')
            return {controller_uuid}

        # Early-exit wait. The comms thread sets the event when all expected
        # pongs have arrived; otherwise we wake on timeout with whatever did.
        self._pong_event.wait(timeout=timeout)

        with self._cluster_lock:
            alive = set(self._pong_responses)
        alive.add(controller_uuid)
        Logger.debug(
            f'Cluster probe: expected={sorted(expected)} '
            f'alive_remote={sorted(alive - {controller_uuid})}'
        )
        return alive

    def _resolve_cluster_state(self) -> None:
        """Probe the cluster and decide which nodes the GO gate must wait for.

        Snapshots three sets at load time:
          - adopted: UUIDs from network_map.xml
          - alive:   UUIDs that ponged within the probe window
          - project: UUIDs referenced by any cue's output_name in self.script

        Computes self._required_nodes = adopted & alive & project, then
        always adds the controller's own UUID (its node-engine always
        processes the load and sends armed_ready, so this is reachable —
        and prevents the degenerate set() >= set() = True case from
        flipping armed=yes before any node has loaded).

        Categorizes each adopted node and logs info / warning / error.
        Resets _armed_nodes / _finished_nodes for the new project.
        """
        controller_uuid = self._controller_uuid()
        adopted = self._adopted_uuids_from_network_map()
        try:
            project = self._collect_project_nodes(self.script.cuelist)
        except Exception as e:
            Logger.warning(f'Could not collect project nodes: {e}')
            project = set()
        alive = self._probe_cluster_liveness()

        # Categorize for operator visibility.
        for uuid in sorted(adopted):
            in_alive = uuid in alive
            in_project = uuid in project
            if in_alive and in_project:
                continue  # the silent happy path — tracked via armed_ready
            if in_alive and not in_project:
                Logger.info(f'node {uuid} online but unused by this project')
            elif not in_alive and in_project:
                Logger.error(
                    f'node {uuid} required by this project but did not '
                    f'respond to ping; cues for it will not play. GO blocked.'
                )
            else:
                Logger.warning(
                    f'node {uuid} is adopted but did not respond to ping; '
                    f'not required by this project — investigate why it is offline'
                )

        # Project nodes that are NOT adopted at all — script is broken for
        # this cluster.
        for uuid in sorted(project - adopted):
            Logger.warning(
                f'project references node {uuid} which is not in the cluster; '
                f'cues for it will not fire'
            )

        required = (adopted & alive & project) | {controller_uuid}

        with self._cluster_lock:
            self._adopted_nodes = set(adopted)
            self._required_nodes = required
            self._armed_nodes.clear()
            self._finished_nodes.clear()

        Logger.info(
            f'Cluster state resolved: required={sorted(required)} '
            f'alive={sorted(alive)} adopted={sorted(adopted)} '
            f'project={sorted(project)}'
        )

        # The probe's `alive` set is a runtime liveness snapshot (sub-second,
        # used here for GO gating). The <online> field in network_map.xml is
        # a different concept: nodeconf's startup-discovery flag, persisted
        # so that adopted-but-currently-absent nodes keep their identity
        # records instead of being dropped from the map. The engine must NOT
        # overwrite that with its runtime view. See CLAUDE.md "Node identity"
        # / "<online> field ownership" for the full rationale.

        self._arm_arm_watchdog()

    _ARM_WATCHDOG_S = 120.0

    def _arm_arm_watchdog(self) -> None:
        """(Re)start the watchdog timer that surfaces a stalled load.

        Fires _ARM_WATCHDOG_S seconds after a load if not all required nodes
        have reported armed_ready by then. Logs an error naming the
        still-pending UUIDs so operator sees what's wedged.
        """
        self._cancel_arm_watchdog()
        timer = threading.Timer(self._ARM_WATCHDOG_S, self._on_arm_watchdog_fire)
        timer.daemon = True
        self._arm_watchdog = timer
        timer.start()

    def _cancel_arm_watchdog(self) -> None:
        timer = self._arm_watchdog
        if timer is not None:
            timer.cancel()
            self._arm_watchdog = None

    def _on_arm_watchdog_fire(self) -> None:
        with self._cluster_lock:
            armed = set(self._armed_nodes)
            required = set(self._required_nodes)
        pending = required - armed
        if not pending:
            return
        Logger.error(
            f'Load stalled: nodes still pending armed_ready after '
            f'{self._ARM_WATCHDOG_S:.0f}s: {sorted(pending)}'
        )

    def stop_script(self, value):
        """Handle STOP command - stop timecode, update status and forward to nodes."""
        if self.get_status('running') != "yes":
            Logger.info('Script not running, nothing to stop.')
            return

        self.go_offset = None
        self.set_status('running', "no")
        self._clear_playback_state()

        # Reset all cue statuses to unplayed (0) and broadcast to UI.
        for cid in self.cue_status:
            self.cue_status[cid] = 0
            self._broadcast_cue_status(cid, 0, force=True)

        self._forward_command_to_nodes('/engine/command/stop', value)

        Logger.info('STOP command processed - timecode stopped; nodes will re-arm')
        return True

    def get_project_status(self, value, context=None):
        """Return current project playback status."""
        running = self.get_status('running') == "yes"
        return {
            "status": "running" if running else "none",
            "project_uuid": str(self.script.id) if running and self.script else ""
        }

    def unload_project(self, value, context=None):
        """Unload the current project. Rejects if playback is running."""
        if self.get_status('running') == "yes":
            raise RuntimeError("Cannot unload while running. Stop playback first.")
        self._clear_playback_state()
        self.reset_script()
        self.cue_status = {}
        self.cue_enabled_status = {}
        self.set_status('load', '')
        # No project loaded → no required/adopted snapshot. Without this, a
        # late armed_ready from a slow node could flip armed=yes on an
        # unloaded project.
        with self._cluster_lock:
            self._required_nodes.clear()
            self._adopted_nodes.clear()
        self._forward_command_to_nodes('/engine/command/stop', value)
        Logger.info('Project unloaded')
        return True

cue_operation_callback(operation)

Callback invoked when cues are received from nodes.

Handles three action types: - ADD: cue started playing on a node → status 1, broadcast immediately - REMOVE: cue finished playing on a node → status 100, broadcast immediately - UPDATE: percentage progress from a node (future) → throttled broadcast

Source code in src/cuemsengine/ControllerEngine.py

def cue_operation_callback(self, operation: NodeOperation):
    """Callback invoked when cues are received from nodes.

    Handles three action types:
    - ADD:    cue started playing on a node → status 1, broadcast immediately
    - REMOVE: cue finished playing on a node → status 100, broadcast immediately
    - UPDATE: percentage progress from a node (future) → throttled broadcast
    """
    Logger.info(f'Cue operation received: {operation}')
    cue_id = operation.data.get('id') if operation.data else None

    # Drop operations for cues not belonging to the current project.
    # This prevents stale REMOVE/ADD notifications from the NodeEngine
    # (sent when it disarms the previous project) from being broadcast
    # to the UI as unknown UUIDs.
    if cue_id and cue_id not in self.cue_status:
        Logger.debug(f'Ignoring cue operation for unknown/stale cue_id {cue_id} (action={operation.action})')
        return

    if operation.action == ActionType.ADD:
        # Cue started playing: mark as playing (1) and broadcast immediately.
        if cue_id:
            self.cue_status[cue_id] = 1
            self._broadcast_cue_status(cue_id, 1, force=True)
        try:
            self.status.currentcue = [operation.data['id'], operation.data['offset']]
            Logger.debug(f"Current cue updated: {self.status.currentcue}")
        except Exception as e:
            Logger.error(f'Error updating currentcue: {e}')

    elif operation.action == ActionType.REMOVE:
        # Cue finished playing: mark as played (100) and broadcast immediately.
        # Only transition to 100 if the cue was actually playing (status == 1).
        # REMOVEs that arrive while status is 0 (e.g. NodeEngine disarming the
        # previous project after a reload) are stale and must be silently dropped.
        if cue_id:
            if self.cue_status.get(cue_id) == 1:
                self.cue_status[cue_id] = 100
                self._broadcast_cue_status(cue_id, 100, force=True)
            else:
                Logger.debug(f'Ignoring stale REMOVE for cue {cue_id} (status={self.cue_status.get(cue_id)}, expected 1)')
        self.status.remove_currentcue(operation.data['id'])
        Logger.debug(f"Cue removed from currentcue: {operation.data['id']}")

    elif operation.action == ActionType.UPDATE:
        # Future: percentage progress updates from loop_cue() during playback.
        # Throttled by _broadcast_cue_status (Tier 2 / controller-side).
        # The node-side Tier 1 throttle (CUE_STATUS_UPDATE_HZ) limits NNG traffic.
        if cue_id:
            pct = operation.data.get('percentage', 1)
            self.cue_status[cue_id] = pct
            self._broadcast_cue_status(cue_id, pct)  # throttled
        Logger.debug(f"Cue percentage update: {cue_id} = {operation.data.get('percentage')}")

    else:
        Logger.warning(f'Unknown cue action: {operation.action}')

get_project_status(value, context=None)

Return current project playback status.

Source code in src/cuemsengine/ControllerEngine.py

def get_project_status(self, value, context=None):
    """Return current project playback status."""
    running = self.get_status('running') == "yes"
    return {
        "status": "running" if running else "none",
        "project_uuid": str(self.script.id) if running and self.script else ""
    }

on_timecode_change(value)

Broadcast timecode to UI as integer ms (whole seconds only), once per second.

Source code in src/cuemsengine/ControllerEngine.py

def on_timecode_change(self, value) -> None:
    """Broadcast timecode to UI as integer ms (whole seconds only), once per second."""
    try:
        ms = int(value) if value is not None else 0
    except (TypeError, ValueError):
        return
    current_second = ms // 1000
    if current_second != self._last_timecode_second:
        self._last_timecode_second = current_second
        self._broadcast_status('timecode', current_second * 1000)
        Logger.debug(f'Timecode broadcast {current_second}s')

player_operation_callback(operation)

Callback invoked when players are received from nodes.

Parameters: - operation: NodeOperation with sender, target (player_id), and action

Source code in src/cuemsengine/ControllerEngine.py

def player_operation_callback(self, operation: NodeOperation):
    """
    Callback invoked when players are received from nodes.

    Parameters:
    - operation: NodeOperation with sender, target (player_id), and action
    """
    Logger.info(f'Player operation received: {operation}')

set_oscquery_values(values)

Stub for OSCQuery value setting - OSCQuery server has been removed.

Status updates are now handled via internal state tracking. TODO: Implement WebSocket status push if UI needs real-time status.

Source code in src/cuemsengine/ControllerEngine.py

def set_oscquery_values(self, values: dict):
    """Stub for OSCQuery value setting - OSCQuery server has been removed.

    Status updates are now handled via internal state tracking.
    TODO: Implement WebSocket status push if UI needs real-time status.
    """
    for key, value in values.items():
        Logger.debug(f"Status update (no-op): {key} = {repr(value)}")

set_status(property, value, strict=False)

Set status and push to UI via WebSocket when running, armed, or load.

Source code in src/cuemsengine/ControllerEngine.py

def set_status(self, property: str, value: str, strict: bool = False) -> None:
    """Set status and push to UI via WebSocket when running, armed, or load."""
    super().set_status(property, value, strict)
    if property in ('running', 'armed', 'load', 'nextcue'):
        self._broadcast_status(property, value)

status_operation_callback(operation)

Callback invoked when status updates are received from nodes.

Handles script_finished, armed_ready, nextcue, cue_enabled, and pong notifications.

Source code in src/cuemsengine/ControllerEngine.py

def status_operation_callback(self, operation: NodeOperation):
    """Callback invoked when status updates are received from nodes.

    Handles script_finished, armed_ready, nextcue, cue_enabled, and pong
    notifications.
    """
    Logger.info(f'Status operation received: {operation}')
    if operation.target == 'pong':
        # Cluster liveness probe reply. Runs on the AsyncCommsThread event
        # loop; _probe_cluster_liveness on the main thread waits on the event.
        with self._cluster_lock:
            self._pong_responses.add(operation.sender)
            if self._pong_responses >= self._pong_expected:
                self._pong_event.set()
        Logger.debug(f'Pong from {operation.sender}')
        return
    if operation.target == 'script_finished':
        if operation.data and operation.data.get('running') == 'no':
            # Aggregate per-node: only flip running=no when all required
            # nodes have reported. Filter foreign senders to keep the
            # tracker bounded.
            sender = operation.sender
            if sender not in self._adopted_nodes:
                Logger.debug(
                    f'Ignoring script_finished from non-adopted node {sender}'
                )
                return
            with self._cluster_lock:
                self._finished_nodes.add(sender)
                finished_now = set(self._finished_nodes)
                required = set(self._required_nodes)
            Logger.info(
                f'Node {sender} script_finished '
                f'({len(finished_now)}/{len(required)})'
            )
            if finished_now >= required and self.get_status('running') == 'yes':
                Logger.info('All required nodes finished — updating running status')
                self.set_status('running', 'no')
    elif operation.target == 'armed_ready':
        if operation.data and operation.data.get('armed') == 'yes':
            # Aggregate per-node: only flip armed=yes when all required
            # nodes have reported. Filter foreign senders to keep the
            # tracker bounded.
            sender = operation.sender
            if sender not in self._adopted_nodes:
                Logger.debug(
                    f'Ignoring armed_ready from non-adopted node {sender}'
                )
                return
            with self._cluster_lock:
                self._armed_nodes.add(sender)
                armed_now = set(self._armed_nodes)
                required = set(self._required_nodes)
            Logger.info(
                f'Node {sender} armed ({len(armed_now)}/{len(required)})'
            )
            if armed_now >= required and self.get_status('armed') != 'yes':
                if self.go_offset is None:
                    Logger.info(
                        'Re-arm after stop complete — restarting timecode and enabling GO'
                    )
                    self.start_timecode()
                    self.go_offset = 0
                else:
                    Logger.info('All required nodes armed — enabling GO')
                self.set_status('armed', 'yes')
                self._cancel_arm_watchdog()
    elif operation.target == 'nextcue':
        nextcue_id = operation.data.get('nextcue', '') if operation.data else ''
        self.set_status('nextcue', nextcue_id)
        Logger.info(f'Next cue updated: {nextcue_id or "(none)"}')
    elif operation.target == 'cue_enabled':
        cue_id = operation.data.get('cue_id') if operation.data else None
        enabled = operation.data.get('enabled', True) if operation.data else True
        if cue_id and cue_id in self.cue_enabled_status:
            self.cue_enabled_status[cue_id] = enabled
            self._broadcast_cue_enabled(cue_id, enabled)
            Logger.info(f'Cue {cue_id} enabled status updated from node: {enabled}')
    else:
        Logger.debug(f'Unknown status target: {operation.target}')

stop_script(value)

Handle STOP command - stop timecode, update status and forward to nodes.

Source code in src/cuemsengine/ControllerEngine.py

def stop_script(self, value):
    """Handle STOP command - stop timecode, update status and forward to nodes."""
    if self.get_status('running') != "yes":
        Logger.info('Script not running, nothing to stop.')
        return

    self.go_offset = None
    self.set_status('running', "no")
    self._clear_playback_state()

    # Reset all cue statuses to unplayed (0) and broadcast to UI.
    for cid in self.cue_status:
        self.cue_status[cid] = 0
        self._broadcast_cue_status(cid, 0, force=True)

    self._forward_command_to_nodes('/engine/command/stop', value)

    Logger.info('STOP command processed - timecode stopped; nodes will re-arm')
    return True

unload_project(value, context=None)

Unload the current project. Rejects if playback is running.

Source code in src/cuemsengine/ControllerEngine.py

def unload_project(self, value, context=None):
    """Unload the current project. Rejects if playback is running."""
    if self.get_status('running') == "yes":
        raise RuntimeError("Cannot unload while running. Stop playback first.")
    self._clear_playback_state()
    self.reset_script()
    self.cue_status = {}
    self.cue_enabled_status = {}
    self.set_status('load', '')
    # No project loaded → no required/adopted snapshot. Without this, a
    # late armed_ready from a slow node could flip armed=yes on an
    # unloaded project.
    with self._cluster_lock:
        self._required_nodes.clear()
        self._adopted_nodes.clear()
    self._forward_command_to_nodes('/engine/command/stop', value)
    Logger.info('Project unloaded')
    return True

NodeEngine

Bases: BaseEngine

This engine manages players for each node

Communicates with the ControllerEngine via OSCQuery

Interacts with Player objects via OSC

It is responsible for

• Starting and stopping players
• Monitoring player status
• Restarting players
• Updating player configurations
• Handling player failures
• Providing a clean interface for starting and stopping players
• Providing a clean interface for monitoring player status

Source code in src/cuemsengine/NodeEngine.py

class NodeEngine(BaseEngine):
    """
    This engine manages players for each node

    Communicates with the ControllerEngine via OSCQuery

    Interacts with Player objects via OSC

    It is responsible for:
      - Starting and stopping players
      - Monitoring player status
      - Restarting players
      - Updating player configurations
      - Handling player failures
      - Providing a clean interface for starting and stopping players
      - Providing a clean interface for monitoring player status

    """
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._command_lock = threading.Lock()
        self._loading_lock = threading.Lock()
        self._loading = False
        self._project_generation: int = 0
        self.nng_hub_address = f"tcp://{self.controller_ip}:{self.cm.node_conf['nng_hub_port']}"
        PORT_HANDLER.add_system_ports()
        if hasattr(self, 'cm'):
            PORT_HANDLER.add_config_ports(
                get_config_ports(self.cm.node_conf)
            )
            self.deploy_manager = CuemsDeploy(
                library_path=self.cm.library_path,
                tmp_path=self.cm.tmp_path,
                controller_ip=self.controller_ip,  # set by BaseEngine.set_cm()
            )
            PLAYER_HANDLER.add_media_folder(
                self.cm.library_path
            )
            PLAYER_HANDLER.set_player_endpoints_generator(
                self.add_player_endpoints,
                # TODO: Use node host from config
                prefix = '/players'
            )

    def start(self):
        CUE_HANDLER.set_nng_comms(self.nng_hub_address, self.cm.node_uuid)
        self.deploy_manager.loop = CUE_HANDLER.communications_thread.event_loop
        self.set_oscquery_comms()  # Creates command dictionary and OSCQuery client
        self.set_players()  # Creates player devices - must be before NNG callback
        self._setup_nng_command_callback()  # Set up NNG command receiving (after players ready)
        self.mtc_listener.start()
        super().start()

    def _setup_nng_command_callback(self):
        """Set up the callback for receiving commands via NNG from ControllerEngine.

        This provides push-based command delivery as an alternative to HTTP polling.
        Commands are received via the NNG bus and routed to the appropriate handlers.
        """
        if hasattr(CUE_HANDLER, 'communications_thread') and CUE_HANDLER.communications_thread:
            CUE_HANDLER.communications_thread.set_command_callback(self._handle_nng_command)
            Logger.info("NNG command callback registered for NodeEngine")
        else:
            Logger.warning("CUE_HANDLER communications thread not available for command callback")

        from .cues.ActionHandler import ACTION_HANDLER

        ACTION_HANDLER.finalize_node_layer_bindings()
        ACTION_HANDLER.set_result_sink(self._action_result_sink)


    def _handle_nng_command(self, command_name: str, value, address: str = None):
        """Handle a command received via NNG from ControllerEngine.

        Args:
            command_name: The command name (e.g., 'go', 'load', 'stop', 'player_control')
            value: The command value
            address: The original OSC address (optional)
        """
        Logger.info(f"NNG command received: {command_name} = {repr(value)}")

        if command_name == 'player_control' and address:
            # Handle player control messages (mixer volumes, video controls, etc.)
            self._handle_player_control_message(address, value)
        else:
            # Handle standard commands (go, load, stop)
            self.run_command(command_name, value)

    def _handle_player_control_message(self, address: str, value):
        """Handle player control messages received via NNG.

        Routes to appropriate player handlers based on the OSC address.
        Supports two formats:
        1. Engine format: /engine/players/<uuid>/<type>/...
        2. Direct format: /<uuid>/<type>/... (from UI)

        Args:
            address: The OSC address
            value: The value to set
        """
        parts = address.strip('/').split('/')

        # Determine format and extract node_uuid, player_type, path_parts
        if len(parts) >= 4 and parts[0] == 'engine' and parts[1] == 'players':
            # Engine format: /engine/players/<node_uuid>/<type>/...
            node_uuid = parts[2]
            player_type = parts[3]
            path_parts = parts[4:] if len(parts) > 4 else []
        elif len(parts) >= 2:
            # Direct format: /<node_uuid>/<type>/...
            node_uuid = parts[0]
            player_type = parts[1]
            path_parts = parts[2:] if len(parts) > 2 else []
        else:
            Logger.warning(f"Invalid player control address: {address}")
            return

        # Only handle messages for this node
        if node_uuid != self.cm.node_uuid:
            Logger.debug(f"Ignoring player message for other node: {node_uuid}")
            return

        Logger.debug(f"Handling player control: type={player_type}, path={path_parts}, value={value}")

        # Route to appropriate handler based on player type
        if player_type == 'video':
            redirect_video_cmd(path_parts, value)
        elif player_type == 'audio':
            CUE_HANDLER.route_audio_message(path_parts, value)
        elif player_type == 'dmx':
            CUE_HANDLER.route_dmx_message(path_parts, value)
        elif player_type == 'audiomixer':
            # Legacy: pre-2026 OSC format /{uuid}/audiomixer/{channel}.
            # The current UI sends /{uuid}/audio/mixer/{output_index}/{channel}/volume,
            # which routes via player_type == 'audio' → CueHandler.route_audio_message().
            # Kept for backwards compatibility with any external tool still using
            # the old format; new code should not target this path.
            # Direct audiomixer command: /<uuid>/audiomixer/<channel>
            # path_parts[0] is channel (e.g., '0', 'master')
            self._handle_audiomixer_command(path_parts, value)
        elif player_type == 'jadeo':
            # Direct video command: /<uuid>/jadeo/<cmd>
            redirect_video_cmd(['jadeo'] + path_parts, value)
        else:
            Logger.debug(f"Unknown player type in control message: {player_type}")

    def _handle_audiomixer_command(self, path_parts: list, value):
        """Handle direct audiomixer OSC command.

        Args:
            path_parts: Remaining path parts after /<uuid>/audiomixer/
                       e.g., ['0'] for channel 0, ['master'] for master
            value: Volume value (0.0 to 1.0)
        """
        if not path_parts:
            Logger.warning("Empty audiomixer command path")
            return

        channel = path_parts[0]
        # jack-volume expects /audiomixer/<client_name>/<channel>
        mixer_cmd = f'/audiomixer/0_mixer/{channel}'

        try:
            PLAYER_HANDLER.get_audio_mixer_client().set_value(mixer_cmd, value)
            Logger.debug(f"Audiomixer command: {mixer_cmd} = {value}")
        except Exception as e:
            Logger.error(f"Error sending audiomixer command: {e}")

    @logged
    def stop(self):
        self.stop_requested = True
        self.stop_node_engine()
        super().stop()

    def stop_node_engine(self):
        """Stop the NodeEngine elements"""
        CUE_HANDLER.disarm_all()
        self.stop_video_devs()

    def stop_video_devs(self):
        try:
            self.unload_video_devs()
            Logger.info('Video devs stopped')
        except Exception as e:
            Logger.warning(f'Exception raised when stopping video devs: {e}')

    def quit_video_devs(self):
        try:
            PLAYER_HANDLER.quit_videocomposer()
            Logger.info('Videocomposer quit successfully')
        except Exception as e:
            Logger.exception(e)

    def unload_video_devs(self):
        try:
            PLAYER_HANDLER.reset_videocomposer()
            Logger.info('Videocomposer reset successfully')
        except Exception as e:
            Logger.exception(e)

    #########################
    # OSCQuery logic
    #########################
    def add_player_endpoints(self, cue: Cue, prefix: str = '/players'):
        """Add player endpoints from a cue to the OSCQuery server

        Args:
            cue: The cue containing the player client
            prefix: Prefix to add to all endpoint paths (default: '/players')
        """
        if not hasattr(cue, '_osc') or cue._osc is None:
            Logger.warning(f'Cue {cue.id} has no OSC client, cannot add endpoints')
            return

        try:
            # Get endpoints from the player client
            endpoints = cue._osc.get_endpoints()
            if not endpoints:
                Logger.warning(f'No endpoints found for cue {cue.id}')
                return

            # Add prefix to all endpoints
            prefixed_endpoints = add_prefix_to_all(endpoints, f"{prefix}/{self.cm.node_uuid}/{cue.id}")

            # Add endpoints to OSCQuery server
            if hasattr(self, 'oscquery_server') and self.oscquery_server:
                self.oscquery_server.add_endpoints(prefixed_endpoints)
                Logger.debug(f'Added {len(prefixed_endpoints)} endpoints for cue {cue.id}')
            else:
                Logger.warning('OSCQuery server not initialized, cannot add endpoints')
        except Exception as e:
            Logger.error(f'Error adding player endpoints for cue {cue.id}: {e}')
            Logger.exception(e)

    def set_oscquery_comms(self):
        """Set up the command dictionary for the NodeEngine.

        Commands are received via NNG from ControllerEngine.
        OSCQuery client is no longer used since pyossia server was removed.
        """
        self.commands_dict = {
            'deploy': self.ready_project,
            'load': self.load_project,
            'loadcue': None,
            'go': self.go_script,
            'gocue': self.go_script,
            'pause': None,
            'resetall': None,
            'stop': self.stop_playback,
            'setnextcue': self.set_next_cue,
            'cue_enabled': self._handle_cue_enabled,
            'test': None,
            'unload': None,
            'update': None,
        }

    def route_message(self, parameter, value):
        # Exclude 'engine' common node
        path_elements = str(parameter.node).split('/')[2:]
        if path_elements[0] == 'command':
            self.run_command(path_elements[1], value)
        elif path_elements[0] == 'status':
            Logger.debug(f'Status update received: {path_elements[1]} = {repr(value)}')
        elif path_elements[0] == 'players':
            # Exclude other nodes' players
            if path_elements[1] != self.cm.node_uuid:
                Logger.debug(f'Ignoring player message for other node: {path_elements[1]}')
                return
            # Route the message to the appropriate player handler
            if path_elements[2] == 'video':
                redirect_video_cmd(path_elements[3:], value)
            if path_elements[2] == 'audio':
                CUE_HANDLER.route_audio_message(path_elements[3:], value)
            if path_elements[2] == 'dmx':
                CUE_HANDLER.route_dmx_message(path_elements[3:], value)
        else:
            Logger.debug(f'Recieved unused OSCQuery path: {str(parameter.node)}')
            return

    def run_command(self, command, value):
        with self._command_lock:
            Logger.debug(f'NodeEngine executing command: {command}({repr(value)})')
            if command in self.commands_dict.keys():
                handler = self.commands_dict[command]
                if handler is not None:
                    handler(value)
                    return True
                else:
                    Logger.warning(f'Command {command} has no handler')
                    return False
            else:
                Logger.error(f'Command {command} not found')
                return False

    #########################
    # Player logic
    #########################
    def set_players(self):
        self.set_video_players()
        self.set_audio_players()
        self.set_dmx_players()
        self.set_gradient_client()

    def set_gradient_client(self) -> None:
        """Wire GradientClient into PLAYER_HANDLER using settings from node_conf."""
        port = int(self.cm.node_conf['gradient_osc_port'])
        PLAYER_HANDLER.set_gradient_client(port=port, node_uuid=self.cm.node_uuid)

    # Audio functions
    def set_audio_players(self):
        """Set the audio players and audio mixer"""
        # Initialize the audio mixer for this node
        if self.cm.node_hw_outputs.get('audio_outputs'):
            audio_outputs = self.cm.node_hw_outputs['audio_outputs']
            Logger.info(f'Initializing audio mixer with {len(audio_outputs)} outputs')

            # Assign a port for the audio mixer
            mixer_id = '0' # TODO: make this a unique identifier for the mixer
            mixer_ports = PORT_HANDLER.assign_ports(['audio_mixer'])
            PORT_HANDLER.add_config_ports(mixer_ports)
            # Start the audio mixer
            try:
                PLAYER_HANDLER.start_audio_mixer(
                    audio_outputs=audio_outputs,
                    port=mixer_ports['audio_mixer'],
                    mixer_id=mixer_id,
                    path=self.cm.node_conf['audiomixer']['path'],
                    args=self.cm.node_conf['audiomixer']['args']
                )
                Logger.info(f'Audio mixer started successfully for mixer {mixer_id}')
                # Register mixer with Controller via NNG
                try:
                    CUE_HANDLER.communications_thread.add_player(f'audiomixer_{mixer_id}', None, timeout=0.1)
                    Logger.info(f'Audio mixer {mixer_id} registered with Controller')
                except Exception as e:
                    Logger.warning(f'Could not register mixer with Controller: {e}')
            except Exception as e:
                Logger.error(f'Error starting audio mixer: {e}')
                Logger.exception(e)
        else:
            Logger.info('No audio outputs detected, skipping audio mixer initialization')

        # Build audio output lookup keyed by <id> (mirrors video output pattern)
        audio_outputs = {}
        for port_type_dict in self.cm.node_mappings.get('audio', []):
            for port_type_list in port_type_dict.values():
                for port in port_type_list:
                    for _, output_data in port.items():
                        output_id = str(output_data.get('id', output_data['name']))
                        mappings = output_data.get('mappings', [])
                        mapped_to = mappings[0]['mapped_to'] if mappings else output_data['name']
                        audio_outputs[output_id] = {
                            'name': output_data['name'],
                            'mapped_to': mapped_to,
                        }
        PLAYER_HANDLER.set_audio_outputs(audio_outputs)

        # Set the audio player generator. Append --output-latency-ms
        # from settings.xml when the operator supplied an integer
        # override (isinstance int); "auto" or absent ⇒ audioplayer
        # runs its Phase-3 JACK-latency query path.
        audio_args = _append_output_latency_flag(
            self.cm.node_conf['audioplayer']['args'],
            self.cm.node_conf['audioplayer'],
        )
        PLAYER_HANDLER.set_audio_output_generator(
            self.cm.node_conf['audioplayer']['path'],
            audio_args,
        )

    # Video functions
    def set_video_players(self):
        """Set the video players"""
        Logger.info(f'Setting video players with: {self.cm.node_conf["videoplayer"]}')
        if not self.cm.node_hw_outputs['video_outputs']:
            Logger.info('No video outputs detected.')
            return

        PLAYER_HANDLER.add_node_uuid(self.cm.node_uuid)
        vc_conf = self.cm.node_conf.get('videoplayer', {})
        osc_video_port = int(vc_conf.get('osc_port', VIDEOCOMPOSER_OSC_PORT_DEFAULT))
        PLAYER_HANDLER.set_video_client(osc_video_port)
        PORT_HANDLER.add_config_ports({'videocomposer': osc_video_port})

        # Canvas geometry comes from /run/cuems/display.conf, written by
        # cuems-generate-display-conf (videocomposer's ExecStartPre). It's the
        # same file the videocomposer reads, so engine + VC agree on canvas
        # size and per-output regions without a handshake. The XML's optional
        # <canvas_region> is a UI-template hint (normalized [0,1]) and is
        # ignored here — engine never sources physical layout from XML.
        display_regions, (canvas_w, canvas_h) = read_display_conf()

        video_outputs = {}
        for port_type_dict in self.cm.node_mappings.get('video', []):
            for port_type_list in port_type_dict.values():
                for port in port_type_list:
                    for _, output_data in port.items():
                        output_id = str(output_data.get('id', output_data['name']))
                        name = output_data['name']
                        mappings = output_data.get('mappings', [])
                        mapped_to = mappings[0]['mapped_to'] if mappings else name
                        region = display_regions.get(mapped_to)
                        if region is None:
                            Logger.warning(
                                f"DISPLAY_MISMATCH: XML output id={output_id} "
                                f"name={name!r} maps to {mapped_to!r} which is "
                                f"not in display.conf; skipping. Available: "
                                f"{sorted(display_regions.keys())}"
                            )
                            continue
                        video_outputs[output_id] = {
                            'name': name,
                            'mapped_to': mapped_to,
                            'x': region['x'],
                            'y': region['y'],
                            'width': region['width'],
                            'height': region['height'],
                            'canvas_region': dict(region),
                        }
        PLAYER_HANDLER.start_video_outputs(
            video_outputs, canvas_override=(canvas_w, canvas_h)
        )


    # DMX functions
    def set_dmx_players(self):
        """Set the DMX player for this node and register its endpoints."""
        # Assign a port for the DMX player
        dmx_ports = PORT_HANDLER.assign_ports(['dmx_player'])
        PORT_HANDLER.add_config_ports(dmx_ports)

        # Get node UUID for player naming
        node_uuid = self.cm.node_conf.get('uuid', 'default_node')

        # Start the DMX player
        try:
            # Append --output-latency-ms from settings.xml when an
            # integer override is present. Dmx has no "auto" form —
            # absent ⇒ dmxplayer's 35 ms Phase-5A default stands.
            dmx_args = _append_output_latency_flag(
                self.cm.node_conf['dmxplayer']['args'],
                self.cm.node_conf['dmxplayer'],
            )
            PLAYER_HANDLER.start_dmx_player(
                port=dmx_ports['dmx_player'],
                node_uuid=node_uuid,
                path=self.cm.node_conf['dmxplayer']['path'],
                args=dmx_args,
            )
            try:
                CUE_HANDLER.communications_thread.add_player(f'dmxplayer_{node_uuid}', None, timeout=0.1)
            except Exception:
                pass  # Ignore - NNG is for distributed nodes
            Logger.info(f'DMX player started successfully for node {node_uuid}')
        except Exception as e:
            Logger.error(f'Error starting DMX player: {e}')
            Logger.exception(e)
            return

    def quit_dmx_devs(self):
        """Quit the DMX player if it exists"""
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            try:
                dmx_client.set_value('/quit', 1)
            except Exception as e:
                Logger.exception(e)
        CUE_HANDLER.communications_thread.remove_player(f'dmxplayer_{self.cm.node_uuid}')


    #########################
    # Project logic
    #########################
    def ready_project(self, project):
        """Prepare the project to be played.

        deploy_project() runs in _load_project_inner BEFORE this point,
        before the teardown of the previous project — so that a deploy
        failure aborts the load without destroying running state.
        Media deploy runs here because it is best-effort: a failure
        logs but does not abort (cached media may already be on disk).
        """
        self.cm.load_project_config(project)
        self.read_script(project)
        self.deploy_media(project)
        self.ensure_video_indexes()
        self.outputs_map = self.map_cue_outputs()
        PLAYER_HANDLER.set_outputs_map(self.outputs_map)
        PORT_HANDLER.clean_random_ports()

    def map_cue_outputs(self, cuelist: CueList = None):
        """Load the output mappings for the project"""
        outputs_map = {}
        if cuelist is None:
            cuelist = self.script.cuelist
        for cue in cuelist.contents:
            if isinstance(cue, CueList):
                outputs_map.update(self.map_cue_outputs(cue))
            elif not isinstance(cue, MediaCue):
                continue

            outputs = [x[1] for x in cue.get_all_output_names() if x[0] == self.cm.node_uuid]
            if outputs:
                outputs_map[cue.id] = outputs
        Logger.debug(f'Outputs map: {outputs_map}')
        return outputs_map

    def load_project(self, project):
        """Load the project files to the node"""
        with self._loading_lock:
            if self._loading:
                Logger.warning(f'Load already in progress, ignoring duplicate load of {project}')
                return
            self._loading = True

        try:
            return self._load_project_inner(project)
        finally:
            with self._loading_lock:
                self._loading = False

    def _load_project_inner(self, project):
        # Don't allow loading while script is running
        if self.get_status('running') == "yes":
            Logger.warning(f'Cannot load project {project} while script is running. Stop first.')
            return False

        # Deploy the critical project files (script.xml, mappings.xml,
        # settings.xml) BEFORE tearing down the previous project. If the
        # controller is unreachable we abort here with the previous
        # project still armed and usable — better than ending up with
        # everything stopped and no new project loaded.
        if not self.deploy_project(project):
            Logger.error(
                f'Project deploy FAILED for {project} — aborting load; '
                f'previous project remains unchanged'
            )
            return False

        gradient_client = PLAYER_HANDLER.get_gradient_client()
        if gradient_client:
            try:
                gradient_client.send_cancel_all()
            except Exception as exc:
                Logger.error(f'gradient send_cancel_all failed on project load: {exc}')
        else:
            Logger.debug('gradient_client not initialised, skipping cancel_all on project load')

        # Stop any running cue threads from the previous project first,
        # so they can't interfere with cleanup (same logic as stop_playback).
        CUE_HANDLER.stop_all_cues()

        # DMX: stop following MTC, blackout all universes.
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            try:
                dmx_client.disable_mtcfollow()
            except Exception as e:
                Logger.warning(f'DMX disable mtcfollow failed: {e}')
            try:
                dmx_client.send_blackout()
            except Exception as e:
                Logger.warning(f'DMX blackout failed: {e}')

        # Video: reset videocomposer (remove all layers, cancel loads, reset master).
        self.unload_video_devs()

        # Audio: reset mixer volumes, kill all players, clean up JACK.
        mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
        if mixer_client:
            try:
                mixer_client.reset_volumes()
            except Exception as e:
                Logger.warning(f'JACK volume reset failed: {e}')
        PLAYER_HANDLER.kill_all_audio_players()
        PLAYER_HANDLER.kill_orphaned_audio_processes()
        PLAYER_HANDLER.cleanup_zombie_jack_clients()

        # Disarm all cues from the previous project.
        CUE_HANDLER.disarm_all()

        # Obtain the project files (this replaces self.script with new project)
        self.ready_project(project)

        # Prepare the script to be played (arms new cues)
        self.ready_script()

        # Start cue dependencies
        # self.set_players()

        # Confirm the project is loaded
        self.set_show_lock_file()
        self.script.unix_name = project
        self.set_status('load', project)
        Logger.info(f'Project {project} loaded')

        # Notify Controller that arming is complete (GO button can go green)
        try:
            from .comms.NodesHub import NodeOperation, OperationType, ActionType
            operation = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=self.cm.node_uuid,
                target='armed_ready',
                data={'armed': 'yes'}
            )
            CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
            Logger.debug('Notified Controller that arming after load is complete')
        except Exception as e:
            Logger.warning(f'Could not notify Controller of armed_ready: {e}')

        # Broadcast initial nextcue to UI
        self._broadcast_nextcue()

        return True

    def deploy_project(self, project):
        """Deploy the project files (script.xml, mappings.xml, settings.xml).

        Critical path: if these fail to sync, the local copy may be stale
        and arming cues against it is unsafe. Caller is expected to abort
        the load on False.
        """
        return self.deploy_manager.sync_files(project, 'project')

    def deploy_media(self, project):
        """Deploy the media files (and their .idx sidecar indexes).

        Best-effort: a failure here is recoverable if media is already
        cached on disk. Returns False to surface the failure to logs,
        but the caller continues the load.
        """
        if not self.script:
            Logger.error('No script loaded')
            return False
        bare_names = self.script.get_own_media_filenames(config=self.cm)
        if len(bare_names) == 0:
            Logger.info('No media files to deploy')
            return True
        if not self.deploy_manager.sync_files(project, 'media', bare_names):
            Logger.error(
                f'Media deploy failed for {project} — continuing with cached '
                f'files; cues whose media is missing locally will fail on GO'
            )
            return False
        return True

    def ensure_video_indexes(self):
        """Run cuems-videoindexer on any video files that are missing a .idx sidecar.

        This is a safety net for files that were copied manually or deployed to a
        node that never ran the editor upload hook. For normally-uploaded files the
        index was already created by the editor and this is a no-op.
        """
        if not self.script:
            return
        file_names = self.script.get_own_media_filenames(config=self.cm)
        video_exts = {'.mp4', '.mov', '.avi', '.mkv', '.mpg'}
        unindexed = []
        for name in file_names:
            ext = os.path.splitext(name)[1].lower()
            if ext not in video_exts:
                continue
            full_path = PLAYER_HANDLER.media_path(name)
            idx_dir = os.path.join(os.path.dirname(full_path), 'indexes')
            idx_path = os.path.join(idx_dir, os.path.basename(full_path) + '.idx')
            if not os.path.exists(idx_path):
                unindexed.append(full_path)
        if unindexed:
            Logger.info(
                f'ensure_video_indexes: indexing {len(unindexed)} video(s) missing .idx: '
                f'{[os.path.basename(p) for p in unindexed]}'
            )
            try:
                result = subprocess.run(
                    ['cuems-videoindexer'] + unindexed,
                    timeout=600,
                    capture_output=True,
                    text=True,
                )
            except Exception as e:
                Logger.warning(f'ensure_video_indexes: indexer failed: {e}')
                return
            if result.returncode == 0:
                Logger.info(f'ensure_video_indexes: indexer ok (rc=0)')
                if result.stdout:
                    Logger.debug(f'ensure_video_indexes stdout: {result.stdout.strip()}')
            else:
                Logger.warning(
                    f'ensure_video_indexes: indexer returned rc={result.returncode}. '
                    f'stdout={result.stdout.strip()!r} stderr={result.stderr.strip()!r}'
                )

    #########################
    # Nextcue
    #########################
    def _broadcast_nextcue(self):
        """Send the current next_cue_pointer UUID to the Controller via NNG."""
        cue_id = self.next_cue_pointer.id if self.next_cue_pointer else ""
        try:
            CUE_HANDLER.communications_thread.update_nextcue(cue_id, timeout=0.1)
            Logger.debug(f'Broadcast nextcue: {cue_id or "(none)"}')
        except Exception as e:
            Logger.warning(f'Could not broadcast nextcue: {e}')

    def _arm_with_enabled_guard(self, cue, project_gen: int):
        """Arm a cue and disarm if it was disabled or project changed while arming.

        Runs in a daemon thread. After arm() completes, re-checks
        cue.enabled and project generation to handle races where:
        - A disable command arrived while arm_cue() was loading media
        - A stop/reload invalidated this project's cues
        """
        if self._project_generation != project_gen:
            Logger.info(f'Aborting arm of {cue.id} — project generation changed')
            return
        CUE_HANDLER.arm(cue, init=True)
        # If project changed during arm, disarm the stale cue.
        if self._project_generation != project_gen:
            if CUE_HANDLER.find_armed_cue(cue):
                CUE_HANDLER.disarm(cue)
            Logger.info(f'Disarmed cue {cue.id} — project changed during async arm')
            return
        # If cue was disabled while we were arming, disarm now.
        if not cue.enabled and CUE_HANDLER.find_armed_cue(cue):
            CUE_HANDLER.disarm(cue)
            Logger.info(f'Disarmed cue {cue.id} — disabled during async arm')

    def _action_result_sink(self, outcome: dict):
        """Custom result sink for ActionHandler — extends default with cue_enabled sync."""
        from .cues.ActionHandler import ACTION_HANDLER
        # Always run default behavior (sends action_cue_outcome via NNG)
        ACTION_HANDLER._default_result_sink(outcome)

        # If an enable/disable action was applied, notify Controller
        action_type = outcome.get('action_type')
        status = outcome.get('status')
        if action_type in ('enable', 'disable') and status == 'applied':
            target_id = outcome.get('target_id')
            if target_id:
                self._notify_cue_enabled(target_id, action_type == 'enable')

    def _notify_cue_enabled(self, cue_id: str, enabled: bool):
        """Send cue enabled status to Controller via NNG."""
        from .comms.NodesHub import NodeOperation, OperationType, ActionType
        try:
            operation = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=self.cm.node_uuid if hasattr(self, 'cm') and self.cm else 'node',
                target='cue_enabled',
                data={'cue_id': cue_id, 'enabled': enabled}
            )
            CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
        except Exception as e:
            Logger.warning(f'Could not notify cue_enabled: {e}')

    def set_next_cue(self, value):
        """Handle setnextcue command from the UI — override next_cue_pointer."""
        if not self.script:
            Logger.warning('No script loaded, cannot set next cue.')
            return
        cue = self.script.find(value)
        if cue:
            self.next_cue_pointer = cue
            if not CUE_HANDLER.find_armed_cue(cue):
                Logger.info(f'Re-arming cue {cue.id} selected as next cue')
                CUE_HANDLER.arm(cue, init=True)
            CUE_HANDLER._arm_ahead(cue)  # extend window from selected cue
            self._broadcast_nextcue()
            Logger.info(f'Next cue overridden by UI: {value}')
        else:
            Logger.warning(f'setnextcue: cue {value} not found in script')

    def _handle_cue_enabled(self, value):
        """Handle cue_enabled toggle from Controller.

        Value format: "<cue_id> <0|1>" (space-separated UUID and enabled flag).
        """
        if not self.script:
            Logger.warning('No script loaded, cannot toggle cue enabled')
            return

        if not value or not isinstance(value, str):
            Logger.warning(f'Invalid cue_enabled value: {repr(value)}')
            return

        parts = value.split(' ', 1)
        if len(parts) != 2 or parts[1] not in ('0', '1'):
            Logger.warning(f'Invalid cue_enabled format: {repr(value)}')
            return

        cue_id, enabled_str = parts
        enabled = enabled_str == '1'

        cue = self.script.find(cue_id)
        if not cue:
            Logger.warning(f'cue_enabled: cue {cue_id} not found in script')
            return

        cue.enabled = enabled

        if not enabled:
            # Disarm only if armed and NOT currently playing.
            # A playing cue has a running go thread (_go_generation > 0) and is still loaded.
            is_playing = (getattr(cue, '_go_generation', 0) > 0
                          and getattr(cue, 'loaded', False))
            if CUE_HANDLER.find_armed_cue(cue) and not is_playing:
                CUE_HANDLER.disarm(cue)
                Logger.info(f'Disarmed disabled cue {cue_id}')
            # Recalculate next_cue_pointer if the disabled cue was next
            if self.next_cue_pointer and self.next_cue_pointer.id == cue_id:
                self.next_cue_pointer = cue.get_next_cue()
                self._broadcast_nextcue()
                Logger.info(f'Next cue was disabled, advanced to {self.next_cue_pointer.id if self.next_cue_pointer else "none"}')
        else:
            # Re-arm in a daemon thread to avoid blocking _command_lock
            # (arm() is slow — media loading, process spawning).
            if cue._local and not CUE_HANDLER.find_armed_cue(cue):
                gen = self._project_generation
                threading.Thread(
                    target=self._arm_with_enabled_guard,
                    args=(cue, gen),
                    daemon=True,
                    name=f'ReArm:{cue_id}'
                ).start()
                Logger.info(f'Re-arming enabled cue {cue_id} (async)')

        self._notify_cue_enabled(cue_id, enabled)
        Logger.info(f'Cue {cue_id} set to {"enabled" if enabled else "disabled"}')

    #########################
    # Script logic
    #########################
    def ready_script(self):
        """Check if the script is ready to be played"""
        if not self.script:
            Logger.warning('No script loaded, cannot process GO command.')
            return

        self.ongoing_cue = None
        self.next_cue_pointer = None
        self.go_offset = 0
        self._project_generation += 1  # Abort in-flight daemon arm threads
        self.unload_video_devs()
        CUE_HANDLER.disarm_all()

        # Reset mixer volumes to default when preparing script
        mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
        if mixer_client:
            mixer_client.reset_volumes()

        self.initial_cuelist_process()

        # Set initial nextcue to the first enabled cue in the script
        if self.script.cuelist.contents:
            first_enabled = None
            for c in self.script.cuelist.contents:
                if c.enabled:
                    first_enabled = c
                    break
            self.next_cue_pointer = first_enabled

        Logger.info(f'Script {self.script.name} loaded and ready to be played')

    def go_script(self, value):
        if not self.script:
            Logger.warning('No script loaded, cannot process GO command.')
            return

        if not self.with_mtc:
            Logger.warning('No MTC listener, cannot process GO command.')
            return

        # Determine the cue to go
        if not self.ongoing_cue:
            # First GO - use next_cue_pointer (may have been overridden by setnextcue)
            cue_to_go = self.next_cue_pointer or self.script.cuelist.contents[0]
            Logger.info(f'GO command received. Starting script {self.script.name}')
        else:
            # Successive GO - advance to next cue
            if self.next_cue_pointer:
                cue_to_go = self.next_cue_pointer
                Logger.info(f'GO command received. Advancing to next cue: {cue_to_go.id}')
            else:
                # No next cue - script has finished. Do not stop timecode or reset state.
                Logger.info('No more cues. Press STOP to restart.')
                return

        if not cue_to_go._local:
            # First cue in the GO target isn't ours, but a post_go='go' chain
            # may include local cues further down. Walk forward through the
            # chain to find our first local cue and start there. Each cue in a
            # post_go='go' chain shares the same frozen MTC snapshot, so every
            # node fires its own local cues from the same GO press in parallel.
            # Stop walking if the chain breaks (post_go != 'go' on a non-local
            # cue) — that's an explicit hand-off point waiting for the next GO.
            original = cue_to_go
            walked = 0
            while cue_to_go is not None and not cue_to_go._local:
                if cue_to_go.post_go != 'go':
                    cue_to_go = None
                    break
                cue_to_go = getattr(cue_to_go, '_target_object', None)
                walked += 1
                if walked > 1024:
                    Logger.error('GO chain walk hit safety limit; aborting')
                    cue_to_go = None
                    break

            if cue_to_go is None:
                Logger.info(
                    f'No local cues in post_go="go" chain from {original.id}; '
                    f'nothing to play on this node for this GO')
                return

            Logger.info(
                f'GO: skipped {walked} non-local cue(s); starting from local '
                f'cue {cue_to_go.id}')

        if not cue_to_go.enabled:
            Logger.info(f'Cue {cue_to_go.id} is disabled, advancing to next enabled cue')
            self.next_cue_pointer = cue_to_go.get_next_cue()
            self._broadcast_nextcue()
            return

        if not CUE_HANDLER.find_armed_cue(cue_to_go):
            Logger.info(f'Cue {cue_to_go.id} not armed, re-arming before GO')
            CUE_HANDLER.arm(cue_to_go, init=True)
            if not CUE_HANDLER.find_armed_cue(cue_to_go):
                Logger.error(f'Failed to re-arm cue {cue_to_go.id}, cannot GO')
                return

        # Update state
        self.set_status('running', "yes")
        self.ongoing_cue = cue_to_go

        # Start the cue
        main_thread = CUE_HANDLER.go(
            cue_to_go,
            self.mtc_listener
        )

        # Update next cue pointer
        self.next_cue_pointer = self.ongoing_cue.get_next_cue()
        # Drift baseline; consumed by BaseEngine.timecode = mtc - go_offset.
        # _exact for sub-ms precision at NTSC framerates.
        self.go_offset = self.mtc_listener.main_tc.milliseconds_exact

        # Broadcast nextcue to UI
        self._broadcast_nextcue()

        Logger.info(f'Cue {cue_to_go.id} started. Next cue: {self.next_cue_pointer.id if self.next_cue_pointer else "none"}')

    def stop_playback(self, value=None):
        """Stop playback, full cleanup, then re-arm so GO is available again.

        Does the cleanup that ready_script() doesn't handle (DMX blackout,
        disconnect video, kill audio), then delegates reset + re-arm to
        ready_script(). Notifies Controller when armed (GO button green).
        """
        Logger.info('STOP command received. Stopping playback.')

        self.set_status('running', "no")

        gradient_client = PLAYER_HANDLER.get_gradient_client()
        if gradient_client:
            try:
                gradient_client.send_cancel_all()
            except Exception as exc:
                Logger.error(f'gradient send_cancel_all failed on stop: {exc}')
        else:
            Logger.debug('gradient_client not initialised, skipping cancel_all on stop')

        # Signal all running cue threads to stop immediately.
        # Must happen BEFORE blackout/reset so loop_cue threads don't
        # re-push DMX frames or send /visible after cleanup.
        CUE_HANDLER.stop_all_cues()
        sleep(0.05)  # 50ms — loop_cue polls every 20ms

        # DMX: disable MTC following first (freezes the playhead so queued
        # scenes can't fire), then blackout via OLA for instant visual reset.
        dmx_client = PLAYER_HANDLER.get_dmx_player_client()
        if dmx_client:
            try:
                dmx_client.disable_mtcfollow()
            except Exception as e:
                Logger.warning(f'DMX disable mtcfollow failed: {e}')
            try:
                dmx_client.send_blackout()
            except Exception as e:
                Logger.warning(f'DMX blackout failed: {e}')

        # Unload all video layers (instant visual blackout)
        self.unload_video_devs()

        # Kill all audio players (ready_script does not do this)
        PLAYER_HANDLER.kill_all_audio_players()
        PLAYER_HANDLER.cleanup_zombie_jack_clients()

        # Reset state + disarm + volume reset + re-arm cues
        if self.script:
            self.ready_script()
            Logger.info(f'Project {self.script.name} reset and ready for GO.')

            # Notify Controller that re-arm is complete (GO button can go green)
            try:
                from .comms.NodesHub import NodeOperation, OperationType, ActionType
                operation = NodeOperation(
                    type=OperationType.STATUS,
                    action=ActionType.UPDATE,
                    sender=self.cm.node_uuid,
                    target='armed_ready',
                    data={'armed': 'yes'}
                )
                CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
                Logger.debug('Notified Controller that re-arm is complete')
            except Exception as e:
                Logger.warning(f'Could not notify Controller of armed_ready: {e}')

            # Broadcast nextcue (reset to first cue after stop)
            self._broadcast_nextcue()
        else:
            Logger.info('Playback stopped (no script loaded).')

        Logger.info('Playback stopped.')

add_player_endpoints(cue, prefix='/players')

Add player endpoints from a cue to the OSCQuery server

Parameters:

Name	Type	Description	Default
cue	Cue	The cue containing the player client	required
prefix	str	Prefix to add to all endpoint paths (default: '/players')	'/players'

Source code in src/cuemsengine/NodeEngine.py

def add_player_endpoints(self, cue: Cue, prefix: str = '/players'):
    """Add player endpoints from a cue to the OSCQuery server

    Args:
        cue: The cue containing the player client
        prefix: Prefix to add to all endpoint paths (default: '/players')
    """
    if not hasattr(cue, '_osc') or cue._osc is None:
        Logger.warning(f'Cue {cue.id} has no OSC client, cannot add endpoints')
        return

    try:
        # Get endpoints from the player client
        endpoints = cue._osc.get_endpoints()
        if not endpoints:
            Logger.warning(f'No endpoints found for cue {cue.id}')
            return

        # Add prefix to all endpoints
        prefixed_endpoints = add_prefix_to_all(endpoints, f"{prefix}/{self.cm.node_uuid}/{cue.id}")

        # Add endpoints to OSCQuery server
        if hasattr(self, 'oscquery_server') and self.oscquery_server:
            self.oscquery_server.add_endpoints(prefixed_endpoints)
            Logger.debug(f'Added {len(prefixed_endpoints)} endpoints for cue {cue.id}')
        else:
            Logger.warning('OSCQuery server not initialized, cannot add endpoints')
    except Exception as e:
        Logger.error(f'Error adding player endpoints for cue {cue.id}: {e}')
        Logger.exception(e)

deploy_media(project)

Deploy the media files (and their .idx sidecar indexes).

Best-effort: a failure here is recoverable if media is already cached on disk. Returns False to surface the failure to logs, but the caller continues the load.

Source code in src/cuemsengine/NodeEngine.py

def deploy_media(self, project):
    """Deploy the media files (and their .idx sidecar indexes).

    Best-effort: a failure here is recoverable if media is already
    cached on disk. Returns False to surface the failure to logs,
    but the caller continues the load.
    """
    if not self.script:
        Logger.error('No script loaded')
        return False
    bare_names = self.script.get_own_media_filenames(config=self.cm)
    if len(bare_names) == 0:
        Logger.info('No media files to deploy')
        return True
    if not self.deploy_manager.sync_files(project, 'media', bare_names):
        Logger.error(
            f'Media deploy failed for {project} — continuing with cached '
            f'files; cues whose media is missing locally will fail on GO'
        )
        return False
    return True

deploy_project(project)

Deploy the project files (script.xml, mappings.xml, settings.xml).

Critical path: if these fail to sync, the local copy may be stale and arming cues against it is unsafe. Caller is expected to abort the load on False.

Source code in src/cuemsengine/NodeEngine.py

def deploy_project(self, project):
    """Deploy the project files (script.xml, mappings.xml, settings.xml).

    Critical path: if these fail to sync, the local copy may be stale
    and arming cues against it is unsafe. Caller is expected to abort
    the load on False.
    """
    return self.deploy_manager.sync_files(project, 'project')

ensure_video_indexes()

Run cuems-videoindexer on any video files that are missing a .idx sidecar.

This is a safety net for files that were copied manually or deployed to a node that never ran the editor upload hook. For normally-uploaded files the index was already created by the editor and this is a no-op.

Source code in src/cuemsengine/NodeEngine.py

def ensure_video_indexes(self):
    """Run cuems-videoindexer on any video files that are missing a .idx sidecar.

    This is a safety net for files that were copied manually or deployed to a
    node that never ran the editor upload hook. For normally-uploaded files the
    index was already created by the editor and this is a no-op.
    """
    if not self.script:
        return
    file_names = self.script.get_own_media_filenames(config=self.cm)
    video_exts = {'.mp4', '.mov', '.avi', '.mkv', '.mpg'}
    unindexed = []
    for name in file_names:
        ext = os.path.splitext(name)[1].lower()
        if ext not in video_exts:
            continue
        full_path = PLAYER_HANDLER.media_path(name)
        idx_dir = os.path.join(os.path.dirname(full_path), 'indexes')
        idx_path = os.path.join(idx_dir, os.path.basename(full_path) + '.idx')
        if not os.path.exists(idx_path):
            unindexed.append(full_path)
    if unindexed:
        Logger.info(
            f'ensure_video_indexes: indexing {len(unindexed)} video(s) missing .idx: '
            f'{[os.path.basename(p) for p in unindexed]}'
        )
        try:
            result = subprocess.run(
                ['cuems-videoindexer'] + unindexed,
                timeout=600,
                capture_output=True,
                text=True,
            )
        except Exception as e:
            Logger.warning(f'ensure_video_indexes: indexer failed: {e}')
            return
        if result.returncode == 0:
            Logger.info(f'ensure_video_indexes: indexer ok (rc=0)')
            if result.stdout:
                Logger.debug(f'ensure_video_indexes stdout: {result.stdout.strip()}')
        else:
            Logger.warning(
                f'ensure_video_indexes: indexer returned rc={result.returncode}. '
                f'stdout={result.stdout.strip()!r} stderr={result.stderr.strip()!r}'
            )

load_project(project)

Load the project files to the node

Source code in src/cuemsengine/NodeEngine.py

def load_project(self, project):
    """Load the project files to the node"""
    with self._loading_lock:
        if self._loading:
            Logger.warning(f'Load already in progress, ignoring duplicate load of {project}')
            return
        self._loading = True

    try:
        return self._load_project_inner(project)
    finally:
        with self._loading_lock:
            self._loading = False

map_cue_outputs(cuelist=None)

Load the output mappings for the project

Source code in src/cuemsengine/NodeEngine.py

def map_cue_outputs(self, cuelist: CueList = None):
    """Load the output mappings for the project"""
    outputs_map = {}
    if cuelist is None:
        cuelist = self.script.cuelist
    for cue in cuelist.contents:
        if isinstance(cue, CueList):
            outputs_map.update(self.map_cue_outputs(cue))
        elif not isinstance(cue, MediaCue):
            continue

        outputs = [x[1] for x in cue.get_all_output_names() if x[0] == self.cm.node_uuid]
        if outputs:
            outputs_map[cue.id] = outputs
    Logger.debug(f'Outputs map: {outputs_map}')
    return outputs_map

quit_dmx_devs()

Quit the DMX player if it exists

Source code in src/cuemsengine/NodeEngine.py

def quit_dmx_devs(self):
    """Quit the DMX player if it exists"""
    dmx_client = PLAYER_HANDLER.get_dmx_player_client()
    if dmx_client:
        try:
            dmx_client.set_value('/quit', 1)
        except Exception as e:
            Logger.exception(e)
    CUE_HANDLER.communications_thread.remove_player(f'dmxplayer_{self.cm.node_uuid}')

ready_project(project)

Prepare the project to be played.

deploy_project() runs in _load_project_inner BEFORE this point, before the teardown of the previous project — so that a deploy failure aborts the load without destroying running state. Media deploy runs here because it is best-effort: a failure logs but does not abort (cached media may already be on disk).

Source code in src/cuemsengine/NodeEngine.py

def ready_project(self, project):
    """Prepare the project to be played.

    deploy_project() runs in _load_project_inner BEFORE this point,
    before the teardown of the previous project — so that a deploy
    failure aborts the load without destroying running state.
    Media deploy runs here because it is best-effort: a failure
    logs but does not abort (cached media may already be on disk).
    """
    self.cm.load_project_config(project)
    self.read_script(project)
    self.deploy_media(project)
    self.ensure_video_indexes()
    self.outputs_map = self.map_cue_outputs()
    PLAYER_HANDLER.set_outputs_map(self.outputs_map)
    PORT_HANDLER.clean_random_ports()

ready_script()

Check if the script is ready to be played

Source code in src/cuemsengine/NodeEngine.py

def ready_script(self):
    """Check if the script is ready to be played"""
    if not self.script:
        Logger.warning('No script loaded, cannot process GO command.')
        return

    self.ongoing_cue = None
    self.next_cue_pointer = None
    self.go_offset = 0
    self._project_generation += 1  # Abort in-flight daemon arm threads
    self.unload_video_devs()
    CUE_HANDLER.disarm_all()

    # Reset mixer volumes to default when preparing script
    mixer_client = PLAYER_HANDLER.get_audio_mixer_client()
    if mixer_client:
        mixer_client.reset_volumes()

    self.initial_cuelist_process()

    # Set initial nextcue to the first enabled cue in the script
    if self.script.cuelist.contents:
        first_enabled = None
        for c in self.script.cuelist.contents:
            if c.enabled:
                first_enabled = c
                break
        self.next_cue_pointer = first_enabled

    Logger.info(f'Script {self.script.name} loaded and ready to be played')

set_audio_players()

Set the audio players and audio mixer

Source code in src/cuemsengine/NodeEngine.py

def set_audio_players(self):
    """Set the audio players and audio mixer"""
    # Initialize the audio mixer for this node
    if self.cm.node_hw_outputs.get('audio_outputs'):
        audio_outputs = self.cm.node_hw_outputs['audio_outputs']
        Logger.info(f'Initializing audio mixer with {len(audio_outputs)} outputs')

        # Assign a port for the audio mixer
        mixer_id = '0' # TODO: make this a unique identifier for the mixer
        mixer_ports = PORT_HANDLER.assign_ports(['audio_mixer'])
        PORT_HANDLER.add_config_ports(mixer_ports)
        # Start the audio mixer
        try:
            PLAYER_HANDLER.start_audio_mixer(
                audio_outputs=audio_outputs,
                port=mixer_ports['audio_mixer'],
                mixer_id=mixer_id,
                path=self.cm.node_conf['audiomixer']['path'],
                args=self.cm.node_conf['audiomixer']['args']
            )
            Logger.info(f'Audio mixer started successfully for mixer {mixer_id}')
            # Register mixer with Controller via NNG
            try:
                CUE_HANDLER.communications_thread.add_player(f'audiomixer_{mixer_id}', None, timeout=0.1)
                Logger.info(f'Audio mixer {mixer_id} registered with Controller')
            except Exception as e:
                Logger.warning(f'Could not register mixer with Controller: {e}')
        except Exception as e:
            Logger.error(f'Error starting audio mixer: {e}')
            Logger.exception(e)
    else:
        Logger.info('No audio outputs detected, skipping audio mixer initialization')

    # Build audio output lookup keyed by <id> (mirrors video output pattern)
    audio_outputs = {}
    for port_type_dict in self.cm.node_mappings.get('audio', []):
        for port_type_list in port_type_dict.values():
            for port in port_type_list:
                for _, output_data in port.items():
                    output_id = str(output_data.get('id', output_data['name']))
                    mappings = output_data.get('mappings', [])
                    mapped_to = mappings[0]['mapped_to'] if mappings else output_data['name']
                    audio_outputs[output_id] = {
                        'name': output_data['name'],
                        'mapped_to': mapped_to,
                    }
    PLAYER_HANDLER.set_audio_outputs(audio_outputs)

    # Set the audio player generator. Append --output-latency-ms
    # from settings.xml when the operator supplied an integer
    # override (isinstance int); "auto" or absent ⇒ audioplayer
    # runs its Phase-3 JACK-latency query path.
    audio_args = _append_output_latency_flag(
        self.cm.node_conf['audioplayer']['args'],
        self.cm.node_conf['audioplayer'],
    )
    PLAYER_HANDLER.set_audio_output_generator(
        self.cm.node_conf['audioplayer']['path'],
        audio_args,
    )

set_dmx_players()

Set the DMX player for this node and register its endpoints.

Source code in src/cuemsengine/NodeEngine.py

def set_dmx_players(self):
    """Set the DMX player for this node and register its endpoints."""
    # Assign a port for the DMX player
    dmx_ports = PORT_HANDLER.assign_ports(['dmx_player'])
    PORT_HANDLER.add_config_ports(dmx_ports)

    # Get node UUID for player naming
    node_uuid = self.cm.node_conf.get('uuid', 'default_node')

    # Start the DMX player
    try:
        # Append --output-latency-ms from settings.xml when an
        # integer override is present. Dmx has no "auto" form —
        # absent ⇒ dmxplayer's 35 ms Phase-5A default stands.
        dmx_args = _append_output_latency_flag(
            self.cm.node_conf['dmxplayer']['args'],
            self.cm.node_conf['dmxplayer'],
        )
        PLAYER_HANDLER.start_dmx_player(
            port=dmx_ports['dmx_player'],
            node_uuid=node_uuid,
            path=self.cm.node_conf['dmxplayer']['path'],
            args=dmx_args,
        )
        try:
            CUE_HANDLER.communications_thread.add_player(f'dmxplayer_{node_uuid}', None, timeout=0.1)
        except Exception:
            pass  # Ignore - NNG is for distributed nodes
        Logger.info(f'DMX player started successfully for node {node_uuid}')
    except Exception as e:
        Logger.error(f'Error starting DMX player: {e}')
        Logger.exception(e)
        return

set_gradient_client()

Wire GradientClient into PLAYER_HANDLER using settings from node_conf.

Source code in src/cuemsengine/NodeEngine.py

def set_gradient_client(self) -> None:
    """Wire GradientClient into PLAYER_HANDLER using settings from node_conf."""
    port = int(self.cm.node_conf['gradient_osc_port'])
    PLAYER_HANDLER.set_gradient_client(port=port, node_uuid=self.cm.node_uuid)

set_next_cue(value)

Handle setnextcue command from the UI — override next_cue_pointer.

Source code in src/cuemsengine/NodeEngine.py

def set_next_cue(self, value):
    """Handle setnextcue command from the UI — override next_cue_pointer."""
    if not self.script:
        Logger.warning('No script loaded, cannot set next cue.')
        return
    cue = self.script.find(value)
    if cue:
        self.next_cue_pointer = cue
        if not CUE_HANDLER.find_armed_cue(cue):
            Logger.info(f'Re-arming cue {cue.id} selected as next cue')
            CUE_HANDLER.arm(cue, init=True)
        CUE_HANDLER._arm_ahead(cue)  # extend window from selected cue
        self._broadcast_nextcue()
        Logger.info(f'Next cue overridden by UI: {value}')
    else:
        Logger.warning(f'setnextcue: cue {value} not found in script')

set_oscquery_comms()

Set up the command dictionary for the NodeEngine.

Commands are received via NNG from ControllerEngine. OSCQuery client is no longer used since pyossia server was removed.

Source code in src/cuemsengine/NodeEngine.py

def set_oscquery_comms(self):
    """Set up the command dictionary for the NodeEngine.

    Commands are received via NNG from ControllerEngine.
    OSCQuery client is no longer used since pyossia server was removed.
    """
    self.commands_dict = {
        'deploy': self.ready_project,
        'load': self.load_project,
        'loadcue': None,
        'go': self.go_script,
        'gocue': self.go_script,
        'pause': None,
        'resetall': None,
        'stop': self.stop_playback,
        'setnextcue': self.set_next_cue,
        'cue_enabled': self._handle_cue_enabled,
        'test': None,
        'unload': None,
        'update': None,
    }

set_video_players()

Set the video players

Source code in src/cuemsengine/NodeEngine.py

def set_video_players(self):
    """Set the video players"""
    Logger.info(f'Setting video players with: {self.cm.node_conf["videoplayer"]}')
    if not self.cm.node_hw_outputs['video_outputs']:
        Logger.info('No video outputs detected.')
        return

    PLAYER_HANDLER.add_node_uuid(self.cm.node_uuid)
    vc_conf = self.cm.node_conf.get('videoplayer', {})
    osc_video_port = int(vc_conf.get('osc_port', VIDEOCOMPOSER_OSC_PORT_DEFAULT))
    PLAYER_HANDLER.set_video_client(osc_video_port)
    PORT_HANDLER.add_config_ports({'videocomposer': osc_video_port})

    # Canvas geometry comes from /run/cuems/display.conf, written by
    # cuems-generate-display-conf (videocomposer's ExecStartPre). It's the
    # same file the videocomposer reads, so engine + VC agree on canvas
    # size and per-output regions without a handshake. The XML's optional
    # <canvas_region> is a UI-template hint (normalized [0,1]) and is
    # ignored here — engine never sources physical layout from XML.
    display_regions, (canvas_w, canvas_h) = read_display_conf()

    video_outputs = {}
    for port_type_dict in self.cm.node_mappings.get('video', []):
        for port_type_list in port_type_dict.values():
            for port in port_type_list:
                for _, output_data in port.items():
                    output_id = str(output_data.get('id', output_data['name']))
                    name = output_data['name']
                    mappings = output_data.get('mappings', [])
                    mapped_to = mappings[0]['mapped_to'] if mappings else name
                    region = display_regions.get(mapped_to)
                    if region is None:
                        Logger.warning(
                            f"DISPLAY_MISMATCH: XML output id={output_id} "
                            f"name={name!r} maps to {mapped_to!r} which is "
                            f"not in display.conf; skipping. Available: "
                            f"{sorted(display_regions.keys())}"
                        )
                        continue
                    video_outputs[output_id] = {
                        'name': name,
                        'mapped_to': mapped_to,
                        'x': region['x'],
                        'y': region['y'],
                        'width': region['width'],
                        'height': region['height'],
                        'canvas_region': dict(region),
                    }
    PLAYER_HANDLER.start_video_outputs(
        video_outputs, canvas_override=(canvas_w, canvas_h)
    )

stop_node_engine()

Stop the NodeEngine elements

Source code in src/cuemsengine/NodeEngine.py

def stop_node_engine(self):
    """Stop the NodeEngine elements"""
    CUE_HANDLER.disarm_all()
    self.stop_video_devs()

stop_playback(value=None)

Stop playback, full cleanup, then re-arm so GO is available again.

Does the cleanup that ready_script() doesn't handle (DMX blackout, disconnect video, kill audio), then delegates reset + re-arm to ready_script(). Notifies Controller when armed (GO button green).

Source code in src/cuemsengine/NodeEngine.py

def stop_playback(self, value=None):
    """Stop playback, full cleanup, then re-arm so GO is available again.

    Does the cleanup that ready_script() doesn't handle (DMX blackout,
    disconnect video, kill audio), then delegates reset + re-arm to
    ready_script(). Notifies Controller when armed (GO button green).
    """
    Logger.info('STOP command received. Stopping playback.')

    self.set_status('running', "no")

    gradient_client = PLAYER_HANDLER.get_gradient_client()
    if gradient_client:
        try:
            gradient_client.send_cancel_all()
        except Exception as exc:
            Logger.error(f'gradient send_cancel_all failed on stop: {exc}')
    else:
        Logger.debug('gradient_client not initialised, skipping cancel_all on stop')

    # Signal all running cue threads to stop immediately.
    # Must happen BEFORE blackout/reset so loop_cue threads don't
    # re-push DMX frames or send /visible after cleanup.
    CUE_HANDLER.stop_all_cues()
    sleep(0.05)  # 50ms — loop_cue polls every 20ms

    # DMX: disable MTC following first (freezes the playhead so queued
    # scenes can't fire), then blackout via OLA for instant visual reset.
    dmx_client = PLAYER_HANDLER.get_dmx_player_client()
    if dmx_client:
        try:
            dmx_client.disable_mtcfollow()
        except Exception as e:
            Logger.warning(f'DMX disable mtcfollow failed: {e}')
        try:
            dmx_client.send_blackout()
        except Exception as e:
            Logger.warning(f'DMX blackout failed: {e}')

    # Unload all video layers (instant visual blackout)
    self.unload_video_devs()

    # Kill all audio players (ready_script does not do this)
    PLAYER_HANDLER.kill_all_audio_players()
    PLAYER_HANDLER.cleanup_zombie_jack_clients()

    # Reset state + disarm + volume reset + re-arm cues
    if self.script:
        self.ready_script()
        Logger.info(f'Project {self.script.name} reset and ready for GO.')

        # Notify Controller that re-arm is complete (GO button can go green)
        try:
            from .comms.NodesHub import NodeOperation, OperationType, ActionType
            operation = NodeOperation(
                type=OperationType.STATUS,
                action=ActionType.UPDATE,
                sender=self.cm.node_uuid,
                target='armed_ready',
                data={'armed': 'yes'}
            )
            CUE_HANDLER.communications_thread.send_operation(operation, timeout=0.1)
            Logger.debug('Notified Controller that re-arm is complete')
        except Exception as e:
            Logger.warning(f'Could not notify Controller of armed_ready: {e}')

        # Broadcast nextcue (reset to first cue after stop)
        self._broadcast_nextcue()
    else:
        Logger.info('Playback stopped (no script loaded).')

    Logger.info('Playback stopped.')

get_config_ports(node_conf)

Create a dict of ports from the config

Source code in src/cuemsengine/NodeEngine.py

def get_config_ports(node_conf: dict) -> dict:
    """Create a dict of ports from the config"""
    k = [i for i in node_conf.keys() if 'port' in i and is_int(node_conf[i])]
    v = [int(node_conf[i]) for i in k]
    return dict(zip(k, v))

is_int(value)

Check if a value is an integer

Source code in src/cuemsengine/NodeEngine.py

def is_int(value: any) -> bool:
    """Check if a value is an integer"""
    try:
        int(value)
        return True
    except ValueError:
        return False

redirect_audio_cmd(path_parts, value)

Redirect the audio command to the audio player

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio command to the audio player"""
    if path_parts[0] == 'mixer':
        redirect_audio_mixer_cmd(path_parts[1:], value)
    elif path_parts[0] == 'cue':
        redirect_audio_player_cmd(path_parts[1:], value)
    else:
        Logger.error(f'Invalid audio command: {path_parts}')
        return

redirect_audio_mixer_cmd(path_parts, value)

Redirect the audio mixer command to the audio mixer Follows the logic: /master/volume -> /audiomixer/0_mixer/master /0/volume -> /audiomixer/0_mixer/0 /1/volume -> /audiomixer/0_mixer/1 ... Args: path_parts: List of path parts value: Value to set

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_mixer_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio mixer command to the audio mixer
     Follows the logic:
     <output_index>/master/volume -> /audiomixer/0_mixer/master
     <output_index>/0/volume -> /audiomixer/0_mixer/0
     <output_index>/1/volume -> /audiomixer/0_mixer/1
     ...
    Args:
        path_parts: List of path parts
        value: Value to set
    """
    output_index, channel, _ = path_parts
    mixer_cmd = f'/audiomixer/0_mixer/{channel}'
    PLAYER_HANDLER.get_audio_mixer_client().set_value(mixer_cmd, value)

redirect_audio_player_cmd(path_parts, value)

Redirect the audio mixer command to the audio mixer Follows the logic: /master/volume -> /volmaster /0/volume -> /vol0 /1/volume -> /vol1 ...

Parameters:

Name	Type	Description	Default
path_parts	list[str]	List of path parts	required
value	str	Value to set	required

Source code in src/cuemsengine/NodeEngine.py

def redirect_audio_player_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the audio mixer command to the audio mixer
     Follows the logic:
     <cue_uuid>/master/volume -> /volmaster
     <cue_uuid>/0/volume -> /vol0
     <cue_uuid>/1/volume -> /vol1
     ...

    Args:
        path_parts: List of path parts
        value: Value to set
    """
    cue_uuid, channel, _ = path_parts
    audio_cmd = f'/vol{channel}'
    cue = CUE_HANDLER.get_armed_cue(cue_uuid)
    if not cue:
        Logger.error(f'Cue {cue_uuid} not found')
        return
    client: AudioClient = cue._osc
    client.set_value(audio_cmd, value)

redirect_dmx_cmd(path_parts, value)

Redirect the DMX command to the DMX player

Source code in src/cuemsengine/NodeEngine.py

def redirect_dmx_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the DMX command to the DMX player"""
    dmx_index = path_parts.index('mixer') + 1 # +1 to skip the 'mixer' keyword
    dmx_cmd = '/' + '/'.join(path_parts[dmx_index:])
    client: DmxClient = PLAYER_HANDLER.get_dmx_player_client()
    client.set_value(dmx_cmd, value)

redirect_video_cmd(path_parts, value)

Redirect the video command to the video client

Source code in src/cuemsengine/NodeEngine.py

def redirect_video_cmd(path_parts: list[str], value: str) -> None:
    """Redirect the video command to the video client"""
    videocomposer_index = path_parts.index('videocomposer')
    videocomposer_cmd = '/' + '/'.join(path_parts[videocomposer_index:])
    client: VideoClient = PLAYER_HANDLER.get_video_client()
    client.set_value(videocomposer_cmd, value)