Upload and Retrieve Data with Viam's Data Client API

The data client API allows you to upload and retrieve data to and from the Viam app.

Establish a connection

To use the Viam data client API, you first need to instantiate a ViamClient and then instantiate an DataClient. See the following example for reference.

Use the Viam CLI to generate an API key to authenticate.

import asyncio

from viam.rpc.dial import DialOptions, Credentials
from viam.app.viam_client import ViamClient


async def connect() -> ViamClient:
    dial_options = DialOptions(
      credentials=Credentials(
        type="api-key",
        # Replace "<API-KEY>" (including brackets) with your machine's API key
        payload='<API-KEY>',
      ),
      # Replace "<API-KEY-ID>" (including brackets) with your machine's
      # API key ID
      auth_entity='<API-KEY-ID>'
    )
    return await ViamClient.create_from_dial_options(dial_options)


async def main():
    # Make a ViamClient
    viam_client = await connect()
    # Instantiate a DataClient to run data client API methods on
    data_client = viam_client.data_client

    viam_client.close()

if __name__ == '__main__':
    asyncio.run(main())

Once you have instantiated a DataClient, you can run the following API methods against the DataClient object (named data_client in the examples).

Find part ID

To find the ID of your machine part, navigate to its Setup tab in the Viam app. Keep architecture selection at default. In Step 1, grab the part id from the second string of the generated command as the token following id=. For example:

Part ID displayed in the Viam app.

API

The data client API supports the following methods (among others):

Method NameDescription
TabularDataByFilterFilter and download tabular data.
BinaryDataByFilterFilter and download binary data.
BinaryDataByIDsDownload binary data by IDs.
DeleteTabularDataDelete tabular data older than a specified number of days.
DeleteBinaryDataByFilterFilter and delete binary data.
DeleteBinaryDataByIdsFilter and delete binary data by ids.
AddTagsToBinaryDataByIdsAdd tags to binary data by ids.
AddTagsToBinaryDataByFilterAdd tags to binary data by filter.
RemoveTagsFromBinaryDataByIdsRemove tags from binary data by ids.
RemoveTagsFromBinaryDataByFilterRemove tags from binary data by filter.
TagsByFilterGet tags from data by filter.
BoundingBoxLabelsByFilterGet a list of bounding box labels using a Filter.
GetDatabaseConnectionGet a connection to access a MongoDB Atlas Data federation instance.
BinaryDataCaptureUploadUpload binary data collected on your machine through a specific component and the relevant metadata to the Viam app.
TabularDataCaptureUploadUpload tabular data collected on your machine through a specific component and the relevant metadata to the Viam app.
StreamingDataCaptureUploadUpload the contents of streaming binary data and the relevant metadata to the Viam app.
FileUploadUpload file data stored on your machine and the relevant metadata to the Viam app.
FileUploadFromPathUpload file data stored on your machine from the specified filepath and the relevant metadata to the Viam app.
AddBoundingBoxToImageByIdAdd a bounding box to an image specified by its BinaryID.
RemoveBoundingBoxFromImageByIdRemoves a bounding box from an image specified by its BinaryID.

TabularDataByFilter

Retrieve optionally filtered tabular data from the Viam app. You can also find your tabular data under the Sensors subtab of the app’s Data tab.

Parameters:

Returns:

from viam.proto.app.data import Filter

my_filter = Filter(component_name="left_motor")
tabular_data = await data_client.tabular_data_by_filter(my_filter)

For more information, see the Python SDK Docs.

BinaryDataByFilter

Retrieve optionally filtered binary data from the Viam app. You can also find your binary data under the Images, Point clouds, or Files subtab of the app’s Data tab, depending on the type of data that you have uploaded.

Parameters:

  • filter (Optional[viam.proto.app.data.Filter]): Optional Filter specifying binary data to retrieve. Specify no filter to download all binary data.
  • dest (Optional[str]): Filepath to write retrieved data to. If not populated, writes to your current directory.
  • include_file_data (bool): Boolean specifying whether to include the binary file data with each retrieved file. Defaults to true, where both the files’ data and metadata are returned.
  • num_files (Optional[str]): Number of binary data to return. Passing 0 returns all binary data matching the filter. Defaults to 100 if no binary data is requested, otherwise 10.

