from collections.abc import Generator
from dataclasses import asdict
from datetime import datetime
from chariot import _apis
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_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,
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
]
if capture_timestamp_range is not None:
capture_timestamp_range = asdict(capture_timestamp_range, dict_factory=_utils.dict_factory)
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,
metadata=metadata,
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,
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
]
if capture_timestamp_range is not None:
capture_timestamp_range = asdict(capture_timestamp_range, dict_factory=_utils.dict_factory)
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,
metadata=metadata,
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,
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
]
if capture_timestamp_range is not None:
capture_timestamp_range = asdict(capture_timestamp_range, dict_factory=_utils.dict_factory)
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,
metadata=metadata,
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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
) -> 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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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 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,
asof_timestamp=asof_timestamp,
unannotated=unannotated,
datum_ids=datum_ids,
approval_status=approval_status,
annotation_metadata=annotation_metadata,
)
request_count = openapi_models.InputGetDatumCountRequest(**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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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_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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
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,
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: Filter by datum metadata values
:type metadata: Optional[Dict[str, str]]
: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,
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,
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,
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]
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)