procedure

def __and__(
    self: Procedure[Bool],
    other: Procedure[Bool],
) -> Procedure[Bool]:
    """``a & b`` — a ``BoolAnd`` node."""
    from capturegraph.procedures.nodes.process.operators.boolean import BoolAnd

    return BoolAnd(
        left=self,
        right=other,
    )

def __eq__(self: Procedure, other: object) -> bool:
    """Two procedures are equal iff they are the same node (same ``uuid``)."""
    from capturegraph.procedures.procedure import (
        Procedure,
    )

    return isinstance(other, Procedure) and self._uuid == other._uuid

def __ge__(
    self: Procedure[Number],
    other: Procedure[Number],
) -> Procedure[Bool]:
    """``a >= b`` — ``BoolNot(NumberLessThan(a, b))``."""
    return ~(self < other)

def navigate(self: Procedure[Path], name: str) -> Procedure[Path]:
    """Read a struct field; bound as ``Procedure.__getattr__``.

    E.g. ``root.reference``, ``CapturePanorama().preview``. Only missing attributes
    reach here, and dunders never navigate.
    """
    if name.startswith("__") and name.endswith("__"):
        raise AttributeError(name)

    from capturegraph.procedures.procedure.authoring.dispatch import (
        navigable,
    )

    nav = navigable(self)
    if nav is None or not nav.has_field(name):
        raise AttributeError(name)

    if nav.is_path:
        require_no_saved_value(self)
    child = memoized_child(self, ("field", name), lambda: nav.nodes.field(source=self, field=name))
    return cast("Procedure[Path]", child)

def __getitem__(
    self: Procedure[Path] | Procedure[Array] | Procedure[Map],
    key: int | Procedure[CGType] | JSONScalar | bool | float | str,
) -> Procedure[CGType]:
    """Index a navigable position or a value-access array / map.

    On a navigated ``Path`` it selects an array element / map entry position
    (another ``Path``). On a map *value* a node returns it selects the entry's
    value (``MapKey``); on the ``Array`` that a map's ``keys()``/``values()``
    returns it selects the element *value* — a key scalar, or a ``Path[V]`` to
    navigate. The runtime dispatches all of these through the same ``navigable`` view.
    """
    from capturegraph.procedures.procedure.authoring.dispatch import (
        navigable,
    )

    nav = navigable(self)
    if nav is None or not (nav.is_array or nav.is_map):
        raise TypeError(f"{self} is not indexable (it is neither an array nor a map)")

    if nav.is_path:
        require_no_saved_value(cast("Procedure[Path]", self))
    child = subscript_child(cast("Procedure[Path]", self), nav, key, create=True)
    assert child is not None  # create=True always returns a child or raises
    return child

def __gt__(
    self: Procedure[Number],
    other: Procedure[Number],
) -> Procedure[Bool]:
    """``a > b`` — ``NumberLessThan`` with the operands swapped."""
    return other < self

def __hash__(self: Procedure) -> int:
    """Hash by node identity (the ``uuid``), consistent with ``__eq__``."""
    return hash(self._uuid)

def __iand__[T: CGType](
    self: Procedure[Path[T]],
    value: Procedure[T],
) -> CacheProcedure[T]:
    """``path &= value`` — cache the value, overwriting any stored value."""
    return record_cache(self, value, skip_if_exists=False)

def __init_subclass__(cls: type[Procedure]) -> None:
    """Record the subclass's return type from its ``Procedure[X]`` base."""
    from capturegraph.procedures.procedure import (
        Procedure,
    )

    super(Procedure, cls).__init_subclass__()
    declared = declared_return_type(cls)
    if declared is not None:
        cls._return_type = declared

def __invert__(self: Procedure[Bool]) -> Procedure[Bool]:
    """``~a`` — a ``BoolNot`` node."""
    from capturegraph.procedures.nodes.process.operators.boolean import BoolNot

    return BoolNot(value=self)

def __ior__[T: CGType](
    self: Procedure[Path[T]],
    value: Procedure[T],
) -> CacheProcedure[T]:
    """``path |= value`` — cache the value, skipping if already stored."""
    return record_cache(self, value, skip_if_exists=True)

def __le__(
    self: Procedure[Number],
    other: Procedure[Number],
) -> Procedure[Bool]:
    """``a <= b`` — ``BoolNot(NumberLessThan(b, a))``."""
    return ~(other < self)

def __lt__(
    self: Procedure[Number],
    other: Procedure[Number],
) -> Procedure[Bool]:
    """``a < b`` — a ``NumberLessThan`` node."""
    from capturegraph.procedures.nodes.process.operators.number import (
        NumberLessThan,
    )

    return NumberLessThan(
        left=self,
        right=other,
    )

def __or__(
    self: Procedure[Bool],
    other: Procedure[Bool],
) -> Procedure[Bool]:
    """``a | b`` — a ``BoolOr`` node."""
    from capturegraph.procedures.nodes.process.operators.boolean import BoolOr

    return BoolOr(
        left=self,
        right=other,
    )

def __post_init__(self: Procedure) -> None:
    """Stamp identity, type-check the node, and register it with the context.

    Constructing a node registers it with the active recording context (which
    is how a procedure body is built — see
    [capturegraph.procedures.procedure.authoring.contexts][]). There must be one: a node
    built with no active context raises, unless under ``cg.detached()``.
    """
    object.__setattr__(self, "_uuid", _new_uuid())
    object.__setattr__(self, "_children", {})
    validation.validate(self)
    from capturegraph.procedures.procedure.authoring.contexts.base import (
        ProcedureContext,
    )

    ProcedureContext.current()._collect(self)

def __repr__(self: Procedure) -> str:
    """The node's DAG as pretty-printed JSON.

    Equivalent to this node's
    [to_json][capturegraph.procedures.procedure.methods.export.to_json].
    """
    return to_json(self)

def __setattr__(self, name: str, value: CacheProcedure[Any]) -> None:
    """Type-checker-only stub; ``make_procedure`` installs the real version."""
    ...

def __setitem__(
    self: Procedure[Path],
    key: int | Procedure[CGType] | JSONScalar | bool | float | str,
    value: CacheProcedure[CGType],
) -> None:
    """Absorb the ``arr[i] |= value`` / ``map[k] &= value`` rebind.

    ``arr[i] |= value`` desugars to ``arr.__setitem__(i, arr[i].__ior__(value))``;
    the cache was already recorded by ``__ior__``, so this swallows it — but only
    when it is *this exact slot's* own cache (its ``destination`` is the navigation
    child for ``key``), mirroring the strict struct-field ``__setattr__`` guard. A
    plain ``arr[i] = value``, or a cache built for a different slot, is rejected.
    """
    from capturegraph.procedures.nodes.destination.cache import (
        CacheProcedure,
    )
    from capturegraph.procedures.procedure.authoring.dispatch import (
        navigable,
    )

    nav = navigable(self)
    if nav is not None and isinstance(value, CacheProcedure):
        if value.destination is subscript_child(self, nav, key, create=False):
            return
    raise TypeError(
        "cache at an indexed slot with '|=' (skip if exists) or '&=' (overwrite), "
        f"not plain assignment (tried to set [{key!r}])"
    )

def __str__(self: Procedure) -> str:
    """``"label (Kind)"``, or just ``"Kind"`` when unlabeled."""
    return f"{self.label} ({self.kind.__name__})" if self.label else self.kind.__name__

def exists(self: Procedure[Path]) -> Procedure[Bool]:
    """``path.exists()`` — a ``Bool`` that is true when a value is stored here.

    Reads the position from disk (``LoadPath``) and reports whether that read
    completes (``ProcedureCompleted``), so a missing value yields ``false`` instead
    of aborting the run. Pair it with ``cg.when(...)`` to run a block only once the
    position has been written. The read goes to disk regardless of any value cached
    at this slot this run, so it tests what was already present before the run.
    """
    from capturegraph.procedures.nodes.destination.access import (
        LoadPath,
    )
    from capturegraph.procedures.nodes.process.control import (
        ProcedureCompleted,
    )

    return ProcedureCompleted(procedure=LoadPath(source=self))

def keys(self: Procedure[Path]) -> Procedure[Array]:
    """A map position's keys as a read-only ``Array[K]`` value (index to select one)."""
    from capturegraph.procedures.procedure.authoring.dispatch import (
        MAP_KEYS,
    )

    require_map_position(self, "keys")
    return cast(
        "Procedure[Array]",
        memoized_child(self, ("keys",), lambda: MAP_KEYS(source=self)),
    )

