Source code for

from typing import List, Mapping, Optional

from grpclib.client import Channel

from viam.proto.common import DoCommandRequest, DoCommandResponse
from viam.proto.service.slam import (
from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
from viam.utils import ValueTypes, dict_to_struct, struct_to_dict

from . import Pose
from .slam import SLAM

[docs]class SLAMClient(SLAM, ReconfigurableResourceRPCClientBase): """ Connect to the SLAMService, which allows the robot to create a map of its surroundings and find its location in that map. """ client: SLAMServiceStub def __init__(self, name: str, channel: Channel): = channel self.client = SLAMServiceStub(channel) super().__init__(name)
[docs] async def get_position(self, *, timeout: Optional[float] = None) -> Pose: request = GetPositionRequest( response: GetPositionResponse = await self.client.GetPosition(request, timeout=timeout) return response.pose
[docs] async def get_point_cloud_map(self, return_edited_map: bool = False, *, timeout: Optional[float] = None) -> List[bytes]: request = GetPointCloudMapRequest(, return_edited_map=return_edited_map) response: List[GetPointCloudMapResponse] = await self.client.GetPointCloudMap(request, timeout=timeout) return [r.point_cloud_pcd_chunk for r in response]
[docs] async def get_internal_state(self, *, timeout: Optional[float] = None) -> List[bytes]: request = GetInternalStateRequest( response: List[GetInternalStateResponse] = await self.client.GetInternalState(request, timeout=timeout) return [r.internal_state_chunk for r in response]
[docs] async def get_properties(self, *, timeout: Optional[float] = None) -> SLAM.Properties: request = GetPropertiesRequest( return await self.client.GetProperties(request, timeout=timeout)
[docs] async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None, **__) -> Mapping[str, ValueTypes]: request = DoCommandRequest(, command=dict_to_struct(command)) response: DoCommandResponse = await self.client.DoCommand(request, timeout=timeout) return struct_to_dict(response.result)