viam.app.data_client

Module Contents

Classes

DataClient

gRPC client for uploading and retrieving data from app.

Attributes

LOGGER

viam.app.data_client.LOGGER
class viam.app.data_client.DataClient(channel: grpclib.client.Channel, metadata: Mapping[str, str])[source]

gRPC client for uploading and retrieving data from app.

Constructor is used by ViamClient to instantiate relevant service stubs. Calls to DataClient 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 DataClient to run data client API methods on
    data_client = viam_client.data_client

    viam_client.close()

if __name__ == '__main__':
    asyncio.run(main())
class TabularData[source]

Class representing a piece of tabular data and associated metadata.

data: Mapping[str, Any]

The requested data

metadata: viam.proto.app.data.CaptureMetadata

The metadata associated with the data

time_requested: datetime.datetime

The time the data were requested

time_received: datetime.datetime

The time the data were received

__str__() → str[source]

Return str(self).

__eq__(other: object) → bool[source]

Return self==value.

class BinaryData[source]

Class representing a piece of binary data and associated metadata.

data: bytes

The request data

metadata: viam.proto.app.data.BinaryMetadata

The metadata associated with the data

__str__() → str[source]

Return str(self).

__eq__(other: object) → bool[source]

Return self==value.

async tabular_data_by_filter(filter: viam.proto.app.data.Filter | None = None, limit: int | None = None, sort_order: viam.proto.app.data.Order.ValueType | None = None, last: str | None = None, count_only: bool = False, include_internal_data: bool = False, dest: str | None = None) → Tuple[List[TabularData], int, str][source]

Filter and download tabular data. The data will be paginated into pages of limit items, and the pagination ID will be included in the returned tuple. If a destination is provided, the data will be saved to that file. If the file is not empty, it will be overwritten.

from viam.proto.app.data import Filter

my_data = []
last = None
my_filter = Filter(component_name="left_motor")
while True:
    tabular_data, count, last = await data_client.tabular_data_by_filter(my_filter, last)
    if not tabular_data:
        break
    my_data.extend(tabular_data)
Parameters:

  • filter (viam.proto.app.data.Filter) – Optional Filter specifying tabular data to retrieve. No Filter implies all tabular data.

  • limit (int) – The maximum number of entries to include in a page. Defaults to 50 if unspecified.

  • sort_order (viam.proto.app.data.Order) – The desired sort order of the data.

  • last (str) – Optional string indicating the ID of the last-returned data. If provided, the server will return the next data entries after the last ID.

  • count_only (bool) – Whether to return only the total count of entries.

  • include_internal_data (bool) – Whether to return the internal data. Internal data is used for Viam-specific data ingestion, like cloud SLAM. Defaults to False

  • dest (str) – Optional filepath for writing retrieved data.

Returns:

The tabular data. int: The count (number of entries) str: The last-returned page ID.

Return type:

List[TabularData]

async tabular_data_by_sql(organization_id: str, sql_query: str) → List[Dict[str, viam.utils.ValueTypes]][source]

Obtain unified tabular data and metadata, queried with SQL.

data = await data_client.tabular_data_by_sql(org_id="<your-org-id>", sql_query="<sql-query>")
Parameters:

  • organization_id (str) – The ID of the organization that owns the data.

  • sql_query (str) – The SQL query to run.

Returns:

An array of data objects.

Return type:

List[Dict[str, ValueTypes]]

async tabular_data_by_mql(organization_id: str, mql_binary: List[bytes]) → List[Dict[str, viam.utils.ValueTypes]][source]

Obtain unified tabular data and metadata, queried with MQL.

data = await data_client.tabular_data_by_mql(org_id="<your-org-id>", mql_binary=[<mql-bytes-1>, <mql-bytes-2>])
Parameters:

  • organization_id (str) – The ID of the organization that owns the data.

  • mql_binary (List[bytes]) – The MQL query to run as a list of BSON documents.

Returns:

An array of data objects.

Return type:

List[Dict[str, ValueTypes]]

