Source code for luna_handlers.classes.event

"""Module contains event structure."""
from typing import Dict, Optional, Union
from uuid import uuid4

from luna3.common.http_objs import RawDescriptor
from lunavl.sdk.descriptors.descriptors import BodyDescriptor, FaceDescriptor
from lunavl.sdk.estimators.face_estimators.basic_attributes import BasicAttributes
from vlutils.descriptors.data import DescriptorsEnum, DescriptorType
from vlutils.structures.dataclasses import dataclass

from classes.event_parts import AggregateEstimations, ImageDetection

FACE_DESCRIPTOR_VERSIONS = [x.value.version for x in DescriptorsEnum if x.value.type == DescriptorType.face]
BODY_DESCRIPTOR_VERSIONS = [x.value.version for x in DescriptorsEnum if x.value.type == DescriptorType.body]


[docs]class ExtractedAttribute: """ Attribute. Attributes: attributeId (Optional[str]): attribute id descriptor (RawDescriptorData): descriptor sampleIds (list(str)): sample ids """ __slots__ = ("attributeId", "descriptor", "sampleIds") def __init__( self, descriptor: Optional[Union[FaceDescriptor, BodyDescriptor, RawDescriptor]] = None, sampleIds: Optional[list[str]] = None, ): self.attributeId = None self.descriptor = descriptor self.sampleIds = sampleIds or []
[docs]class FaceExtractedAttribute(ExtractedAttribute): """ Face extracted attribute. """ __slots__ = ("basicAttributes", "url") def __init__( self, descriptor: Optional[Union[FaceDescriptor, RawDescriptor]] = None, basicAttributes: Optional[BasicAttributes] = None, sampleIds: Optional[list[str]] = None, ): super().__init__(descriptor=descriptor, sampleIds=sampleIds) self.basicAttributes = basicAttributes self.url = None
[docs] def asDict(self): """Face attributes in api format.""" res = {"attribute_id": self.attributeId, "url": self.url, "samples": self.sampleIds} if self.basicAttributes: res["basic_attributes"] = self.basicAttributes.asDict() if self.descriptor: res["score"] = self.descriptor.garbageScore if isinstance(self.descriptor, FaceDescriptor) else 1.0 return res
[docs]class BodyExtractedAttribute(ExtractedAttribute): """ Body extracted attribute. """ def __init__( self, descriptor: Optional[Union[BodyDescriptor, RawDescriptor]] = None, sampleIds: Optional[list[str]] = None, ): super().__init__(descriptor=descriptor, sampleIds=sampleIds)
[docs] def asDict(self): """Body attributes in api format.""" res = {"samples": self.sampleIds} if self.descriptor: res["score"] = self.descriptor.garbageScore if isinstance(self.descriptor, BodyDescriptor) else 1.0 return res
[docs]@dataclass(withSlots=True) class GeoPosition: """ Event coordinates Attributes: latitude: longitude coordinate longitude: longitude coordinate """ latitude: Optional[float] = None longitude: Optional[float] = None def __post_init__(self): if not self.isValid(): raise ValueError("Not valid")
[docs] def asDict(self) -> Dict[str, Optional[float]]: """ Convert coordinates to dict. Returns: dict. Keys with non None value. """ res = {} if self.latitude is not None: res["latitude"] = self.latitude if self.longitude is not None: res["longitude"] = self.longitude return res
[docs] def isValid(self) -> bool: """ Check that both of coordinates are specified Returns: true if both of coordinates are set or no one otherwise false """ return (self.latitude is None) is (self.longitude is None)
[docs] def isNotEmpty(self) -> bool: """ Check geo position is not empty Returns: true if not empty else false """ return self.latitude is not None and self.longitude is not None
[docs] def copy(self) -> "GeoPosition": """ Deep copy coordinates. Returns: event coordinates """ return GeoPosition(latitude=self.latitude, longitude=self.longitude)
[docs]@dataclass(withSlots=True) class Location: """ Event location Attributes: street: street houseNumber: house number district: district area: area city: city geoPosition: geo coordinates """ street: Optional[str] houseNumber: Optional[str] district: Optional[str] area: Optional[str] city: Optional[str] geoPosition: GeoPosition
[docs] def asDict(self) -> Dict[str, str]: """ Convert location to dict. Returns: dict. Keys with non None value. """ res = dict() if self.geoPosition.isNotEmpty(): res["geo_position"] = self.geoPosition.asDict() if self.district is not None: res["district"] = self.district if self.city is not None: res["city"] = self.city if self.area is not None: res["area"] = self.area if self.street is not None: res["street"] = self.street if self.houseNumber is not None: res["house_number"] = self.houseNumber return res
[docs] def copy(self) -> "Location": """ Deep copy location. Returns: event location """ return Location( street=self.street, houseNumber=self.houseNumber, district=self.district, area=self.area, city=self.city, geoPosition=self.geoPosition.copy(), )
[docs]@dataclass(withSlots=True) class EventMetadata: """User defined event metadata.""" accountId: str handlerId: str createEventTime: str endEventTime: str source: str = None tags: list[str] = None userData: str = None externalId: str = None location: Location = None trackId: str = None
[docs]class Event: """ Structure for storage an event Attributes: eventId (str): event id eventUrl: (Optional[str]) event url sdkEstimations (list[ImageDetection]): face & body estimations aggregateEstimations (AggregateEstimations): event aggregate estimations faceAttributes (FaceExtractedAttribute): face extracted attribute bodyAttributes (BodyExtractedAttribute): body extracted attribute sampleIds (list[str]): event attribute sample ids meta (EventMetadata): user defined metadata tags (list[str]): event tags (user defined & conditional ones) avatar: event avatar faceId (Optional[str]): face id associated with the event faceUrl (Optional[str]): face url linkedLists (Optional[List[str]]): list that contains the face matches (List[EventMatchResult]): list of match results """ __slots__ = ( "eventId", "eventUrl", "sdkEstimations", "aggregateEstimations", "faceAttributes", "bodyAttributes", "sampleIds", "meta", "tags", "avatar", "faceId", "faceUrl", "linkedLists", "matches", ) def __init__( self, sdkEstimations: Optional[list[ImageDetection]] = None, aggregatedEstimations: Optional[AggregateEstimations] = None, faceAttributes: Optional[FaceExtractedAttribute] = None, bodyAttributes: Optional[BodyExtractedAttribute] = None, sampleIds: Optional[list[str]] = None, meta: EventMetadata = None, ): self.sdkEstimations = sdkEstimations or [] self.aggregateEstimations = aggregatedEstimations or AggregateEstimations() self.faceAttributes = faceAttributes self.bodyAttributes = bodyAttributes self.sampleIds = sampleIds or [] self.eventId: str = str(uuid4()) self.eventUrl: Optional[str] = None self.meta: EventMetadata = meta self.tags = self.meta.tags if self.meta and self.meta.tags else [] self.avatar: str = "" self.faceId: Optional[str] = None self.faceUrl: Optional[str] = None self.linkedLists: list[str] = [] self.matches: List["EventMatchResult"] = [] # noqa: F821
[docs] def asDict(self) -> dict: """ Prepares dict representation. """ res = { "face_attributes": self.faceAttributes.asDict() if self.faceAttributes else None, "body_attributes": self.bodyAttributes.asDict() if self.bodyAttributes else None, "tags": self.tags, "source": self.meta.source, "event_id": self.eventId, "url": self.eventUrl, "matches": [match.asDict() for match in self.matches], "external_id": self.meta.externalId, "user_data": self.meta.userData, "location": self.meta.location.asDict(), "detections": [detection.asDict() for detection in self.sdkEstimations], "aggregate_estimations": self.aggregateEstimations.asDict(), "track_id": self.meta.trackId, } if self.faceId is not None: face = { "external_id": self.meta.externalId, "face_id": self.faceId, "user_data": self.meta.userData, "url": self.faceUrl, "lists": self.linkedLists or [], "avatar": self.avatar, "event_id": self.eventId, } else: face = None res["face"] = face return res