Returns:

from viam.proto.app.data import Filter

my_filter = Filter(component_type="camera")
binary_data = await data_client.binary_data_by_filter(my_filter)

For more information, see the Python SDK Docs.

BinaryDataByIDs

Retrieve binary data from the Viam app by BinaryID. You can also find your binary data under the Images, Point clouds, or Files subtab of the app’s Data tab, depending on the type of data that you have uploaded.

Parameters:

Returns:

from viam.proto.app.data import BinaryID

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
    )

my_ids = []

for obj in binary_metadata:
    my_ids.append(
        BinaryID(
            file_id=obj.metadata.id,
            organization_id=obj.metadata.capture_metadata.organization_id,
            location_id=obj.metadata.capture_metadata.location_id
            )
        )

binary_data = await data_client.binary_data_by_ids(my_ids)

For more information, see the Python SDK Docs.

DeleteTabularData

Delete tabular data older than a specified number of days.

Parameters:

  • organization_id (str): ID of organization to delete data from. You can obtain your organization id from the organization settings page.
  • delete_older_than_days (int): Delete data that was captured up to this many days ago. For example if delete_older_than_days is 10, this deletes any data that was captured up to 10 days ago. If it is 0, all existing data is deleted.

Returns:

  • None.
from viam.proto.app.data import Filter

my_filter = Filter(component_name="left_motor")
days_of_data_to_delete = 10
tabular_data = await data_client.delete_tabular_data(
    "a12b3c4e-1234-1abc-ab1c-ab1c2d345abc", days_of_data_to_delete)

For more information, see the Python SDK Docs.

DeleteBinaryDataByFilter

Filter and delete binary data.

Parameters:

  • filter (viam.proto.app.data.Filter): Optional Filter specifying binary data to delete. Passing an empty Filter will lead to all data being deleted. Exercise caution when using this option.

Returns:

  • None.
from viam.proto.app.data import Filter

my_filter = Filter(component_name="left_motor")
res = await data_client.delete_binary_data_by_filter(my_filter)

For more information, see the Python SDK Docs.

DeleteBinaryDataByIds

Filter and delete binary data by ids.

Parameters:

Returns:

  • (int): The number of items deleted.

Raises:

  • GRPCError – This error is raised if no BinaryID objects are provided.
from viam.proto.app.data import BinaryID

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
    )

my_ids = []

for obj in binary_metadata:
    my_ids.append(
        BinaryID(
            file_id=obj.metadata.id,
            organization_id=obj.metadata.capture_metadata.organization_id,
            location_id=obj.metadata.capture_metadata.location_id
            )
        )

binary_data = await data_client.delete_binary_data_by_ids(my_ids)

For more information, see the Python SDK Docs.

AddTagsToBinaryDataByIds

Add tags to binary data by ids.

Parameters:

  • tags (List[str]): List of tags to add to specified binary data. Must be non-empty.
  • binary_ids (List[viam.app.proto.BinaryID]): List of BinaryID objects specifying binary data to tag. Must be non-empty.

Returns:

  • None.

Raises:

  • GRPCError – This error is raised if no BinaryID objects or tags are provided.
from viam.proto.app.data import BinaryID

tags = ["tag1", "tag2"]

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
    )

my_ids = []

for obj in binary_metadata:
    my_ids.append(
        BinaryID(
            file_id=obj.metadata.id,
            organization_id=obj.metadata.capture_metadata.organization_id,
            location_id=obj.metadata.capture_metadata.location_id
            )
        )

binary_data = await data_client.add_tags_to_binary_data_by_ids(tags, my_ids)

For more information, see the Python SDK Docs.

AddTagsToBinaryDataByFilter

Add tags to binary data by filter.

