viam.services.navigation.client

Classes

NavigationClient

Connect to the NavigationService, which allows the robot to navigate to specified locations.

Module Contents

class viam.services.navigation.client.NavigationClient(name: str, channel: grpclib.client.Channel)[source]

Bases: viam.services.navigation.navigation.Navigation, viam.resource.rpc_client_base.ReconfigurableResourceRPCClientBase

Connect to the NavigationService, which allows the robot to navigate to specified locations.

client: viam.proto.service.navigation.NavigationServiceStub
channel
async add_location(location_name: str, position: viam.proto.service.navigation.Position | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Location[source]

Save a named Location on the active map.

Parameters:

  • location_name (str) – The name of the location to save.

  • position (Optional[navigation.Position]) – The position to save. Omit to use the robot’s current position.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The saved location.

Return type:

navigation.Location

async get_location(location_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Location[source]

Get a previously saved Location by name.

Parameters:

  • location_name (str) – The name of the location to retrieve.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The requested location.

Return type:

navigation.Location

async list_locations(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → List[viam.proto.service.navigation.Location][source]

List all previously saved Locations.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

All saved locations.

Return type:

List[navigation.Location]

async update_location(location_name: str, position: viam.proto.service.navigation.Position | None = None, new_name: str | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Location[source]

Update a previously saved Location.

Parameters:

  • location_name (str) – The name of the location to update.

  • position (Optional[navigation.Position]) – The new position. Omit to leave the position unchanged.

  • new_name (Optional[str]) – The new name for the location. Omit to leave the name unchanged.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The updated location.

Return type:

navigation.Location

async delete_location(location_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → None[source]

Delete a previously saved Location.

Parameters:

  • location_name (str) – The name of the location to delete.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

async add_route(route_name: str, waypoints: List[viam.proto.service.navigation.Waypoint] | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Route[source]

Save a named Route, an ordered list of waypoints.

Parameters:

  • route_name (str) – The name of the route to save.

  • waypoints (Optional[List[navigation.Waypoint]]) – The ordered waypoints that make up the route.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The saved route.

Return type:

navigation.Route

async get_route(route_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Route[source]

Get a previously saved Route by name.

Parameters:

  • route_name (str) – The name of the route to retrieve.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The requested route.

Return type:

navigation.Route

async list_routes(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → List[viam.proto.service.navigation.Route][source]

List all previously saved Routes.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

All saved routes.

Return type:

List[navigation.Route]

async update_route(route_name: str, new_name: str | None = None, waypoints: List[viam.proto.service.navigation.Waypoint] | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Route[source]

Update a previously saved Route.

Parameters:

  • route_name (str) – The name of the route to update.

  • new_name (Optional[str]) – The new name for the route. Omit to leave the name unchanged.

  • waypoints (Optional[List[navigation.Waypoint]]) – The new ordered waypoints. Omit to leave the waypoints unchanged; pass an empty list to clear them.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The updated route.

Return type:

navigation.Route

async delete_route(route_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → None[source]

Delete a previously saved Route.

Parameters:

  • route_name (str) – The name of the route to delete.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

async add_zone(zone_name: str, type: str, geometry: Dict[str, Any], max_speed: float | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Zone[source]

Save a named Zone, a region on the active map.

Parameters:

  • zone_name (str) – The name of the zone to save.

  • type (str) – The type of zone, either “keepout” or “speed_limit”.

  • geometry (Dict[str, Any]) – The shape of the zone (circle, box, or polygon).

  • max_speed (Optional[float]) – Max speed in meters/sec; required when type is “speed_limit”.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The saved zone.

Return type:

navigation.Zone

async get_zone(zone_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Zone[source]

Get a previously saved Zone by name.

Parameters:

  • zone_name (str) – The name of the zone to retrieve.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The requested zone.

Return type:

navigation.Zone

async list_zones(type: str | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → List[viam.proto.service.navigation.Zone][source]

List all previously saved Zones.

Parameters:

  • type (Optional[str]) – Filter by zone type, either “keepout” or “speed_limit”.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

All saved zones matching the filter, if provided.

Return type:

List[navigation.Zone]

async update_zone(zone_name: str, type: str | None = None, geometry: Dict[str, Any] | None = None, max_speed: float | None = None, new_name: str | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Zone[source]

Update a previously saved Zone.

Parameters:

  • zone_name (str) – The name of the zone to update.

  • type (Optional[str]) – The new zone type. Omit to leave unchanged.

  • geometry (Optional[Dict[str, Any]]) – The new shape of the zone. Omit to leave unchanged.

  • max_speed (Optional[float]) – The new max speed in meters/sec. Omit to leave unchanged.

  • new_name (Optional[str]) – The new name for the zone. Omit to leave the name unchanged.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The updated zone.

Return type:

navigation.Zone

async delete_zone(zone_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → None[source]

Delete a previously saved Zone.

Parameters:

  • zone_name (str) – The name of the zone to delete.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

async navigate_to_location(location_name: str, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Start navigating to a previously saved Location.

Parameters:

  • location_name (str) – The name of the location to navigate to.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async navigate_to_point(destination: viam.proto.service.navigation.Position, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Start navigating to a destination Position.

Parameters:

  • destination (navigation.Position) – The destination to navigate to.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async navigate_route(route_name: str | None = None, waypoints: List[viam.proto.service.navigation.Waypoint] | None = None, start_index: int = 0, start_nearest: bool = False, loop: bool = False, wait: bool = False, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Start navigating a Route, either a previously saved one or an inline list of waypoints.

Parameters:

  • route_name (Optional[str]) – The name of a previously saved route.

  • waypoints (Optional[List[navigation.Waypoint]]) – An inline list of waypoints.

  • start_index (int) – The index of the waypoint to start at.

  • start_nearest (bool) – Start at the waypoint nearest to the robot’s current position, overriding start_index.

  • loop (bool) – Loop the route indefinitely.

  • wait (bool) – With loop, only return once the robot has stopped.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async plan_to_point(destination: viam.proto.service.navigation.Position, start: viam.proto.service.navigation.Position | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Plan[source]

Preview a path to a destination Position without moving the robot.

Parameters:

  • destination (navigation.Position) – The destination to plan to.

  • start (Optional[navigation.Position]) – The position to plan from. Defaults to the robot’s current position.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The planned path.

Return type:

navigation.Plan

async plan_to_location(location_name: str, start: viam.proto.service.navigation.Position | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Plan[source]

Preview a path to a previously saved Location without moving the robot.

Parameters:

  • location_name (str) – The name of the location to plan to.

  • start (Optional[navigation.Position]) – The position to plan from. Defaults to the robot’s current position.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The planned path.

Return type:

navigation.Plan

async execute_plan(plan: viam.proto.service.navigation.Plan, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Start navigating a previously previewed Plan.

Parameters:

  • plan (navigation.Plan) – The plan to execute, as returned by plan_to_point or plan_to_location.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async go_to_location(location_name: str, wait: bool | None = None, linear_m_per_sec: float | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Drive directly to a previously saved Location, without planning around zones or obstacles.

Parameters:

  • location_name (str) – The name of the location to drive to.

  • wait (Optional[bool]) – Whether to wait until the robot has arrived. Defaults to true.

  • linear_m_per_sec (Optional[float]) – The linear speed to drive at.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async go_to_point(destination: viam.proto.service.navigation.Position, wait: bool | None = None, linear_m_per_sec: float | None = None, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Drive directly to a destination Position, without planning around zones or obstacles.

Parameters:

  • destination (navigation.Position) – The destination to drive to.

  • wait (Optional[bool]) – Whether to wait until the robot has arrived. Defaults to true.

  • linear_m_per_sec (Optional[float]) – The linear speed to drive at.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async suspend(reason: str = '', *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Suspend the current goal, holding position until resume is called.

Parameters:

  • reason (str) – The reason for suspending, reported on the suspended goal.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async resume(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Resume a previously suspended goal.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async cancel(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Cancel the current goal.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The resulting navigation status.

Return type:

navigation.Status

async get_position(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Position[source]

Get the current position of the robot in the navigation service.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The current position of the robot.

Return type:

navigation.Position

async get_navigation_status(*, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Status[source]

Get the current navigation state of the robot.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The current navigation status.

Return type:

navigation.Status

async get_properties(*, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.Properties[source]

Get information about the navigation service that does not change while the service is running.

Parameters:

timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

Information about the navigation service.

Return type:

navigation.Properties

async get_costmap(layer: str = '', stride: int = 0, *, extra: Dict[str, Any] | None = None, timeout: float | None = None, **kwargs) → viam.proto.service.navigation.GetCostmapResponse[source]

Get a costmap describing the obstacles known to the navigation service.

Parameters:

  • layer (str) – Which costmap layer to retrieve, one of “auto”, “local”, or “global”. Defaults to “auto”.

  • stride (int) – Downsample factor. Defaults to 2.

  • timeout (Optional[float]) – An option to set how long to wait (in seconds) before calling a time-out and closing the underlying RPC call.

Returns:

The requested costmap.

Return type:

navigation.GetCostmapResponse

async do_command(command: Mapping[str, viam.utils.ValueTypes], *, timeout: float | None = None, **kwargs) → Mapping[str, viam.utils.ValueTypes][source]

Send/receive arbitrary commands.

service = SERVICE.from_robot(robot=machine, "builtin")  # replace SERVICE with the appropriate class

my_command = {
  "cmnd": "dosomething",
  "someparameter": 52
}

# Can be used with any resource, using the motion service as an example
await service.do_command(command=my_command)
Parameters:

command (Dict[str, ValueTypes]) – The command to execute

Returns:

Result of the executed command

Return type:

Dict[str, ValueTypes]

async get_status(*, timeout: float | None = None, **kwargs) → Mapping[str, viam.utils.ValueTypes][source]

Get the current status of the service as a dictionary.

status = await service.get_status()
Returns:

The status of the service

Return type:

Mapping[str, ValueTypes]

classmethod from_robot(robot: viam.robot.client.RobotClient, name: str) → Self

Get the service named name from the provided robot.

async def connect() -> RobotClient:
    # Replace "<API-KEY>" (including brackets) with your API key and "<API-KEY-ID>" with your API key ID
    options = RobotClient.Options.with_api_key("<API-KEY>", "<API-KEY-ID>")
    # Replace "<MACHINE-URL>" (included brackets) with your machine's connection URL or FQDN
    return await RobotClient.at_address("<MACHINE-URL>", options)

async def main():
    robot = await connect()

    # Can be used with any resource, using the motion service as an example
    motion = MotionClient.from_robot(robot=machine, name="builtin")

    robot.close()
Parameters:

  • robot (RobotClient) – The robot

  • name (str) – The name of the service

Returns:

The service, if it exists on the robot

Return type:

Self

classmethod get_resource_name(name: str) → viam.proto.common.ResourceName

Get the ResourceName for this Resource with the given name

# Can be used with any resource, using an arm as an example
my_arm_name = Arm.get_resource_name("my_arm")
Parameters:

name (str) – The name of the Resource

Returns:

The ResourceName of this Resource

Return type:

ResourceName

get_operation(kwargs: Mapping[str, Any]) → viam.operations.Operation

Get the Operation associated with the currently running function.

When writing custom resources, you should get the Operation by calling this function and check to see if it’s cancelled. If the Operation is cancelled, then you can perform any necessary (terminating long running tasks, cleaning up connections, etc. ).

Parameters:

kwargs (Mapping[str, Any]) – The kwargs object containing the operation

Returns:

The operation associated with this function

Return type:

viam.operations.Operation

async close()

Safely shut down the resource and prevent further use.

Close must be idempotent. Later configuration may allow a resource to be “open” again. If a resource does not want or need a close function, it is assumed that the resource does not need to return errors when future non-Close methods are called.

await component.close()