xnano.core.runtime

@classmethod
def live(
    cls,
    *,
    state: StateT | None = None,
    title: str | None = None,
    tick_interval: int = 16,
    mouse_events: bool = False,
) -> "Runtime[StateT]":
    """Create a runtime backed by the active terminal."""
    session = CoreSession.init(tick_rate_ms=None)
    if mouse_events:
        session.enable_mouse_capture()
    return cls(
        session,
        live=True,
        state=state,
        title=title,
        tick_interval=tick_interval,
    )

@classmethod
def offscreen(
    cls,
    width: int = 80,
    height: int = 24,
    *,
    state: StateT | None = None,
    title: str | None = None,
) -> "Runtime[StateT]":
    """Create an active in-memory runtime."""
    runtime = cls(
        CoreSession.offscreen(width, height),
        live=False,
        state=state,
        title=title,
        surface="offscreen",
    )
    return runtime.enter()

@staticmethod
def supports_live_terminal() -> bool:
    """Return whether this build can claim a live terminal."""
    return CoreSession.supports_live_terminal()

def enter(self) -> "Runtime[StateT]":
    """Bind this runtime to the current context."""
    if self._token is None:
        self._token = _ACTIVE_RUNTIME.set(self)
    if self._live:
        self._install_signal_handlers()
    return self

def close(self) -> None:
    """Restore the native session and release the active binding."""
    if self._closed:
        return
    self._closed = True
    if self._token is not None:
        _ACTIVE_RUNTIME.reset(self._token)
        self._token = None
    # Restore the host terminal first (raw mode, mouse tracking,
    # alternate screen, SGR) so a failure restoring signal handlers
    # never leaves the screen corrupted.
    try:
        self._session.restore()
    finally:
        # Runtime owns cursor/device/action proxies that point back to it,
        # so the Python object can remain in a reference cycle after
        # close. Release the unsendable PyO3 session now, on its owner
        # thread, instead of leaving its destructor to a later GC pass.
        del self._session
        self._restore_signal_handlers()

def set_root(self, root: Any) -> None:
    """Set the renderable used by subsequent empty renders."""
    self._root = root

def render(
    self,
    *renderables: Any,
    foreground: ColorLike | None = None,
    background: ColorLike | None = None,
    modifiers: Sequence[CharacterModifier] | None = None,
    horizontal_align: Alignment | None = None,
    vertical_align: VerticalAlignment | None = None,
    border: Border | None = None,
    border_sides: Sequence[Side] | None = None,
    border_color: ColorLike | None = None,
    title: str | None = None,
    title_position: FrameTitlePosition | None = None,
    padding: PaddingLike | None = None,
    gap: int = 0,
    direction: Direction = "vertical",
    color: ColorLike | None = None,
    align: Alignment | None = None,
) -> Frame:
    """Render one frame and return its immutable snapshot."""
    foreground = resolve_color_alias(foreground, color, stacklevel=3)
    horizontal_align = resolve_renamed_alias(
        horizontal_align,
        align,
        old="align",
        new="horizontal_align",
        stacklevel=3,
    )
    self._render(
        *renderables,
        foreground=foreground,
        background=background,
        modifiers=modifiers,
        horizontal_align=horizontal_align,
        vertical_align=vertical_align,
        border=border,
        border_sides=border_sides,
        border_color=border_color,
        title=title,
        title_position=title_position,
        padding=padding,
        gap=gap,
        direction=direction,
    )
    return self._snapshot_frame()

def call_soon(self, callback: Callable[..., Any], *args: Any) -> None:
    """Schedule ``callback`` to run on the UI thread before the next pump.

    Thread-safe: worker threads enqueue here and the runtime drains the
    queue on its own thread, so background work can mutate grid state
    without racing the renderer.
    """
    with self._call_soon_lock:
        self._call_soon_queue.append((callback, args))

def pump(self, timeout: float = 0.0) -> bool:
    """Poll and dispatch at most one event."""
    self._drain_call_soon()
    if self._should_exit:
        return False
    timeout_ms = max(0, int(timeout * 1000))
    if self._live and timeout_ms == 0:
        timeout_ms = self._tick_interval
    native_event = self._session.poll_event(timeout_ms)
    if native_event is not None:
        self.dispatch(event_from_core(native_event))
    elif self._root is not None:
        from xnano.core.dispatch import dispatch_idle

        dispatch_idle(self._root, self)
    if self._root is not None:
        from xnano.events import Event, TickEventData

        now = time.monotonic() * 1000
        elapsed_ms = max(0, int(now - self._last_tick_ms))
        self._last_tick_ms = now
        self.dispatch(
            Event.from_data(TickEventData(elapsed_ms=elapsed_ms))
        )
    return not self._should_exit

def dispatch(self, event: Any) -> None:
    """Dispatch one event to the root grid or component."""
    from xnano.core.dispatch import dispatch_event
    from xnano.utils.focus import (
        apply_text_keyboard,
        cycle_field_focus,
        focused_component,
        move_field_focus,
        spatial_focus_enabled,
    )

    consumed = False
    mouse = getattr(event, "mouse_event", None)
    if mouse is not None and mouse.field is None:
        from xnano.core.dispatch import iter_grids
        from xnano.events import Event, MouseEventData

        for grid in iter_grids(self._root):
            for hit in reversed(getattr(grid, "_grid_field_hits", ())):
                if hit.area.contains((mouse.x, mouse.y)):
                    field = hit.grid._grid_field_info(hit.field_name)
                    self._auto_scroll_wheel(hit, field, mouse.kind)
                    self._click_to_focus(hit, field, mouse.kind)
                    component = getattr(hit.grid, hit.field_name, None)
                    hover = getattr(component, "handle_hover", None)
                    if callable(hover):
                        hover(mouse.x, mouse.y)
                    event = Event.from_data(
                        MouseEventData(
                            kind=mouse.kind,
                            x=mouse.x,
                            y=mouse.y,
                            button=mouse.button,
                            field=hit.field_name,
                            group=field.group,
                        )
                    )
                    break
    keyboard = getattr(event, "keyboard_event", None)
    if keyboard is not None:
        if keyboard.matches("ctrl+c"):
            self.request_exit()
            return
        if keyboard.matches("tab"):
            consumed = cycle_field_focus(self)
        elif keyboard.matches("shift+tab"):
            consumed = cycle_field_focus(self, -1)
        else:
            consumed = apply_text_keyboard(
                focused_component(self),
                keyboard,
            )
            if not consumed and spatial_focus_enabled(self):
                for direction in ("up", "down", "left", "right"):
                    if keyboard.matches(direction):
                        consumed = move_field_focus(self, direction)
                        break
    clipboard = getattr(event, "clipboard_event", None)
    tick = getattr(event, "tick_event", None)
    if tick is not None:
        self._elapsed_ms += tick.elapsed_ms
    component = focused_component(self)
    if clipboard is not None and component is not None:
        paste_handler = getattr(component, "handle_paste", None)
        if callable(paste_handler):
            consumed = (
                bool(paste_handler(clipboard.text or "")) or consumed
            )
    if self._root is not None and not consumed:
        dispatch_event(self._root, self, event)
    root_dispatch = getattr(self._root, "dispatch_event", None)
    if callable(root_dispatch):
        root_dispatch(event, self)

def play_effect(
    self,
    effect: Any,
    *,
    fields: list[str] | None = None,
    repeat: bool = False,
) -> list[str]:
    """Play an effect over fields recorded by the latest render.

    Args:
        effect: Effect description.
        fields: Field names whose rendered areas receive the effect.
        repeat: Loop the effect until it is cancelled, instead of
            running once for its duration.

    Returns:
        The session keys registered, one per field that had a
        rendered area. Empty when nothing was targeted.
    """
    import xnano_core.rust.native as native

    from xnano.core.effects import resolve_native_effect

    keys: list[str] = []
    lowered = None
    for field in fields or ():
        area = self._session.effect_area_for(field)
        if area is None:
            continue
        if lowered is None:
            # Lower once, not once per target field.
            lowered = resolve_native_effect(effect)
            if repeat:
                lowered = native.repeating_effect(lowered)
        key = f"{getattr(effect, 'key', None) or field}:{field}"
        self._session.add_unique_effect(key, lowered.with_area(area))
        keys.append(key)
    return keys