Parameters:

  • tags (List[str]): List of tags to add to specified binary data. Must be non-empty.
  • filter (viam.proto.app.data.Filter): Filter specifying binary data to tag. If no Filter is provided, all data will be tagged.

Returns:

  • None.

Raises:

  • GRPCError – This error is raised if no Btags are provided.
from viam.proto.app.data import Filter

my_filter = Filter(component_name="my_camera")
tags = ["tag1", "tag2"]
res = await data_client.add_tags_to_binary_data_by_filter(tags, my_filter)

For more information, see the Python SDK Docs.

RemoveTagsFromBinaryDataByIds

Remove tags from binary by ids.

Parameters:

  • tags (List[str]): List of tags to remove from specified binary data. Must be non-empty.
  • file_ids (List[str]): List of BinaryID objects specifying binary data to untag. Must be non-empty.

Returns:

  • (int): The number of tags removed.

Raises:

  • GRPCError – This error is raised if no BinaryID objects or tags are provided.
from viam.proto.app.data import BinaryID

tags = ["tag1", "tag2"]

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
    )

my_ids = []

for obj in binary_metadata:
    my_ids.append(
        BinaryID(
            file_id=obj.metadata.id,
            organization_id=obj.metadata.capture_metadata.organization_id,
            location_id=obj.metadata.capture_metadata.location_id
            )
        )

binary_data = await data_client.remove_tags_from_binary_data_by_ids(
    tags, my_ids)

For more information, see the Python SDK Docs.

RemoveTagsFromBinaryDataByFilter

Remove tags from binary data by filter.

Parameters:

  • tags (List[str]): List of tags to remove from specified binary data.
  • filter (viam.proto.app.data.Filter): Filter specifying binary data to untag. If no Filter is provided, all data will be untagged.

Returns:

  • (int): The number of tags removed.

Raises:

  • GRPCError – This error is raised if no tags are provided.
from viam.proto.app.data import Filter

my_filter = Filter(component_name="my_camera")
tags = ["tag1", "tag2"]
res = await data_client.remove_tags_from_binary_data_by_filter(tags, my_filter)

For more information, see the Python SDK Docs.

TagsByFilter

Get a list of tags using a filter.

Parameters:

  • filter (viam.proto.app.data.Filter): Filter specifying data to retrieve from. If no Filter is provided, all data tags are returned.

Returns:

from viam.proto.app.data import Filter

my_filter = Filter(component_name="my_camera")
tags = await data_client.tags_by_filter(my_filter)

For more information, see the Python SDK Docs.

BoundingBoxLabelsByFilter

Get a list of bounding box labels using a Filter.

Parameters:

Returns:

  • (List[str]): The list of bounding box labels.
from viam.proto.app.data import Filter

my_filter = Filter(component_name="my_camera")
bounding_box_labels = await data_client.bounding_box_labels_by_filter(
    my_filter)

For more information, see the Python SDK Docs.

GetDatabaseConnection

Get a connection to access a MongoDB Atlas Data federation instance.

Parameters:

Returns:

  • (str): The hostname of the federated database.
data_client.get_database_connection("a12b3c4e-1234-1abc-ab1c-ab1c2d345abc")

For more information, see the Python SDK Docs.

BinaryDataCaptureUpload

Upload binary data collected on your machine through a specific component and the relevant metadata to the Viam app. Uploaded binary data can be found under the Images, Point clouds, or Files subtab of the app’s Data tab, depending on the type of data that you upload.

Parameters:

  • binary_data (bytes): The data to be uploaded, represented in bytes.
  • part_id (str): Part ID of the component used to capture the data. See Find Part ID for instructions on retrieving this value.
  • component_type (str): Type of the component used to capture the data.
  • component_name (str): Name of the component used to capture the data.
  • method_name (str): Name of the method used to capture the data.
  • tags (Optional[List[str]]): Optional list of image tags to allow for tag-based data filtering when retrieving data.
  • data_request_times (Optional[Tuple[datetime.datetime, datetime.datetime]]): Optional tuple containing datetime objects denoting the times this data was requested and received by the appropriate sensor.
  • file_extension (Optional[str]): The file extension of binary data including the period. For example, ".jpg", ".png", or ".pcd". Specify this to route the binary data to its corresponding mime type in storage in the Viam app.