async binary_data_by_filter(filter: viam.proto.app.data.Filter | None = None, limit: int | None = None, sort_order: viam.proto.app.data.Order.ValueType | None = None, last: str | None = None, include_binary_data: bool = True, count_only: bool = False, include_internal_data: bool = False, dest: str | None = None) → Tuple[List[BinaryData], int, str][source]

Filter and download binary data. The data will be paginated into pages of limit items, and the pagination ID will be included in the returned tuple. If a destination is provided, the data will be saved to that file. If the file is not empty, it will be overwritten.

from viam.proto.app.data import Filter


my_data = []
last = None
my_filter = Filter(component_name="camera")
while True:
    data, count, last = await data_client.binary_data_by_filter(my_filter, last)
    if not data:
        break
    my_data.extend(data)
Parameters:

  • filter (viam.proto.app.data.Filter) – Optional Filter specifying tabular data to retrieve. No Filter implies all binary data.

  • limit (int) – The maximum number of entries to include in a page. Defaults to 50 if unspecified.

  • sort_order (viam.proto.app.data.Order) – The desired sort order of the data.

  • last (str) – Optional string indicating the ID of the last-returned data. If provided, the server will return the next data entries after the last ID.

  • include_binary_data (bool) – Boolean specifying whether to actually include the binary file data with each retrieved file. Defaults to true (i.e., both the files’ data and metadata are returned).

  • count_only (bool) – Whether to return only the total count of entries.

  • include_internal_data (bool) – Whether to return the internal data. Internal data is used for Viam-specific data ingestion, like cloud SLAM. Defaults to False

  • dest (str) – Optional filepath for writing retrieved data.

Returns:

The binary data. int: The count (number of entries) str: The last-returned page ID.

Return type:

List[BinaryData]

async binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dest: str | None = None) → List[BinaryData][source]

Filter and download binary data.

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)
Parameters:

  • binary_ids (List[viam.proto.app.data.BinaryID]) – BinaryID objects specifying the desired data. Must be non-empty.

  • dest (str) – Optional filepath for writing retrieved data.

Raises:

GRPCError – If no BinaryID objects are provided.

Returns:

The binary data.

Return type:

List[BinaryData]

async delete_tabular_data(organization_id: str, delete_older_than_days: int) → int[source]

Delete tabular data older than a specified number of days.

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(
    org_id="a12b3c4e-1234-1abc-ab1c-ab1c2d345abc", days_of_data_to_delete)
Parameters:

  • organization_id (str) – ID of organization to delete data from.

  • 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:

The number of items deleted.

Return type:

int

abstract async delete_tabular_data_by_filter(filter: viam.proto.app.data.Filter | None) → int[source]

Deprecated: use delete_tabular_data instead.

async delete_binary_data_by_filter(filter: viam.proto.app.data.Filter | None) → int[source]

Filter and delete binary data.

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)
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:

The number of items deleted.

Return type:

int

async delete_binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID]) → int[source]

Filter and delete binary data.

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)
Parameters:

binary_ids (List[viam.proto.app.data.BinaryID]) – BinaryID objects specifying the data to be deleted. Must be non-empty.

Raises:

GRPCError – If no BinaryID objects are provided.

Returns:

The number of items deleted.

Return type:

int

async add_tags_to_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID]) → None[source]

Add tags to binary data.

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)
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.

Raises:

GRPCError – If no BinaryID objects or tags are provided.

async add_tags_to_binary_data_by_filter(tags: List[str], filter: viam.proto.app.data.Filter | None = None) → None[source]

Add tags to binary data.

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)
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.

Raises:

GRPCError – If no tags are provided.

async remove_tags_from_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID]) → int[source]

Remove tags from binary.

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)
Parameters:

  • tags (List[str]) – List of tags to remove from specified binary data. Must be non-empty.

  • binary_ids (List[BinaryID]) – List of BinaryID objects specifying binary data to untag. Must be non-empty.

Raises:

GRPCError – If no binary_ids or tags are provided.

Returns:

The number of tags removed.

Return type:

int

async remove_tags_from_binary_data_by_filter(tags: List[str], filter: viam.proto.app.data.Filter | None = None) → int[source]

