"""Summary types for platform catalog APIs."""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Optional
from olo_protos.platform.v1 import platform_pb2
[docs]
class NavConfigSource(Enum):
UNSPECIFIED = "unspecified"
CUSTOM = "custom"
GLOBAL = "global"
[docs]
class CatalogLocation(Enum):
UNSPECIFIED = "unspecified"
CLOUD = "cloud"
REGISTRY_ONLY = "registry_only"
[docs]
class NavigationMode(Enum):
UNSPECIFIED = "unspecified"
SLAM = "slam"
LOCALIZATION = "localization"
GPS = "gps"
[docs]
class NavigationDesiredState(Enum):
UNSPECIFIED = "unspecified"
RUNNING = "running"
STOPPED = "stopped"
[docs]
class NavigationObservedState(Enum):
UNSPECIFIED = "unspecified"
STOPPED = "stopped"
APPLYING = "applying"
READY = "ready"
FAILED = "failed"
OFFLINE = "offline"
_NAV_SOURCE = {
platform_pb2.NAV_CONFIG_SOURCE_UNSPECIFIED: NavConfigSource.UNSPECIFIED,
platform_pb2.NAV_CONFIG_SOURCE_CUSTOM: NavConfigSource.CUSTOM,
platform_pb2.NAV_CONFIG_SOURCE_GLOBAL: NavConfigSource.GLOBAL,
}
_CATALOG_LOCATION = {
platform_pb2.CATALOG_LOCATION_UNSPECIFIED: CatalogLocation.UNSPECIFIED,
platform_pb2.CATALOG_LOCATION_CLOUD: CatalogLocation.CLOUD,
platform_pb2.CATALOG_LOCATION_REGISTRY_ONLY: CatalogLocation.REGISTRY_ONLY,
}
_NAVIGATION_MODE = {
platform_pb2.NAVIGATION_MODE_UNSPECIFIED: NavigationMode.UNSPECIFIED,
platform_pb2.NAVIGATION_MODE_SLAM: NavigationMode.SLAM,
platform_pb2.NAVIGATION_MODE_LOCALIZATION: NavigationMode.LOCALIZATION,
platform_pb2.NAVIGATION_MODE_GPS: NavigationMode.GPS,
}
_NAVIGATION_MODE_TO_PROTO = {
NavigationMode.UNSPECIFIED: platform_pb2.NAVIGATION_MODE_UNSPECIFIED,
NavigationMode.SLAM: platform_pb2.NAVIGATION_MODE_SLAM,
NavigationMode.LOCALIZATION: platform_pb2.NAVIGATION_MODE_LOCALIZATION,
NavigationMode.GPS: platform_pb2.NAVIGATION_MODE_GPS,
}
_DESIRED_STATE = {
platform_pb2.NAVIGATION_DESIRED_STATE_UNSPECIFIED: NavigationDesiredState.UNSPECIFIED,
platform_pb2.NAVIGATION_DESIRED_STATE_RUNNING: NavigationDesiredState.RUNNING,
platform_pb2.NAVIGATION_DESIRED_STATE_STOPPED: NavigationDesiredState.STOPPED,
}
_OBSERVED_STATE = {
platform_pb2.NAVIGATION_OBSERVED_STATE_UNSPECIFIED: NavigationObservedState.UNSPECIFIED,
platform_pb2.NAVIGATION_OBSERVED_STATE_STOPPED: NavigationObservedState.STOPPED,
platform_pb2.NAVIGATION_OBSERVED_STATE_APPLYING: NavigationObservedState.APPLYING,
platform_pb2.NAVIGATION_OBSERVED_STATE_READY: NavigationObservedState.READY,
platform_pb2.NAVIGATION_OBSERVED_STATE_FAILED: NavigationObservedState.FAILED,
platform_pb2.NAVIGATION_OBSERVED_STATE_OFFLINE: NavigationObservedState.OFFLINE,
}
def _optional_timestamp(msg: object, field: str) -> Optional[datetime]:
if not msg.HasField(field): # type: ignore[attr-defined]
return None
return msg.__getattribute__(field).ToDatetime()
[docs]
@dataclass(frozen=True)
class NavConfigInfo:
id: str
name: str
is_default: bool
source: NavConfigSource
updated_at: Optional[datetime]
supported_modes: tuple[NavigationMode, ...] = ()
[docs]
@dataclass(frozen=True)
class MapInfo:
id: str
name: str
description: str
map_type: str
file_size: int
is_active: bool
created_at: Optional[datetime]
updated_at: Optional[datetime]
[docs]
@dataclass(frozen=True)
class BagInfo:
name: str
robot_id: str
robot_name: str
total_bytes: int
file_count: int
category: str
location: CatalogLocation
updated_at: Optional[datetime]
@dataclass(frozen=True)
class VideoCatalogInfo:
filename: str
robot_id: str
robot_name: str
file_size: int
location: CatalogLocation
updated_at: Optional[datetime]
@dataclass(frozen=True)
class ImageCatalogInfo:
filename: str
robot_id: str
robot_name: str
file_size: int
mime_type: str
location: CatalogLocation
updated_at: Optional[datetime]
def nav_config_from_proto(msg: platform_pb2.NavConfigSummary) -> NavConfigInfo:
supported_modes = tuple(
_NAVIGATION_MODE.get(mode, NavigationMode.UNSPECIFIED)
for mode in msg.supported_modes
if mode in _NAVIGATION_MODE
)
return NavConfigInfo(
id=msg.id,
name=msg.name,
is_default=bool(msg.is_default),
source=_NAV_SOURCE.get(msg.source, NavConfigSource.UNSPECIFIED),
updated_at=_optional_timestamp(msg, "updated_at"),
supported_modes=supported_modes,
)
def map_from_proto(msg: platform_pb2.MapSummary) -> MapInfo:
return MapInfo(
id=msg.id,
name=msg.name,
description=msg.description,
map_type=msg.map_type,
file_size=int(msg.file_size),
is_active=bool(msg.is_active),
created_at=_optional_timestamp(msg, "created_at"),
updated_at=_optional_timestamp(msg, "updated_at"),
)
def bag_from_proto(msg: platform_pb2.BagCatalogSummary) -> BagInfo:
return BagInfo(
name=msg.name,
robot_id=msg.robot_id,
robot_name=msg.robot_name,
total_bytes=int(msg.total_bytes),
file_count=int(msg.file_count),
category=msg.category,
location=_CATALOG_LOCATION.get(msg.location, CatalogLocation.UNSPECIFIED),
updated_at=_optional_timestamp(msg, "updated_at"),
)
def video_catalog_from_proto(msg: platform_pb2.VideoCatalogSummary) -> VideoCatalogInfo:
return VideoCatalogInfo(
filename=msg.filename,
robot_id=msg.robot_id,
robot_name=msg.robot_name,
file_size=int(msg.file_size),
location=_CATALOG_LOCATION.get(msg.location, CatalogLocation.UNSPECIFIED),
updated_at=_optional_timestamp(msg, "updated_at"),
)
def image_catalog_from_proto(msg: platform_pb2.ImageCatalogSummary) -> ImageCatalogInfo:
return ImageCatalogInfo(
filename=msg.filename,
robot_id=msg.robot_id,
robot_name=msg.robot_name,
file_size=int(msg.file_size),
mime_type=msg.mime_type or "image/jpeg",
location=_CATALOG_LOCATION.get(msg.location, CatalogLocation.UNSPECIFIED),
updated_at=_optional_timestamp(msg, "updated_at"),
)
[docs]
@dataclass(frozen=True)
class Pose2D:
"""Planar pose. ``yaw`` maps to proto ``theta``."""
x: float
y: float
yaw: float
[docs]
@dataclass(frozen=True)
class NavigationDesiredSpec:
state: NavigationDesiredState
mode: NavigationMode
config_id: str
map_id: str = ""
initial_pose: Optional[Pose2D] = None
lidar3d: bool = False
has_laser_scan: bool = False
generation: int = 0
[docs]
@dataclass(frozen=True)
class NavigationObservedStatus:
state: NavigationObservedState
mode: NavigationMode = NavigationMode.UNSPECIFIED
config_id: str = ""
map_id: str = ""
generation: int = 0
error_message: str = ""
[docs]
@dataclass(frozen=True)
class NavigationDeploymentInfo:
robot_id: str
namespace: str
desired: Optional[NavigationDesiredSpec]
observed: Optional[NavigationObservedStatus]
def navigation_mode_to_proto(mode: NavigationMode) -> int:
"""Convert SDK NavigationMode enum to proto enum value."""
return _NAVIGATION_MODE_TO_PROTO.get(mode, platform_pb2.NAVIGATION_MODE_UNSPECIFIED)
def _pose_from_proto(msg: platform_pb2.Pose2D) -> Pose2D:
return Pose2D(x=msg.x, y=msg.y, yaw=msg.theta)
def _desired_from_proto(msg: platform_pb2.NavigationDesiredSpec) -> NavigationDesiredSpec:
pose = None
if msg.HasField("initial_pose"):
pose = _pose_from_proto(msg.initial_pose)
return NavigationDesiredSpec(
state=_DESIRED_STATE.get(msg.state, NavigationDesiredState.UNSPECIFIED),
mode=_NAVIGATION_MODE.get(msg.mode, NavigationMode.UNSPECIFIED),
config_id=msg.config_id,
map_id=msg.map_id,
initial_pose=pose,
lidar3d=msg.lidar3d,
has_laser_scan=msg.has_laser_scan,
generation=int(msg.generation),
)
def _observed_from_proto(msg: platform_pb2.NavigationObservedStatus) -> NavigationObservedStatus:
return NavigationObservedStatus(
state=_OBSERVED_STATE.get(msg.state, NavigationObservedState.UNSPECIFIED),
mode=_NAVIGATION_MODE.get(msg.mode, NavigationMode.UNSPECIFIED),
config_id=msg.config_id,
map_id=msg.map_id,
generation=int(msg.generation),
error_message=msg.error_message,
)
def deployment_from_proto(msg: platform_pb2.NavigationDeployment) -> NavigationDeploymentInfo:
"""Convert a proto NavigationDeployment to the SDK dataclass."""
desired = None
if msg.HasField("desired"):
desired = _desired_from_proto(msg.desired)
observed = None
if msg.HasField("observed"):
observed = _observed_from_proto(msg.observed)
return NavigationDeploymentInfo(
robot_id=msg.robot_id,
namespace=msg.namespace or "default",
desired=desired,
observed=observed,
)