Source code for luna_handlers.classes.event

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