"""Module contains event structure."""
import uuid
from typing import List, Optional, Dict
from vlutils.structures.dataclasses import dataclass
from app.api_sdk_adaptors.extractor import _APIFaceAttributeForExtract, _APIBodyAttributeForExtract
from classes.event_parts import ImageDetection, AggregateEstimations
[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 a event
Attributes:
source (str): source of event
tags (List[str]): tags which is associated with the event
externalId (Optional[str]): external id created face
userData (str): user data created face
matches (List[EventMatchResult]): list of match results
faceAttributes (_APIAttributeForExtract): extracted face attributes
bodyAttributes (_APIAttributeForExtract): extracted body attributes
eventId (str): event id
faceId (str): face id associated with the event
faceUrl (str): face url
linkedLists (Optional[List[str]]): list that contains the face
location (Location): event location
detections (List[dict]): face & body detections
aggregateEstimations (AggregateEstimations): detection aggregated estimations
eventUrl: (str) event url
trackId: (str) event track id
"""
__slots__ = (
"matches",
"faceAttributes",
"bodyAttributes",
"linkedLists",
"faceId",
"faceUrl",
"source",
"externalId",
"tags",
"userData",
"eventId",
"avatar",
"location",
"detections",
"aggregateEstimations",
"eventUrl",
"trackId",
)
def __init__(
self,
faceAttributes: Optional[_APIFaceAttributeForExtract] = None,
bodyAttributes: Optional[_APIBodyAttributeForExtract] = None,
aggregateEstimations: Optional[AggregateEstimations] = None,
detections: Optional[List[ImageDetection]] = None,
):
self.matches: List["EventMatchResult"] = [] # noqa: F821
self.faceAttributes = faceAttributes
self.bodyAttributes = bodyAttributes
self.linkedLists: List[str]
self.faceId: Optional[str] = None
self.faceUrl: Optional[str] = None
self.source: str
self.externalId: str
self.tags: List[str] = []
self.userData: str
self.eventId = str(uuid.uuid4())
self.avatar: str
self.detections: List[ImageDetection] = detections or []
self.aggregateEstimations: AggregateEstimations = aggregateEstimations or AggregateEstimations()
self.location: Location
self.eventUrl: Optional[str] = None
self.trackId: Optional[str] = None
[docs]def eventAsDict(event: Event) -> dict:
"""
Return event in the response format.
Args:
event: event
Returns:
event as dict
"""
res = {
"face_attributes": event.faceAttributes.json if event.faceAttributes else None,
"body_attributes": event.bodyAttributes.json if event.bodyAttributes else None,
"source": event.source,
"tags": event.tags,
"event_id": event.eventId,
"url": event.eventUrl,
"matches": [match.asDict() for match in event.matches],
"external_id": event.externalId,
"user_data": event.userData,
"location": event.location.asDict(),
"detections": [detection.asDict() for detection in event.detections],
"aggregate_estimations": event.aggregateEstimations.asDict(),
"track_id": event.trackId,
}
if event.faceId is not None:
face = {
"external_id": event.externalId,
"face_id": event.faceId,
"user_data": event.userData,
"url": event.faceUrl,
"lists": event.linkedLists,
"avatar": event.avatar,
"event_id": event.eventId,
}
else:
face = None
res["face"] = face
return res
