"""
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