:py:mod:`viam.app.data_client` ============================== .. py:module:: viam.app.data_client Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: viam.app.data_client.DataClient Attributes ~~~~~~~~~~ .. autoapisummary:: viam.app.data_client.LOGGER .. py:data:: LOGGER .. py:class:: DataClient(channel: grpclib.client.Channel, metadata: Mapping[str, str]) 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 "" (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()) .. 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 Return str(self). .. py:method:: __eq__(other: object) -> bool Return self==value. .. 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, 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) :param filter: Optional `Filter` specifying tabular data to retrieve. No `Filter` implies 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 ID of the last-returned data. If provided, the server will return the next data entries after the `last` ID. :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: The tabular data. int: The count (number of entries) str: The last-returned page ID. :rtype: List[TabularData] .. py:method:: tabular_data_by_sql(organization_id: str, sql_query: str) -> List[Dict[str, viam.utils.ValueTypes]] :async: Obtain unified tabular data and metadata, queried with SQL. :: data = await data_client.tabular_data_by_sql(org_id="", sql_query="SELECT * FROM readings LIMIT 5") :param organization_id: The ID of the organization that owns the data. :type organization_id: str :param sql_query: The SQL query to run. :type sql_query: str :returns: An array of data objects. :rtype: List[Dict[str, ValueTypes]] .. py:method:: tabular_data_by_mql(organization_id: str, mql_binary: List[bytes]) -> List[Dict[str, viam.utils.ValueTypes]] :async: Obtain unified tabular data and metadata, queried with MQL. :: # using bson import bson tabular_data = await data_client.tabular_data_by_mql(org_id="", mql_binary=[ bson.dumps({ '$match': { 'location_id': '' } }), bson.dumps({ "$limit": 5 }) ]) # using pymongo import bson tabular_data = await data_client.tabular_data_by_mql(org_id="", mql_binary=[ bson.encode({ '$match': { 'location_id': '' } }), bson.encode({ "$limit": 5 }) ]) :param organization_id: The ID of the organization that owns the data. :type organization_id: str :param mql_binary: The MQL query to run as a list of BSON queries. You can encode your bson queries using a library like `pymongo` or `bson`. :type mql_binary: List[bytes] :returns: An array of data objects. :rtype: List[Dict[str, ValueTypes]] .. 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. 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) :param filter: Optional `Filter` specifying tabular data to retrieve. No `Filter` implies 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 ID of the last-returned data. If provided, the server will return the next data entries after the `last` ID. :type last: str :param include_binary_data: 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). :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: The binary data. int: The count (number of entries) str: The last-returned page ID. :rtype: List[viam.proto.app.data.BinaryData] .. py:method:: binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dest: Optional[str] = None) -> List[viam.proto.app.data.BinaryData] :async: 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) :param binary_ids: `BinaryID` objects specifying the desired data. Must be non-empty. :type binary_ids: List[viam.proto.app.data.BinaryID] :param dest: Optional filepath for writing retrieved data. :type dest: str :raises GRPCError: If no `BinaryID` objects are provided. :returns: The binary data. :rtype: List[viam.proto.app.data.BinaryData] .. 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. :: 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) :param organization_id: ID of organization to delete data from. :type organization_id: str :param delete_older_than_days: 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. :type delete_older_than_days: int :returns: The number of items deleted. :rtype: int .. py:method:: delete_tabular_data_by_filter(filter: Optional[viam.proto.app.data.Filter]) -> int :abstractmethod: :async: Deprecated: use 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.proto.app.data import Filter my_filter = Filter(component_name="left_motor") res = await data_client.delete_binary_data_by_filter(my_filter) :param 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. :type filter: viam.proto.app.data.Filter :returns: The number of items deleted. :rtype: int .. py:method:: delete_binary_data_by_ids(binary_ids: List[viam.proto.app.data.BinaryID]) -> int :async: 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) :param binary_ids: `BinaryID` objects specifying the data to be deleted. Must be non-empty. :type binary_ids: List[viam.proto.app.data.BinaryID] :raises GRPCError: If no `BinaryID` objects are provided. :returns: The number of items deleted. :rtype: int .. py:method:: add_tags_to_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID]) -> None :async: 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) :param tags: List of tags to add to specified binary data. Must be non-empty. :type tags: List[str] :param binary_ids: List of `BinaryID` objects specifying binary data to tag. Must be non-empty. :type binary_ids: List[viam.app.proto.BinaryID] :raises GRPCError: If no `BinaryID` objects or tags are provided. .. 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.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) :param tags: List of tags to add to specified binary data. Must be non-empty. :type tags: List[str] :param filter: `Filter` specifying binary data to tag. If no `Filter` is provided, all data will be tagged. :type filter: viam.proto.app.data.Filter :raises GRPCError: If no tags are provided. .. py:method:: remove_tags_from_binary_data_by_ids(tags: List[str], binary_ids: List[viam.proto.app.data.BinaryID]) -> int :async: 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) :param tags: List of tags to remove from specified binary data. Must be non-empty. :type tags: List[str] :param binary_ids: List of `BinaryID` objects specifying binary data to untag. Must be non-empty. :type binary_ids: List[BinaryID] :raises GRPCError: If no binary_ids or tags are provided. :returns: The number of tags removed. :rtype: int .. 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.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) :param tags: List of tags to remove from specified binary data. :type tags: List[str] :param filter: `Filter` specifying binary data to untag. If no `Filter` is provided, all data will be untagged. :type filter: viam.proto.app.data.Filter :raises GRPCError: If no tags are provided. :returns: The number of tags removed. :rtype: int .. 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.proto.app.data import Filter my_filter = Filter(component_name="my_camera") tags = await data_client.tags_by_filter(my_filter) :param filter: `Filter` specifying data to retrieve from. If no `Filter` is provided, all data tags will return. :type filter: viam.proto.app.data.Filter :returns: The list of tags. :rtype: List[str] .. py:method:: 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 :async: 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) :param binary_id: The ID of the image to add the bounding box to. :type binary_id: viam.proto.app.data.BinaryID :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 .. py:method:: remove_bounding_box_from_image_by_id(bbox_id: str, binary_id: viam.proto.app.data.BinaryID) -> None :async: 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" ) :param bbox_id: The ID of the bounding box to remove. :type bbox_id: str :param Binary_id: Binary ID of the image to to remove the bounding box from :type Binary_id: viam.proto.arr.data.BinaryID .. 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.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) :param filter: `Filter` specifying data to retrieve from. If no `Filter` is provided, all labels will return. :type filter: viam.proto.app.data.Filter :returns: The list of bounding box labels. :rtype: List[str] .. py:method:: get_database_connection(organization_id: str) -> str :async: Get a connection to access a MongoDB Atlas Data federation instance. :: data_client.get_database_connection(org_id="a12b3c4e-1234-1abc-ab1c-ab1c2d345abc") :param organization_id: Organization to retrieve the connection for. :type organization_id: str :returns: The hostname of the federated database. :rtype: str .. 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. :param organization_id: The ID of the organization. :type organization_id: str :param password: The password of the user. :type password: str .. py:method:: create_dataset(name: str, organization_id: str) -> str :async: Create a new dataset. :: name = await data_client.create_dataset( name="", organization_id="" ) print(name) :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. :type organization_id: str :returns: The dataset ID of the created dataset. :rtype: str .. 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=["abcd-1234xyz-8765z-123abc"] ) print(datasets) :param ids: The IDs of the datasets being called for. :type ids: List[str] :returns: The list of datasets. :rtype: Sequence[Dataset] .. 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_dataset_by_ids( ids=["abcd-1234xyz-8765z-123abc"] ) print(datasets) :param organization_id: The ID of the organization. :type organization_id: str :returns: The list of datasets in the organization. :rtype: Sequence[Dataset] .. py:method:: rename_dataset(id: str, name: str) -> None :async: Rename a dataset specified by the dataset ID. :: await data_client.rename_dataset( id="abcd-1234xyz-8765z-123abc", name="" ) :param id: The ID of the dataset. :type id: str :param name: The new name of the dataset. :type name: str .. py:method:: delete_dataset(id: str) -> None :async: Delete a dataset. :: await data_client.delete_dataset( id="abcd-1234xyz-8765z-123abc" ) :param id: The ID of the dataset. :type id: str .. py:method:: add_binary_data_to_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dataset_id: str) -> None :async: 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" ) :param binary_ids: The IDs of binary data to add to dataset. :type binary_ids: List[BinaryID] :param dataset_id: The ID of the dataset to be added to. :type dataset_id: str .. py:method:: remove_binary_data_from_dataset_by_ids(binary_ids: List[viam.proto.app.data.BinaryID], dataset_id: str) -> None :async: 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" ) :param binary_ids: The IDs of binary data to remove from dataset. :type binary_ids: List[BinaryID] :param dataset_id: The ID of the dataset to be removed from. :type dataset_id: str .. 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 (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" ) :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 (e.g., "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, 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. :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`s 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 file_id of the uploaded data. :rtype: str .. py:method:: tabular_data_capture_upload(tabular_data: List[Mapping[str, Any]], part_id: str, component_type: str, component_name: str, method_name: str, method_parameters: Optional[Mapping[str, Any]] = None, tags: Optional[List[str]] = None, data_request_times: Optional[List[Tuple[datetime.datetime, datetime.datetime]]] = None) -> str :async: 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}] ) :param tabular_data: List of the data to be uploaded, represented tabularly as a collection of dictionaries. :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 (e.g., "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 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 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. :type data_request_times: Optional[List[Tuple[datetime.datetime, datetime.datetime]]] :raises GRPCError: If an invalid part ID is passed. :raises 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. :rtype: str .. 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 (e.g., "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`s 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 file_id of the uploaded data. :rtype: str .. 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 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" ) :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 (e.g., "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: ID of the new file. :rtype: str .. 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 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///" ) :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 (e.g., "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: ID of the new file. :rtype: str .. 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: