Source code for olo

"""OLO Python SDK — appliance gRPC control plane + platform Connect catalogs."""

from typing import TYPE_CHECKING, Optional

from olo._channel import DEFAULT_TARGET, ENV_TARGET, ChannelSession
from olo.robot import Robot

if TYPE_CHECKING:
    from olo.ai import Ai
    from olo.archive import Archive
    from olo.core import Core
    from olo.kinematics import Kinematics
    from olo.locomotion import Locomotion
    from olo.navigation import Navigation
    from olo.platform import Platform
    from olo.spatial import Spatial
from olo.errors import (
    OloAmbiguousNamespace,
    OloAmbiguousPlanningGroup,
    OloError,
    OloFailedPrecondition,
    OloInternalError,
    OloInvalidArgument,
    OloNotFound,
    OloPermissionDenied,
    OloPlatformNotConfigured,
    OloTimeoutError,
    OloUnauthenticated,
    OloUnavailable,
)




class Client:
    """Entry point for the sync OLO Python SDK."""



    def __init__(
        self,
        target: Optional[str] = None,
        *,
        platform_url: Optional[str] = None,
        platform_token: Optional[str] = None,
    ):
        """Connect to the appliance gRPC target; optionally configure platform."""
        self._session = ChannelSession(target)
        self._platform_url = platform_url
        self._platform_token = platform_token
        self._ai: Optional["Ai"] = None
        self._locomotion: Optional["Locomotion"] = None
        self._core: Optional["Core"] = None
        self._kinematics: Optional["Kinematics"] = None
        self._navigation: Optional["Navigation"] = None
        self._spatial: Optional["Spatial"] = None
        self._platform: Optional["Platform"] = None
        self._archive: Optional["Archive"] = None


    @property
    def ai(self) -> "Ai":
        """Return the AI API wrapper."""
        if self._ai is None:
            from olo.ai import Ai

            self._ai = Ai(self._session)
        return self._ai

    @property
    def locomotion(self) -> "Locomotion":
        """Return the local locomotion API wrapper."""
        if self._locomotion is None:
            from olo.locomotion import Locomotion

            self._locomotion = Locomotion(self._session)
        return self._locomotion

    @property
    def core(self) -> "Core":
        """Return the Core API wrapper."""
        if self._core is None:
            from olo.core import Core

            self._core = Core(self._session)
        return self._core

    @property
    def kinematics(self) -> "Kinematics":
        """Return the Kinematics API wrapper."""
        if self._kinematics is None:
            from olo.kinematics import Kinematics

            self._kinematics = Kinematics(self._session)
        return self._kinematics

    @property
    def navigation(self) -> "Navigation":
        """Return the Navigation API wrapper."""
        if self._navigation is None:
            from olo.navigation import Navigation

            self._navigation = Navigation(self._session)
        return self._navigation

    @property
    def spatial(self) -> "Spatial":
        """Return the Spatial API wrapper."""
        if self._spatial is None:
            from olo.spatial import Spatial

            self._spatial = Spatial(self._session)
        return self._spatial

    @property
    def archive(self) -> "Archive":
        """Return the local archive API wrapper."""
        if self._archive is None:
            from olo.archive import Archive

            self._archive = Archive(self._session)
        return self._archive

    @property
    def platform(self) -> "Platform":
        """Return the platform catalog API (server-direct, PAT-authenticated)."""
        if self._platform is None:
            from olo.platform import Platform

            self._platform = Platform(
                self._platform_url,
                self._platform_token,
                namespace_resolver=self._session.resolve_namespace,
            )
        return self._platform



    def robot(self, namespace: Optional[str] = None) -> "Robot":
        """Return a namespace-scoped robot handle."""
        resolved = self._session.resolve_namespace(namespace)
        return Robot(self, resolved)




    def close(self) -> None:
        """Close the underlying gRPC channel."""
        self._session.close()


    def __enter__(self) -> "Client":
        """Enter a context manager."""
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        """Close the client on context exit."""
        self.close()



__all__ = [
    "Client",
    "DEFAULT_TARGET",
    "ENV_TARGET",
    "OloAmbiguousNamespace",
    "OloAmbiguousPlanningGroup",
    "OloError",
    "OloFailedPrecondition",
    "OloInternalError",
    "OloInvalidArgument",
    "Robot",
    "OloNotFound",
    "OloPermissionDenied",
    "OloPlatformNotConfigured",
    "OloTimeoutError",
    "OloUnauthenticated",
    "OloUnavailable",
]