Source code for luna_lambda_tools.public.handlers.schemas

"""
Module contains pydantic schemas for handlers lambda
"""

from typing import Literal, final

from pydantic import Field, StrictBytes, conlist
from vlutils.structures.pydantic import BaseModel as _BaseModel

from luna_lambda_tools.public.schemas.base import BoundingBox, Location


[docs] @final class ImageOrigin(_BaseModel): """Image origin""" # image origin as bytes body: str | bytes # image body meta bodyMeta: dict | None = None
[docs] @final class SourceData(_BaseModel): """Source data""" # source (if present) source: str = Field(default_factory=lambda: None) # stream id streamId: str = Field(default_factory=lambda: None) # tags (if present) tags: list[str] = Field(default_factory=lambda: None) # user data userData: str = "" # external id externalId: str = "" # track id trackId: str = Field(default_factory=lambda: None) # meta information provided by user (if present) meta: dict = Field(default_factory=lambda: None) # luna-event time (if present) eventTime: str = Field(default_factory=lambda: None) # luna event end time (if present) eventEndTime: str = Field(default_factory=lambda: None) # luna event location location: Location | None = None
[docs] class EventSourceBase(_BaseModel): """Event source base""" filename: str = Field(default_factory=lambda: None) detectTime: str = Field(default_factory=lambda: None) detectTs: str = Field(default_factory=lambda: None) imageOrigin: ImageOrigin | str = Field(default_factory=lambda: None)
[docs] class RawImageAggregated(_BaseModel): """Raw image source""" body: StrictBytes faceBoundingBoxes: conlist(BoundingBox, min_length=1, max_length=1) = Field(default_factory=lambda: None) bodyBoundingBoxes: conlist(BoundingBox, min_length=1, max_length=1) = Field(default_factory=lambda: None) meta: dict | None = Field(default_factory=lambda: None)
[docs] class RawDetection(_BaseModel): """Raw detection""" warp: StrictBytes = Field(default_factory=lambda: None) boundingBox: BoundingBox = Field(default_factory=lambda: None) meta: dict | None = Field(default_factory=lambda: None)
[docs] class RawDetectionsAggregated(_BaseModel): """Raw detections pair""" face: RawDetection = Field(default_factory=lambda: None) body: RawDetection = Field(default_factory=lambda: None)
[docs] class RawImageNonAggregated(RawImageAggregated): """Raw image non aggregated""" sourceData: SourceData = SourceData()
[docs] class RawDetectionsNonAggregated(RawDetectionsAggregated): """Raw detections non aggregated""" sourceData: SourceData = SourceData()
[docs] class EventSourceAggregatedRawImage(EventSourceBase): sourceType: Literal["raw_image"] source: RawImageAggregated
[docs] class EventSourceAggregatedDetections(EventSourceBase): sourceType: Literal["detections"] source: conlist(RawDetectionsAggregated)
[docs] class EventSourceNonAggregatedRawImage(EventSourceBase): sourceType: Literal["raw_image"] source: RawImageNonAggregated
[docs] class EventSourceNonAggregatedDetections(EventSourceBase): sourceType: Literal["detections"] source: conlist(RawDetectionsNonAggregated)
[docs] class EventSourceSchema(_BaseModel): """Event source non aggregated schema""" sources: ( list[EventSourceNonAggregatedRawImage | EventSourceNonAggregatedDetections] | list[EventSourceAggregatedRawImage | EventSourceAggregatedDetections] ) aggregateAttributes: Literal[0, 1] useExifInfo: bool = True sourceData: dict | None = None