Remove tags from binary data.

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)
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.

Raises:

GRPCError – If no tags are provided.

Returns:

The number of tags removed.

Return type:

int

async tags_by_filter(filter: viam.proto.app.data.Filter | None = None) → List[str][source]

Get a list of tags using a filter.

from viam.proto.app.data import Filter

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

filter (viam.proto.app.data.Filter) – Filter specifying data to retrieve from. If no Filter is provided, all data tags will return.

Returns:

The list of tags.

Return type:

List[str]

async add_bounding_box_to_image_by_id(binary_id: viam.proto.app.data.BinaryID, label: str, x_min_normalized: float, y_min_normalized: float, x_max_normalized: float, y_max_normalized: float) → str[source]

Add a bounding box to an image.

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)
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) – Min X value of the bounding box normalized from 0 to 1.

  • y_min_normalized (float) – Min Y value of the bounding box normalized from 0 to 1.

  • x_max_normalized (float) – Max X value of the bounding box normalized from 0 to 1.

  • y_max_normalized (float) – Max Y value of the bounding box normalized from 0 to 1.

Raises:

GRPCError – If the X or Y values are outside of the [0, 1] range.

Returns:

The bounding box ID

Return type:

str

async remove_bounding_box_from_image_by_id(bbox_id: str, binary_id: viam.proto.app.data.BinaryID) → None[source]

Removes a bounding box from an image.

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"
)
Parameters:

  • bbox_id (str) – The ID of the bounding box to remove.

  • Binary_id (viam.proto.arr.data.BinaryID) – Binary ID of the image to to remove the bounding box from

async bounding_box_labels_by_filter(filter: viam.proto.app.data.Filter | None = None) → List[str][source]

Get a list of bounding box labels using a Filter.

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)
Parameters:

filter (viam.proto.app.data.Filter) – Filter specifying data to retrieve from. If no Filter is provided, all labels will return.

Returns:

The list of bounding box labels.

Return type:

List[str]

async get_database_connection(organization_id: str) → str[source]

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

data_client.get_database_connection(org_id="a12b3c4e-1234-1abc-ab1c-ab1c2d345abc")
Parameters:

organization_id (str) – Organization to retrieve the connection for.

Returns:

The hostname of the federated database.

Return type:

str

async configure_database_user(organization_id: str, password: str) → None[source]

Configure a database user for the Viam organization’s MongoDB Atlas Data Federation instance. It can also be used to reset the password of the existing database user.

Parameters:

  • organization_id (str) – The ID of the organization.

  • password (str) – The password of the user.

async create_dataset(name: str, organization_id: str) → str[source]

Create a new dataset.

name = await data_client.create_dataset(
    name="<dataset-name>",
    organization_id="<your-org-id>"
)
print(name)
Parameters:

  • name (str) – The name of the dataset being created.

  • organization_id (str) – The ID of the organization where the dataset is being created.

Returns:

The dataset ID of the created dataset.

Return type:

str

async list_dataset_by_ids(ids: List[str]) → Sequence[viam.proto.app.dataset.Dataset][source]

Get a list of datasets using their IDs.

datasets = await data_client.list_dataset_by_ids(
    ids=["abcd-1234xyz-8765z-123abc"]
)
print(datasets)
Parameters:

ids (List[str]) – The IDs of the datasets being called for.

Returns:

The list of datasets.

Return type:

Sequence[Dataset]

async list_datasets_by_organization_id(organization_id: str) → Sequence[viam.proto.app.dataset.Dataset][source]

Get the datasets in an organization.

datasets = await data_client.list_dataset_by_ids(
    ids=["abcd-1234xyz-8765z-123abc"]
)
print(datasets)
Parameters:

organization_id (str) – The ID of the organization.

Returns:

The list of datasets in the organization.

Return type:

Sequence[Dataset]

async rename_dataset(id: str, name: str) → None[source]

Rename a dataset specified by the dataset ID.

await data_client.rename_dataset(
    id="abcd-1234xyz-8765z-123abc",
    name="<dataset-name>"
)
Parameters:

  • id (str) – The ID of the dataset.

  • name (str) – The new name of the dataset.

