"""
Module contains pydantic schemas for handlers lambda
"""
from typing import final
from pydantic import Field
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 EventSourceAggregated(_BaseModel):
"""Aggregated event source"""
# image body or descriptor as bytes
body: bytes | None = None
# image body meta
bodyMeta: dict | None = None
# image filename (if present)
filename: str = Field(default_factory=lambda: None)
# source type: 0 - usual image, 1 - vl face warp, 2 - vl body warp, 3 - vl face descriptor, 4 - vl body descriptor
sourceType: str
# face bounding boxes (if present)
faceBoundingBoxes: list[BoundingBox] = Field(default_factory=lambda: None)
# body bounding boxes (if present)
bodyBoundingBoxes: list[BoundingBox] = Field(default_factory=lambda: None)
# image detect time (if present)
detectTime: str = Field(default_factory=lambda: None)
# image detect timestamp (if present)
detectTs: str = Field(default_factory=lambda: None)
# image origin (if present)
imageOrigin: ImageOrigin | str = Field(default_factory=lambda: None)
[docs]
@final
class EventSource(EventSourceAggregated):
"""Event source schema"""
# event source
sourceData: SourceData | None = None
[docs]
@final
class EventSourceSchema(_BaseModel):
"""Event source schema"""
# sources list
sources: list[EventSourceAggregated] | list[EventSource]
# whether to aggregate attributes
aggregateAttributes: int
# whether to use image exif info
useExifInfo: bool = True
# event source | available if aggregation is enabled
sourceData: SourceData | None = None