Application API

@abstractmethod
def init(self, commandline_args: Sequence[str]) -> None:
    """Parse application arguments and start preparation."""
    ...

def default_client_window_mode(self) -> str | None:
    """Return the preferred CLI mode when ``--mode`` is omitted.

    Asked before :meth:`init`, so this must not load models or acquire
    resources. Return an installed client-window mode name, or ``None``
    to retain the runtime's MP4 default. An explicit ``--mode`` wins.
    """
    return None

def resolve_client_window_mode(self, mode: str) -> str:
    """Return the installed window mode implementing the requested output.

    An application may select a registered mode that adds its input source
    while preserving the requested presentation backend. The runtime still
    validates arguments and creates the window. Called before :meth:`init`;
    do not load models or acquire resources here. Help and application
    preload skip this hook.

    Args:
        mode: Explicit CLI mode, or the resolved application/runtime default.

    Returns:
        Installed mode name. The default returns ``mode`` unchanged.

    Raises:
        ValueError: The application cannot support the requested output.
    """
    return mode

def session_desc(self) -> SessionDesc | None:
    """Return the description of a session this application would generate.

    A caller has to describe a session before there is one to describe, and
    only the application knows what its model was trained for. Asked before
    :meth:`init`, so describing a session costs nothing.

    Returns:
        The session to create when nobody asks for another, or ``None``,
        the default, from an application that generates whatever it is
        asked for. Its caller describes the session instead.
    """
    return None

@abstractmethod
def create_session(self, session_desc: SessionDesc) -> ISession:
    """Create one isolated, uninitialized session for ``session_desc``.

    Args:
        session_desc: Session the runtime is asking for.

    Returns:
        A session for ``session_desc``, resolved to what this application can
        actually produce.

    Raises:
        ValueError: The application cannot honour ``session_desc``.
    """
    ...

def close(self) -> None:
    """Release whatever the application holds.

    Not abstract, and does nothing by default, so an application with nothing
    to release does not implement it.
    """
    return

@abstractmethod
def init(self) -> None:
    """Initialize state and register this session's loops.

    A model loop is required. Registering a UI loop is optional; without one
    the runtime uses :class:`BlitModelOutputToScreenLoop`.

    Raises:
        RuntimeError: No model loop was registered by the time the runtime
            asks for the loops.
    """
    ...

@final
def register_ui_loop(
    self,
    loop_type: type[IUILoop[Any]],
    *,
    state: Any = None,
    **kwargs: Any,
) -> IUILoop[Any]:
    """Create and register the UI loop, and return it.

    Omit this call to use the default UI. The loop is paced at the session's
    ``frames_per_second_for_ui``.

    Args:
        loop_type: UI loop class to instantiate.
        state: State the loop owns. Unlike a model loop, a UI loop is
            allowed to hold none.
        **kwargs: Passed to ``loop_type``.

    Raises:
        RuntimeError: The runtime already took the loops, or a UI loop was
            registered already.
        TypeError: ``loop_type`` does not derive from :class:`IUILoop`.
    """
    if self._registrations_frozen:
        raise RuntimeError("Loop registrations are already in use.")
    if self._registered_ui_loop is not None:
        raise RuntimeError("The session already registered a UI loop.")
    if not issubclass(loop_type, IUILoop):
        raise TypeError("The UI loop must derive from IUILoop.")
    loop = loop_type(**kwargs)
    loop.register_session_loop_objects(
        state=state,
        frequency=self.session_desc.frames_per_second_for_ui,
        shutdown_event=self._shutdown_event,
        failure_queue=self._failure_queue,
        profiler=self._profiler,
    )
    loop.register_session_ui_loop_objects(
        session_desc=self.session_desc,
        presentation_manager=self._presentation_manager,
    )
    self._registered_ui_loop = loop
    return loop

@final
def register_model_loop(
    self,
    loop_type: type[IModelLoop[Any]],
    *,
    state: Any,
    **kwargs: Any,
) -> IModelLoop[Any]:
    """Create and register the model loop, and return it.

    The loop is paced at the session's ``frames_per_second_for_step``.

    Args:
        loop_type: Model loop class to instantiate.
        state: State the loop owns. Required, since a model loop with no
            state has nothing to generate from.
        **kwargs: Passed to ``loop_type``.

    Raises:
        RuntimeError: The runtime already took the loops, or a model loop
            was registered already.
        TypeError: ``loop_type`` does not derive from :class:`IModelLoop`.
    """
    if self._registrations_frozen:
        raise RuntimeError("Loop registrations are already in use.")
    if self._registered_model_loop is not None:
        raise RuntimeError("The session already registered a model loop.")
    if not issubclass(loop_type, IModelLoop):
        raise TypeError("The model loop must derive from IModelLoop.")
    loop = loop_type(**kwargs)
    loop.register_session_loop_objects(
        state=state,
        frequency=self.session_desc.frames_per_second_for_step,
        shutdown_event=self._shutdown_event,
        failure_queue=self._failure_queue,
        profiler=self._profiler,
    )
    self._registered_model_loop = loop
    return loop

def close(self) -> None:
    """Release resources owned by the session."""
    return

@abstractmethod
def step(self, step_index: int, events: UserInputEvents) -> list[StepResult]:
    """We expect IModelLoop's to return a list of StepResults, one result per channel."""
    ...

@abstractmethod
def step(self, step_index: int, events: UserInputEvents) -> list[StepResult]:
    """Return zero or one :class:`StepResult` to present this tick."""
    ...