async delete_dataset(id: str) → None[source]

Delete a dataset.

await data_client.delete_dataset(
    id="abcd-1234xyz-8765z-123abc"
)
Parameters:

id (str) – The ID of the dataset.

async add_binary_data_to_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dataset_id: str) → None[source]

Add the BinaryData to the provided dataset.

This BinaryData will be tagged with the VIAM_DATASET_{id} label.

from viam.proto.app.data import BinaryID

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
)

my_binary_ids = []

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

await data_client.add_binary_data_to_dataset_by_ids(
    binary_ids=my_binary_ids,
    dataset_id="abcd-1234xyz-8765z-123abc"
)
Parameters:

  • binary_ids (List[BinaryID]) – The IDs of binary data to add to dataset.

  • dataset_id (str) – The ID of the dataset to be added to.

async remove_binary_data_from_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dataset_id: str) → None[source]

Remove the BinaryData from the provided dataset.

This BinaryData will lose the VIAM_DATASET_{id} tag.

from viam.proto.app.data import BinaryID

binary_metadata = await data_client.binary_data_by_filter(
    include_file_data=False
)

my_binary_ids = []

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

await data_client.remove_binary_data_from_dataset_by_ids(
    binary_ids=my_binary_ids,
    dataset_id="abcd-1234xyz-8765z-123abc"
)
Parameters:

  • binary_ids (List[BinaryID]) – The IDs of binary data to remove from dataset.

  • dataset_id (str) – The ID of the dataset to be removed from.

async binary_data_capture_upload(binary_data: bytes, part_id: str, component_type: str, component_name: str, method_name: str, file_extension: str, method_parameters: Mapping[str, Any] | None = None, tags: List[str] | None = None, data_request_times: Tuple[datetime.datetime, datetime.datetime] | None = None) → str[source]

Upload binary sensor data.

Upload binary data collected on a robot through a specific component (e.g., a motor) along with the relevant metadata to app.viam.com. Binary data can be found under the “Files” subtab of the Data tab on app.viam.com.

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"
)
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.

  • component_type (str) – Type of the component used to capture the data (e.g., “movement_sensor”).

  • component_name (str) – Name of the component used to capture the data.

  • method_name (str) – Name of the method used to capture the data.

  • file_extension (str) – The file extension of binary data including the period, e.g. .jpg, .png, .pcd. The backend will route the binary to its corresponding mime type based on this extension. Files with a .jpeg, .jpg, or .png extension will be saved to the images tab.

  • method_parameters (Optional[Mapping[str, Any]]) – Optional dictionary of method parameters. No longer in active use.

  • tags (Optional[List[str]]) – Optional list of tags to allow for tag-based data filtering when retrieving data.

  • data_request_times (Optional[Tuple[datetime.datetime, datetime.datetime]]) – Optional tuple containing `datetime`s objects denoting the times this data was requested[0] by the robot and received[1] from the appropriate sensor.

Raises:

GRPCError – If an invalid part ID is passed.

Returns:

the file_id of the uploaded data.

Return type:

str

async tabular_data_capture_upload(tabular_data: List[Mapping[str, Any]], part_id: str, component_type: str, component_name: str, method_name: str, method_parameters: Mapping[str, Any] | None = None, tags: List[str] | None = None, data_request_times: List[Tuple[datetime.datetime, datetime.datetime]] | None = None) → str[source]

Upload tabular sensor data.

Upload tabular data collected on a robot through a specific component (e.g., a motor) along with the relevant metadata to app.viam.com. Tabular data can be found under the “Sensors” subtab of the Data tab on app.viam.com.

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}]
)
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.

  • component_type (str) – Type of the component used to capture the data (e.g., “movement_sensor”).

  • component_name (str) – Name of the component used to capture the data.

  • method_name (str) – Name of the method used to capture the data.

  • method_parameters (Optional[Mapping[str, Any]]) – Optional dictionary of method parameters. No longer in active use.

  • tags (Optional[List[str]]) – Optional list of tags to allow for tag-based data filtering when retrieving data.

  • data_request_times (Optional[List[Tuple[datetime.datetime, datetime.datetime]]]) – Optional list of tuples, each containing datetime objects denoting the times this data was requested[0] by the robot and received[1] from the appropriate sensor. Passing a list of tabular data and Timestamps with length n > 1 will result in n datapoints being uploaded, all tied to the same metadata.

