"""Namespace-scoped TF frame handle."""
from datetime import datetime
from typing import TYPE_CHECKING
from olo.spatial._convert import transform_from_proto
from olo.spatial._timeouts import _DEFAULT_TIMEOUT_SEC
from olo.spatial._tree import build_frame_tree
from olo.spatial.tree_subscription import TreeSubscription
from olo.spatial.types import FrameTree, Transform
if TYPE_CHECKING:
from olo.spatial import Spatial, TransformSubscription
[docs]
class RobotFrames:
"""Namespace-scoped TF frame handle for a robot."""
[docs]
def __init__(self, spatial: "Spatial", namespace: str) -> None:
"""Bind to a spatial client and normalized robot namespace."""
self._spatial = spatial
self.namespace = namespace
[docs]
def tree(
self,
*,
include_ancestors: bool = False,
timeout: float = _DEFAULT_TIMEOUT_SEC,
) -> FrameTree:
"""Return the robot's TF subtree as a multi-root tree."""
response = self._spatial._get_tree(
self.namespace,
include_ancestors=include_ancestors,
timeout=timeout,
)
return build_frame_tree(
(transform_from_proto(edge) for edge in response.edges),
response.root_frames,
self.namespace,
)
[docs]
def base_frame(self, *, timeout: float = _DEFAULT_TIMEOUT_SEC) -> str | None:
"""Return the detected base frame for this robot, or None when unknown."""
response = self._spatial._get_tree(self.namespace, timeout=timeout)
return response.base_frame or None
[docs]
def lookup(
self,
target_frame: str,
source_frame: str,
*,
at: datetime | None = None,
timeout: float = _DEFAULT_TIMEOUT_SEC,
) -> Transform:
"""Look up a transform between two frames, resolving bare names for this robot."""
return self._spatial.lookup(
target_frame,
source_frame,
at=at,
timeout=timeout,
robot_namespace=self.namespace,
)
[docs]
def subscribe(
self,
target_frame: str,
source_frame: str,
*,
timeout: float | None = None,
) -> "TransformSubscription":
"""Subscribe to transform updates, resolving bare names for this robot."""
return self._spatial.subscribe(
target_frame,
source_frame,
timeout=timeout,
robot_namespace=self.namespace,
)
[docs]
def subscribe_tree(
self,
*,
include_ancestors: bool = False,
timeout: float | None = None,
) -> TreeSubscription:
"""Subscribe to TF tree updates for this robot's subtree."""
return self._spatial._subscribe_tree(
self.namespace,
include_ancestors=include_ancestors,
timeout=timeout,
)
[docs]
def publish_static(self, transform: Transform, *, timeout: float = _DEFAULT_TIMEOUT_SEC) -> None:
"""Publish a static transform, resolving bare frame ids for this robot."""
self._spatial.publish_static(
transform,
timeout=timeout,
robot_namespace=self.namespace,
)
__all__ = ["RobotFrames"]