Tools

CuemsDeploy: async rsync-based media and project deployer for CUEMS nodes.

Async model: _sync(), _check_mandatory_sources(), _kill(), and _pump() are coroutines scheduled on the event loop injected via NodeEngine.start() late-bind. sync_files() remains a synchronous blocking API; it submits _deploy_all_async() via run_coroutine_threadsafe() and blocks until completion. Watchdogs inside the coroutine bound all wait times.

Late-bind protocol

NodeEngine.init() creates CuemsDeploy(loop=None). NodeEngine.start() calls CUE_HANDLER.set_nng_comms(...), which starts AsyncCommsThread (creates the event_loop), then: self.deploy_manager.loop = CUE_HANDLER.communications_thread.event_loop Any sync_files() before that bind returns False immediately.

Why _avahi_resolve stays synchronous: it is called from init() before any asyncio loop exists, so subprocess.run() (short timeout) is correct.

CuemsDeploy

Source code in src/cuemsengine/tools/CuemsDeploy.py

class CuemsDeploy:
    _RSYNC_PASSWORD: ClassVar[str] = "f48t5eL2kLHw2Wfw"

    def __init__(
        self,
        library_path="/opt/cuems_library/",
        tmp_path="/tmp/cuems_library/",
        controller_ip: str | None = None,
        hostname: str | None = None,
        log_file: str = "/run/cuems/rsync.log",
        on_progress: Callable[[dict], None] | None = None,
        loop: asyncio.AbstractEventLoop | None = None,
        is_async: bool = False,
    ):
        """Construct a deploy manager.

        Args:
            controller_ip: IP of the controller's rsync daemon (preferred).
                Pass BaseEngine.controller_ip. If falsy, manager runs disabled.
            hostname: Legacy fallback resolved via avahi when controller_ip is
                not provided. Kept for backwards compatibility.
            log_file: Where rsync writes its log.
            on_progress: Optional callback fired for each rsync progress update
                (parsed from --info=progress2). Receives a dict with keys
                bytes, pct, rate, eta, and optionally xfr, remaining, total.
                Must be non-blocking — invoked from the asyncio loop.
            loop: The asyncio event loop for run_coroutine_threadsafe().
            Defaults
                to None; late-bind via NodeEngine.start().
        """
        self.library_path = library_path
        self.tmp_path = tmp_path
        self.log_file = log_file
        self.errors = []
        self.encoding = sys.getfilesystemencoding()
        self._on_progress = on_progress or (lambda parsed: None)
        self.loop = loop
        self._is_async = is_async

        # TODO: rebuild on network_map reload to pick up IP changes without
        # restarting.
        if controller_ip:
            self.main_ip = controller_ip
        elif hostname:
            self.main_ip = self._avahi_resolve(hostname)
        else:
            self.main_ip = None

        if self.main_ip:
            self.address = f"rsync://cuems_library_rsync@{self.main_ip}/cuems"
            self.enabled = True
        else:
            self.address = None
            self.enabled = False
            Logger.warning(
                f"CuemsDeploy disabled: no valid controller IP "
                f"(controller_ip={controller_ip!r}, hostname={hostname!r}, "
                f"resolved={self.main_ip!r}). Project deploys will be skipped."
            )

    def sync_files(
        self, project: str, tag: str, file_names: list[str] | None = None
    ) -> bool:
        """Sync files from the controller to the node.

        Dispatches to _sync_files_async (event-loop path) or _sync_files_blocking
        (subprocess path) based on self._is_async.

        Args:
            project: Project identifier used to build paths and log file names.
            tag: Transfer type — ``'project'`` or ``'media'``.
            file_names: Explicit list of rsync-relative paths to transfer.

        Returns:
            True on success; False on any failure. self.errors populated on failure.
        """
        if not self.enabled:
            Logger.error(
                f"CuemsDeploy is disabled (no controller IP) — "
                f"skipping {tag} sync for project {project!r}"
            )
            return False

        file_names = list(file_names or [])
        if self._is_async:
            return self._sync_files_async(project, tag, file_names)
        return self._sync_files_blocking(project, tag, file_names)

    def _sync_files_async(self, project: str, tag: str, file_names: list[str]) -> bool:
        """Async deploy path — requires self.loop to be bound."""
        if self.loop is None:
            Logger.error(
                f"CuemsDeploy event loop not bound (NodeEngine.start() not "
                f"called yet) — skipping {tag} sync for project {project!r}"
            )
            self.errors = ["event loop not bound"]
            return False

        if tag == "project" and len(file_names) == 0:
            file_names = self._project_files(project)
        elif tag == "media" and len(file_names) > 0:
            file_names = self._media_files(file_names)

        mandatory_paths = self._mandatory_paths(project, tag)
        log_file = self._deploy_log_path(project, tag)

        try:
            coro = self._deploy_all_async(log_file, file_names, mandatory_paths)
            synced = asyncio.run_coroutine_threadsafe(coro, self.loop).result()
        except Exception as e:
            Logger.error(f"Unexpected error during deploy of {project!r}: {e}")
            self.errors = [str(e)]
            return False

        if synced:
            self._reset_deploy_log(log_file)
            self.errors = []
        else:
            Logger.error(
                f"Failed to sync {tag} files for project {project!r} "
                f"from {self.address} (log: {log_file})"
            )
            for error in self.errors:
                Logger.error(error)
        return synced

    def _sync_files_blocking(
        self, project: str, tag: str, file_names: list[str]
    ) -> bool:
        """Synchronous deploy path — no event loop required."""
        if not self.enabled:
            Logger.error(
                f"CuemsDeploy is disabled (no controller IP) — "
                f"skipping {tag} sync for project {project!r}"
            )
            return False

        if tag == "project" and len(file_names) == 0:
            file_names = self._project_files(project)
        elif tag == "media" and len(file_names) > 0:
            file_names = self._media_files(file_names)

        log_file = self._deploy_log_path(project, tag)
        self._create_deploy_log(log_file, file_names)
        synced = self._sync_blocking(log_file)

        if synced:
            self._reset_deploy_log(log_file)
            self.errors = []
        else:
            Logger.error(
                f"Failed to sync {tag} files for project {project!r} "
                f"from {self.address} (log: {log_file})"
            )
            for error in self.errors:
                Logger.error(error)
        return synced

    def _kill_blocking(self, proc) -> None:
        """Terminate proc gracefully, escalating to SIGKILL after 2 s."""
        proc.terminate()
        try:
            proc.wait(timeout=2)
        except subprocess.TimeoutExpired:
            proc.kill()
            try:
                proc.wait(timeout=2)
            except subprocess.TimeoutExpired:
                pass

    def _sync_blocking(self, path: str) -> bool:
        """Blocking rsync with selectors-based I/O supervision and watchdog."""
        try:
            os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
        except OSError as e:
            Logger.warning(f"Could not create rsync log directory: {e}")

        cmd = [
            "rsync",
            "-rt",
            "--delete",
            "--delete-delay",
            "--info=progress2,name0",
            "--stats",
            "--contimeout=2",
            "--timeout=5",
            "--ignore-missing-args",
            f"--files-from={path}",
            f"--log-file={self.log_file}",
            self.address,
            self.library_path,
        ]
        env = dict(os.environ, RSYNC_PASSWORD=self._RSYNC_PASSWORD)

        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
            bufsize=0,
        )

        for fd in (proc.stdout.fileno(), proc.stderr.fileno()):
            flags = fcntl.fcntl(fd, fcntl.F_GETFL)
            fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)

        sel = selectors.DefaultSelector()
        sel.register(proc.stdout, selectors.EVENT_READ, "out")
        sel.register(proc.stderr, selectors.EVENT_READ, "err")

        bufs = {"out": "", "err": ""}
        stderr_lines: list[str] = []
        started = False
        deadline = time.monotonic() + _STARTUP_DEADLINE_S
        rc: int | None = None

        try:
            while sel.get_map():
                budget = (deadline - time.monotonic()) if not started else _INACTIVITY_S
                events = sel.select(timeout=max(budget, 0.1))
                if not events:
                    reason = (
                        "no output within startup deadline"
                        if not started
                        else "no output within inactivity threshold"
                    )
                    self._kill_blocking(proc)
                    self.errors = [f"rsync {reason} (target: {self.address})"]
                    return False

                for key, _ in events:
                    tag = key.data
                    try:
                        chunk = os.read(key.fd, 4096)
                    except BlockingIOError:
                        continue
                    if not chunk:
                        sel.unregister(key.fileobj)
                        if bufs[tag]:
                            self._dispatch_line(tag, bufs[tag], stderr_lines)
                            bufs[tag] = ""
                        continue
                    started = True
                    bufs[tag] += chunk.decode(errors="replace")
                    *parts, bufs[tag] = re.split(r"[\r\n]", bufs[tag])
                    for p in parts:
                        if p:
                            self._dispatch_line(tag, p, stderr_lines)

            try:
                rc = proc.wait(timeout=_INACTIVITY_S)
            except subprocess.TimeoutExpired:
                self._kill_blocking(proc)
                self.errors = [
                    f"rsync closed pipes but did not exit within "
                    f"{_INACTIVITY_S}s (target: {self.address})"
                ]
                return False
        finally:
            if proc.poll() is None:
                self._kill_blocking(proc)
            sel.close()

        if rc == 0:
            self.errors = []
            return True
        self.errors = (
            stderr_lines[:-1]
            if stderr_lines and "rsync error:" in stderr_lines[-1]
            else stderr_lines
        )
        return False

    def _mandatory_paths(self, project: str, tag: str) -> list[str]:
        if tag != "project":
            return []
        return [f"/projects/{project}/script.xml"]

    async def _check_mandatory_sources(
        self, mandatory_paths: list[str]
    ) -> tuple[bool, list[str]]:
        """Verify mandatory remote paths exist before bulk transfer.

        Uses one rsync --list-only probe with all mandatory paths. Output is
        consumed via proc.communicate() — the probe is short-lived and has
        bounded output, so streaming is unnecessary (proc.communicate()
        rationale: see research.md §2).
        """
        mandatory_paths = [p.strip() for p in mandatory_paths if p.strip()]
        if not mandatory_paths:
            return True, []

        env = dict(os.environ, RSYNC_PASSWORD=self._RSYNC_PASSWORD)
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding=self.encoding,
            delete=False,
            prefix="rsync_mandatory_",
            suffix=".lst",
        ) as probe_list:
            for source_path in mandatory_paths:
                probe_list.write(f"{source_path}\n")
            probe_list_path = probe_list.name

        try:
            proc = await asyncio.create_subprocess_exec(
                "rsync",
                "-r",
                "--list-only",
                "--contimeout=2",
                "--timeout=5",
                f"--files-from={probe_list_path}",
                self.address,
                self.library_path,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env=env,
            )
            _, stderr_bytes = await proc.communicate()
        finally:
            try:
                os.remove(probe_list_path)
            except OSError:
                pass

        if proc.returncode == 0:
            return True, []

        stderr = stderr_bytes.decode(errors="replace").strip()
        missing = []
        for source_path in mandatory_paths:
            if source_path in stderr and (
                "No such file or directory" in stderr
                or "link_stat" in stderr
                or "failed to stat" in stderr
            ):
                missing.append(source_path)

        if missing:
            return False, missing

        self.errors = [
            f"rsync mandatory precheck failed: "
            f'{stderr or f"exit code {proc.returncode}"}'
        ]
        return False, []

    def _avahi_resolve(self, hostname: str) -> str | None:
        """Resolve a hostname via avahi-resolve-host-name.

        Stays synchronous: called from __init__() before any asyncio loop
        exists. Returns the IP string on success, or None on failure.
        """
        try:
            result = subprocess.run(
                ["avahi-resolve-host-name", "-n", hostname],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=5,
            )
            result.check_returncode()
            ip = result.stdout.decode(self.encoding).replace(hostname, "").strip()
            return ip if ip else None
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
            return None

    async def _pump(
        self, stream: asyncio.StreamReader, tag: str, queue: asyncio.Queue
    ) -> None:
        """Read 4096-byte chunks until EOF; push (tag, chunk) then (tag, None).

        (tag, None) is the EOF sentinel consumed by _sync's driver loop.
        """
        while True:
            chunk = await stream.read(4096)
            if not chunk:
                await queue.put((tag, None))
                return
            await queue.put((tag, chunk))

    async def _sync(self, path: str) -> bool:
        """Async rsync transfer with two concurrent reader tasks and watchdogs.

        Watchdog state machine (data-model.md):
          STARTUP → empty asyncio.wait result before first queue item → KILLED
          ACTIVE  → empty asyncio.wait result after receiving data    → KILLED
          DONE    → both pipes closed, proc exits rc=0                → True
          DONE    → both pipes closed, proc exits rc≠0                → False

        Queue pattern: two _pump tasks push to a shared asyncio.Queue; the
        main driver loop drains it after each asyncio.wait() call. The
        watchdog deadline resets on every received chunk.
        """
        try:
            os.makedirs(os.path.dirname(self.log_file), exist_ok=True)
        except OSError as e:
            Logger.warning(f"Could not create rsync log directory: {e}")

        # -t: rsync shows >f..T...... without it; also breaks .idx mtime cache
        # (2026-05-19).
        cmd = [
            "rsync",
            "-rt",
            "--delete",
            "--delete-delay",
            "--info=progress2,name0",
            "--stats",
            "--contimeout=2",
            "--timeout=5",
            "--ignore-missing-args",
            f"--files-from={path}",
            f"--log-file={self.log_file}",
            self.address,
            self.library_path,
        ]
        env = dict(os.environ, RSYNC_PASSWORD=self._RSYNC_PASSWORD)

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=env,
        )

        queue: asyncio.Queue = asyncio.Queue()
        t_out = asyncio.create_task(self._pump(proc.stdout, "out", queue))
        t_err = asyncio.create_task(self._pump(proc.stderr, "err", queue))

        bufs = {"out": "", "err": ""}
        stderr_lines: list[str] = []
        started = False
        deadline = asyncio.get_event_loop().time() + _STARTUP_DEADLINE_S
        pipes_done = 0
        rc: int | None = None
        # Only pending tasks: avoids busy-loop once one pump finishes before
        # the other.
        pending = {t_out, t_err}

        try:
            while pipes_done < 2:
                budget = deadline - asyncio.get_event_loop().time()
                done, pending = await asyncio.wait(pending, timeout=max(budget, 0.1))
                # Drain before watchdog: pump can push data without completing
                # its task.
                got_data = False
                while not queue.empty():
                    tag, chunk = queue.get_nowait()
                    if chunk is None:
                        if bufs[tag]:
                            self._dispatch_line(tag, bufs[tag], stderr_lines)
                            bufs[tag] = ""
                        pipes_done += 1
                        continue
                    got_data = True
                    started = True
                    deadline = asyncio.get_event_loop().time() + _INACTIVITY_S
                    bufs[tag] += chunk.decode(errors="replace")
                    *parts, bufs[tag] = re.split(r"[\r\n]", bufs[tag])
                    for p in parts:
                        if p:
                            self._dispatch_line(tag, p, stderr_lines)
                if not done and not got_data:
                    reason = (
                        "no output within startup deadline"
                        if not started
                        else "no output within inactivity threshold"
                    )
                    t_out.cancel()
                    t_err.cancel()
                    await asyncio.gather(t_out, t_err, return_exceptions=True)
                    await self._kill(proc)
                    self.errors = [f"rsync {reason} (target: {self.address})"]
                    return False

            try:
                rc = await asyncio.wait_for(proc.wait(), timeout=_INACTIVITY_S)
            except asyncio.TimeoutError:
                await self._kill(proc)
                self.errors = [
                    f"rsync closed pipes but did not exit within "
                    f"{_INACTIVITY_S}s (target: {self.address})"
                ]
                return False
        finally:
            if proc.returncode is None:
                await self._kill(proc)

        if rc == 0:
            self.errors = []
            return True
        # Drop the positional "rsync error: ... at main.c(NNN)" trailer if
        # present.
        self.errors = (
            stderr_lines[:-1]
            if stderr_lines and "rsync error:" in stderr_lines[-1]
            else stderr_lines
        )
        return False

    def _dispatch_line(self, tag: str, line: str, stderr_lines: list[str]) -> None:
        if tag == "out":
            Logger.debug(f"rsync: {line}")
            parsed = self._parse_progress(line)
            if parsed:
                self._on_progress(parsed)
        else:
            Logger.warning(f"rsync: {line}")
            stderr_lines.append(line)

    async def _kill(self, proc: asyncio.subprocess.Process) -> None:
        """Terminate proc gracefully, escalating to SIGKILL after 2 s."""
        proc.terminate()
        try:
            await asyncio.wait_for(proc.wait(), timeout=2.0)
        except asyncio.TimeoutError:
            proc.kill()
            try:
                await asyncio.wait_for(proc.wait(), timeout=2.0)
            except asyncio.TimeoutError:
                pass

    async def _deploy_all_async(
        self, log_file: str, file_names: list[str], mandatory_paths: list[str]
    ) -> bool:
        """Full deploy flow: precheck → log creation → rsync transfer.

        Early-fails on precheck failure; the log file is only created when
        precheck passes (preserving the pre-refactor invariant).
        """
        if mandatory_paths:
            mandatory_ok, missing = await self._check_mandatory_sources(mandatory_paths)
            if not mandatory_ok:
                if missing:
                    self.errors = [
                        f"mandatory project files are missing at source "
                        f'{self.address}: {", ".join(missing)}'
                    ]
                Logger.error("Failed mandatory precheck for project files")
                return False
        self._create_deploy_log(log_file, file_names)
        return await self._sync(log_file)

    def _parse_progress(self, line: str) -> dict[str, int | str]:
        """Parse a rsync --info=progress2 line.

        Returns {} for non-progress lines (stats block, file names, blank).
        Returns a structured dict with keys bytes, pct, rate, eta, and
        optionally xfr, remaining, total. Keep keys stable — UI consumers
        depend on them.
        """
        m = _PROGRESS2_RE.match(line)
        if not m:
            return {}
        bytes_str, pct, rate, eta, xfr, done, total = m.groups()
        out = {
            "bytes": int(bytes_str.replace(",", "")),
            "pct": int(pct),
            "rate": rate,
            "eta": eta,
        }
        if xfr is not None:
            out.update(
                {
                    "xfr": int(xfr),
                    "remaining": int(done),
                    "total": int(total),
                }
            )
        return out

    def _deploy_log_path(self, project: str, tag: str = "project") -> str:
        return os.path.join(self.tmp_path, f"rsync_request_{project}_{tag}.log")

    def _create_deploy_log(self, log_file: str, file_names: list[str] = []) -> bool:
        """Create the rsync --files-from list file for a deploy request."""
        try:
            os.makedirs(os.path.dirname(log_file), exist_ok=True)
            with open(log_file, "w") as f:
                # Normalize to one-path-per-line; callers may omit the trailing
                # newline.
                for name in file_names:
                    if not name.endswith("\n"):
                        name = name + "\n"
                    f.write(name)
        except Exception as e:
            Logger.error(f"Exception raised when writing rsync request log file: {e}")
            return False
        return True

    def _reset_deploy_log(self, log_file: str) -> None:
        with open(log_file, "w"):
            pass
        Logger.info(f"rsync Deploy log file {log_file} emptied")

    def _project_files(self, project: str) -> list[str]:
        return [
            "/projects/" + project + "/script.xml\n",
            "/projects/" + project + "/mappings.xml\n",
            "/projects/" + project + "/settings.xml\n",
        ]

    def _media_files(self, bare_names: list[str]) -> list[str]:
        """
        Expand bare media filenames to rsync-relative paths for --files-from.

        Every file gets a media/<name> entry. Video files (.mp4 .mov .avi
        .mkv .mpg) also get a media/indexes/<name>.idx sidecar entry.
        """
        _VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".mpg"}
        result = []
        for name in bare_names:
            result.append(f"media/{name}")
            if os.path.splitext(name)[1].lower() in _VIDEO_EXTS:
                result.append(f"media/indexes/{name}.idx")
        return result

__init__(library_path='/opt/cuems_library/', tmp_path='/tmp/cuems_library/', controller_ip=None, hostname=None, log_file='/run/cuems/rsync.log', on_progress=None, loop=None, is_async=False)

Construct a deploy manager.

Parameters:

Name	Type	Description	Default
controller_ip	str | None	IP of the controller's rsync daemon (preferred). Pass BaseEngine.controller_ip. If falsy, manager runs disabled.	None
hostname	str | None	Legacy fallback resolved via avahi when controller_ip is not provided. Kept for backwards compatibility.	None
log_file	str	Where rsync writes its log.	'/run/cuems/rsync.log'
on_progress	Callable[[dict], None] | None	Optional callback fired for each rsync progress update (parsed from --info=progress2). Receives a dict with keys bytes, pct, rate, eta, and optionally xfr, remaining, total. Must be non-blocking — invoked from the asyncio loop.	None
loop	AbstractEventLoop | None	The asyncio event loop for run_coroutine_threadsafe().	None

Source code in src/cuemsengine/tools/CuemsDeploy.py

def __init__(
    self,
    library_path="/opt/cuems_library/",
    tmp_path="/tmp/cuems_library/",
    controller_ip: str | None = None,
    hostname: str | None = None,
    log_file: str = "/run/cuems/rsync.log",
    on_progress: Callable[[dict], None] | None = None,
    loop: asyncio.AbstractEventLoop | None = None,
    is_async: bool = False,
):
    """Construct a deploy manager.

    Args:
        controller_ip: IP of the controller's rsync daemon (preferred).
            Pass BaseEngine.controller_ip. If falsy, manager runs disabled.
        hostname: Legacy fallback resolved via avahi when controller_ip is
            not provided. Kept for backwards compatibility.
        log_file: Where rsync writes its log.
        on_progress: Optional callback fired for each rsync progress update
            (parsed from --info=progress2). Receives a dict with keys
            bytes, pct, rate, eta, and optionally xfr, remaining, total.
            Must be non-blocking — invoked from the asyncio loop.
        loop: The asyncio event loop for run_coroutine_threadsafe().
        Defaults
            to None; late-bind via NodeEngine.start().
    """
    self.library_path = library_path
    self.tmp_path = tmp_path
    self.log_file = log_file
    self.errors = []
    self.encoding = sys.getfilesystemencoding()
    self._on_progress = on_progress or (lambda parsed: None)
    self.loop = loop
    self._is_async = is_async

    # TODO: rebuild on network_map reload to pick up IP changes without
    # restarting.
    if controller_ip:
        self.main_ip = controller_ip
    elif hostname:
        self.main_ip = self._avahi_resolve(hostname)
    else:
        self.main_ip = None

    if self.main_ip:
        self.address = f"rsync://cuems_library_rsync@{self.main_ip}/cuems"
        self.enabled = True
    else:
        self.address = None
        self.enabled = False
        Logger.warning(
            f"CuemsDeploy disabled: no valid controller IP "
            f"(controller_ip={controller_ip!r}, hostname={hostname!r}, "
            f"resolved={self.main_ip!r}). Project deploys will be skipped."
        )

sync_files(project, tag, file_names=None)

Sync files from the controller to the node.

Dispatches to _sync_files_async (event-loop path) or _sync_files_blocking (subprocess path) based on self._is_async.

Parameters:

Name	Type	Description	Default
project	str	Project identifier used to build paths and log file names.	required
tag	str	Transfer type — 'project' or 'media'.	required
file_names	list[str] | None	Explicit list of rsync-relative paths to transfer.	None

Returns:

Type	Description
bool	True on success; False on any failure. self.errors populated on failure.

Source code in src/cuemsengine/tools/CuemsDeploy.py

def sync_files(
    self, project: str, tag: str, file_names: list[str] | None = None
) -> bool:
    """Sync files from the controller to the node.

    Dispatches to _sync_files_async (event-loop path) or _sync_files_blocking
    (subprocess path) based on self._is_async.

    Args:
        project: Project identifier used to build paths and log file names.
        tag: Transfer type — ``'project'`` or ``'media'``.
        file_names: Explicit list of rsync-relative paths to transfer.

    Returns:
        True on success; False on any failure. self.errors populated on failure.
    """
    if not self.enabled:
        Logger.error(
            f"CuemsDeploy is disabled (no controller IP) — "
            f"skipping {tag} sync for project {project!r}"
        )
        return False

    file_names = list(file_names or [])
    if self._is_async:
        return self._sync_files_async(project, tag, file_names)
    return self._sync_files_blocking(project, tag, file_names)

MtcListener

Bases: Thread

Source code in src/cuemsengine/tools/MtcListener.py

class MtcListener(Thread):
    def __init__(
        self,
        step_callback: Callable | None = None,
        reset_callback: Callable | None = None,
        port: str | None = None,
    ):
        # self.main_tc = CTimecode('0:0:0:0')
        self.main_tc = CTimecode()
        self.main_tc.set_fractional(True)

        self.__quarter_frames = [0, 0, 0, 0, 0, 0, 0, 0]
        self.port = None
        self.port_name = None
        # monotonic() of the last MTC message received (quarter frame or
        # timecode full frame), None until the first one. See is_receiving.
        self.last_rx_monotonic: float | None = None
        self.__open_port(port)

        self.step_callback = step_callback
        self.reset_callback = reset_callback

        # 24h MTC rollover state (closes 869cpdbzy):
        # MIDI MTC encodes hours in a 5-bit field (0-23) and real SMPTE senders
        # reset to 00:00:00:00 after 24h. This listener detects that wrap by
        # comparing decoded TC frames against the previous one (a backward jump
        # of more than 1 hour indicates a rollover, not a manual seek) and
        # accumulates a 24h offset that is added to every subsequent decoded
        # TC. CTimecode itself is monotonic past 24h post-cuemsutils 0.1.0rc7
        # (PR #10 layer 1), but without this listener-side accumulation the
        # received MTC would reset to ~frames=1 every 24h regardless.
        self._24h_offset_frames: int = 0
        self._last_decoded_frames: int | None = None
        # Guards _24h_offset_frames + _last_decoded_frames against the race
        # between this listener's mido callback thread (running _apply_24h_offset
        # on every authoritative decode) and the engine's load thread (calling
        # reset_24h_state on project load). The read-modify-write
        # `_24h_offset_frames +=` is NOT atomic under the GIL. Mirrors the C++
        # receiver's wrapStateMutex_.
        self._wrap_lock = Lock()

        super().__init__(name="mtclistener")
        self.daemon = True

    def _apply_24h_offset(self, decoded: CTimecode) -> CTimecode:
        """Detect 24h MTC rollover and apply accumulated offset.

        Heuristic: a real 24h wrap goes from 23:59:59:F (frames ≈ 24h - 1f)
        to 00:00:00:00 (frames ≈ 0). We treat a backward jump as a 24h wrap
        only when both:
        - delta < -1h (large backward jump, not a small seek), AND
        - prev_frames was within the last hour of the 24h boundary.

        The second condition is critical: without it, a manual seek back to
        00:00:00:00 from ANY high-watermark MTC time (e.g. after the engine
        has been running for 4h and the user reloads a project, which sends
        a Full-Frame SYSEX SEEK to frame 0) is mistakenly treated as a 24h
        wrap, adding a phantom 2,160,000-frame offset that corrupts every
        downstream timestamp (cue offsets become -2160k, video layers try
        to seek to frame -2.5M of a 300-frame clip, layers stay in
        awaitingFrame forever, monitor goes black).

        For environments with manual seeking, deltas under 1h are treated
        as seeks (no offset accumulated; existing reset detection still
        fires when main_tc.milliseconds_rounded reaches 0).

        Returns the offset-adjusted CTimecode (or the original if no offset
        is active).
        """
        with self._wrap_lock:
            if self._last_decoded_frames is not None:
                delta = decoded.frames - self._last_decoded_frames
                # Real-rate constants, NOT decoded._int_framerate (the LABEL rate,
                # 30 for 29.97). At 29.97 the label rate would make a 24h offset of
                # 30*86400 = 2,592,000 frames, which CTimecode reconverts (÷ the
                # real 29.97) to 86,486,453 ms instead of a true 86,400,000 ms — an
                # 86.5s divergence from the C++ ms-domain receiver. round(...*real)
                # keeps 25/30 fps identical and corrects 29.97. (Plan 4 / audit)
                fps = float(decoded.framerate)
                frames_per_hour = round(3600.0 * fps)
                frames_per_24h = round(86400.0 * fps)
                near_24h_boundary = (
                    self._last_decoded_frames > frames_per_24h - frames_per_hour
                )
                if delta < -frames_per_hour and near_24h_boundary:
                    # 24h MTC rollover: prev was in the last hour of the day and
                    # the head jumped back > 1h → accumulate a full day.
                    self._24h_offset_frames += frames_per_24h
                    Logger.info(
                        f"MtcListener: detected 24h MTC rollover "
                        f"(prev frames={self._last_decoded_frames}, "
                        f"new={decoded.frames}, delta={delta}); "
                        f"accumulated offset = {self._24h_offset_frames} frames "
                        f"({self._24h_offset_frames / fps / 3600:.1f}h)"
                    )
                elif delta < -frames_per_hour and decoded.frames < frames_per_hour:
                    # Transport reset: an authoritative return to ~0 (new pos in
                    # the first hour) from a NON-boundary previous position. The
                    # exact inverse of the wrap predicate, so wrap vs reset stay
                    # mutually exclusive (wrap is tested first). The
                    # `decoded.frames < frames_per_hour` guard is MANDATORY — it
                    # stops a mid-range seek (e.g. 14h→3h) from wrongly zeroing.
                    # Mirrors the C++ applyWrap reset branch (wireMs < HOUR_MS).
                    self._24h_offset_frames = 0
                    Logger.info(
                        f"MtcListener: transport reset to ~0 "
                        f"(prev frames={self._last_decoded_frames}, "
                        f"new={decoded.frames}); cleared 24h offset"
                    )
                elif delta < -frames_per_hour:
                    Logger.info(
                        f"MtcListener: large backward MTC jump ignored as "
                        f"manual seek (prev frames={self._last_decoded_frames}, "
                        f"new={decoded.frames}, delta={delta}); not a 24h wrap "
                        f"(prev < {frames_per_24h - frames_per_hour})"
                    )
            self._last_decoded_frames = decoded.frames

            if self._24h_offset_frames > 0:
                return CTimecode(
                    framerate=decoded.framerate,
                    frames=decoded.frames + self._24h_offset_frames,
                )
            return decoded

    def reset_24h_state(self) -> None:
        """Clear the 24h-wrap accumulator (control-plane reset).

        Called from the engine's project-load/reset orchestration (NOT off the
        wire): a graceful reload sends MTC back toward 0 with only a SMALL
        backward delta, which the wire-driven reset in _apply_24h_offset cannot
        catch (it requires delta < -1h). Without this, a project loaded after a
        >24h run would start at hour 24+ and the cue-sequence reset_callback
        (gated on milliseconds_rounded == 0) would stay silenced forever.

        Zeroes both the offset and the last-position memory (None sentinel,
        mirroring the C++ resetWrapOffset() → lastWireMs_ = -1) under the same
        lock the decode path uses, so it cannot interleave with a concurrent
        wrap accumulation on the callback thread.
        """
        with self._wrap_lock:
            self._24h_offset_frames = 0
            self._last_decoded_frames = None

    def timecode(self):
        return self.main_tc

    def milliseconds_rounded(self) -> int:
        """Integer milliseconds from the current MTC."""
        return self.main_tc.milliseconds_rounded

    def milliseconds_exact(self) -> float:
        """Float milliseconds from the current MTC."""
        return self.main_tc.milliseconds_exact

    def milliseconds(self) -> int:
        """
        Deprecated alias of milliseconds_rounded
        (CTimecode.milliseconds is deprecated).
        """
        return self.milliseconds_rounded()

    def is_receiving(self, max_age_s: float = 0.5) -> bool:
        """True if an MTC message arrived within the last max_age_s seconds.

        A listener that failed to open its port (or whose sender stopped)
        keeps a main_tc that looks like a valid reading -- 0.0 if it never
        received anything. ControllerEngine.go_script ships that reading as
        every node's GO anchor, so it must be able to tell a live reading
        from a dead one (test2, 2026-09-25, 869f79ecc). While MTC rolls,
        quarter frames arrive every ~10ms at 25fps, so 0.5s is generous.
        """
        last = self.last_rx_monotonic
        return last is not None and (monotonic() - last) <= max_age_s

    def __update_timecode(self, timecode):
        self.main_tc = timecode
        if self.main_tc.milliseconds_rounded == 0:
            if self.step_callback is not None and self.reset_callback is not None:
                self.reset_callback()
        if self.step_callback is not None:
            self.step_callback(self.main_tc)

    def __open_port(self, port):
        # HEADLESS/CLOUD: get_input_names() can throw when no MIDI subsystem is
        # present; catch and treat as empty list so the engine keeps running.
        # port_name is left as None and re-detected later in
        # ControllerEngine.start()
        # once the timecode sender has created the virtual MIDI port.
        if port is not None and not isinstance(port, str):
            raise TypeError(
                f"port must be a MIDI port name (str), got {type(port).__name__}"
            )
        try:
            ports = mido.get_input_names()  # type: ignore[attr-defined]
        except Exception as e:
            Logger.warning(f"Could not list MIDI input ports: {e}")
            ports = []

        if port is not None:
            # Exact match first; fall back to substring match because ALSA/JACK
            # port names include the client name and ID suffix
            # e.g. "Midi Through Port-0" → "Midi Through:Midi Through Port-0
            # 14:0"
            if port in ports:
                self.port_name = port
            else:
                matches = [p for p in ports if port in p]
                if matches:
                    self.port_name = matches[0]
                    Logger.info(f'MIDI port "{port}" matched as "{self.port_name}"')
                else:
                    Logger.warning(f'MIDI port "{port}" not found, auto-detecting...')
                    port = None  # fall through to auto-detect

        if port is None:
            # Prefer ports whose name contains "mtc" (e.g. MtcMaster:MTCPort)
            mtc_ports = [s for s in ports if "mtc" in s.lower()]
            if mtc_ports:
                self.port_name = mtc_ports[-1]
            elif ports:
                self.port_name = ports[-1]
            else:
                # HEADLESS/CLOUD: no ports yet; caller must retry after the
                # virtual MIDI sender port has been created.
                self.port_name = None
                Logger.warning("No MIDI input ports available")
        if self.port_name:
            Logger.info(f"MtcListener will use MIDI port: {self.port_name}")

    def run(self):
        Logger.debug("Starting MTC listener")
        self.port = mido.open_input(  # type: ignore[attr-defined]
            self.port_name, callback=self.__handle_message
        )
        Logger.info(f"Listening to MIDI messages on > {self.port_name} <")

    def stop(self):
        if self.port is not None:
            self.port.close()

    def __handle_message(self, message):
        if message.type == "quarter_frame":
            self.last_rx_monotonic = monotonic()
            self.__quarter_frames[message.frame_type] = message.frame_value
            if (message.frame_type == 3) or (message.frame_type == 7):
                self.__update_timecode(self.main_tc + 1)
            #    print('QF+:',self.main_tc)
            if message.frame_type == 7:
                tc = self.__mtc_decode_quarter_frames(self.__quarter_frames)
                #    print('QFC:',tc)
                self.__update_timecode(tc)
        elif message.type == "sysex":
            # check to see if this is a timecode frame
            if len(message.data) == 8 and message.data[0:4] == (
                127,
                127,
                1,
                1,
            ):
                self.last_rx_monotonic = monotonic()
                data = message.data[4:]
                tc = self.__mtc_decode(data)
                Logger.debug("FF:" + tc.__str__())
                self.__update_timecode(tc)
                # Deliberately do NOT clear self.__quarter_frames here. Full
                # frames (a seek, or libmtcmaster's ~2s periodic resync) never
                # touch the QF buffer, and the QF decode fires only at
                # frame_type==7 reading all 8 in-order-refreshed indices — so no
                # stale-nibble decode occurs. The C++ receiver clears its decoder
                # state on every full frame, but that resets C++-only
                # sequence-validity fields (qfCount/direction/flags) this
                # index-based decoder lacks; flushing here would instead corrupt
                # the next decode when a resync lands mid-QF-sequence. (Plan 4)
        else:
            Logger.debug(message)
            raise (NotImplementedError)

    def __mtc_decode(self, mtc_bytes):
        # print(mtc_bytes)
        rhh, mins, secs, frs = mtc_bytes
        rateflag = rhh >> 5
        hrs = rhh & 31
        fps = ["24", "25", "29.97", "30"][rateflag]
        # total_frames = frs + float(fps) * (secs + mins * 60 + hrs * 60 * 60)
        # //  TODO: goes to frame 0 in tc, non existent frame, changed to tc
        # 0:0:0:0 = frame 1
        decoded = CTimecode("{}:{}:{}:{}".format(hrs, mins, secs, frs), framerate=fps)
        # Route through 24h-wrap detection so main_tc stays monotonic past 24h.
        # See _apply_24h_offset docstring for heuristic details (closes
        # 869cpdbzy).
        return self._apply_24h_offset(decoded)

    def __mtc_decode_full_frame(self, full_frame_bytes):
        mtc_bytes = full_frame_bytes[5:-1]
        return self.__mtc_decode(mtc_bytes)

    def __mtc_decode_quarter_frames(self, frame_pieces):
        mtc_bytes = bytearray(4)
        if len(frame_pieces) < 8:
            return None
        for piece in range(8):
            mtc_index = (
                3 - piece // 2
            )  # quarter frame pieces are in reverse order of mtc_encode
            this_frame = frame_pieces[piece]
            if this_frame is bytearray or this_frame is list:
                this_frame = this_frame[1]  # type: ignore[index]
            # ignore the frame_piece marker bits
            data = this_frame & 15  # type: ignore[operator]
            if piece % 2 == 0:
                # 'even' pieces came from the low nibble
                # and the first piece is 0, so it's even
                mtc_bytes[mtc_index] += data
            else:
                # 'odd' pieces came from the high nibble
                mtc_bytes[mtc_index] += data * 16
        return self.__mtc_decode(mtc_bytes)

is_receiving(max_age_s=0.5)

True if an MTC message arrived within the last max_age_s seconds.

A listener that failed to open its port (or whose sender stopped) keeps a main_tc that looks like a valid reading -- 0.0 if it never received anything. ControllerEngine.go_script ships that reading as every node's GO anchor, so it must be able to tell a live reading from a dead one (test2, 2026-09-25, 869f79ecc). While MTC rolls, quarter frames arrive every ~10ms at 25fps, so 0.5s is generous.

Source code in src/cuemsengine/tools/MtcListener.py

def is_receiving(self, max_age_s: float = 0.5) -> bool:
    """True if an MTC message arrived within the last max_age_s seconds.

    A listener that failed to open its port (or whose sender stopped)
    keeps a main_tc that looks like a valid reading -- 0.0 if it never
    received anything. ControllerEngine.go_script ships that reading as
    every node's GO anchor, so it must be able to tell a live reading
    from a dead one (test2, 2026-09-25, 869f79ecc). While MTC rolls,
    quarter frames arrive every ~10ms at 25fps, so 0.5s is generous.
    """
    last = self.last_rx_monotonic
    return last is not None and (monotonic() - last) <= max_age_s

milliseconds()

Deprecated alias of milliseconds_rounded (CTimecode.milliseconds is deprecated).

Source code in src/cuemsengine/tools/MtcListener.py

def milliseconds(self) -> int:
    """
    Deprecated alias of milliseconds_rounded
    (CTimecode.milliseconds is deprecated).
    """
    return self.milliseconds_rounded()

milliseconds_exact()

Float milliseconds from the current MTC.

Source code in src/cuemsengine/tools/MtcListener.py

def milliseconds_exact(self) -> float:
    """Float milliseconds from the current MTC."""
    return self.main_tc.milliseconds_exact

milliseconds_rounded()

Integer milliseconds from the current MTC.

Source code in src/cuemsengine/tools/MtcListener.py

def milliseconds_rounded(self) -> int:
    """Integer milliseconds from the current MTC."""
    return self.main_tc.milliseconds_rounded

reset_24h_state()

Clear the 24h-wrap accumulator (control-plane reset).

Called from the engine's project-load/reset orchestration (NOT off the wire): a graceful reload sends MTC back toward 0 with only a SMALL backward delta, which the wire-driven reset in _apply_24h_offset cannot catch (it requires delta < -1h). Without this, a project loaded after a

24h run would start at hour 24+ and the cue-sequence reset_callback (gated on milliseconds_rounded == 0) would stay silenced forever.

Zeroes both the offset and the last-position memory (None sentinel, mirroring the C++ resetWrapOffset() → lastWireMs_ = -1) under the same lock the decode path uses, so it cannot interleave with a concurrent wrap accumulation on the callback thread.

Source code in src/cuemsengine/tools/MtcListener.py

def reset_24h_state(self) -> None:
    """Clear the 24h-wrap accumulator (control-plane reset).

    Called from the engine's project-load/reset orchestration (NOT off the
    wire): a graceful reload sends MTC back toward 0 with only a SMALL
    backward delta, which the wire-driven reset in _apply_24h_offset cannot
    catch (it requires delta < -1h). Without this, a project loaded after a
    >24h run would start at hour 24+ and the cue-sequence reset_callback
    (gated on milliseconds_rounded == 0) would stay silenced forever.

    Zeroes both the offset and the last-position memory (None sentinel,
    mirroring the C++ resetWrapOffset() → lastWireMs_ = -1) under the same
    lock the decode path uses, so it cannot interleave with a concurrent
    wrap accumulation on the callback thread.
    """
    with self._wrap_lock:
        self._24h_offset_frames = 0
        self._last_decoded_frames = None

PortHandler

Bases: object

Source code in src/cuemsengine/tools/PortHandler.py

class PortHandler(object):
    def __new__(cls):
        """
        Singleton class responsible for handling port objects.

        Ports come in three kinds, and they are NOT interchangeable:

        - *Reservations* — `_all_used_ports` (per cue), `_random_ports` (OSC
          clients) and `_pending` (drawn, not yet registered). These are ports
          this handler handed out. Double-booking one is a real error and may
          raise.
        - *Declarations* — `_config_ports`. Named service/config ports out of
          settings.xml. A declaration is a statement of fact, not a request to
          allocate, so registering one must never raise: two names may
          legitimately share a port, and a declared port may already be bound.
        - *Observations* — `_system_ports`. What `ss` currently sees bound on
          the host. Also never raises.

        All three are excluded from allocation; only reservations are
        validated by check_ports(). Conflating them is what made
        NodeEngine.__init__ crash in every earlier attempt at this fix
        (869ed9wf7).

        Thread-safe: internal state mutations are guarded by an RLock.
        """
        if not hasattr(cls, "_instance"):
            cls._instance = super(PortHandler, cls).__new__(cls)
            cls._instance._lock = RLock()
            cls._instance._ports = {None: {}}
            cls._instance._all_used_ports = []
            cls._instance._all_available_ports = set(range(INITIAL_PORT, MAX_PORT))
            cls._instance._random_ports = []
            cls._instance._pending = set()
            cls._instance._config_ports = set()
            cls._instance._system_ports = set()
        return cls._instance

    def assign_ports(self, names: list[str], cue: CuemsDict = None) -> dict:
        """Assign free ports to a list of names

        This method is thread-safe and should be the preferred way to assign
        ports to a list of names for a cue or config.

        Args:
            names: The names to assign ports to
            cue: The cue to assign ports to
        """
        # No outer lock: get_free_ports() and set_ports()/add_config_ports()
        # each lock internally. Holding it here would serialize every other
        # PORT_HANDLER operation behind the bind probes — and, on the degraded
        # path, behind an `ss` subprocess.
        new_ports = self.get_free_ports(len(names))
        out = {k: new_ports[i] for i, k in enumerate(names)}
        if cue is None:
            self.add_config_ports(out)
        else:
            self.set_ports(cue, out)
        return out

    def get_ports(self, cue: CuemsDict) -> dict | None:
        """
        Get the ports for a cue
        """
        with self._lock:
            return self._ports.get(cue, None)

    @staticmethod
    def _as_dict(ports: dict) -> dict:
        """
        Defensive copy of a caller's port mapping.

        Never alias the stored dict: add_config_ports() used to hand
        set_ports() the very object `_ports[None]` already held, making the
        `previous_ports == ports` comparison an identity check that always
        short-circuited — which is why the exclusion list stayed empty.
        """
        return dict(ports)

    def _release(self, values) -> None:
        """
        Drop these ports from the cue-scope reservations (multiset-safe).
        """
        with self._lock:
            remaining = list(self._all_used_ports)
            for p in values:
                if p in remaining:
                    remaining.remove(p)
            self._all_used_ports = remaining

    def set_ports(self, cue: CuemsDict, ports: dict, check_range: bool = True) -> None:
        """
        Set the ports for a cue, replacing any it previously held.

        Cue scope only — config ports go through add_config_ports(), which
        cannot raise.
        """
        if cue is None:
            raise ValueError(
                "set_ports() is cue-scoped; use add_config_ports() for config"
            )
        with self._lock:
            new_ports = self._as_dict(ports)
            previous_ports = self._ports.get(cue)
            if previous_ports == new_ports:
                return
            # Claim our own draws before validating, or check_ports() would
            # reject them as "already in use" against _pending.
            self._pending -= set(new_ports.values())
            # Release this cue's own reservations first, so re-arming onto a
            # port it already holds validates cleanly.
            if previous_ports:
                self._release(previous_ports.values())
            try:
                ports_list = self.check_ports(new_ports, check_range)
            except ValueError:
                if previous_ports:
                    self._all_used_ports.extend(previous_ports.values())
                raise
            self._all_used_ports.extend(ports_list)
            self._ports[cue] = new_ports

    def remove_ports(self, cue: CuemsDict):
        """
        Remove the ports for a cue
        """
        with self._lock:
            p = self._ports.pop(cue, None)
            if p is None:
                return
            self._release(p.values())

    def _reserved_ports(self) -> set[int]:
        """
        Ports THIS handler handed out — what double-booking is judged against.

        Deliberately excludes config declarations and system observations:
        those coinciding with a port is expected, not an error.
        """
        with self._lock:
            return (
                set[int](self._all_used_ports)
                | set[int](self._random_ports)
                | self._pending
            )

    def get_all_used_ports(self) -> set[int]:
        """
        Every port the allocator must avoid: reservations, config
        declarations and observed system sockets.
        """
        with self._lock:
            Logger.debug(f"All used ports: {self._all_used_ports}")
            Logger.debug(f"Random ports: {self._random_ports}")
            Logger.debug(f"Config ports: {sorted(self._config_ports)}")
            return self._reserved_ports() | self._config_ports | self._system_ports

    def check_ports(self, ports: list | dict, check_range: bool = True) -> list:
        """
        Check the ports for a cue and return the list of ports if they are
        valid

        Args:
            ports: The ports to check
            check_range: Whether to check the port range

        Returns:
            The ports list if they are valid

        Raises:
            ValueError:
            - If duplicate ports are found
            - If ports are already reserved
            - If check_range is True and the port range is invalid
        """
        if isinstance(ports, dict):
            ports = [i for i in ports.values()]
        if len(ports) > len(set[int](ports)):
            raise ValueError("Duplicate ports found")
        # Validate against reservations only — never get_all_used_ports().
        reserved = self._reserved_ports()
        if reserved & set[int](ports):
            raise ValueError(f"Ports already in use: {reserved & set[int](ports)}")
        if check_range:
            self.check_port_range(ports)
        return ports

    @staticmethod
    def check_port_range(ports: list) -> None:
        """
        Check the port range
        """
        for port in ports:
            if port > MAX_PORT:
                raise ValueError(f"Port {port} is too high")
            if port < INITIAL_PORT:
                raise ValueError(f"Port {port} is too low")

    def get_free_port(self) -> int:
        """
        Draw a free port and reserve it.

        The returned port is added to `_pending`, so a concurrent draw cannot
        return it too. It leaves `_pending` when set_ports(),
        add_config_ports() or store_random_port() registers it properly, or
        when release_pending()/clean_random_ports() gives it back.

        Candidates are bind-probed before being handed out — the startup
        snapshot alone cannot see anything that bound after it was taken.

        Returns:
            The free port

        Raises:
            ValueError: If no free port could be found and verified. Never
            returns an unverified port: handing one out is exactly the bug
            this method exists to prevent, and doing it when the pool is
            already degraded is the worst possible moment.
        """
        for attempt in range(3):
            if attempt == 2:
                # The cheap picture is stale. Now it is worth paying for a
                # rescan before giving up.
                Logger.warning(
                    f"No bindable port in {2 * PROBE_ATTEMPTS} draws; "
                    "rescanning system ports"
                )
                self.add_system_ports()
            with self._lock:
                available = self._all_available_ports - self.get_all_used_ports()
            if not available:
                raise ValueError("No free ports found")
            candidates = sample(sorted(available), min(len(available), PROBE_ATTEMPTS))
            # Probe outside the lock: bind/close syscalls must not serialize
            # every other PORT_HANDLER operation on the cue-arm path.
            for port in candidates:
                if not port_is_bindable(port):
                    continue
                with self._lock:
                    if port in self.get_all_used_ports():
                        continue  # raced with a concurrent draw
                    self._pending.add(port)
                    return port
        raise ValueError("No bindable free port found")

    def get_free_ports(self, n: int) -> list:
        """
        Get n free ports, in draw order.

        Distinct by construction: each draw reserves into `_pending` before
        returning, so the next one cannot pick it again.
        """
        return [self.get_free_port() for _ in range(n)]

    def release_pending(self, port: int):
        """
        Give back a port drawn by get_free_port() that will not be registered.
        """
        with self._lock:
            self._pending.discard(port)

    def find_system_ports(self) -> set[int]:
        """
        Find all ports currently bound on the system
        """
        return get_used_ports()

    def add_system_ports(self):
        """
        Snapshot every port currently bound on this host (observation only).
        """
        # Scan outside the lock — it spawns an `ss` subprocess.
        ports = self.find_system_ports()
        with self._lock:
            self._system_ports = ports

    def add_config_ports(self, ports: dict):
        """
        Merge named config/service ports into the config scope.

        NEVER raises. A config port is a *declaration* ("this port belongs to
        that service"), not a request to allocate. Two names may legitimately
        map to one port — stock settings.xml has `osc_in_port_base` 7000 and
        the videocomposer default is also 7000 — and a declared port may
        already be bound, as `oscquery_ws_port` 9190 is on a controller while
        the node engine starts. Neither is an error; both used to crash
        NodeEngine.__init__ once check_ports()'s guards were reachable.
        """
        with self._lock:
            self._ports[None].update(self._as_dict(ports))
            declared = set()
            for value in ports.values():
                try:
                    declared.add(int(value))
                except (TypeError, ValueError):
                    continue  # e.g. mtc_port is an ALSA client name
            self._config_ports |= declared
            self._pending -= declared

    def new_random_port(self) -> int:
        """
        Get a new random port and store it
        """
        port = self.get_free_port()
        self.store_random_port(port)
        return port

    def store_random_port(self, port: int):
        """
        Store a random port to the random ports set
        """
        with self._lock:
            self._random_ports.append(port)
            self._pending.discard(port)

    def remove_random_port(self, port: int):
        """
        Remove a specific port from the random ports list, freeing it for
        reuse.
        Called when an OSC client that owned the port is closed.
        """
        with self._lock:
            try:
                self._random_ports.remove(port)
            except ValueError:
                pass

    def clean_random_ports(self):
        """
        Clean the random ports set by keeping only ports that are in use by the
        system, and refresh the system snapshot with the scan it already pays
        for.

        Note this only started doing real work once get_used_ports() could see
        other users' sockets: the old scan returned almost nothing, so this
        wiped `_random_ports` wholesale on every project load. It now keeps
        genuinely-bound ports, and still self-heals — a port whose client died
        and released the socket drops out on the next load.
        """
        sys_ports = self.find_system_ports()
        with self._lock:
            self._system_ports = sys_ports
            self._random_ports = [i for i in self._random_ports if i in sys_ports]
            # Draws that were never registered are not coming back.
            self._pending.clear()

__new__()

Singleton class responsible for handling port objects.

Ports come in three kinds, and they are NOT interchangeable:

• Reservations — _all_used_ports (per cue), _random_ports (OSC clients) and _pending (drawn, not yet registered). These are ports this handler handed out. Double-booking one is a real error and may raise.
• Declarations — _config_ports. Named service/config ports out of settings.xml. A declaration is a statement of fact, not a request to allocate, so registering one must never raise: two names may legitimately share a port, and a declared port may already be bound.
• Observations — _system_ports. What ss currently sees bound on the host. Also never raises.

All three are excluded from allocation; only reservations are validated by check_ports(). Conflating them is what made NodeEngine.init crash in every earlier attempt at this fix (869ed9wf7).

Thread-safe: internal state mutations are guarded by an RLock.

Source code in src/cuemsengine/tools/PortHandler.py

def __new__(cls):
    """
    Singleton class responsible for handling port objects.

    Ports come in three kinds, and they are NOT interchangeable:

    - *Reservations* — `_all_used_ports` (per cue), `_random_ports` (OSC
      clients) and `_pending` (drawn, not yet registered). These are ports
      this handler handed out. Double-booking one is a real error and may
      raise.
    - *Declarations* — `_config_ports`. Named service/config ports out of
      settings.xml. A declaration is a statement of fact, not a request to
      allocate, so registering one must never raise: two names may
      legitimately share a port, and a declared port may already be bound.
    - *Observations* — `_system_ports`. What `ss` currently sees bound on
      the host. Also never raises.

    All three are excluded from allocation; only reservations are
    validated by check_ports(). Conflating them is what made
    NodeEngine.__init__ crash in every earlier attempt at this fix
    (869ed9wf7).

    Thread-safe: internal state mutations are guarded by an RLock.
    """
    if not hasattr(cls, "_instance"):
        cls._instance = super(PortHandler, cls).__new__(cls)
        cls._instance._lock = RLock()
        cls._instance._ports = {None: {}}
        cls._instance._all_used_ports = []
        cls._instance._all_available_ports = set(range(INITIAL_PORT, MAX_PORT))
        cls._instance._random_ports = []
        cls._instance._pending = set()
        cls._instance._config_ports = set()
        cls._instance._system_ports = set()
    return cls._instance

add_config_ports(ports)

Merge named config/service ports into the config scope.

NEVER raises. A config port is a declaration ("this port belongs to that service"), not a request to allocate. Two names may legitimately map to one port — stock settings.xml has osc_in_port_base 7000 and the videocomposer default is also 7000 — and a declared port may already be bound, as oscquery_ws_port 9190 is on a controller while the node engine starts. Neither is an error; both used to crash NodeEngine.init once check_ports()'s guards were reachable.

Source code in src/cuemsengine/tools/PortHandler.py

def add_config_ports(self, ports: dict):
    """
    Merge named config/service ports into the config scope.

    NEVER raises. A config port is a *declaration* ("this port belongs to
    that service"), not a request to allocate. Two names may legitimately
    map to one port — stock settings.xml has `osc_in_port_base` 7000 and
    the videocomposer default is also 7000 — and a declared port may
    already be bound, as `oscquery_ws_port` 9190 is on a controller while
    the node engine starts. Neither is an error; both used to crash
    NodeEngine.__init__ once check_ports()'s guards were reachable.
    """
    with self._lock:
        self._ports[None].update(self._as_dict(ports))
        declared = set()
        for value in ports.values():
            try:
                declared.add(int(value))
            except (TypeError, ValueError):
                continue  # e.g. mtc_port is an ALSA client name
        self._config_ports |= declared
        self._pending -= declared

add_system_ports()

Snapshot every port currently bound on this host (observation only).

Source code in src/cuemsengine/tools/PortHandler.py

def add_system_ports(self):
    """
    Snapshot every port currently bound on this host (observation only).
    """
    # Scan outside the lock — it spawns an `ss` subprocess.
    ports = self.find_system_ports()
    with self._lock:
        self._system_ports = ports

assign_ports(names, cue=None)

Assign free ports to a list of names

This method is thread-safe and should be the preferred way to assign ports to a list of names for a cue or config.

Parameters:

Name	Type	Description	Default
names	list[str]	The names to assign ports to	required
cue	CuemsDict	The cue to assign ports to	None

Source code in src/cuemsengine/tools/PortHandler.py

def assign_ports(self, names: list[str], cue: CuemsDict = None) -> dict:
    """Assign free ports to a list of names

    This method is thread-safe and should be the preferred way to assign
    ports to a list of names for a cue or config.

    Args:
        names: The names to assign ports to
        cue: The cue to assign ports to
    """
    # No outer lock: get_free_ports() and set_ports()/add_config_ports()
    # each lock internally. Holding it here would serialize every other
    # PORT_HANDLER operation behind the bind probes — and, on the degraded
    # path, behind an `ss` subprocess.
    new_ports = self.get_free_ports(len(names))
    out = {k: new_ports[i] for i, k in enumerate(names)}
    if cue is None:
        self.add_config_ports(out)
    else:
        self.set_ports(cue, out)
    return out

check_port_range(ports) staticmethod

Check the port range

Source code in src/cuemsengine/tools/PortHandler.py

@staticmethod
def check_port_range(ports: list) -> None:
    """
    Check the port range
    """
    for port in ports:
        if port > MAX_PORT:
            raise ValueError(f"Port {port} is too high")
        if port < INITIAL_PORT:
            raise ValueError(f"Port {port} is too low")

check_ports(ports, check_range=True)

Check the ports for a cue and return the list of ports if they are valid

Parameters:

Name	Type	Description	Default
ports	list | dict	The ports to check	required
check_range	bool	Whether to check the port range	True

Returns:

Type	Description
list	The ports list if they are valid

Raises:

Type	Description
ValueError

Source code in src/cuemsengine/tools/PortHandler.py

def check_ports(self, ports: list | dict, check_range: bool = True) -> list:
    """
    Check the ports for a cue and return the list of ports if they are
    valid

    Args:
        ports: The ports to check
        check_range: Whether to check the port range

    Returns:
        The ports list if they are valid

    Raises:
        ValueError:
        - If duplicate ports are found
        - If ports are already reserved
        - If check_range is True and the port range is invalid
    """
    if isinstance(ports, dict):
        ports = [i for i in ports.values()]
    if len(ports) > len(set[int](ports)):
        raise ValueError("Duplicate ports found")
    # Validate against reservations only — never get_all_used_ports().
    reserved = self._reserved_ports()
    if reserved & set[int](ports):
        raise ValueError(f"Ports already in use: {reserved & set[int](ports)}")
    if check_range:
        self.check_port_range(ports)
    return ports

clean_random_ports()

Clean the random ports set by keeping only ports that are in use by the system, and refresh the system snapshot with the scan it already pays for.

Note this only started doing real work once get_used_ports() could see other users' sockets: the old scan returned almost nothing, so this wiped _random_ports wholesale on every project load. It now keeps genuinely-bound ports, and still self-heals — a port whose client died and released the socket drops out on the next load.

Source code in src/cuemsengine/tools/PortHandler.py

def clean_random_ports(self):
    """
    Clean the random ports set by keeping only ports that are in use by the
    system, and refresh the system snapshot with the scan it already pays
    for.

    Note this only started doing real work once get_used_ports() could see
    other users' sockets: the old scan returned almost nothing, so this
    wiped `_random_ports` wholesale on every project load. It now keeps
    genuinely-bound ports, and still self-heals — a port whose client died
    and released the socket drops out on the next load.
    """
    sys_ports = self.find_system_ports()
    with self._lock:
        self._system_ports = sys_ports
        self._random_ports = [i for i in self._random_ports if i in sys_ports]
        # Draws that were never registered are not coming back.
        self._pending.clear()

find_system_ports()

Find all ports currently bound on the system

Source code in src/cuemsengine/tools/PortHandler.py

def find_system_ports(self) -> set[int]:
    """
    Find all ports currently bound on the system
    """
    return get_used_ports()

get_all_used_ports()

Every port the allocator must avoid: reservations, config declarations and observed system sockets.

Source code in src/cuemsengine/tools/PortHandler.py

def get_all_used_ports(self) -> set[int]:
    """
    Every port the allocator must avoid: reservations, config
    declarations and observed system sockets.
    """
    with self._lock:
        Logger.debug(f"All used ports: {self._all_used_ports}")
        Logger.debug(f"Random ports: {self._random_ports}")
        Logger.debug(f"Config ports: {sorted(self._config_ports)}")
        return self._reserved_ports() | self._config_ports | self._system_ports

get_free_port()

Draw a free port and reserve it.

The returned port is added to _pending, so a concurrent draw cannot return it too. It leaves _pending when set_ports(), add_config_ports() or store_random_port() registers it properly, or when release_pending()/clean_random_ports() gives it back.

Candidates are bind-probed before being handed out — the startup snapshot alone cannot see anything that bound after it was taken.

Returns:

Type	Description
int	The free port

Raises:

Type	Description
ValueError	If no free port could be found and verified. Never
returns an unverified port	handing one out is exactly the bug

Source code in src/cuemsengine/tools/PortHandler.py

def get_free_port(self) -> int:
    """
    Draw a free port and reserve it.

    The returned port is added to `_pending`, so a concurrent draw cannot
    return it too. It leaves `_pending` when set_ports(),
    add_config_ports() or store_random_port() registers it properly, or
    when release_pending()/clean_random_ports() gives it back.

    Candidates are bind-probed before being handed out — the startup
    snapshot alone cannot see anything that bound after it was taken.

    Returns:
        The free port

    Raises:
        ValueError: If no free port could be found and verified. Never
        returns an unverified port: handing one out is exactly the bug
        this method exists to prevent, and doing it when the pool is
        already degraded is the worst possible moment.
    """
    for attempt in range(3):
        if attempt == 2:
            # The cheap picture is stale. Now it is worth paying for a
            # rescan before giving up.
            Logger.warning(
                f"No bindable port in {2 * PROBE_ATTEMPTS} draws; "
                "rescanning system ports"
            )
            self.add_system_ports()
        with self._lock:
            available = self._all_available_ports - self.get_all_used_ports()
        if not available:
            raise ValueError("No free ports found")
        candidates = sample(sorted(available), min(len(available), PROBE_ATTEMPTS))
        # Probe outside the lock: bind/close syscalls must not serialize
        # every other PORT_HANDLER operation on the cue-arm path.
        for port in candidates:
            if not port_is_bindable(port):
                continue
            with self._lock:
                if port in self.get_all_used_ports():
                    continue  # raced with a concurrent draw
                self._pending.add(port)
                return port
    raise ValueError("No bindable free port found")

get_free_ports(n)

Get n free ports, in draw order.

Distinct by construction: each draw reserves into _pending before returning, so the next one cannot pick it again.

Source code in src/cuemsengine/tools/PortHandler.py

def get_free_ports(self, n: int) -> list:
    """
    Get n free ports, in draw order.

    Distinct by construction: each draw reserves into `_pending` before
    returning, so the next one cannot pick it again.
    """
    return [self.get_free_port() for _ in range(n)]

get_ports(cue)

Get the ports for a cue

Source code in src/cuemsengine/tools/PortHandler.py

def get_ports(self, cue: CuemsDict) -> dict | None:
    """
    Get the ports for a cue
    """
    with self._lock:
        return self._ports.get(cue, None)

new_random_port()

Get a new random port and store it

Source code in src/cuemsengine/tools/PortHandler.py

def new_random_port(self) -> int:
    """
    Get a new random port and store it
    """
    port = self.get_free_port()
    self.store_random_port(port)
    return port

release_pending(port)

Give back a port drawn by get_free_port() that will not be registered.

Source code in src/cuemsengine/tools/PortHandler.py

def release_pending(self, port: int):
    """
    Give back a port drawn by get_free_port() that will not be registered.
    """
    with self._lock:
        self._pending.discard(port)

remove_ports(cue)

Remove the ports for a cue

Source code in src/cuemsengine/tools/PortHandler.py

def remove_ports(self, cue: CuemsDict):
    """
    Remove the ports for a cue
    """
    with self._lock:
        p = self._ports.pop(cue, None)
        if p is None:
            return
        self._release(p.values())

remove_random_port(port)

Remove a specific port from the random ports list, freeing it for reuse. Called when an OSC client that owned the port is closed.

Source code in src/cuemsengine/tools/PortHandler.py

def remove_random_port(self, port: int):
    """
    Remove a specific port from the random ports list, freeing it for
    reuse.
    Called when an OSC client that owned the port is closed.
    """
    with self._lock:
        try:
            self._random_ports.remove(port)
        except ValueError:
            pass

set_ports(cue, ports, check_range=True)

Set the ports for a cue, replacing any it previously held.

Cue scope only — config ports go through add_config_ports(), which cannot raise.

Source code in src/cuemsengine/tools/PortHandler.py

def set_ports(self, cue: CuemsDict, ports: dict, check_range: bool = True) -> None:
    """
    Set the ports for a cue, replacing any it previously held.

    Cue scope only — config ports go through add_config_ports(), which
    cannot raise.
    """
    if cue is None:
        raise ValueError(
            "set_ports() is cue-scoped; use add_config_ports() for config"
        )
    with self._lock:
        new_ports = self._as_dict(ports)
        previous_ports = self._ports.get(cue)
        if previous_ports == new_ports:
            return
        # Claim our own draws before validating, or check_ports() would
        # reject them as "already in use" against _pending.
        self._pending -= set(new_ports.values())
        # Release this cue's own reservations first, so re-arming onto a
        # port it already holds validates cleanly.
        if previous_ports:
            self._release(previous_ports.values())
        try:
            ports_list = self.check_ports(new_ports, check_range)
        except ValueError:
            if previous_ports:
                self._all_used_ports.extend(previous_ports.values())
            raise
        self._all_used_ports.extend(ports_list)
        self._ports[cue] = new_ports

store_random_port(port)

Store a random port to the random ports set

Source code in src/cuemsengine/tools/PortHandler.py

def store_random_port(self, port: int):
    """
    Store a random port to the random ports set
    """
    with self._lock:
        self._random_ports.append(port)
        self._pending.discard(port)

port_is_bindable(port)

True if nothing currently holds port on either TCP or UDP.

Deliberately does NOT set SO_REUSEADDR: a socket lingering in TIME_WAIT should count as taken. IPv4 only — the whole CUEMS stack (avahi, the in network_map.xml, the NNG bus) is IPv4, and probing a family nothing uses would only mask real conflicts.

Inherently advisory: the probe releases the port before the real consumer binds it. It shrinks the race window from "permanently blind" to microseconds, which is the same bargain tests/helpers.py::free_udp_port already makes.

Source code in src/cuemsengine/tools/PortHandler.py

def port_is_bindable(port: int) -> bool:
    """
    True if nothing currently holds `port` on either TCP or UDP.

    Deliberately does NOT set SO_REUSEADDR: a socket lingering in TIME_WAIT
    should count as taken. IPv4 only — the whole CUEMS stack (avahi, the <ip>
    in network_map.xml, the NNG bus) is IPv4, and probing a family nothing
    uses would only mask real conflicts.

    Inherently advisory: the probe releases the port before the real consumer
    binds it. It shrinks the race window from "permanently blind" to
    microseconds, which is the same bargain tests/helpers.py::free_udp_port
    already makes.
    """
    for sock_type in (socket.SOCK_STREAM, socket.SOCK_DGRAM):
        try:
            with socket.socket(socket.AF_INET, sock_type) as s:
                s.bind(("", port))
        except OSError:
            return False
    return True

Reader for /run/cuems/display.conf.

display.conf is written by cuems-generate-display-conf (ExecStartPre of cuems-videocomposer.service) and read by both the videocomposer (for DRM modeset + canvas layout) and the engine (for canvas geometry + per-output canvas regions, replacing the broken x = index * 1920 heuristic that used to live in NodeEngine.set_video_outputs).

File format is INI-like:

canvas_layout=custom
canvas_size=5760x1080        # optional, overrides bbox

[output:HDMI-A-1]
canvas_region=0,0,1920,1080
resolution=1920x1080         # optional
refresh=60.0                 # optional
enabled=false                # optional; leaves the output dark

The engine consumes canvas_region, enabled and the optional global canvas_size override; resolution + refresh are read by the videocomposer directly.

An output with enabled set to anything but true/1 gets no surface and stays out of the videocomposer's canvas even when it keeps a canvas_region, so the engine leaves it out too (869fgq3xv). Otherwise the engine would route cues to a dark output and compute a wider canvas than the videocomposer's, placing every layer wrong.

DisplayConfNotFoundError

Bases: RuntimeError

display.conf is missing, unreadable, or has no [output:*] sections.

Source code in src/cuemsengine/tools/display_conf.py

class DisplayConfNotFoundError(RuntimeError):
    """display.conf is missing, unreadable, or has no [output:*] sections."""

DisplayConfValueError

Bases: RuntimeError

display.conf is present but contains an invalid value (e.g. a canvas_size override that is malformed, non-positive, or smaller than the per-output region bounding box).

Source code in src/cuemsengine/tools/display_conf.py

class DisplayConfValueError(RuntimeError):
    """display.conf is present but contains an invalid value (e.g. a
    canvas_size override that is malformed, non-positive, or smaller than
    the per-output region bounding box)."""

read_display_conf(path=DEFAULT_DISPLAY_CONF)

Parse display.conf and return (regions, canvas_size).

Returns a 2-tuple:

• regions: {connector_name: {'x', 'y', 'width', 'height'}} with pixel-int values.
• canvas_size: (canvas_width, canvas_height). If the global canvas_size=WIDTHxHEIGHT key is present in the file's preamble, it is used (after validating it is >= the per-region bounding box). Otherwise, computed as max(x + width, y + height) over all regions.

Raises:

• DisplayConfNotFoundError if the file is missing or has no enabled [output:*] section with a valid canvas_region.
• DisplayConfValueError if canvas_size= is malformed, has non-positive values, or is smaller than the per-region bbox.

Source code in src/cuemsengine/tools/display_conf.py

def read_display_conf(
    path: str = DEFAULT_DISPLAY_CONF,
) -> Tuple[dict, Tuple[int, int]]:
    """Parse display.conf and return ``(regions, canvas_size)``.

    Returns a 2-tuple:

    - ``regions``: ``{connector_name: {'x', 'y', 'width', 'height'}}`` with
      pixel-int values.
    - ``canvas_size``: ``(canvas_width, canvas_height)``. If the global
      ``canvas_size=WIDTHxHEIGHT`` key is present in the file's preamble,
      it is used (after validating it is >= the per-region bounding box).
      Otherwise, computed as ``max(x + width, y + height)`` over all regions.

    Raises:

    - ``DisplayConfNotFoundError`` if the file is missing or has no
      enabled ``[output:*]`` section with a valid canvas_region.
    - ``DisplayConfValueError`` if ``canvas_size=`` is malformed, has
      non-positive values, or is smaller than the per-region bbox.
    """
    if not os.path.isfile(path):
        raise DisplayConfNotFoundError(
            f"display.conf not found at {path}; "
            "videocomposer must run first (its ExecStartPre generates it)"
        )

    with open(path) as f:
        body = f.read()

    # Pre-pass: extract global preamble (everything before the first
    # [section] header) so we can scan it for `canvas_size=`. The
    # ConfigParser path below DISCARDS the preamble — without this
    # pre-pass the override is silently lost.
    if body.lstrip().startswith("["):
        preamble = ""
        sectioned = body.lstrip()
    elif "\n[" in body:
        split_at = body.find("\n[") + 1
        preamble = body[:split_at]
        sectioned = body[split_at:]
    else:
        preamble = body
        sectioned = ""

    canvas_override = _parse_canvas_size_override(preamble)
    disabled = _disabled_outputs(sectioned)

    # strict=False: a repeated key (two `enabled=` lines) is legal for the
    # videocomposer, which keeps the last one; strict mode would raise and
    # take the whole read down with it.
    parser = configparser.ConfigParser(strict=False)
    parser.optionxform = str  # preserve key case for forward-compat keys
    if sectioned:
        parser.read_string(sectioned)

    regions: dict = {}
    for section in parser.sections():
        if not section.startswith("output:"):
            continue
        connector = section[len("output:") :]
        if connector in disabled:
            continue
        raw = parser.get(section, "canvas_region", fallback=None)
        if raw is None:
            continue
        parts = [p.strip() for p in raw.split(",")]
        if len(parts) != 4:
            continue
        try:
            x, y, w, h = (int(p) for p in parts)
        except ValueError:
            continue
        regions[connector] = {"x": x, "y": y, "width": w, "height": h}

    if not regions:
        raise DisplayConfNotFoundError(
            f"display.conf at {path} has no [output:*] sections with a "
            "valid canvas_region (disabled outputs do not count)"
        )

    bbox_w = max(r["x"] + r["width"] for r in regions.values())
    bbox_h = max(r["y"] + r["height"] for r in regions.values())

    if canvas_override is not None:
        cw, ch = canvas_override
        if cw < bbox_w or ch < bbox_h:
            raise DisplayConfValueError(
                f"canvas_size={cw}x{ch} is smaller than the per-output "
                f"bounding box {bbox_w}x{bbox_h}; monitors would be cropped"
            )
        return regions, (cw, ch)

    return regions, (bbox_w, bbox_h)

get_used_ports()

Recover every local port currently bound on this host, TCP and UDP.

Uses ss -tuln — deliberately without -p. The owning PID is of no use to the port allocator, and asking for it is actively harmful: ss hides the whole users:(...) field for sockets owned by another user, and the previous implementation only recorded a port when it could also read a pid= out of that field. As a non-owner it therefore returned nothing at all rather than a partial list, leaving the allocator blind to every socket it did not own (869ed9wf7).

Returns:

Type	Description
set[int]	set[int]: Local ports currently bound. Empty if ss is unavailable —
set[int]	that degrades the allocator to its previous behaviour rather than
set[int]	taking the engine down.

Source code in src/cuemsengine/tools/system_ports.py

def get_used_ports() -> set[int]:
    """
    Recover every local port currently bound on this host, TCP and UDP.

    Uses `ss -tuln` — deliberately *without* `-p`. The owning PID is of no use
    to the port allocator, and asking for it is actively harmful: `ss` hides
    the whole `users:(...)` field for sockets owned by another user, and the
    previous implementation only recorded a port when it could also read a
    `pid=` out of that field. As a non-owner it therefore returned nothing at
    all rather than a partial list, leaving the allocator blind to every
    socket it did not own (869ed9wf7).

    Returns:
        set[int]: Local ports currently bound. Empty if `ss` is unavailable —
        that degrades the allocator to its previous behaviour rather than
        taking the engine down.
    """
    try:
        result = subprocess.run(
            ["ss", "-tuln"], capture_output=True, text=True, check=True
        )
    except (subprocess.CalledProcessError, FileNotFoundError, OSError) as e:
        Logger.warning(f"Could not list system ports with 'ss': {e}")
        return set()

    ports = set()
    for line in result.stdout.splitlines():
        # Netid State Recv-Q Send-Q <Local Address:Port> <Peer Address:Port>
        parts = line.split()
        if len(parts) < 5 or parts[0] == "Netid":
            continue
        # rpartition handles both "0.0.0.0:5353" and "[::]:22"; the wildcard
        # "*:*" and any other non-numeric tail fall through the isdigit test.
        port = parts[4].rpartition(":")[2]
        if port.isdigit():
            ports.add(int(port))
    return ports

is_port_in_use(port)

Check if a specific port is currently bound on this host.

Parameters:

Name	Type	Description	Default
port	int	The port number to check	required

Returns:

Name	Type	Description
bool	bool	True if port is in use, False otherwise

Example

if is_port_in_use(8080): ... print("Port 8080 is in use")

Source code in src/cuemsengine/tools/system_ports.py

def is_port_in_use(port: int) -> bool:
    """
    Check if a specific port is currently bound on this host.

    Args:
        port (int): The port number to check

    Returns:
        bool: True if port is in use, False otherwise

    Example:
        >>> if is_port_in_use(8080):
        ...     print("Port 8080 is in use")
    """
    return port in get_used_ports()