def load[T: CGType](self: Procedure[Path[T]]) -> Procedure[T]:
    """``path.load()`` — read the value stored at this position.

    Loading is explicit: an unwritten position used as a value never lowers to a
    disk read on its own. The read waits for a stored value (a prior run's, or
    one an interceptor fills); while nothing is stored it stays pending, so use
    [load_or][] when an absent value should produce a fallback instead. The
    read goes to disk regardless of any value cached at this slot this run, so it
    sees what was already present before the run.
    """
    from capturegraph.procedures.nodes.destination.access import (
        LoadPath,
    )

    # The compile-time validation pass checks the node against the schema
    # position, so the declared target is enforced before execution.
    return cast("Procedure[T]", LoadPath(source=self))

def load_or(
    self: Procedure[Path],
    default: object,
) -> object:
    """``path.load_or(default)`` — the stored value, or ``default``'s value.

    Returned when nothing is stored here. The fallback runs only on a definitive
    "nothing stored" answer — never while the value is still loading — so a slow
    remote read cannot be mistaken for an absent one. ``default`` is a procedure (or
    raw literal) whose type matches the position's — so a typed default names the
    result's type, and the compile-time validation pass enforces the match.
    """
    from capturegraph.procedures.nodes.destination.access import (
        LoadPath,
    )
    from capturegraph.procedures.procedure.authoring.bridge import (
        coerce_input,
    )

    return LoadPath(source=self, default=coerce_input(default))

def preload[T: CGType](
    self: Procedure[T],
    default: object,
) -> Procedure[T]:
    """``node.preload(default)`` — pre-answer this node with an adjustable default.

    ``default`` is a procedure (or a raw literal, lifted to a constant) whose
    type matches this node's; the runtime preloads it in so the node starts
    complete and the user may adjust it before capturing::

        session.photo &= cg.CaptureImage(
            raw_photo=cg.UserInputBool(label="RAW").preload(False),
        )
    """
    from capturegraph.procedures.nodes.process.structural import (
        PreloadProcedure,
    )
    from capturegraph.procedures.procedure.authoring.bridge import (
        coerce_input,
    )

    return cast(
        "Procedure[T]",
        PreloadProcedure(procedure=self, default=coerce_input(default)),
    )

def set_label(self, label: str) -> Self:
    """Set a label on this node (in place) and return it, for fluent chaining."""
    object.__setattr__(self, "label", label)
    return self

def static_map_keys(self) -> frozenset[str] | None:
    """The keys of the ``Map`` this node produces, when known at authoring time.

    E.g. ``CaptureImagesByLabel``'s labels; ``None`` when they are only known at
    runtime. A value-level ``[key]`` access warns when a constant key is
    provably absent from this set; nodes whose keys are dynamic leave it
    ``None`` and are never second-guessed.
    """
    return None

def to_graphviz(
    self: Procedure,
    group_by_depth: bool = False,
) -> Digraph:
    """This node's DAG as a Graphviz diagram for visualization.

    See
    [procedure_to_graphviz][capturegraph.procedures.procedure.exporting.visualizer.procedure_to_graphviz].
    """
    from capturegraph.procedures.procedure.exporting.visualizer import (
        procedure_to_graphviz,
    )

    return procedure_to_graphviz(self, group_by_depth=group_by_depth)

def to_json(
    self: Procedure,
    indent: int = 4,
    target_schema: type[CGType] | None = None,
) -> str:
    """This node's DAG as a JSON string in the cross-platform wire form.

    See
    [procedure_to_json][capturegraph.procedures.procedure.exporting.serializer.procedure_to_json].
    """
    from capturegraph.procedures.procedure.exporting.serializer import (
        procedure_to_json,
    )

    return procedure_to_json(self, indent=indent, target_schema=target_schema)

def values(self: Procedure[Path]) -> Procedure[Array]:
    """A map position's entry positions as ``Array[Path[V]]``.

    Fan out, or index to select one to navigate.
    """
    from capturegraph.procedures.procedure.authoring.dispatch import (
        MAP_VALUES,
    )

    require_map_position(self, "values")
    return cast(
        "Procedure[Array]",
        memoized_child(self, ("values",), lambda: MAP_VALUES(source=self)),
    )