"""Hook primitives for function components.

Provides React-like hooks for managing state, effects, memoisation,

context, and navigation within function components decorated with

:func:`component`.

Usage::

import pythonnative as pn

@pn.component

def counter(initial=0):

count, set_count = pn.use_state(initial)

return pn.Column(

pn.Text(f"Count: {count}"),

pn.Button("+", on_click=lambda: set_count(count + 1)),

)

"""

import inspect

import threading

from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar

from .element import Element

T = TypeVar("T")

_SENTINEL = object()

_hook_context: threading.local = threading.local()

# ======================================================================

# Hook state container

# ======================================================================

class HookState:

"""Stores all hook data for a single function component instance."""

__slots__ = ("states", "effects", "memos", "refs", "hook_index", "_trigger_render")

def __init__(self) -> None:

self.states: List[Any] = []

self.effects: List[Tuple[Any, Any]] = []

self.memos: List[Tuple[Any, Any]] = []

self.refs: List[dict] = []

self.hook_index: int = 0

self._trigger_render: Optional[Callable[[], None]] = None

def reset_index(self) -> None:

self.hook_index = 0

def run_pending_effects(self) -> None:

"""Execute effects whose deps changed during the last render pass."""

for i, (deps, cleanup) in enumerate(self.effects):

if deps is _SENTINEL:

continue

def cleanup_all_effects(self) -> None:

"""Run all outstanding cleanup functions (called on unmount)."""

for i, (deps, cleanup) in enumerate(self.effects):

if callable(cleanup):

try:

cleanup()

except Exception:

pass

self.effects[i] = (_SENTINEL, None)

# ======================================================================

# Thread-local context helpers

# ======================================================================

def _get_hook_state() -> Optional[HookState]:

return getattr(_hook_context, "current", None)

def _set_hook_state(state: Optional[HookState]) -> None:

_hook_context.current = state

def _deps_changed(prev: Any, current: Any) -> bool:

if prev is _SENTINEL:

return True

if prev is None or current is None:

return True

if len(prev) != len(current):

return True

return any(p is not c and p != c for p, c in zip(prev, current))

# ======================================================================

# Public hooks

# ======================================================================

def use_state(initial: Any = None) -> Tuple[Any, Callable]:

"""Return ``(value, setter)`` for component-local state.

If *initial* is callable it is invoked once (lazy initialisation).

The setter accepts a value **or** a ``current -> new`` callable.

"""

ctx = _get_hook_state()

if ctx is None:

raise RuntimeError("use_state must be called inside a @component function")

idx = ctx.hook_index

ctx.hook_index += 1

if idx >= len(ctx.states):

val = initial() if callable(initial) else initial

ctx.states.append(val)

current = ctx.states[idx]

def setter(new_value: Any) -> None:

if callable(new_value):

new_value = new_value(ctx.states[idx])

if ctx.states[idx] is not new_value and ctx.states[idx] != new_value:

ctx.states[idx] = new_value

if ctx._trigger_render:

ctx._trigger_render()

return current, setter

def use_effect(effect: Callable, deps: Optional[list] = None) -> None:

"""Schedule *effect* to run after render.

*deps* controls when the effect re-runs:

- ``None`` -> every render

- ``[]`` -> mount only

- ``[a, b]``-> when *a* or *b* change

*effect* may return a cleanup callable.

"""

ctx = _get_hook_state()

if ctx is None:

raise RuntimeError("use_effect must be called inside a @component function")

idx = ctx.hook_index

ctx.hook_index += 1

if idx >= len(ctx.effects):

ctx.effects.append((_SENTINEL, None))

prev_deps, prev_cleanup = ctx.effects[idx]

if _deps_changed(prev_deps, deps):

if callable(prev_cleanup):

try:

prev_cleanup()

except Exception:

pass

cleanup = effect()

ctx.effects[idx] = (list(deps) if deps is not None else None, cleanup)

else:

ctx.effects[idx] = (prev_deps, prev_cleanup)

def use_memo(factory: Callable[[], T], deps: list) -> T:

"""Return a memoised value, recomputed only when *deps* change."""

ctx = _get_hook_state()

if ctx is None:

raise RuntimeError("use_memo must be called inside a @component function")

idx = ctx.hook_index

ctx.hook_index += 1

if idx >= len(ctx.memos):

value = factory()

ctx.memos.append((list(deps), value))

return value

prev_deps, prev_value = ctx.memos[idx]

if not _deps_changed(prev_deps, deps):

return prev_value

value = factory()

ctx.memos[idx] = (list(deps), value)

return value

def use_callback(callback: Callable, deps: list) -> Callable:

"""Return a stable reference to *callback*, updated only when *deps* change."""

return use_memo(lambda: callback, deps)

def use_ref(initial: Any = None) -> dict:

"""Return a mutable ref dict ``{"current": initial}`` that persists across renders."""

ctx = _get_hook_state()

if ctx is None:

raise RuntimeError("use_ref must be called inside a @component function")

idx = ctx.hook_index

ctx.hook_index += 1

if idx >= len(ctx.refs):

ref: dict = {"current": initial}

ctx.refs.append(ref)

return ref

return ctx.refs[idx]

# ======================================================================

# Context

# ======================================================================

class Context:

"""A context object created by :func:`create_context`."""

def __init__(self, default: Any = None) -> None:

self.default = default

self._stack: List[Any] = []

def _current(self) -> Any:

return self._stack[-1] if self._stack else self.default

def create_context(default: Any = None) -> Context:

"""Create a new context with an optional default value."""

return Context(default)

def use_context(context: Context) -> Any:

"""Read the current value of *context* from the nearest ``Provider`` ancestor."""

ctx = _get_hook_state()

if ctx is None:

raise RuntimeError("use_context must be called inside a @component function")

return context._current()

# ======================================================================

# Provider element helper

# ======================================================================

def Provider(context: Context, value: Any, child: Element) -> Element:

"""Create a context provider element.

All descendants of *child* will read *value* via ``use_context(context)``.

"""

return Element("__Provider__", {"__context__": context, "__value__": value}, [child])

# ======================================================================

# Navigation

# ======================================================================

_NavigationContext: Context = create_context(None)

class NavigationHandle:

"""Object returned by :func:`use_navigation` providing push/pop/get_args.

Navigates by component reference rather than string path, e.g.::

nav = pn.use_navigation()

nav.push(DetailScreen, args={"id": 42})

"""

def __init__(self, host: Any) -> None:

self._host = host

def push(self, page: Any, args: Optional[Dict[str, Any]] = None) -> None:

"""Navigate forward to *page* (a ``@component`` function or class)."""

self._host._push(page, args)

def pop(self) -> None:

"""Navigate back to the previous screen."""

self._host._pop()

def get_args(self) -> Dict[str, Any]:

"""Return arguments passed from the previous screen."""

return self._host._get_nav_args()

def use_navigation() -> NavigationHandle:

"""Return a :class:`NavigationHandle` for the current screen.

Must be called inside a ``@component`` function rendered by PythonNative.

"""

handle = use_context(_NavigationContext)

if handle is None:

raise RuntimeError(

"use_navigation() called outside a PythonNative page. "

"Ensure your component is rendered via create_page()."

)

return handle

# ======================================================================

# @component decorator

# ======================================================================

def component(func: Callable) -> Callable[..., Element]:

"""Decorator that turns a Python function into a PythonNative component.

The decorated function can use hooks (``use_state``, ``use_effect``, etc.)

and returns an ``Element`` tree. Each call site creates an independent

component instance with its own hook state.

"""

sig = inspect.signature(func)

positional_params = [

name

for name, p in sig.parameters.items()

if p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)

]

has_var_positional = any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values())

def wrapper(*args: Any, **kwargs: Any) -> Element:

props: dict = dict(kwargs)

if args:

if has_var_positional:

props["children"] = list(args)

else:

for i, arg in enumerate(args):

if i < len(positional_params):

props[positional_params[i]] = arg

key = props.pop("key", None)

return Element(func, props, [], key=key)

wrapper.__wrapped__ = func # noqa: B010

wrapper.__name__ = func.__name__

wrapper.__qualname__ = func.__qualname__

wrapper._pn_component = True # noqa: B010

return wrapper