def cancel_effect(self, key: str) -> None:
    """Stop the effect registered under ``key``."""
    self._session.cancel_effect(key)

def is_animating(self) -> bool:
    """Whether any effect is currently running in this session."""
    return bool(self._session.is_animating())

def perform(self, action: Any) -> None:
    """Perform a synthetic action through its event representation."""
    from xnano.actions import (
        ClickAction,
        ClipboardAction,
        FocusAction,
        KeyboardAction,
        MouseAction,
        RequestAction,
        ResizeAction,
        TickAction,
    )
    from xnano.events import (
        ClipboardEventData,
        Event,
        FocusEventData,
        KeyboardEventData,
        MouseEventData,
        ResizeEventData,
        TickEventData,
    )

    if isinstance(action, KeyboardAction):
        for binding in action.bindings:
            self.dispatch(
                Event.from_data(
                    KeyboardEventData.from_binding(
                        str(binding),
                        kind=action.kind or "press",
                    )
                )
            )
        return
    if isinstance(action, MouseAction):
        self.dispatch(
            Event.from_data(
                MouseEventData(
                    kind=action.kind or "press",
                    x=0,
                    y=0,
                    button=action.buttons[0]
                    if action.buttons
                    else "unknown",
                )
            )
        )
        return
    if isinstance(action, ClickAction):
        event = Event.from_data(
            MouseEventData(
                kind="press",
                x=0,
                y=0,
                button=action.button,
                field=action.field,
            )
        )
        self.dispatch(event)
        return
    if isinstance(action, FocusAction):
        self.dispatch(
            Event.from_data(
                FocusEventData(
                    kind=action.kind or "gained",
                    field=action.field,
                )
            )
        )
        return
    if isinstance(action, ClipboardAction):
        self.dispatch(
            Event.from_data(ClipboardEventData(text=action.text))
        )
        return
    if isinstance(action, ResizeAction):
        self.dispatch(
            Event.from_data(
                ResizeEventData(
                    width=action.width or self.size[0],
                    height=action.height or self.size[1],
                )
            )
        )
        return
    if isinstance(action, TickAction):
        self.dispatch(
            Event.from_data(
                TickEventData(elapsed_ms=action.interval_ms),
            )
        )
        return
    if isinstance(action, RequestAction) and self._root is not None:
        from xnano.requests import dispatch_request

        dispatch_request(
            self._root,
            action.method,
            action.path,
            runtime=self,
        )
        return
    self.dispatch(action)

def resize(self, width: int, height: int) -> None:
    """Resize support is fixed at offscreen-session construction."""
    if self.size != (width, height):
        raise RuntimeError("Create a new offscreen runtime to resize it.")

def request_exit(self) -> None:
    """Stop the run loop after the current dispatch."""
    self._should_exit = True

def focus(self, group: str) -> bool:
    """Focus a named field group."""
    from xnano.utils.focus import (
        resolve_group_target,
        set_field_focus,
    )

    target = resolve_group_target(self, group)
    if target is None:
        return False
    set_field_focus(self, target)
    return True

def blur(self) -> None:
    """Clear field focus."""
    from xnano.utils.focus import clear_field_focus

    clear_field_focus(self)
    self._focused_group = None

def focus_next(self) -> bool:
    """Move focus through the root grid when supported."""
    from xnano.utils.focus import cycle_field_focus

    return cycle_field_focus(self)

def focus_previous(self) -> bool:
    """Move focus backward through the root grid when supported."""
    from xnano.utils.focus import cycle_field_focus

    return cycle_field_focus(self, -1)

def get_output(self) -> str:
    """Return the current buffer as plain text."""
    return "\n".join(self._session.buffer_snapshot().to_string_lines())

def get_output_as_ansi(self) -> str:
    """Return the current buffer with ANSI styling."""
    return "\n".join(self._session.buffer_snapshot().to_ansi_lines())