from collections.abc import Generator
from dataclasses import fields
from chariot import _apis
from chariot.datasets import _utils, models
from chariot_api._openapi.datasets_v3 import models as openapi_models
__all__ = [
"get_video_reference",
"get_video_references",
]
def _video_reference_from_openapi(data) -> models.VideoReference:
raw = data.model_dump()
names = {f.name for f in fields(models.VideoReference)}
return _utils.convert_to_dataclass(
{k: v for k, v in raw.items() if k in names},
models.VideoReference,
)
[docs]
def get_video_reference(id: str) -> models.VideoReference:
"""Retrieves a VideoReference record by its unique identifier.
A VideoReference is a metadata record that tracks a specific video file's properties (e.g., FPS, resolution, duration).
It serves as the link between the video service record and the Dataset/Datum entities that utilize the video.
:param id: The unique identified of the video reference.
:type id: str
:return: The metadata object containing video properties.
:rtype: models.VideoReference
"""
response = _apis.datasets_v3.videos_api.get_video_reference(video_reference_id=id)
if not response.data:
raise RuntimeError("Received malformed response (missing `data`) from get_video_reference")
return _video_reference_from_openapi(response.data)
def _get_video_references(
*,
project_ids: list[str] | None = None,
dataset_id: str | None = None,
datum_id: str | None = None,
has_klv: bool | None = None,
codec: str | None = None,
video_container_format: str | None = None,
min_duration_ms: int | None = None,
sort: str | None = None,
direction: models.SortDirection | None = None,
limit: int | None = None,
offset: int | None = None,
) -> list[models.VideoReference]:
request = openapi_models.InputGetVideoReferencesRequest(
project_ids=project_ids,
dataset_id=dataset_id,
datum_id=datum_id,
has_klv=has_klv,
codec=codec,
format=video_container_format,
min_duration_ms=min_duration_ms,
sort=sort,
direction=_utils.enum_value(direction),
limit=limit,
offset=offset,
)
response = _apis.datasets_v3.videos_api.get_video_references(body=request)
if not response.data:
return []
return [_video_reference_from_openapi(v) for v in response.data]
[docs]
def get_video_references(
*,
project_ids: list[str] | None = None,
dataset_id: str | None = None,
datum_id: str | None = None,
has_klv: bool | None = None,
codec: str | None = None,
video_container_format: str | None = None,
min_duration_ms: int | None = None,
sort: str | None = None,
direction: models.SortDirection | None = None,
max_items: int | None = None,
) -> Generator[models.VideoReference, None, None]:
"""Get video references with optional filters and sorted pagination.
:param project_ids: Restrict to datasets in these projects
:type project_ids: Optional[list[str]]
:param dataset_id: Restrict to a single dataset id
:type dataset_id: Optional[str]
:param datum_id: Restrict to the video reference linked from this datum id
:type datum_id: Optional[str]
:param has_klv: Filter by presence of KLV metadata
:type has_klv: Optional[bool]
:param codec: Filter by codec
:type codec: Optional[str]
:param video_container_format: Filter by container format (for example mp4)
:type video_container_format: Optional[str]
:param min_duration_ms: Minimum duration in milliseconds
:type min_duration_ms: Optional[int]
:param sort: Sort column (`id` or `creation timestamp`)
:type sort: Optional[str]
:param direction: Sort direction
:type direction: Optional[models.SortDirection]
:param max_items: Maximum number of items to yield
:type max_items: Optional[int]
:return: Generator of video references
:rtype: Generator[models.VideoReference, None, None]
"""
params = {k: v for k, v in locals().items() if k != "max_items"}
return _utils.paginate_items(_get_video_references, params, max_items)