Source code for viam.app.provisioning_client

from typing import List, Mapping, Optional

from grpclib.client import Channel

from viam import logging
from viam.proto.provisioning import (
    CloudConfig,
    GetNetworkListRequest,
    GetNetworkListResponse,
    GetSmartMachineStatusRequest,
    GetSmartMachineStatusResponse,
    NetworkInfo,
    ProvisioningServiceStub,
    SetNetworkCredentialsRequest,
    SetSmartMachineCredentialsRequest,
)

LOGGER = logging.getLogger(__name__)


[docs]class ProvisioningClient: """gRPC client for getting and setting smart machine info. Constructor is used by `ViamClient` to instantiate relevant service stubs. Calls to `ProvisioningClient` methods should be made through `ViamClient`. Establish a connection:: import asyncio from viam.rpc.dial import DialOptions, Credentials from viam.app.viam_client import ViamClient async def connect() -> ViamClient: # Replace "<API-KEY>" (including brackets) with your API key and "<API-KEY-ID>" with your API key ID dial_options = DialOptions.with_api_key("<API-KEY>", "<API-KEY-ID>") return await ViamClient.create_from_dial_options(dial_options) async def main(): # Make a ViamClient viam_client = await connect() # Instantiate a ProvisioningClient to run provisioning client API methods on provisioning_client = viam_client.provisioning_client viam_client.close() if __name__ == '__main__': asyncio.run(main()) """ def __init__(self, channel: Channel, metadata: Mapping[str, str]): """Create a `ProvisioningClient` that maintains a connection to app. Args: channel (grpclib.client.Channel): Connection to app. metadata (Mapping[str, str]): Required authorization token to send requests to app. """ self._metadata = metadata self._provisioning_client = ProvisioningServiceStub(channel) self._channel = channel _provisioning_client: ProvisioningServiceStub _metadata: Mapping[str, str] _channel: Channel
[docs] async def get_network_list(self) -> List[NetworkInfo]: """Returns list of networks that are visible to the Smart Machine.""" request = GetNetworkListRequest() resp: GetNetworkListResponse = await self._provisioning_client.GetNetworkList(request, metadata=self._metadata) return list(resp.networks)
[docs] async def get_smart_machine_status(self) -> GetSmartMachineStatusResponse: """Returns the status of the smart machine.""" request = GetSmartMachineStatusRequest() return await self._provisioning_client.GetSmartMachineStatus(request, metadata=self._metadata)
[docs] async def set_network_credentials(self, network_type: str, ssid: str, psk: str) -> None: """Sets the network credentials of the Smart Machine. Args: network_type (str): The type of the network. ssid (str): The SSID of the network. psk (str): The network's passkey. """ request = SetNetworkCredentialsRequest(type=network_type, ssid=ssid, psk=psk) await self._provisioning_client.SetNetworkCredentials(request, metadata=self._metadata)
[docs] async def set_smart_machine_credentials(self, cloud_config: Optional[CloudConfig] = None) -> None: request = SetSmartMachineCredentialsRequest(cloud=cloud_config) await self._provisioning_client.SetSmartMachineCredentials(request, metadata=self._metadata)