Raises:

  • GRPCError – If an invalid part ID is passed.

  • ValueError – If a list of Timestamp objects is provided and its length does not match the length of the list of tabular data.

Returns:

the file_id of the uploaded data.

Return type:

str

async streaming_data_capture_upload(data: bytes, part_id: str, file_ext: str, component_type: str | None = None, component_name: str | None = None, method_name: str | None = None, method_parameters: Mapping[str, Any] | None = None, data_request_times: Tuple[datetime.datetime, datetime.datetime] | None = None, tags: List[str] | None = None) → str[source]

Uploads the metadata and contents of streaming binary data.

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"]
)
Parameters:

  • data (bytes) – the data to be uploaded.

  • 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 (e.g., “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`s objects denoting the times this data was requested[0] by the robot and received[1] from the appropriate sensor.

  • tags (Optional[List[str]]) – Optional list of tags to allow for tag-based filtering when retrieving data.

Raises:

GRPCError – If an invalid part ID is passed.

Returns:

the file_id of the uploaded data.

Return type:

str

async file_upload(part_id: str, data: bytes, component_type: str | None = None, component_name: str | None = None, method_name: str | None = None, file_name: str | None = None, method_parameters: Mapping[str, Any] | None = None, file_extension: str | None = None, tags: List[str] | None = None) → str[source]

Upload arbitrary file data.

Upload file data that may be stored on a robot along with the relevant metadata to app.viam.com. File data can be found under the “Files” subtab of the Data tab on app.viam.com.

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"
)
Parameters:

  • part_id (str) – Part ID of the resource associated with the file.

  • data (bytes) – Bytes representing file data to upload.

  • component_type (Optional[str]) – Optional type of the component associated with the file (e.g., “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.

  • file_name (Optional[str]) – Optional name of the file. The empty string “” will be assigned as the file name if one isn’t provided.

  • method_parameters (Optional[str]) – Optional dictionary of the method parameters. No longer in active use.

  • file_extension (Optional[str]) – Optional file extension. The empty string “” will be assigned as the file extension if one isn’t provided. Files with a .jpeg, .jpg, or .png extension will be saved to the images tab.

  • tags (Optional[List[str]]) – Optional list of tags to allow for tag-based filtering when retrieving data.

Raises:

GRPCError – If an invalid part ID is passed.

Returns:

ID of the new file.

Return type:

str

async file_upload_from_path(filepath: str, part_id: str, component_type: str | None = None, component_name: str | None = None, method_name: str | None = None, method_parameters: Mapping[str, Any] | None = None, tags: List[str] | None = None) → str[source]

Upload arbitrary file data.

Upload file data that may be stored on a robot along with the relevant metadata to app.viam.com. File data can be found under the “Files” subtab of the Data tab on app.viam.com.

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>"
)
Parameters:

  • filepath (str) – Absolute filepath of file to be uploaded.

  • part_id (str) – Part ID of the component associated with the file.

  • component_type (Optional[str]) – Optional type of the component associated with the file (e.g., “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.

  • tags (Optional[List[str]]) – Optional list of tags to allow for tag-based filtering when retrieving data.

Raises:

  • GRPCError – If an invalid part ID is passed.

  • FileNotFoundError – If the provided filepath is not found.

Returns:

ID of the new file.

Return type:

str

static create_filter(component_name: str | None = None, component_type: str | None = None, method: str | None = None, robot_name: str | None = None, robot_id: str | None = None, part_name: str | None = None, part_id: str | None = None, location_ids: List[str] | None = None, organization_ids: List[str] | None = None, mime_type: List[str] | None = None, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, tags: List[str] | None = None, bbox_labels: List[str] | None = None, dataset_id: str | None = None) → viam.proto.app.data.Filter[source]