from collections.abc import Generator
from dataclasses import asdict
from datetime import datetime
from chariot import _apis, mcp_setting
from chariot.datasets import _utils, models
from chariot_api._openapi.datasets_v3 import models as openapi_models
__all__ = [
"get_datum",
"get_dataset_datums",
"get_dataset_datum_statistics",
"get_dataset_datum_count",
"get_dataset_datum_labels",
"get_snapshot_datums",
"get_snapshot_datum_statistics",
"get_snapshot_datum_count",
"get_snapshot_datum_labels",
"get_upload_datums",
"archive_datum",
]
def _create_get_datums_request(
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: models.SplitName | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> openapi_models.InputGetDatumsRequest:
if task_type_label_filters is not None:
task_type_label_filters = [
asdict(f, dict_factory=_utils.dict_factory) for f in task_type_label_filters
]
if gps_coordinates_circle is not None:
gps_coordinates_circle = asdict(gps_coordinates_circle, dict_factory=_utils.dict_factory)
if gps_coordinates_rectangle is not None:
gps_coordinates_rectangle = asdict(
gps_coordinates_rectangle, dict_factory=_utils.dict_factory
)
if gps_coordinates_polygon is not None:
gps_coordinates_polygon = [
asdict(p, dict_factory=_utils.dict_factory) for p in gps_coordinates_polygon
]
return openapi_models.InputGetDatumsRequest(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range._to_post_body()
if capture_timestamp_range
else None,
metadata=metadata,
metadata_filters=[f._to_post_body() for f in metadata_filters]
if metadata_filters
else None,
split=_utils.enum_value(split),
asof_timestamp=_utils.format_datetime_utc(asof_timestamp) if asof_timestamp else None,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=_utils.enum_value(sort),
direction=_utils.enum_value(direction),
)
def _create_get_datum_labels_request(
task_type_label_filter: models.TaskTypeLabelFilter | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: models.SplitName | None = None,
asof_timestamp: datetime | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> openapi_models.InputGetDatumLabelsRequest:
task_type_label_filters = None
if task_type_label_filter is not None:
task_type_label_filters = [asdict(task_type_label_filter, dict_factory=_utils.dict_factory)]
if gps_coordinates_circle is not None:
gps_coordinates_circle = asdict(gps_coordinates_circle, dict_factory=_utils.dict_factory)
if gps_coordinates_rectangle is not None:
gps_coordinates_rectangle = asdict(
gps_coordinates_rectangle, dict_factory=_utils.dict_factory
)
if gps_coordinates_polygon is not None:
gps_coordinates_polygon = [
asdict(p, dict_factory=_utils.dict_factory) for p in gps_coordinates_polygon
]
return openapi_models.InputGetDatumLabelsRequest(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range._to_post_body()
if capture_timestamp_range
else None,
metadata=metadata,
metadata_filters=[f._to_post_body() for f in metadata_filters]
if metadata_filters
else None,
split=_utils.enum_value(split),
asof_timestamp=_utils.format_datetime_utc(asof_timestamp) if asof_timestamp else None,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=_utils.enum_value(sort),
direction=_utils.enum_value(direction),
)
def _create_get_datum_stats_request(
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
) -> openapi_models.InputGetDatumStatsRequest:
if task_type_label_filters is not None:
task_type_label_filters = [
asdict(f, dict_factory=_utils.dict_factory) for f in task_type_label_filters
]
if gps_coordinates_circle is not None:
gps_coordinates_circle = asdict(gps_coordinates_circle, dict_factory=_utils.dict_factory)
if gps_coordinates_rectangle is not None:
gps_coordinates_rectangle = asdict(
gps_coordinates_rectangle, dict_factory=_utils.dict_factory
)
if gps_coordinates_polygon is not None:
gps_coordinates_polygon = [
asdict(p, dict_factory=_utils.dict_factory) for p in gps_coordinates_polygon
]
return openapi_models.InputGetDatumStatsRequest(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range._to_post_body()
if capture_timestamp_range
else None,
metadata=metadata,
metadata_filters=[f._to_post_body() for f in metadata_filters]
if metadata_filters
else None,
split=_utils.enum_value(split),
asof_timestamp=_utils.format_datetime_utc(asof_timestamp) if asof_timestamp else None,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
[docs]
def get_datum(id: str, *, task_type: models.TaskType | None = None) -> models.Datum:
"""Get a datum by id
:param id: Id of datum to get
:type id: str
:param task_type: Task type annotation filter
:type task_type: Optional[models.TaskType]
:return: Datum details
:rtype: models.Datum
"""
response = _apis.datasets_v3.datums_api.get_datum(
datum_id=id, task_type=_utils.enum_value(task_type)
)
if not response.data:
raise RuntimeError("Received malformed response (missing `data`) from get_datum")
return _utils.convert_to_dataclass(response.data.model_dump(), models.Datum)
[docs]
def get_dataset_datums(
dataset_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
max_items: int | None = None,
) -> Generator[models.Datum, None, None]:
"""Get dataset datums with various criteria
:param dataset_id: Id of dataset to get datums for
:type dataset_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filter by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param asof_timestamp: Filter datums and/or annotations at the timestamp
:type asof_timestamp: Optional[datetime]
:param unannotated: Filter datums without annotation
:type unannotated: Optional[bool]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param max_items: Limit the returned generator to only produce this many items
:type max_items: Optional[int]
:return: Generator over the matching datums
:rtype: Generator[models.Datum, None, None]
"""
params = locals()
if "max_items" in params:
del params["max_items"]
if datum_ids:
return iter(
_get_dataset_datums(
dataset_id,
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
)
else:
return _utils.paginate_items(_get_dataset_datums, params, max_items, True)
def _get_dataset_datums(
dataset_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> list[models.Datum]:
request = _create_get_datums_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=sort,
direction=direction,
)
response = _apis.datasets_v3.datums_api.get_dataset_datums(dataset_id=dataset_id, body=request)
if not response.data:
return []
return [_utils.convert_to_dataclass(d.model_dump(), models.Datum) for d in response.data]
[docs]
def get_dataset_datum_statistics(
dataset_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
) -> models.DatumStatistics:
"""Get dataset datum statistics with various criteria
:param dataset_id: Id of dataset to get datums for
:type dataset_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param asof_timestamp: Filter datums and/or annotations at the timestamp
:type asof_timestamp: Optional[datetime]
:param unannotated: Filter datums without annotation
:type unannotated: Optional[bool]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:return: Datum statistics for matching datums
:rtype: models.DatumStatistics
"""
request = _create_get_datum_stats_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
response = _apis.datasets_v3.datums_api.get_dataset_datum_statistics(
dataset_id=dataset_id, body=request
)
if not response.data:
raise RuntimeError(
"Received malformed response (missing `data`) from get_dataset_datum_statistics"
)
return _utils.convert_to_dataclass(response.data.model_dump(), models.DatumStatistics)
[docs]
def get_dataset_datum_count(
dataset_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
datum_id_before: str = "",
sort: models.DatumSortColumn | None = models.DatumSortColumn.ID,
direction: models.SortDirection | None = models.SortDirection.DESCENDING,
sort_metadata_key_path: list[str] | None = None,
sort_metadata_type: models.MetadataType | None = None,
) -> int:
"""Get dataset datum count with various criteria.
:param dataset_id: Id of dataset to get datums for
:type dataset_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param asof_timestamp: Filter datums and/or annotations at the timestamp
:type asof_timestamp: Optional[datetime]
:param unannotated: Filter datums without annotation
:type unannotated: Optional[bool]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param datum_id_before: Filter datums with ids before the provided id
:type datum_id_before: str
:param sort: sort by column (defaults to "id")
:type sort: Optional[model.DatumSortColumn]
:param direction: sort by direction (defaults to "asc")
:type direction: Optional[models.SortDirection]
:param sort_metadata_key_path: metadata key path if sort datum by metadata value
:type sort_metadata_key_path: Optional[List[str]]
:param sort_metadata_type: metadata type if sort datum by metadata value
:type sort_metadata_type: Optional[models.MetadataType]
:return: Datum count for matching datums
:rtype: int
"""
request = _create_get_datum_stats_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
request_count = openapi_models.InputGetDatumCountRequest(
datum_id_before=datum_id_before,
sort=_utils.enum_value(sort),
sort_metadata_key_path=sort_metadata_key_path,
sort_metadata_type=_utils.enum_value(sort_metadata_type),
direction=_utils.enum_value(direction),
**request.to_dict(),
)
response = _apis.datasets_v3.datums_api.get_dataset_datum_count(
dataset_id=dataset_id, body=request_count
)
return response.data or 0
[docs]
def get_dataset_datum_labels(
dataset_id: str,
*,
task_type_label_filter: models.TaskTypeLabelFilter | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
max_items: int | None = None,
) -> Generator[str, None, None]:
"""Get dataset datum labels with various criteria
:param dataset_id: Id of dataset to get datums for
:type dataset_id: str
:param task_type_label_filter: Filter by a single task type and associated labels
:type task_type_label_filter: Optional[models.TaskTypeLabelFilter]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param asof_timestamp: Filter datums and/or annotations at the timestamp
:type asof_timestamp: Optional[datetime]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param max_items: Limit the returned generator to only produce this many items
:type max_items: Optional[int]
:return: Generator over the matching datum labels
:rtype: Generator[str, None, None]
"""
params = locals()
if "max_items" in params:
del params["max_items"]
if datum_ids:
return iter(
_get_dataset_datum_labels(
dataset_id,
task_type_label_filter=task_type_label_filter,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
)
else:
return _utils.paginate_strings(_get_dataset_datum_labels, params, max_items, True)
def _get_dataset_datum_labels(
dataset_id: str,
*,
task_type_label_filter: models.TaskTypeLabelFilter | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
asof_timestamp: datetime | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> list[str]:
request = _create_get_datum_labels_request(
task_type_label_filter=task_type_label_filter,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
asof_timestamp=asof_timestamp,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=sort,
direction=direction,
)
response = _apis.datasets_v3.datums_api.get_dataset_datum_labels(
dataset_id=dataset_id, body=request
)
if not response.data:
return []
return response.data
[docs]
def get_snapshot_datums(
snapshot_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
max_items: int | None = None,
) -> Generator[models.Datum, None, None]:
"""Get snapshot datums with various criteria
:param snapshot_id: Id of snapshot to get datums for
:type snapshot_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param split: Filter by datum split assignment
:type split: Optional[str]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param max_items: Limit the returned generator to only produce this many items
:type max_items: Optional[int]
:return: Generator over the matching datums
:rtype: Generator[models.Datum, None, None]
"""
params = locals()
if "max_items" in params:
del params["max_items"]
if datum_ids:
return iter(
_get_snapshot_datums(
snapshot_id,
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
)
else:
return _utils.paginate_items(_get_snapshot_datums, params, max_items, True)
def _get_snapshot_datums(
snapshot_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> list[models.Datum]:
request = _create_get_datums_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=sort,
direction=direction,
)
response = _apis.datasets_v3.datums_api.get_snapshot_datums(
snapshot_id=snapshot_id, body=request
)
if not response.data:
return []
return [_utils.convert_to_dataclass(d.model_dump(), models.Datum) for d in response.data]
[docs]
def get_snapshot_datum_statistics(
snapshot_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
) -> models.DatumStatistics:
"""Get snapshot datum statistics with various criteria
:param snapshot_id: Id of snapshot to get datums for
:type snapshot_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param split: Filter by datum split assignment
:type split: Optional[str]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:return: Datum statistics for matching datums
:rtype: models.DatumStatistics
"""
request = _create_get_datum_stats_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
response = _apis.datasets_v3.datums_api.get_snapshot_datum_statistics(
snapshot_id=snapshot_id, body=request
)
if not response.data:
raise RuntimeError(
"Received malformed response (missing `data`) from get_snapshot_datum_statistics"
)
return _utils.convert_to_dataclass(response.data.model_dump(), models.DatumStatistics)
[docs]
def get_snapshot_datum_count(
snapshot_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
datum_id_before: str = "",
sort: models.DatumSortColumn | None = models.DatumSortColumn.ID,
direction: models.SortDirection | None = models.SortDirection.DESCENDING,
sort_metadata_key_path: list[str] | None = None,
sort_metadata_type: models.MetadataType | None = None,
) -> int:
"""Get snapshot datum statistics with various criteria
:param snapshot_id: Id of snapshot to get datums for
:type snapshot_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param split: Filter by datum split assignment
:type split: Optional[str]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param datum_id_before: Filter datums with ids before the provided id
:type datum_id_before: str
:param sort: sort by column (defaults to "id")
:type sort: Optional[model.DatumSortColumn]
:param direction: sort by direction (defaults to "desc")
:type direction: Optional[models.SortDirection]
:param sort_metadata_key_path: metadata key path if sort datum by metadata value
:type sort_metadata_key_path: Optional[List[str]]
:param sort_metadata_type: metadata type if sort datum by metadata value
:type sort_metadata_type: Optional[models.MetadataType]
:return: Datum count for matching datums
:rtype: int
"""
request = _create_get_datum_stats_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
request_count = openapi_models.InputGetDatumCountRequest(
datum_id_before=datum_id_before,
sort=_utils.enum_value(sort),
sort_metadata_key_path=sort_metadata_key_path,
sort_metadata_type=_utils.enum_value(sort_metadata_type),
direction=_utils.enum_value(direction),
**request.to_dict(),
)
response = _apis.datasets_v3.datums_api.get_snapshot_datum_count(
snapshot_id=snapshot_id, body=request_count
)
return response.data or 0
[docs]
def get_snapshot_datum_labels(
snapshot_id: str,
*,
task_type_label_filter: models.TaskTypeLabelFilter | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
max_items: int | None = None,
) -> Generator[str, None, None]:
"""Get snapshot datum labels with various criteria
:param snapshot_id: Id of snapshot to get datums for
:type snapshot_id: str
:param task_type_label_filter: Filter by a single task type and associated labels
:type task_type_label_filter: Optional[models.TaskTypeLabelFilter]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param split: Filter by datum split assignment
:type split: Optional[str]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param max_items: Limit the returned generator to only produce this many items
:type max_items: Optional[int]
:return: Generator over the matching datum labels
:rtype: Generator[str, None, None]
"""
params = locals()
if "max_items" in params:
del params["max_items"]
if datum_ids:
return iter(
_get_snapshot_datum_labels(
snapshot_id,
task_type_label_filter=task_type_label_filter,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
)
else:
return _utils.paginate_strings(_get_snapshot_datum_labels, params, max_items, True)
def _get_snapshot_datum_labels(
snapshot_id: str,
*,
task_type_label_filter: models.TaskTypeLabelFilter | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
split: str | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> list[str]:
request = _create_get_datum_labels_request(
task_type_label_filter=task_type_label_filter,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
split=split,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=sort,
direction=direction,
)
response = _apis.datasets_v3.datums_api.get_snapshot_datum_labels(
snapshot_id=snapshot_id, body=request
)
if not response.data:
return []
return response.data
[docs]
def get_upload_datums(
upload_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
max_items: int | None = None,
) -> Generator[models.Datum, None, None]:
"""Get upload datums with various criteria
:param upload_id: Id of upload to get datums for
:type upload_id: str
:param task_type_label_filters: Filter by task types and associated labels
:type task_type_label_filters: Optional[List[models.TaskTypeLabelFilter]]
:param gps_coordinates_circle: Filter datums within the given circle
:type gps_coordinates_circle: Optional[models.Circle]
:param gps_coordinates_rectangle: Filter datums within the given rectangle
:type gps_coordinates_rectangle: Optional[models.Rectangle]
:param gps_coordinates_polygon: Filter datums within the given polygon
:type gps_coordinates_polygon: Optional[List[models.GeoPoint]]
:param capture_timestamp_range: Filter by datum capture timestamp
:type capture_timestamp_range: Optional[models.TimestampRange]
:param metadata: obsolete, use metadata_filters instead
:type metadata: Optional[Dict[str, str]]
:param metadata_filters: filters by datum metadata values
:type metadata_filters: Optional[List[models.MetadataFilter]]
:param unannotated: Filter datums without annotation
:type unannotated: Optional[bool]
:param datum_ids: Filter datums with a list of datum ids
:type datum_ids: Optional[List[str]]
:param approval_status: Filter by annotation approval status
:type approval_status: Optional[List[str]]
:param annotation_metadata: Filter by annotation metadata values
:type annotation_metadata: Optional[Dict[str, str]]
:param max_items: Limit the returned generator to only produce this many items
:type max_items: Optional[int]
:return: Generator over the matching datums
:rtype: Generator[models.Datum, None, None]
"""
params = locals()
if "max_items" in params:
del params["max_items"]
if datum_ids:
return iter(
_get_upload_datums(
upload_id,
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
)
return _utils.paginate_items(_get_upload_datums, params, max_items, True)
def _get_upload_datums(
upload_id: str,
*,
task_type_label_filters: list[models.TaskTypeLabelFilter] | None = None,
gps_coordinates_circle: models.Circle | None = None,
gps_coordinates_rectangle: models.Rectangle | None = None,
gps_coordinates_polygon: list[models.GeoPoint] | None = None,
capture_timestamp_range: models.TimestampRange | None = None,
metadata: dict[str, str] | None = None,
metadata_filters: list[models.MetadataFilter] | None = None,
unannotated: bool | None = None,
datum_ids: list[str] | None = None,
approval_status: list[str] | None = None,
annotation_metadata: dict[str, str] | None = None,
limit: int | None = None,
offset: int | None = None,
id_after: str | None = None,
sort: models.DatumSortColumn | None = None,
direction: models.SortDirection | None = None,
) -> list[models.Datum]:
request = _create_get_datums_request(
task_type_label_filters=task_type_label_filters,
gps_coordinates_circle=gps_coordinates_circle,
gps_coordinates_rectangle=gps_coordinates_rectangle,
gps_coordinates_polygon=gps_coordinates_polygon,
capture_timestamp_range=capture_timestamp_range,
metadata=metadata,
metadata_filters=metadata_filters,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
limit=limit,
offset=offset,
id_after=id_after,
sort=sort,
direction=direction,
)
response = _apis.datasets_v3.datums_api.get_upload_datums(upload_id=upload_id, body=request)
if not response.data:
return []
return [_utils.convert_to_dataclass(d.model_dump(), models.Datum) for d in response.data]
[docs]
@mcp_setting(mutating=True)
def archive_datum(id: str) -> models.Datum:
"""Archive (soft-delete) a datum by id
:param id: Id of datum to archive
:type id: str
:return: Datum details
:rtype: models.Datum
"""
response = _apis.datasets_v3.datums_api.archive_datum(datum_id=id)
if not response.data:
raise RuntimeError("Received malformed response (missing `data`) from archive_datum")
return _utils.convert_to_dataclass(response.data.model_dump(), models.Datum)