@final
def register_session_ui_loop_objects(
    self,
    *,
    session_desc: SessionDesc,
    presentation_manager: PresentationManager,
) -> None:
    """Store UI objects supplied when this loop is registered with a session.

    Args:
        session_desc: Description of the session this UI presents.
        presentation_manager: Buffer containing model frames.
    """
    self.session_desc = session_desc
    self.output_layout = session_desc.output_layout
    self._presentation_manager = presentation_manager

    self._ui_loop_requests: UILoopRequests | None = None

@final
def request_new_session(self, session_desc: SessionDesc) -> None:
    """Ask the runtime to replace this session after the current UI step.

    Args:
        session_desc: Fully resolved description for the replacement session.
    """
    self.get_or_create_ui_loop_requests().new_session = session_desc

@final
def request_hide_cursor(self, hide_cursor: bool) -> None:
    """Request that the client window show or hide its cursor."""
    if not isinstance(hide_cursor, bool):
        raise TypeError("hide_cursor must be a bool.")
    self.get_or_create_ui_loop_requests().hide_cursor = hide_cursor

@final
def request_lock_cursor_to_window(self, lock_cursor_to_window: bool) -> None:
    """Request that the client window release or capture pointer motion."""
    if not isinstance(lock_cursor_to_window, bool):
        raise TypeError("lock_cursor_to_window must be a bool.")
    self.get_or_create_ui_loop_requests().lock_cursor_to_window = (
        lock_cursor_to_window
    )

@final
def request_new_window_size(self, new_window_size: tuple[int, int]) -> None:
    """Request a positive downstream client-window width and height.

    Args:
        new_window_size: Requested ``(width, height)`` in pixels.

    Raises:
        TypeError: The size is not a pair of integers.
        ValueError: Either dimension is not positive.
    """
    if (
        not isinstance(new_window_size, tuple)
        or len(new_window_size) != 2
        or any(
            isinstance(dimension, bool) or not isinstance(dimension, int)
            for dimension in new_window_size
        )
    ):
        raise TypeError("new_window_size must be a tuple of two integers.")
    if any(dimension <= 0 for dimension in new_window_size):
        raise ValueError("new_window_size dimensions must be > 0.")
    self.get_or_create_ui_loop_requests().new_window_size = new_window_size

@final
def request_links(self, links: tuple[ClientWindowLink, ...]) -> None:
    """Replace up to sixteen clickable regions of the output content.

    Args:
        links: Immutable regions; an empty tuple clears existing links.

    Raises:
        TypeError: The request is not a tuple of ``ClientWindowLink`` objects.
        ValueError: More than sixteen regions are supplied.
    """
    if not isinstance(links, tuple) or any(
        not isinstance(link, ClientWindowLink) for link in links
    ):
        raise TypeError("links must be a tuple of ClientWindowLink objects.")
    if len(links) > 16:
        raise ValueError("At most sixteen client-window links are supported.")
    self.get_or_create_ui_loop_requests().links = links

def flush_ui_loop_requests(self) -> UILoopRequests | None:
    """return UI loop requests and clear the internal state of requests."""
    requests = self._ui_loop_requests
    self._ui_loop_requests = None
    return requests

@final
def presented_model_frame(
    self,
    channel_index: int = 0,
) -> Tensor | None:
    """Return the current frame from one model-result channel.

    Args:
        channel_index: Channel to read, indexed as the model loop returned
            them.

    Returns:
        A ``[C, H, W]`` frame with one, three or four channels, or ``None``
        before the first model result has been presented.

    Raises:
        IndexError: The presented result has no such channel.
        ValueError: The presentation stream is on a different CUDA device
            from the presented result.
    """
    return self._presentation_manager.presented_frame(channel_index)

@final
def presented_model_frames(self) -> tuple[Tensor, ...]:
    """Return the current frame from every model-result channel.

    Returns:
        One ``[C, H, W]`` frame per channel, bottom channel first, or an
        empty tuple before the first model result has been presented.

    Raises:
        ValueError: The presentation stream is on a different CUDA device
            from a presented result.
    """
    return self._presentation_manager.presented_frames()

@final
def has_pending_model_frames(self) -> bool:
    """Return whether another model frame is ready to present."""
    return self._presentation_manager.has_pending_frames()

def resize_ui_loop(self, width: int, height: int) -> None:
    """Resize the ImGui renderer."""
    self.renderer.resize(width, height)

def get_ui_loop_size(self) -> tuple[int, int]:
    """Return the current ImGui render-target size."""
    return self.renderer.width, self.renderer.height

@abstractmethod
def step_ui(
    self,
    imgui: Any,
    step_index: int,
    events: UserInputEvents,
) -> Tensor | None:
    """Draw one immediate-mode UI frame and return its optional back buffer."""
    ...

@final
def step(self, step_index: int, events: UserInputEvents) -> list[StepResult]:
    """Render and composite one ImGui frame."""
    # User choses when to resize the UI, UI
    # must never resize the UI loop itself.
    # User may want to stretch the UI to fill
    # the overly-sized window.
    back_buffer: Tensor | None = None

    def draw(imgui: Any, index: int, current_events: UserInputEvents) -> None:
        nonlocal back_buffer
        back_buffer = self.step_ui(imgui, index, current_events)

    overlay = self.renderer.render(step_index, events, draw)
    back_buffer = prepare_ui_back_buffer(back_buffer, overlay)
    frame = self._presentation_manager.composite(back_buffer, overlay)
    return [
        StepResult(
            step_index=step_index,
            output=frame.unsqueeze(0),
            frame_count=1,
            output_layout=self.output_layout,
        )
    ]

def reset(self) -> None:
    """Reset renderer state after a session reset event."""
    self.renderer.reset()

def close(self) -> None:
    """Release the renderer."""
    self.renderer.close()