viam.app.data_client ==================== .. py:module:: viam.app.data_client Attributes ---------- .. autoapisummary:: viam.app.data_client.LOGGER Classes ------- .. autoapisummary:: viam.app.data_client.DataClient Module Contents --------------- .. py:data:: LOGGER .. py:class:: DataClient(channel: grpclib.client.Channel, metadata: Mapping[str, str]) gRPC client for uploading and retrieving data from app. This class's constructor instantiates relevant service stubs. Always make :class:`DataClient` method calls through an instance of :class:`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 "" (including brackets) with your API key and "" with your API key ID dial_options = DialOptions.with_api_key("", "") 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 `_. .. py:class:: TabularData Class representing a piece of tabular data and associated metadata. .. py:attribute:: data :type: Mapping[str, Any] The requested data .. py:attribute:: metadata :type: viam.proto.app.data.CaptureMetadata The metadata associated with the data .. py:attribute:: time_requested :type: datetime.datetime The time the data were requested .. py:attribute:: time_received :type: datetime.datetime The time the data were received .. py:method:: __str__() -> str .. py:method:: __eq__(other: object) -> bool .. py:class:: TabularDataPoint Represents a tabular data point and its associated metadata. .. py:attribute:: part_id :type: str The robot part ID .. py:attribute:: resource_name :type: str The resource name .. py:attribute:: resource_api :type: str sensor :type: The resource API. For example, rdk :type: component .. py:attribute:: method_name :type: str The method used for data capture. For example, Readings .. py:attribute:: time_captured :type: datetime.datetime The time at which the data point was captured .. py:attribute:: organization_id :type: str The organization ID .. py:attribute:: location_id :type: str The location ID .. py:attribute:: robot_name :type: str The robot name .. py:attribute:: robot_id :type: str The robot ID .. py:attribute:: part_name :type: str The robot part name .. py:attribute:: method_parameters :type: Mapping[str, viam.utils.ValueTypes] Additional parameters associated with the data capture method .. py:attribute:: tags :type: List[str] A list of tags associated with the data point .. py:attribute:: payload :type: Mapping[str, viam.utils.ValueTypes] The captured data .. py:method:: __str__() -> str .. py:method:: __eq__(other: object) -> bool .. py:property:: resource_subtype :type: str .. py:class:: DataPipeline Represents a data pipeline and its associated metadata. .. py:attribute:: id :type: str The ID of the data pipeline .. py:attribute:: organization_id :type: str The organization ID .. py:attribute:: name :type: str The name of the data pipeline .. py:attribute:: mql_binary :type: List[Dict[str, Any]] The MQL binary of the data pipeline .. py:attribute:: schedule :type: str The schedule of the data pipeline .. py:attribute:: created_on :type: datetime.datetime The time the data pipeline was created .. py:attribute:: updated_at :type: datetime.datetime The time the data pipeline was last updated .. py:attribute:: enabled :type: bool Whether the data pipeline is enabled .. py:attribute:: data_source_type :type: viam.proto.app.data.TabularDataSourceType.ValueType The type of data source for the data pipeline .. py:method:: from_proto(data_pipeline: viam.proto.app.datapipelines.DataPipeline) -> typing_extensions.Self :classmethod: .. py:class:: DataPipelineRun Represents a data pipeline run and its associated metadata. .. py:attribute:: id :type: str The ID of the data pipeline run .. py:attribute:: status :type: viam.proto.app.datapipelines.DataPipelineRunStatus.ValueType The status of the data pipeline run .. py:attribute:: start_time :type: datetime.datetime The time the data pipeline run started .. py:attribute:: end_time :type: datetime.datetime The time the data pipeline run ended .. py:attribute:: data_start_time :type: datetime.datetime The start time of the data that was processed in the run. .. py:attribute:: data_end_time :type: datetime.datetime The end time of the data that was processed in the run. .. py:method:: from_proto(data_pipeline_run: viam.proto.app.datapipelines.DataPipelineRun) -> typing_extensions.Self :classmethod: .. py:class:: DataPipelineRunsPage Represents a page of data pipeline runs and provides pagination functionality. .. py:attribute:: pipeline_id :type: str The ID of the pipeline these runs belong to .. py:attribute:: page_size :type: int The number of runs per page .. py:attribute:: runs :type: List[DataClient] The list of runs in this page .. py:attribute:: next_page_token :type: str The token to use to get the next page of results .. py:method:: next_page() -> DataClient :async: Get the next page of data pipeline runs. :returns: The next page of runs, or an empty page if there are no more runs :rtype: DataPipelineRunsPage .. py:method:: from_proto(data_pipeline_runs_page: viam.proto.app.datapipelines.ListDataPipelineRunsResponse, client: DataClient, page_size: int) -> typing_extensions.Self :classmethod: .. py:method:: tabular_data_by_filter(filter: Optional[viam.proto.app.data.Filter] = None, limit: Optional[int] = None, sort_order: Optional[viam.proto.app.data.Order.ValueType] = None, last: Optional[str] = None, count_only: bool = False, include_internal_data: bool = False, dest: Optional[str] = None) -> Tuple[List[TabularData], int, str] :async: 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}") :param filter: Optional, specifies tabular data to retrieve. If missing, matches all tabular data. :type filter: ~viam.proto.app.data.Filter :param limit: The maximum number of entries to include in a page. Defaults to 50 if unspecified. :type limit: int :param sort_order: The desired sort order of the data. :type sort_order: ~viam.proto.app.data.Order :param last: Optional string indicating the object identifier of the last-returned data. Returned by calls to :class:`TabularDataByFilter` as the ``last`` value. If provided, the server returns the next data entries after the last object identifier. :type last: str :param count_only: Whether to return only the total count of entries. :type count_only: bool :param include_internal_data: Whether to return the internal data. Internal data is used for Viam-specific data ingestion, like cloud SLAM. Defaults to ``False``. :type include_internal_data: bool :param dest: Optional filepath for writing retrieved data. :type dest: str :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. :rtype: Tuple[List[TabularData], int, str] For more information, see `Data Client API `_. .. py:method:: tabular_data_by_sql(organization_id: str, sql_query: str) -> List[Dict[str, Union[viam.utils.ValueTypes, datetime.datetime]]] :async: 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="", sql_query="SELECT * FROM readings LIMIT 5" ) :param organization_id: The ID of the organization that owns the data. To find your organization ID, visit the organization settings page. :type organization_id: str :param sql_query: The SQL query to run. :type sql_query: str :returns: An array of decoded BSON data objects. :rtype: List[Dict[str, Union[ValueTypes, datetime]]] For more information, see `Data Client API `_. .. py:method:: tabular_data_by_mql(organization_id: str, query: Union[List[bytes], List[Dict[str, Any]]], use_recent_data: Optional[bool] = None, tabular_data_source_type: viam.proto.app.data.TabularDataSourceType.ValueType = TabularDataSourceType.TABULAR_DATA_SOURCE_TYPE_STANDARD, pipeline_id: Optional[str] = None) -> List[Dict[str, Union[viam.utils.ValueTypes, datetime.datetime]]] :async: Obtain unified tabular data and metadata, queried with MQL. :: import bson tabular_data = await data_client.tabular_data_by_mql(organization_id="", query=[ { '$match': { 'location_id': '' } }, { "$limit": 5 } ]) print(f"Tabular Data: {tabular_data}") :param organization_id: The ID of the organization that owns the data. To find your organization ID, visit the organization settings page. :type organization_id: str :param query: 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. :type query: Union[List[bytes], List[Dict[str, Any]]] :param use_recent_data: Whether to query blob storage or your recent data store. Defaults to ``False``.. Deprecated, use `tabular_data_source_type` instead. :type use_recent_data: bool :param tabular_data_source_type: The data source to query. Defaults to `TABULAR_DATA_SOURCE_TYPE_STANDARD`. :type tabular_data_source_type: viam.proto.app.data.TabularDataSourceType :param pipeline_id: The ID of the data pipeline to query. Defaults to `None`. Required if `tabular_data_source_type` is `TABULAR_DATA_SOURCE_TYPE_PIPELINE_SINK`. :type pipeline_id: str :returns: An array of decoded BSON data objects. :rtype: List[Dict[str, Union[ValueTypes, datetime]]] For more information, see `Data Client API `_. .. py:method:: get_latest_tabular_data(part_id: str, resource_name: str, resource_api: str, method_name: str) -> Optional[Tuple[datetime.datetime, datetime.datetime, Dict[str, viam.utils.ValueTypes]]] :async: 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}") :param part_id: The ID of the part that owns the data. :type part_id: str :param resource_name: The name of the requested resource that captured the data. For example, "my-sensor". :type resource_name: str :param resource_api: The API of the requested resource that captured the data. For example, "rdk:component:sensor". :type resource_api: str :param method_name: The data capture method name. For exampe, "Readings". :type method_name: str :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. :rtype: Optional[Tuple[datetime, datetime, Dict[str, ValueTypes]]] For more information, see `Data Client API `_. .. py:method:: export_tabular_data(part_id: str, resource_name: str, resource_api: str, method_name: str, start_time: Optional[datetime.datetime] = None, end_time: Optional[datetime.datetime] = None) -> List[TabularDataPoint] :async: Obtain unified tabular data and metadata from the specified data source. :: tabular_data = await data_client.export_tabular_data( part_id="", resource_name="", resource_api="", method_name="", start_time="" end_time="" ) print(f"My data: {tabular_data}") :param part_id: The ID of the part that owns the data. :type part_id: str :param resource_name: The name of the requested resource that captured the data. :type resource_name: str :param resource_api: The API of the requested resource that captured the data. :type resource_api: str :param method_name: The data capture method name. :type method_name: str :param start_time: Optional start time for requesting a specific range of data. :type start_time: datetime :param end_time: Optional end time for requesting a specific range of data. :type end_time: datetime :returns: The unified tabular data and metadata. :rtype: List[TabularDataPoint] For more information, see `Data Client API `_. .. py:method:: binary_data_by_filter(filter: Optional[viam.proto.app.data.Filter] = None, limit: Optional[int] = None, sort_order: Optional[viam.proto.app.data.Order.ValueType] = None, last: Optional[str] = None, include_binary_data: bool = True, count_only: bool = False, include_internal_data: bool = False, dest: Optional[str] = None) -> Tuple[List[viam.proto.app.data.BinaryData], int, str] :async: 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 as ``last``. 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) :param filter: Optional, specifies tabular data to retrieve. An empty filter matches all binary data. :type filter: ~viam.proto.app.data.Filter :param limit: The maximum number of entries to include in a page. Defaults to 50 if unspecified. :type limit: int :param sort_order: The desired sort order of the data. :type sort_order: ~viam.proto.app.data.Order :param last: Optional string indicating the object identifier of the last-returned data. This object identifier is returned by calls to :meth:`binary_data_by_filter` as the ``last`` value. If provided, the server will return the next data entries after the last object identifier. :type last: str :param include_binary_data: 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). :type include_binary_data: bool :param count_only: Whether to return only the total count of entries. :type count_only: bool :param include_internal_data: Whether to return the internal data. Internal data is used for Viam-specific data ingestion, like cloud SLAM. Defaults to ``False``. :type include_internal_data: bool :param dest: Optional filepath for writing retrieved data. :type dest: str :returns: A tuple containing the following: - ``data`` (*List[* :class:`~viam.proto.app.data.BinaryData` *]*): The binary data. - ``count`` (*int*): The count (number of entries). - ``last`` (*str*): The last-returned page ID. :rtype: Tuple[List[~viam.proto.app.data.BinaryData], int, str] For more information, see `Data Client API `_. .. py:method:: binary_data_by_ids(binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]], dest: Optional[str] = None) -> List[viam.proto.app.data.BinaryData] :async: 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) :param binary_ids: Binary data ID strings specifying the desired data or :class:`BinaryID` objects. Must be non-empty. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_ids: Union[List[~viam.proto.app.data.BinaryID], List[str]] :param dest: Optional filepath for writing retrieved data. :type dest: str :raises GRPCError: If no binary data ID strings or :class:`BinaryID` objects are provided. :returns: The binary data. :rtype: List[~viam.proto.app.data.BinaryData] For more information, see `Data Client API `_. .. py:method:: delete_tabular_data(organization_id: str, delete_older_than_days: int) -> int :async: Delete tabular data older than a specified number of days. :: tabular_data = await data_client.delete_tabular_data( organization_id="", delete_older_than_days=150 ) :param organization_id: The ID of the organization to delete the data from. To find your organization ID, visit the organization settings page. :type organization_id: str :param delete_older_than_days: 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. :type delete_older_than_days: int :returns: The number of items deleted. :rtype: int For more information, see `Data Client API `_. .. py:method:: delete_tabular_data_by_filter(filter: Optional[viam.proto.app.data.Filter]) -> int :abstractmethod: :async: Deprecated: use :meth:`delete_tabular_data` instead. .. py:method:: delete_binary_data_by_filter(filter: Optional[viam.proto.app.data.Filter]) -> int :async: Filter and delete binary data. :: from viam.utils import create_filter my_filter = create_filter(component_name="left_motor", organization_ids=[""]) res = await data_client.delete_binary_data_by_filter(my_filter) :param filter: Optional, specifies binary data to delete. **CAUTION: Passing an empty** ``Filter`` **deletes all binary data!** You must specify an organization ID with ``organization_ids`` when using this option. To find your organization ID, visit the organization settings page. :type filter: ~viam.proto.app.data.Filter :returns: The number of items deleted. :rtype: int For more information, see `Data Client API `_. .. py:method:: delete_binary_data_by_ids(binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]]) -> int :async: 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=[""]) 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) :param binary_ids: Binary data ID strings specifying the data to be deleted or :class:`BinaryID` objects. Must be non-empty. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_ids: Union[List[~viam.proto.app.data.BinaryID], List[str]] :raises GRPCError: If no binary data ID strings or :class:`BinaryID` objects are provided. :returns: The number of items deleted. :rtype: int For more information, see `Data Client API `_. .. py:method:: add_tags_to_binary_data_by_ids(tags: List[str], binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]]) -> None :async: Add tags to binary data. :: from viam.utils import create_filter tags = ["tag1", "tag2"] my_filter = create_filter(component_name="camera-1", organization_ids=[""]) 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) :param tags: List of tags to add to specified binary data. Must be non-empty. :type tags: List[str] :param binary_ids: Binary data ID strings specifying the data to be tagged or :class:`BinaryID` objects. Must be non-empty. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_ids: Union[List[~viam.proto.app.data.BinaryID], List[str]] :raises GRPCError: If no binary data ID strings or :class:`BinaryID` objects are provided. For more information, see `Data Client API `_. .. py:method:: add_tags_to_binary_data_by_filter(tags: List[str], filter: Optional[viam.proto.app.data.Filter] = None) -> None :async: 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) :param tags: List of tags to add to specified binary data. Must be non-empty. :type tags: List[str] :param filter: Specifies binary data to tag. If none is provided, tags all data. :type filter: ~viam.proto.app.data.Filter :raises GRPCError: If no tags are provided. For more information, see `Data Client API `_. .. py:method:: remove_tags_from_binary_data_by_ids(tags: List[str], binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]]) -> int :async: 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) :param tags: List of tags to remove from specified binary data. Must be non-empty. :type tags: List[str] :param binary_ids: Binary data ID strings specifying the data to be untagged or `BinaryID` objects. Must be non-empty. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_ids: Union[List[~viam.proto.app.data.BinaryID], List[str]] :raises GRPCError: If no binary data ID strings, :class:`BinaryID` objects, or tags are provided. :returns: The number of tags removed. :rtype: int For more information, see `Data Client API `_. .. py:method:: remove_tags_from_binary_data_by_filter(tags: List[str], filter: Optional[viam.proto.app.data.Filter] = None) -> int :async: 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) :param tags: List of tags to remove from specified binary data. :type tags: List[str] :param filter: Specifies binary data to untag. If none is provided, removes tags from all data. :type filter: ~viam.proto.app.data.Filter :raises GRPCError: If no tags are provided. :returns: The number of tags removed. :rtype: int For more information, see `Data Client API `_. .. py:method:: tags_by_filter(filter: Optional[viam.proto.app.data.Filter] = None) -> List[str] :async: 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) :param filter: Specifies subset ofdata to retrieve tags from. If none is provided, returns all tags. :type filter: ~viam.proto.app.data.Filter :returns: The list of tags. :rtype: List[str] For more information, see `Data Client API `_. .. py:method:: add_bounding_box_to_image_by_id(binary_id: Union[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 :async: Add a bounding box to an image. :: bbox_id = await data_client.add_bounding_box_to_image_by_id( binary_id="", label="label", x_min_normalized=0, y_min_normalized=.1, x_max_normalized=.2, y_max_normalized=.3 ) print(bbox_id) :param binary_id: The binary data ID or :class:`BinaryID` of the image to add the bounding box to. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_id: Union[~viam.proto.app.data.BinaryID, str] :param label: A label for the bounding box. :type label: str :param x_min_normalized: Min X value of the bounding box normalized from 0 to 1. :type x_min_normalized: float :param y_min_normalized: Min Y value of the bounding box normalized from 0 to 1. :type y_min_normalized: float :param x_max_normalized: Max X value of the bounding box normalized from 0 to 1. :type x_max_normalized: float :param y_max_normalized: Max Y value of the bounding box normalized from 0 to 1. :type y_max_normalized: float :raises GRPCError: If the X or Y values are outside of the [0, 1] range. :returns: The bounding box ID. :rtype: str For more information, see `Data Client API `_. .. py:method:: remove_bounding_box_from_image_by_id(bbox_id: str, binary_id: Union[viam.proto.app.data.BinaryID, str]) -> None :async: Removes a bounding box from an image. :: await data_client.remove_bounding_box_from_image_by_id( binary_id="", bbox_id="your-bounding-box-id-to-delete" ) :param bbox_id: The ID of the bounding box to remove. :type bbox_id: str :param binary_id: The binary data ID or :class:`BinaryID` of the image to remove the bounding box from. *DEPRECATED:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_id: Union[~viam.proto.app.data.BinaryID, str] For more information, see `Data Client API `_. .. py:method:: bounding_box_labels_by_filter(filter: Optional[viam.proto.app.data.Filter] = None) -> List[str] :async: 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) :param filter: Specifies data to retrieve bounding box labels from. If none is provided, returns labels :type filter: ~viam.proto.app.data.Filter :param from all data.: :returns: The list of bounding box labels. :rtype: List[str] For more information, see `Data Client API `_. .. py:method:: get_database_connection(organization_id: str) -> str :async: Get a connection to access a MongoDB Atlas Data federation instance. :: hostname = await data_client.get_database_connection(organization_id="") :param organization_id: The ID of the organization you'd like to connect to. To find your organization ID, visit the organization settings page. :type organization_id: str :returns: The hostname of the federated database. :rtype: str For more information, see `Data Client API `_. .. py:method:: configure_database_user(organization_id: str, password: str) -> None :async: 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="", password="Your_Password@1234" ) :param organization_id: The ID of the organization you'd like to configure a database user for. To find your organization ID, visit the organization settings page. :type organization_id: str :param password: The password of the user. :type password: str For more information, see `Data Client API `_. .. py:method:: create_dataset(name: str, organization_id: str) -> str :async: Create a new dataset. :: dataset_id = await data_client.create_dataset( name="", organization_id="" ) print(dataset_id) :param name: The name of the dataset being created. :type name: str :param organization_id: The ID of the organization where the dataset is being created. To find your organization ID, visit the organization settings page. :type organization_id: str :returns: The dataset ID of the created dataset. :rtype: str For more information, see `Data Client API `_. .. py:method:: list_dataset_by_ids(ids: List[str]) -> Sequence[viam.proto.app.dataset.Dataset] :async: Get a list of datasets using their IDs. :: datasets = await data_client.list_dataset_by_ids( ids=[", "] ) print(datasets) :param ids: 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**. :type ids: List[str] :returns: The list of datasets. :rtype: Sequence[Dataset] For more information, see `Data Client API `_. .. py:method:: list_datasets_by_organization_id(organization_id: str) -> Sequence[viam.proto.app.dataset.Dataset] :async: Get the datasets in an organization. :: datasets = await data_client.list_datasets_by_organization_id( organization_id="" ) print(datasets) :param organization_id: The ID of the organization you'd like to retrieve datasets from. To find your organization ID, visit the organization settings page. :type organization_id: str :returns: The list of datasets in the organization. :rtype: Sequence[Dataset] For more information, see `Data Client API `_. .. py:method:: rename_dataset(id: str, name: str) -> None :async: Rename a dataset specified by the dataset ID. :: await data_client.rename_dataset( id="", name="MyDataset" ) :param id: 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**. :type id: str :param name: The new name of the dataset. :type name: str For more information, see `Data Client API `_. .. py:method:: delete_dataset(id: str) -> None :async: Delete a dataset. :: await data_client.delete_dataset( id="" ) :param id: 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**. :type id: str For more information, see `Data Client API `_. .. py:method:: add_binary_data_to_dataset_by_ids(binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]], dataset_id: str) -> None :async: 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" ) :param binary_ids: 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. :type binary_ids: List[~viam.proto.app.data.BinaryID] :param dataset_id: 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**. :type dataset_id: str For more information, see `Data Client API `_. .. py:method:: remove_binary_data_from_dataset_by_ids(binary_ids: Union[List[viam.proto.app.data.BinaryID], List[str]], dataset_id: str) -> None :async: 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" ) :param binary_ids: 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:* :class:`BinaryID` *is deprecated and will be removed in a future release. Instead, pass binary data IDs as a list of strings.* :type binary_ids: Union[List[~viam.proto.app.data.BinaryID], List[str]] :param dataset_id: 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**. :type dataset_id: str For more information, see `Data Client API `_. .. py:method:: binary_data_capture_upload(binary_data: bytes, part_id: str, component_type: str, component_name: str, method_name: str, file_extension: str, method_parameters: Optional[Mapping[str, Any]] = None, tags: Optional[List[str]] = None, data_request_times: Optional[Tuple[datetime.datetime, datetime.datetime]] = None) -> str :async: 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. :: 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" ) :param binary_data: The data to be uploaded, represented in bytes. :type binary_data: bytes :param part_id: Part ID of the component used to capture the data. :type part_id: str :param component_type: Type of the component used to capture the data (for example, "movement_sensor"). :type component_type: str :param component_name: Name of the component used to capture the data. :type component_name: str :param method_name: Name of the method used to capture the data. :type method_name: str :param file_extension: 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. :type file_extension: str :param method_parameters: Optional dictionary of method parameters. No longer in active use. :type method_parameters: Optional[Mapping[str, Any]] :param tags: Optional list of tags to allow for tag-based data filtering when retrieving data. :type tags: Optional[List[str]] :param data_request_times: Optional tuple containing datetime objects denoting the times this data was requested ``[0]`` by the robot and received ``[1]`` from the appropriate sensor. :type data_request_times: Optional[Tuple[datetime.datetime, datetime.datetime]] :raises GRPCError: If an invalid part ID is passed. :returns: The binary data ID of the uploaded data. :rtype: str For more information, see `Data Client API `_. .. py:method:: 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: Optional[Mapping[str, Any]] = None, tags: Optional[List[str]] = None) -> str :async: 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} } }] ) :param tabular_data: List of the data to be uploaded, represented tabularly as a collection of dictionaries. Must include the key ``readings`` for sensors. :type tabular_data: List[Mapping[str, Any]] :param part_id: Part ID of the component used to capture the data. :type part_id: str :param component_type: Type of the component used to capture the data (for example, ``rdk:component:movement_sensor``). :type component_type: str :param component_name: Name of the component used to capture the data. :type component_name: str :param method_name: Name of the method used to capture the data. :type method_name: str :param data_request_times: 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 length ``n > 1`` to upload ``n`` datapoints, all with the same metadata. :type data_request_times: List[Tuple[datetime.datetime, datetime.datetime]] :param method_parameters: Optional dictionary of method parameters. No longer in active use. :type method_parameters: Optional[Mapping[str, Any]] :param tags: Optional list of tags to allow for tag-based data filtering when retrieving data. :type tags: Optional[List[str]] :raises GRPCError: If an invalid part ID is passed. :raises 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. :rtype: str For more information, see `Data Client API `_. .. py:method:: streaming_data_capture_upload(data: bytes, part_id: str, file_ext: str, component_type: Optional[str] = None, component_name: Optional[str] = None, method_name: Optional[str] = None, method_parameters: Optional[Mapping[str, Any]] = None, data_request_times: Optional[Tuple[datetime.datetime, datetime.datetime]] = None, tags: Optional[List[str]] = None) -> str :async: 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"] ) :param data: The data to be uploaded. :type data: bytes :param part_id: Part ID of the resource associated with the file. :type part_id: str :param file_ext: File extension type for the data. required for determining MIME type. :type file_ext: str :param component_type: Optional type of the component associated with the file (for example, "movement_sensor"). :type component_type: Optional[str] :param component_name: Optional name of the component associated with the file. :type component_name: Optional[str] :param method_name: Optional name of the method associated with the file. :type method_name: Optional[str] :param method_parameters: Optional dictionary of the method parameters. No longer in active use. :type method_parameters: Optional[str] :param data_request_times: Optional tuple containing datetime objects denoting the times this data was requested ``[0]`` by the robot and received ``[1]`` from the appropriate sensor. :type data_request_times: Optional[Tuple[datetime.datetime, datetime.datetime]] :param tags: Optional list of tags to allow for tag-based filtering when retrieving data. :type tags: Optional[List[str]] :raises GRPCError: If an invalid part ID is passed. :returns: The binary data ID of the uploaded data. :rtype: str For more information, see `Data Client API `_. .. py:method:: file_upload(part_id: str, data: bytes, component_type: Optional[str] = None, component_name: Optional[str] = None, method_name: Optional[str] = None, file_name: Optional[str] = None, method_parameters: Optional[Mapping[str, Any]] = None, file_extension: Optional[str] = None, tags: Optional[List[str]] = None) -> str :async: 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" ) :param part_id: Part ID of the resource associated with the file. :type part_id: str :param data: Bytes representing file data to upload. :type data: bytes :param component_type: Optional type of the component associated with the file (for example, "movement_sensor"). :type component_type: Optional[str] :param component_name: Optional name of the component associated with the file. :type component_name: Optional[str] :param method_name: Optional name of the method associated with the file. :type method_name: Optional[str] :param file_name: Optional name of the file. The empty string ``""`` will be assigned as the file name if one isn't provided. :type file_name: Optional[str] :param method_parameters: Optional dictionary of the method parameters. No longer in active use. :type method_parameters: Optional[str] :param file_extension: 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. :type file_extension: Optional[str] :param tags: Optional list of tags to allow for tag-based filtering when retrieving data. :type tags: Optional[List[str]] :raises GRPCError: If an invalid part ID is passed. :returns: Binary data ID of the new file. :rtype: str For more information, see `Data Client API `_. .. py:method:: file_upload_from_path(filepath: str, part_id: str, component_type: Optional[str] = None, component_name: Optional[str] = None, method_name: Optional[str] = None, method_parameters: Optional[Mapping[str, Any]] = None, tags: Optional[List[str]] = None) -> str :async: 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///" ) :param filepath: Absolute filepath of file to be uploaded. :type filepath: str :param part_id: Part ID of the component associated with the file. :type part_id: str :param component_type: Optional type of the component associated with the file (for example, "movement_sensor"). :type component_type: Optional[str] :param component_name: Optional name of the component associated with the file. :type component_name: Optional[str] :param method_name: Optional name of the method associated with the file. :type method_name: Optional[str] :param method_parameters: Optional dictionary of the method parameters. No longer in active use. :type method_parameters: Optional[str] :param tags: Optional list of tags to allow for tag-based filtering when retrieving data. :type tags: Optional[List[str]] :raises GRPCError: If an invalid part ID is passed. :raises FileNotFoundError: If the provided filepath is not found. :returns: Binary data ID of the new file. :rtype: str For more information, see `Data Client API `_. .. py:method:: get_data_pipeline(id: str) -> DataPipeline :async: Get a data pipeline by its ID. :: data_pipeline = await data_client.get_data_pipeline(id="") :param id: The ID of the data pipeline to get. :type id: str :returns: The data pipeline with the given ID. :rtype: DataPipeline .. py:method:: list_data_pipelines(organization_id: str) -> List[DataPipeline] :async: List all of the data pipelines for an organization. :: data_pipelines = await data_client.list_data_pipelines(organization_id="") :param organization_id: The ID of the organization that owns the pipelines. You can obtain your organization ID from the organization settings page. :type organization_id: str :returns: A list of all of the data pipelines for the given organization. :rtype: List[DataPipeline] .. py:method:: create_data_pipeline(organization_id: str, name: str, mql_binary: List[Dict[str, Any]], schedule: str, data_source_type: viam.proto.app.data.TabularDataSourceType.ValueType = TabularDataSourceType.TABULAR_DATA_SOURCE_TYPE_STANDARD) -> str :async: Create a new data pipeline. :: data_pipeline_id = await data_client.create_data_pipeline( organization_id="", name="", mql_binary=[], schedule="", data_source_type=TabularDataSourceType.TABULAR_DATA_SOURCE_TYPE_STANDARD, ) :param organization_id: The ID of the organization that will own the pipeline. You can obtain your organization ID from the organization settings page. :type organization_id: str :param name: The name of the pipeline. :type name: str :param mql_binary: The MQL pipeline to run, as a list of MongoDB aggregation pipeline stages. :type mql_binary: List[Dict[str, Any]] :param schedule: A cron expression representing the expected execution schedule in UTC (note this also defines the input time window; an hourly schedule would process 1 hour of data at a time). :type schedule: str :param data_source_type: The type of data source to use for the pipeline. Defaults to TabularDataSourceType.TABULAR_DATA_SOURCE_TYPE_STANDARD. :type data_source_type: TabularDataSourceType :returns: The ID of the newly created pipeline. :rtype: str .. py:method:: delete_data_pipeline(id: str) -> None :async: Delete a data pipeline by its ID. :: await data_client.delete_data_pipeline(id="") :param id: The ID of the data pipeline to delete. :type id: str .. py:method:: list_data_pipeline_runs(id: str, page_size: int = 10) -> DataPipelineRunsPage :async: List all of the data pipeline runs for a data pipeline. :: data_pipeline_runs = await data_client.list_data_pipeline_runs(id="") while len(data_pipeline_runs.runs) > 0: data_pipeline_runs = await data_pipeline_runs.next_page() :param id: The ID of the pipeline to list runs for :type id: str :param page_size: The number of runs to return per page. Defaults to 10. :type page_size: int :returns: A page of data pipeline runs with pagination support :rtype: DataPipelineRunsPage .. py:method:: create_filter(component_name: Optional[str] = None, component_type: Optional[str] = None, method: Optional[str] = None, robot_name: Optional[str] = None, robot_id: Optional[str] = None, part_name: Optional[str] = None, part_id: Optional[str] = None, location_ids: Optional[List[str]] = None, organization_ids: Optional[List[str]] = None, mime_type: Optional[List[str]] = None, start_time: Optional[datetime.datetime] = None, end_time: Optional[datetime.datetime] = None, tags: Optional[List[str]] = None, bbox_labels: Optional[List[str]] = None, dataset_id: Optional[str] = None) -> viam.proto.app.data.Filter :staticmethod: