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