chatomon.orbitals.base ¶

@classmethod
def default_name(cls) -> str:
    """Return the name an instance gets when none is given.

    It is the class name in snake case without the `Orbital` suffix:
    `ScopeGuardOrbital` becomes `scope_guard`. A subclass may override it.
    [FunctionOrbital][chatomon.orbitals.base.FunctionOrbital] does not use
    it: each instance is named after its function.

    Returns:
        The default name.
    """
    base = cls.__name__
    if base.endswith("Orbital") and base != "Orbital":
        base = base[: -len("Orbital")]
    return re.sub(r"(?<!^)(?=[A-Z])", "_", base).lower()

@abstractmethod
async def check(self, ctx: OrbitalContext) -> Verdict:
    """Check one message and return a verdict.

    Raising an exception is allowed: the engine applies `on_error`.

    Args:
        ctx: The text under check, the history and earlier annotations.

    Returns:
        The verdict.
    """

def bind(self, connector: Connector[Any]) -> None:
    """Receive the atom's connector; called once by `Atom.build()`.

    The default does nothing. Orbitals that call a model, such as the judge,
    use the connector to reach it.

    Args:
        connector: The connector of the atom being built.
    """
    del connector

def config(self) -> dict[str, Any]:
    """Return the public configuration that identifies this orbital.

    Subclasses extend the dictionary with their own settings. It feeds
    [fingerprint][chatomon.orbitals.base.Orbital.fingerprint].

    Returns:
        A JSON-serializable dictionary.
    """
    return {
        "kind": type(self).__qualname__,
        "name": self.name,
        "reads": list(self.reads),
        "timeout": self.timeout,
        "on_error": self.on_error,
    }

def fingerprint(self) -> str:
    """Return a stable hash of the configuration, recorded in audit reports.

    Returns:
        `"sha256:"` followed by the hex digest of `config()` as canonical JSON.
    """
    return sha256_text(stable_json(self.config()))

async def check(self, ctx: OrbitalContext) -> Verdict:
    """Call the wrapped function.

    Args:
        ctx: The context passed to the function.

    Returns:
        The function's verdict.

    Raises:
        TypeError: If the function does not return a Verdict.
    """
    if inspect.iscoroutinefunction(self.func):
        result: object = self.func(ctx)
    else:
        result = await asyncio.to_thread(self.func, ctx)
    if inspect.isawaitable(result):
        result = await result
    if not isinstance(result, Verdict):
        raise TypeError(
            f"orbital {self.name!r} returned {type(result).__name__}, expected Verdict"
        )
    return result

def config(self) -> dict[str, Any]:
    """Return the configuration, including the wrapped function's path.

    Returns:
        A JSON-serializable dictionary.
    """
    module = getattr(self.func, "__module__", "?")
    qualname = getattr(self.func, "__qualname__", repr(self.func))
    return {**super().config(), "function": f"{module}.{qualname}"}