Returns:

  • (str): ID of the new file.
time_requested = datetime(2023, 6, 5, 11)
time_received = datetime(2023, 6, 5, 11, 0, 3)

file_id = await data_client.binary_data_capture_upload(
    part_id="INSERT YOUR PART ID",
    component_type='camera',
    component_name='my_camera',
    method_name='GetImages',
    method_parameters=None,
    tags=["tag_1", "tag_2"],
    data_request_times=[time_requested, time_received],
    file_extension=".jpg",
    binary_data=b"Encoded image bytes"
)

For more information, see the Python SDK Docs.

TabularDataCaptureUpload

Upload tabular data collected on your machine through a specific component to the Viam app. Uploaded tabular data can be found under the Sensors subtab of the app’s Data tab.

Parameters:

  • tabular_data (List[Mapping[str, Any]]): List of the data to be uploaded, represented tabularly as a collection of dictionaries.
  • part_id (str): Part ID of the component used to capture the data. See Find Part ID for instructions on retrieving this value.
  • component_type (str): Type of the component used to capture the data.
  • component_name (str): Name of the component used to capture the data.
  • method_name (str): Name of the method used to capture the data.
  • tags (Optional[List[str]]): Optional list of image tags to allow for tag-based data filtering when retrieving data.
  • data_request_times (Optional[Tuple[datetime.datetime, datetime.datetime]]): Optional tuple containing datetime objects denoting the times this data was requested and received by the appropriate sensor.

Returns:

  • (str): ID of the new file.
time_requested = datetime(2023, 6, 5, 11)
time_received = datetime(2023, 6, 5, 11, 0, 3)

file_id = await data_client.tabular_data_capture_upload(
    part_id="INSERT YOUR PART ID",
    component_type='motor',
    component_name='left_motor',
    method_name='IsPowered',
    tags=["tag_1", "tag_2"],
    data_request_times=[(time_requested, time_received)],
    tabular_data=[{'PowerPCT': 0, 'IsPowered': False}]
)

For more information, see the Python SDK Docs.

StreamingDataCaptureUpload

Upload the contents of streaming binary data and the relevant metadata to the Viam app. Uploaded streaming data can be found under the Data tab.

Parameters:

  • data (bytes): Data to be uploaded, represented in bytes.
  • part_id (str): Part ID of the resource associated with the file.
  • file_ext (str): File extension type for the data. required for determining MIME type.
  • component_type (Optional[str]): Optional type of the component associated with the file (For example, “movement_sensor”).
  • component_name (Optional[str]): Optional name of the component associated with the file.
  • method_name (Optional[str]): Optional name of the method associated with the file.
  • method_parameters (Optional[str]): Optional dictionary of the method parameters. No longer in active use.
  • data_request_times (Optional[Tuple[datetime.datetime, datetime.datetime]]): Optional tuple containing datetime objects denoting the times this data was requested and received by the appropriate sensor.
  • tags (Optional[List[str]]): Optional list of image tags to allow for tag-based data filtering when retrieving data.

Returns:

  • (str): The file_id of the uploaded data.

Raises:

  • GRPCError – If an invalid part ID is passed.
time_requested = datetime(2023, 6, 5, 11)
time_received = datetime(2023, 6, 5, 11, 0, 3)

file_id = await data_client.streaming_data_capture_upload(
    data="byte-data-to-upload",
    part_id="INSERT YOUR PART ID",
    file_ext="png",
    component_type='motor',
    component_name='left_motor',
    method_name='IsPowered',
    data_request_times=[(time_requested, time_received)],
    tags=["tag_1", "tag_2"]
)

print(file_id)

For more information, see the Python SDK Docs.

FileUpload

