viam.app.data_client
Attributes
Classes
gRPC client for uploading and retrieving data from app. |
Module Contents
- 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.
This class’s constructor instantiates relevant service stubs. Always make
DataClient
method calls through an instance ofViamClient
.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())
For more information, see Data Client API.
- 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
- class TabularDataPoint[source]
Represents a tabular data point and its associated metadata.
- part_id: str
The robot part ID
- resource_name: str
The resource name
- resource_api: str
sensor
- Type:
The resource API. For example, rdk
- Type:
component
- method_name: str
The method used for data capture. For example, Readings
- time_captured: datetime.datetime
The time at which the data point was captured
- organization_id: str
The organization ID
- location_id: str
The location ID
- robot_name: str
The robot name
- robot_id: str
The robot ID
- part_name: str
The robot part name
- method_parameters: Mapping[str, viam.utils.ValueTypes]
Additional parameters associated with the data capture method
- tags: List[str]
A list of tags associated with the data point
- payload: Mapping[str, viam.utils.ValueTypes]
The captured data
- property resource_subtype: str
- 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; the returned tuple will include the pagination ID. If a destination is provided, this method saves returned data to that file, overwriting any existing file content.from viam.utils import create_filter my_data = [] my_filter = create_filter(component_name="motor-1") last = None while True: tabular_data, count, last = await data_client.tabular_data_by_filter(my_filter, last=last) if not tabular_data: break my_data.extend(tabular_data) print(f"My data: {my_data}")
- Parameters:
filter (Filter) – Optional, specifies tabular data to retrieve. If missing, matches all tabular data.
limit (int) – The maximum number of entries to include in a page. Defaults to 50 if unspecified.
sort_order (Order) – The desired sort order of the data.
last (str) – Optional string indicating the object identifier of the last-returned data. Returned by calls to
TabularDataByFilter
as thelast
value. If provided, the server returns the next data entries after the last object identifier.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:
A tuple containing the following:
tabular_data
(List[TabularData]): The tabular data.count
(int): The count (number of entries).last
(str): The last-returned page ID.
- Return type:
Tuple[List[TabularData], int, str]
For more information, see Data Client API.
- async tabular_data_by_sql(organization_id: str, sql_query: str) List[Dict[str, viam.utils.ValueTypes | datetime.datetime]] [source]
Obtain unified tabular data and metadata, queried with SQL. Make sure your API key has permissions at the organization level in order to use this.
data = await data_client.tabular_data_by_sql( organization_id="<YOUR-ORG-ID>", sql_query="SELECT * FROM readings LIMIT 5" )
- Parameters:
organization_id (str) – The ID of the organization that owns the data. To find your organization ID, visit the organization settings page in the Viam app.
sql_query (str) – The SQL query to run.
- Returns:
An array of decoded BSON data objects.
- Return type:
List[Dict[str, Union[ValueTypes, datetime]]]
For more information, see Data Client API.
- async tabular_data_by_mql(organization_id: str, query: List[bytes] | List[Dict[str, Any]], use_recent_data: bool | None = None) List[Dict[str, viam.utils.ValueTypes | datetime.datetime]] [source]
Obtain unified tabular data and metadata, queried with MQL.
import bson tabular_data = await data_client.tabular_data_by_mql(organization_id="<YOUR-ORG-ID>", query=[ { '$match': { 'location_id': '<YOUR-LOCATION-ID>' } }, { "$limit": 5 } ]) print(f"Tabular Data: {tabular_data}")
- Parameters:
organization_id (str) – The ID of the organization that owns the data. To find your organization ID, visit the organization settings page in the Viam app.
query (Union[List[bytes], List[Dict[str, Any]]]) – The MQL query to run, as a list of MongoDB aggregation pipeline stages. Each stage can be provided as either a dictionary or raw BSON bytes, but support for bytes will be removed in the future, so prefer the dictionary option.
use_recent_data (bool) – Whether to query blob storage or your recent data store. Defaults to
False
.
- Returns:
An array of decoded BSON data objects.
- Return type:
List[Dict[str, Union[ValueTypes, datetime]]]
For more information, see Data Client API.
- async get_latest_tabular_data(part_id: str, resource_name: str, resource_api: str, method_name: str) Tuple[datetime.datetime, datetime.datetime, Dict[str, viam.utils.ValueTypes]] | None [source]
Gets the most recent tabular data captured from the specified data source, as long as it was synced within the last year.
tabular_data = await data_client.get_latest_tabular_data( part_id="77ae3145-7b91-123a-a234-e567cdca8910", resource_name="camera-1", resource_api="rdk:component:camera", method_name="GetImage" ) if tabular_data: time_captured, time_synced, payload = tabular_data print(f"Time Captured: {time_captured}") print(f"Time Synced: {time_synced}") print(f"Payload: {payload}") else: print(f"No data returned: {tabular_data}")
- Parameters:
part_id (str) – The ID of the part that owns the data.
resource_name (str) – The name of the requested resource that captured the data. For example, “my-sensor”.
resource_api (str) – The API of the requested resource that captured the data. For example, “rdk:component:sensor”.
method_name (str) – The data capture method name. For exampe, “Readings”.
- Returns:
A return value of
None
means that this data source has not synced data in the last year. Otherwise, the data source has synced some data in the last year, so the returned tuple contains the following:time_captured
(datetime): The time captured.time_synced
(datetime): The time synced.payload
(Dict[str, ValueTypes]): The latest tabular data captured from the specified data source.
- Return type:
Optional[Tuple[datetime, datetime, Dict[str, ValueTypes]]]
For more information, see Data Client API.
- async export_tabular_data(part_id: str, resource_name: str, resource_api: str, method_name: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None) List[TabularDataPoint] [source]
Obtain unified tabular data and metadata from the specified data source.
tabular_data = await data_client.export_tabular_data( part_id="<PART-ID>", resource_name="<RESOURCE-NAME>", resource_api="<RESOURCE-API>", method_name="<METHOD-NAME>", start_time="<START_TIME>" end_time="<END_TIME>" ) print(f"My data: {tabular_data}")
- Parameters:
part_id (str) – The ID of the part that owns the data.
resource_name (str) – The name of the requested resource that captured the data.
resource_api (str) – The API of the requested resource that captured the data.
method_name (str) – The data capture method name.
start_time (datetime) – Optional start time for requesting a specific range of data.
end_time (datetime) – Optional end time for requesting a specific range of data.
- Returns:
The unified tabular data and metadata.
- Return type:
List[TabularDataPoint]
For more information, see Data Client API.
- 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[viam.proto.app.data.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 aslast
. If a destination is provided, this method saves returned data to that file, overwriting any existing file content.from viam.utils import create_filter from viam.proto.app.data import Filter, TagsFilter, TagsFilterType # Get data captured from camera components my_data = [] last = None my_filter = create_filter(component_name="camera-1") while True: data, count, last = await data_client.binary_data_by_filter( my_filter, limit=1, last=last) if not data: break my_data.extend(data) print(f"My data: {my_data}") # Get untagged data from a dataset my_untagged_data = [] last = None tags_filter = TagsFilter(type=TagsFilterType.TAGS_FILTER_TYPE_UNTAGGED) my_filter = Filter( dataset_id="66db6fe7d93d1ade24cd1dc3", tags_filter=tags_filter ) while True: data, count, last = await data_client.binary_data_by_filter( my_filter, last=last, include_binary_data=False) if not data: break my_untagged_data.extend(data)
- Parameters:
filter (Filter) – Optional, specifies tabular data to retrieve. An empty filter matches all binary data.
limit (int) – The maximum number of entries to include in a page. Defaults to 50 if unspecified.
sort_order (Order) – The desired sort order of the data.
last (str) – Optional string indicating the object identifier of the last-returned data. This object identifier is returned by calls to
binary_data_by_filter()
as thelast
value. If provided, the server will return the next data entries after the last object identifier.include_binary_data (bool) – Boolean specifying whether to actually include the binary file data with each retrieved file. Defaults to true (that is, 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:
A tuple containing the following:
data
(List[BinaryData
]): The binary data.count
(int): The count (number of entries).last
(str): The last-returned page ID.
- Return type:
Tuple[List[BinaryData], int, str]
For more information, see Data Client API.
- async binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID] | List[str], dest: str | None = None) List[viam.proto.app.data.BinaryData] [source]
Filter and download binary data.
binary_metadata, count, last = await data_client.binary_data_by_filter( include_binary_data=False ) my_ids = [] for obj in binary_metadata: my_ids.append(obj.metadata.binary_data_id) binary_data = await data_client.binary_data_by_ids(my_ids)
- Parameters:
binary_ids (Union[List[BinaryID], List[str]]) – Binary data ID strings specifying the desired data or
BinaryID
objects. Must be non-empty. DEPRECATED:BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.dest (str) – Optional filepath for writing retrieved data.
- Raises:
GRPCError – If no binary data ID strings or
BinaryID
objects are provided.- Returns:
The binary data.
- Return type:
List[BinaryData]
For more information, see Data Client API.
- async delete_tabular_data(organization_id: str, delete_older_than_days: int) int [source]
Delete tabular data older than a specified number of days.
tabular_data = await data_client.delete_tabular_data( organization_id="<YOUR-ORG-ID>", delete_older_than_days=150 )
- Parameters:
organization_id (str) – The ID of the organization to delete the data from. To find your organization ID, visit the organization settings page in the Viam app.
delete_older_than_days (int) – Delete data that was captured up to this many days ago. For example, a value of 10 deletes any data that was captured up to 10 days ago. A value of 0 deletes all existing data.
- Returns:
The number of items deleted.
- Return type:
int
For more information, see Data Client API.
- abstract delete_tabular_data_by_filter(filter: viam.proto.app.data.Filter | None) int [source]
- Async:
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.utils import create_filter my_filter = create_filter(component_name="left_motor", organization_ids=["<YOUR-ORG-ID>"]) res = await data_client.delete_binary_data_by_filter(my_filter)
- Parameters:
filter (Filter) – Optional, specifies binary data to delete. CAUTION: Passing an empty
Filter
deletes all binary data! You must specify an organization ID withorganization_ids
when using this option. To find your organization ID, visit the organization settings page in the Viam app.- Returns:
The number of items deleted.
- Return type:
int
For more information, see Data Client API.
- async delete_binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID] | List[str]) int [source]
Filter and delete binary data.
from viam.proto.app.data import BinaryID from viam.utils import create_filter my_filter = create_filter(component_name="camera-1", organization_ids=["<YOUR-ORG-ID>"]) binary_metadata, count, last = await data_client.binary_data_by_filter( filter=my_filter, limit=20, include_binary_data=False ) my_ids = [] for obj in binary_metadata: my_ids.append( obj.metadata.binary_data_id ) binary_data = await data_client.delete_binary_data_by_ids(my_ids)
- Parameters:
binary_ids (Union[List[BinaryID], List[str]]) – Binary data ID strings specifying the data to be deleted or
BinaryID
objects. Must be non-empty. DEPRECATED:BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.- Raises:
GRPCError – If no binary data ID strings or
BinaryID
objects are provided.- Returns:
The number of items deleted.
- Return type:
int
For more information, see Data Client API.
- async add_tags_to_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID] | List[str]) None [source]
Add tags to binary data.
from viam.utils import create_filter tags = ["tag1", "tag2"] my_filter = create_filter(component_name="camera-1", organization_ids=["<YOUR-ORG-ID>"]) binary_metadata, count, last = await data_client.binary_data_by_filter( filter=my_filter, limit=20, include_binary_data=False ) my_ids = [] for obj in binary_metadata: my_ids.append( obj.metadata.binary_data_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 (Union[List[BinaryID], List[str]]) – Binary data ID strings specifying the data to be tagged or
BinaryID
objects. Must be non-empty. DEPRECATED:BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.
- Raises:
GRPCError – If no binary data ID strings or
BinaryID
objects are provided.
For more information, see Data Client API.
- 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.utils import create_filter my_filter = create_filter(component_name="my_camera") tags = ["tag1", "tag2"] 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 (Filter) – Specifies binary data to tag. If none is provided, tags all data.
- Raises:
GRPCError – If no tags are provided.
For more information, see Data Client API.
- async remove_tags_from_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID] | List[str]) int [source]
Remove tags from binary data by IDs.
from viam.utils import create_filter tags = ["tag1", "tag2"] my_filter = create_filter(component_name="camera-1") binary_metadata, count, last = await data_client.binary_data_by_filter( filter=my_filter, limit=50, include_binary_data=False ) my_ids = [] for obj in binary_metadata: my_ids.append( obj.metadata.binary_data_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 (Union[List[BinaryID], List[str]]) – Binary data ID strings specifying the data to be untagged or BinaryID objects. Must be non-empty. DEPRECATED:
BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.
- Raises:
GRPCError – If no binary data ID strings,
BinaryID
objects, or tags are provided.- Returns:
The number of tags removed.
- Return type:
int
For more information, see Data Client API.
- 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.utils import create_filter my_filter = create_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 (Filter) – Specifies binary data to untag. If none is provided, removes tags from all data.
- Raises:
GRPCError – If no tags are provided.
- Returns:
The number of tags removed.
- Return type:
int
For more information, see Data Client API.
- 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.utils import create_filter my_filter = create_filter(component_name="my_camera") tags = await data_client.tags_by_filter(my_filter)
- Parameters:
filter (Filter) – Specifies subset ofdata to retrieve tags from. If none is provided, returns all tags.
- Returns:
The list of tags.
- Return type:
List[str]
For more information, see Data Client API.
- async add_bounding_box_to_image_by_id(binary_id: viam.proto.app.data.BinaryID | str, 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.
bbox_id = await data_client.add_bounding_box_to_image_by_id( binary_id="<YOUR-BINARY-DATA-ID>", label="label", x_min_normalized=0, y_min_normalized=.1, x_max_normalized=.2, y_max_normalized=.3 ) print(bbox_id)
- Parameters:
binary_id (Union[BinaryID, str]) – The binary data ID or
BinaryID
of the image to add the bounding box to. DEPRECATED:BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.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
For more information, see Data Client API.
- async remove_bounding_box_from_image_by_id(bbox_id: str, binary_id: viam.proto.app.data.BinaryID | str) None [source]
Removes a bounding box from an image.
await data_client.remove_bounding_box_from_image_by_id( binary_id="<YOUR-BINARY-DATA-ID>", bbox_id="your-bounding-box-id-to-delete" )
- Parameters:
bbox_id (str) – The ID of the bounding box to remove.
binary_id (Union[BinaryID, str]) – The binary data ID or
BinaryID
of the image to remove the bounding box from. DEPRECATED:BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.
For more information, see Data Client API.
- 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.utils import create_filter my_filter = create_filter(component_name="my_camera") bounding_box_labels = await data_client.bounding_box_labels_by_filter( my_filter) print(bounding_box_labels)
- Parameters:
filter (Filter) – Specifies data to retrieve bounding box labels from. If none is provided, returns labels
data. (from all) –
- Returns:
The list of bounding box labels.
- Return type:
List[str]
For more information, see Data Client API.
- async get_database_connection(organization_id: str) str [source]
Get a connection to access a MongoDB Atlas Data federation instance.
hostname = await data_client.get_database_connection(organization_id="<YOUR-ORG-ID>")
- Parameters:
organization_id (str) – The ID of the organization you’d like to connect to. To find your organization ID, visit the organization settings page in the Viam app.
- Returns:
The hostname of the federated database.
- Return type:
str
For more information, see Data Client API.
- 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.
await data_client.configure_database_user( organization_id="<YOUR-ORG-ID>", password="Your_Password@1234" )
- Parameters:
organization_id (str) – The ID of the organization you’d like to configure a database user for. To find your organization ID, visit the organization settings page in the Viam app.
password (str) – The password of the user.
For more information, see Data Client API.
- async create_dataset(name: str, organization_id: str) str [source]
Create a new dataset.
dataset_id = await data_client.create_dataset( name="<DATASET-NAME>", organization_id="<YOUR-ORG-ID>" ) print(dataset_id)
- Parameters:
name (str) – The name of the dataset being created.
organization_id (str) – The ID of the organization where the dataset is being created. To find your organization ID, visit the organization settings page in the Viam app.
- Returns:
The dataset ID of the created dataset.
- Return type:
str
For more information, see Data Client API.
- 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=["<YOUR-DATASET-ID-1>, <YOUR-DATASET-ID-2>"] ) print(datasets)
- Parameters:
ids (List[str]) –
The IDs of the datasets that you would like to retrieve information about. To retrieve a dataset ID:
Navigate to the DATASETS tab of the DATA page.
Click on the dataset.
Click the … menu.
Select Copy dataset ID.
- Returns:
The list of datasets.
- Return type:
Sequence[Dataset]
For more information, see Data Client API.
- 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_datasets_by_organization_id( organization_id="<YOUR-ORG-ID>" ) print(datasets)
- Parameters:
organization_id (str) – The ID of the organization you’d like to retrieve datasets from. To find your organization ID, visit the organization settings page in the Viam app.
- Returns:
The list of datasets in the organization.
- Return type:
Sequence[Dataset]
For more information, see Data Client API.
- async rename_dataset(id: str, name: str) None [source]
Rename a dataset specified by the dataset ID.
await data_client.rename_dataset( id="<YOUR-DATASET-ID>", name="MyDataset" )
- Parameters:
id (str) –
The ID of the dataset. To retrieve the dataset ID:
Navigate to the DATASETS tab of the DATA page.
Click on the dataset.
Click the … menu.
Select Copy dataset ID.
name (str) – The new name of the dataset.
For more information, see Data Client API.
- async delete_dataset(id: str) None [source]
Delete a dataset.
await data_client.delete_dataset( id="<YOUR-DATASET-ID>" )
- Parameters:
id (str) –
The ID of the dataset. To retrieve the dataset ID:
Navigate to the DATASETS tab of the DATA page.
Click on the dataset.
Click the … menu.
Select Copy dataset ID.
For more information, see Data Client API.
- async add_binary_data_to_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID] | List[str], dataset_id: str) None [source]
Add the BinaryData to the provided dataset.
This BinaryData will be tagged with the VIAM_DATASET_{id} label.
binary_metadata, count, last = await data_client.binary_data_by_filter( include_binary_data=False ) my_binary_data_ids = [] for obj in binary_metadata: my_binary_data_ids.append( obj.metadata.binary_data_id ) await data_client.add_binary_data_to_dataset_by_ids( binary_ids=my_binary_data_ids, dataset_id="abcd-1234xyz-8765z-123abc" )
- Parameters:
binary_ids (List[BinaryID]) – Unique identifiers for binary data to add to the dataset. To retrieve these IDs, navigate to the DATA page, click on an image, and copy its Binary Data ID from the details tab.
dataset_id (str) –
The ID of the dataset to be added to. To retrieve the dataset ID:
Navigate to the DATASETS tab of the DATA page.
Click on the dataset.
Click the … menu.
Select Copy dataset ID.
For more information, see Data Client API.
- async remove_binary_data_from_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID] | List[str], dataset_id: str) None [source]
Remove the BinaryData from the provided dataset.
This BinaryData will lose the VIAM_DATASET_{id} tag.
binary_metadata, count, last = await data_client.binary_data_by_filter( include_binary_data=False ) my_binary_data_ids = [] for obj in binary_metadata: my_binary_data_ids.append( obj.metadata.binary_data_id ) await data_client.remove_binary_data_from_dataset_by_ids( binary_ids=my_binary_data_ids, dataset_id="abcd-1234xyz-8765z-123abc" )
- Parameters:
binary_ids (Union[List[BinaryID], List[str]]) – Unique identifiers for the binary data to remove from the dataset. To retrieve these IDs, navigate to the DATA page, click on an image and copy its Binary Data ID from the details tab. DEPRECATED:
BinaryID
is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.dataset_id (str) –
The ID of the dataset to be removed from. To retrieve the dataset ID:
Navigate to the DATASETS tab of the DATA page.
Click on the dataset.
Click the … menu.
Select Copy dataset ID.
For more information, see Data Client API.
- 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 (for example, a motor), along with the relevant metadata. Binary data can be found on the DATA page of the Viam app.
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 (for example, “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, for example
.jpg
,.png
,.pcd
. The backend routes the binary to its corresponding mime type based on this extension. Files with a.jpeg
,.jpg
, or.png
extension will appear in 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 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 binary data ID of the uploaded data.
- Return type:
str
For more information, see Data Client API.
- async tabular_data_capture_upload(tabular_data: List[Mapping[str, Any]], part_id: str, component_type: str, component_name: str, method_name: str, data_request_times: List[Tuple[datetime.datetime, datetime.datetime]], method_parameters: Mapping[str, Any] | None = None, tags: List[str] | None = None) str [source]
Upload tabular sensor data.
Upload tabular data collected on a robot through a specific component (for example, a motor), along with the relevant metadata. Tabular data can be found under the Sensors tab of the DATA page.
from datetime import datetime 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='rdk:component:movement_sensor', component_name='my_movement_sensor', method_name='Readings', tags=["sensor_data"], data_request_times=[(time_requested, time_received)], tabular_data=[{ 'readings': { 'linear_velocity': {'x': 0.5, 'y': 0.0, 'z': 0.0}, 'angular_velocity': {'x': 0.0, 'y': 0.0, 'z': 0.1} } }] )
- Parameters:
tabular_data (List[Mapping[str, Any]]) – List of the data to be uploaded, represented tabularly as a collection of dictionaries. Must include the key
readings
for sensors.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 (for example,
rdk:component: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.
data_request_times (List[Tuple[datetime.datetime, datetime.datetime]]) – 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. Pass a list of tabular data and timestamps with lengthn > 1
to uploadn
datapoints, all with the same metadata.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.
- Raises:
GRPCError – If an invalid part ID is passed.
ValueError – If the provided list of Timestamp objects has a length that does not match the length of the list of tabular data.
- Returns:
The file ID of the uploaded data.
- Return type:
str
For more information, see Data Client API.
- 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 (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
[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 binary data ID of the uploaded data.
- Return type:
str
For more information, see Data Client API.
- 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. File data can be found in the Files tab of the DATA page.
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 (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.
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:
Binary data ID of the new file.
- Return type:
str
For more information, see Data Client API.
- 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. File data can be found in the Files tab of the DATA page.
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 (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.
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:
Binary data ID of the new file.
- Return type:
str
For more information, see Data Client API.
- 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]