Upload arbitrary files stored on your machine to the Viam app by file name. If uploaded with a file extension of .jpeg/.jpg/.png, uploaded files can be found in the Images subtab of the app’s Data tab. If .pcd, the uploaded files can be found in the Point clouds subtab. All other types of uploaded files can be found under the Files subtab of the app’s Data tab.

Parameters:

  • data (bytes): Bytes representing the file data to upload.
  • part_id (str): Part ID of the component used to capture the data. See Find Part ID for instructions on retrieving this value.
  • component_type (Optional[str]): Type of the component used to capture the data.
  • component_name (Optional[str]): Name of the component used to capture the data.
  • file_name (Optional[str]): Optional name of the file. The empty string "" will be assigned as the filename if one isn’t provided.
  • file_extension (Optional[str]): Optional file extension. The empty string "" will be assigned as the file extension if one isn’t provided.
  • tags (Optional[List[str]]): Optional list of image tags to allow for tag-based data filtering when retrieving data.

Returns:

  • (str): ID of the new file.
file_id = await data_client.file_upload(
    data=b"Encoded image bytes",
    part_id="INSERT YOUR PART ID",
    tags=["tag_1", "tag_2"],
    file_name="your-file",
    file_extension=".txt"
)

For more information, see the Python SDK Docs.

FileUploadFromPath

Upload files stored on your machine to the Viam app by filepath. Uploaded files can be found under the Files subtab of the app’s Data tab.

Parameters:

  • filepath (str): The absolute filepath of the file to be uploaded.
  • part_id (str): Part ID of the component used to capture the data. See Find Part ID for instructions on retrieving this value.
  • component_type (Optional[str]): Type of the component used to capture the data.
  • component_name (Optional[str]): Name of the component used to capture the data.
  • tags (Optional[List[str]]): Optional list of image tags to allow for tag-based data filtering when retrieving data.

Returns:

  • (str): ID of the new file.
file_id = await data_client.file_upload_from_path(
    part_id="INSERT YOUR PART ID",
    tags=["tag_1", "tag_2"],
    filepath="/Users/<your-username>/<your-directory>/<your-file.txt>"
)

For more information, see the Python SDK Docs.

AddBoundingBoxToImageById

Add a bounding box to an image specified by its BinaryID.

Parameters:

  • binary_id (viam.proto.app.data.BinaryID): The ID of the image to add the bounding box to.
  • label (str): A label for the bounding box.
  • x_min_normalized (float): Minimum X value of the bounding box normalized from 0 to 1.
  • y_min_normalized (float): Minimum Y value of the bounding box normalized from 0 to 1.
  • x_max_normalized (float): Maximum X value of the bounding box normalized from 0 to 1.
  • y_max_normalized (float): Maximum Y value of the bounding box normalized from 0 to 1.

Returns:

  • (str): The bounding box ID of the image.

Raises:

  • GRPCError – If the X or Y values are outside of the [0, 1] range.
from viam.proto.app.data import BinaryID

MY_BINARY_ID = BinaryID(
    file_id=your-file_id,
    organization_id=your-org-id,
    location_id=your-location-id
)

bbox_label = await data_client.add_bounding_box_to_image_by_id(
  binary_id=MY_BINARY_ID,
  label="label",
  x_min_normalized=0,
  y_min_normalized=.1,
  x_max_normalized=.2,
  y_max_normalized=.3
)

print(bbox_label)

For more information, see the Python SDK Docs.

RemoveBoundingBoxFromImageById

Removes a bounding box from an image specified by its BinaryID.

Parameters:

Returns:

  • None.
from viam.proto.app.data import BinaryID

MY_BINARY_ID = BinaryID(
    file_id=your-file_id,
    organization_id=your-org-id,
    location_id=your-location-id
)

await data_client.remove_bounding_box_from_image_by_id(
  binary_id=MY_BINARY_ID,
  bbox_id="your-bounding-box-id-to-delete"
)

For more information, see the Python SDK Docs.



Have questions, or want to meet other people working on robots? Join our Community Discord.

If you notice any issues with the documentation, feel free to file an issue or edit this file.