"""
Module contains luna-events client adapted for usage in lambda
"""
from typing import Awaitable, Literal
from luna3.common.http_objs import StatQuery
from luna3.common.luna_response import LunaResponse
from luna3.events.events import EventsApi
from luna3.events.http_objs import CLOTHING_COLOR, EVENT_TARGET, PRIMITIVE_CLOTHING_COLOR, Event
[docs]
class Events:
"""Luna-events client"""
[docs]
def __init__(self, eventsClient: EventsApi, accountId: str):
self._accountId = accountId
self._eventsClient = eventsClient
[docs]
def getAddress(self) -> str:
"""Get luna events address"""
return self._eventsClient.baseUri
[docs]
def saveEvents(
self,
events: list[Event],
waitEventsSaving: bool | None = None,
raiseError=True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Save events to luna-events.
Args:
events: list of events
waitEventsSaving: whether to wait events saving (response will be received only after events will be saved)
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
"""
return self._eventsClient.saveEvents(
accountId=self._accountId,
events=events,
waitEventsSaving=waitEventsSaving,
raiseError=raiseError,
**kwargs,
)
[docs]
def getEvents( # pylint: disable=R0914
self,
target: list[EVENT_TARGET] | None = None,
createTimeGte: str | None = None,
createTimeLt: str | None = None,
endTimeGte: str | None = None,
endTimeLt: str | None = None,
eventIds: list[str] | None = None,
eventIdGte: str | None = None,
eventIdLt: str | None = None,
handlerIds: list[str] | None = None,
externalIds: list[str] | None = None,
topMatchingCandidatesLabel: str | None = None,
topSimilarObjectIds: list[str] | None = None,
topSimilarExternalIds: list[str] | None = None,
topSimilarObjectSimilarityGte: float | None = None,
topSimilarObjectSimilarityLt: float | None = None,
ageLt: int | None = None,
ageGte: int | None = None,
gender: int | Literal["null"] | None = None,
emotions: list[int | Literal["null"]] | None = None,
masks: list[int | Literal["null"]] | None = None,
ethnicGroups: list[int | Literal["null"]] | None = None,
liveness: list[int | Literal["null"]] | None = None,
apparentAgeGte: int | None = None,
apparentAgeLt: int | None = None,
apparentGender: list[Literal[0, 1, 2]] | Literal["null"] = None,
headwearStates: list[Literal[0, 1, 2] | Literal["null"]] | None = None,
headwearApparentColors: list[PRIMITIVE_CLOTHING_COLOR | Literal["null"]] | None = None,
sleeveLengths: list[Literal["short", "long", "undefined", "null"]] | None = None,
upperClothingColors: list[CLOTHING_COLOR | Literal["null"]] | None = None,
lowerGarmentTypes: list[Literal["undefined", "trousers", "shorts", "skirt", "null"]] | None = None,
lowerGarmentColors: list[CLOTHING_COLOR | Literal["null"]] | None = None,
shoesApparentColors: list[PRIMITIVE_CLOTHING_COLOR | Literal["null"]] | None = None,
backpackStates: list[Literal[0, 1, 2] | Literal["null"]] | None = None,
faceIds: list[str] | None = None,
userData: str | None = None,
page: int | None = 1,
pageSize: int | None = 10,
order: str | None = "desc",
sources: list[str | Literal["null"]] | None = None,
tags: list[str] | None = None,
cities: list[str | Literal["null"]] | None = None,
areas: list[str | Literal["null"]] | None = None,
districts: list[str | Literal["null"]] | None = None,
streets: list[str | Literal["null"]] | None = None,
houseNumbers: list[str | Literal["null"]] | None = None,
originLongitude: float | Literal["null"] | None = None,
originLatitude: float | Literal["null"] | None = None,
longitudeDelta: float | None = None,
latitudeDelta: float | None = None,
trackIds: list[str | Literal["null"]] | None = None,
meta: dict | None = None,
raiseError=True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get events by filters.
Args:
target: list of events' fields to receive instead of full events
createTimeGte: lower including create time boundary
createTimeLt: upper excluding create time boundary
endTimeGte: lower including end time boundary
endTimeLt: upper excluding end time boundary
faceIds: ids of faces linked with an event
eventIds: event ids
eventIdGte: event id lower including bound
eventIdLt: event id upper excluding bound
handlerIds: handler ids
externalIds: external ids
userData: event user data
sources: list of source
tags: users tags
topMatchingCandidatesLabel: top matching candidates label
topSimilarObjectIds: top similar object ids
topSimilarExternalIds: top similar external ids
topSimilarObjectSimilarityGte: lower including top similarity boundary
topSimilarObjectSimilarityLt: upper excluding top similarity boundary
ageLt: upper excluding age boundary
ageGte: lower including age boundary
gender: gender of event (0 or 1)
emotions: dominant emotions in event (one of (1, 2, 3, 4, 5, 6, 7) -
matches "anger", "disgust", "fear", "happiness", "neutral", "sadness", "surprise")
masks: dominant masks in event (one of (1, 2, 3) -
matches "missing", "medical_mask", "occluded")
ethnicGroups: dominant ethnicity of the face ((1, 2, 3, 4) -
means (indian, asian, african_american, caucasian)
liveness: dominant liveness of the face ((0, 1, 2) - means ("spoof", "real", "unknown"))
apparentAgeLt: apparent age upper excluding boundary
apparentAgeGte: apparent age upper including boundary
apparentGender: apparent gender estimation values (0 - female, 1 - male, 2 - undefined)
headwearStates: headwear states (0 - absent, 1 - present, 2 - undefined)
headwearApparentColors: headwear apparent colors
sleeveLengths: sleeve lengths
upperClothingColors: upper clothing colors
lowerGarmentTypes: lower garment types
lowerGarmentColors: lower garment colors
shoesApparentColors: shoes apparent colors
backpackStates: backpack states (0 - absent, 1 - present, 2 - undefined)
areas: list of areas
streets: list of streets
districts: list of districts
cities: list of cities
houseNumbers: list of house numbers
originLongitude: origin longitude in degrees
originLatitude: origin latitude in degrees
longitudeDelta: longitude delta in degrees
latitudeDelta: latitude delta in degrees
trackIds: list of track ids
meta: user-defined event meta's filters in format {meta.<path.to.field>__<operator>:<type>: <value>}
page: number of result page
pageSize: count of results in page
order: result sort order (ask or desc)
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with events will be returned.
"""
return self._eventsClient.getEvents(
target=target,
createTimeGte=createTimeGte,
createTimeLt=createTimeLt,
endTimeGte=endTimeGte,
endTimeLt=endTimeLt,
eventIds=eventIds,
eventIdGte=eventIdGte,
eventIdLt=eventIdLt,
handlerIds=handlerIds,
externalIds=externalIds,
topMatchingCandidatesLabel=topMatchingCandidatesLabel,
topSimilarObjectIds=topSimilarObjectIds,
topSimilarExternalIds=topSimilarExternalIds,
topSimilarObjectSimilarityGte=topSimilarObjectSimilarityGte,
topSimilarObjectSimilarityLt=topSimilarObjectSimilarityLt,
ageLt=ageLt,
ageGte=ageGte,
gender=gender,
emotions=emotions,
masks=masks,
ethnicGroups=ethnicGroups,
liveness=liveness,
apparentAgeGte=apparentAgeGte,
apparentAgeLt=apparentAgeLt,
apparentGender=apparentGender,
headwearStates=headwearStates,
headwearApparentColors=headwearApparentColors,
sleeveLengths=sleeveLengths,
upperClothingColors=upperClothingColors,
lowerGarmentTypes=lowerGarmentTypes,
lowerGarmentColors=lowerGarmentColors,
shoesApparentColors=shoesApparentColors,
backpackStates=backpackStates,
faceIds=faceIds,
userData=userData,
page=page,
pageSize=pageSize,
order=order,
sources=sources,
tags=tags,
cities=cities,
areas=areas,
districts=districts,
streets=streets,
houseNumbers=houseNumbers,
originLongitude=originLongitude,
originLatitude=originLatitude,
longitudeDelta=longitudeDelta,
latitudeDelta=latitudeDelta,
trackIds=trackIds,
meta=meta,
accountId=self._accountId,
raiseError=raiseError,
**kwargs,
)
[docs]
def getEventDescriptor(
self,
eventId: str | None = None,
descriptorVersion: int | None = None,
descriptorType: str | None = None,
raiseError=True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get event descriptor by filters.
Args:
eventId: event id
descriptorVersion: descriptor version
descriptorType: Kind of the descriptor (body or face)
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with event descriptor will be returned.
"""
return self._eventsClient.getEventDescriptor(
eventId=eventId,
accountId=self._accountId,
descriptorVersion=descriptorVersion,
descriptorType=descriptorType,
raiseError=raiseError,
**kwargs,
)
[docs]
def getEvent(
self,
eventId: str,
target: list[EVENT_TARGET] | None = None,
raiseError: bool = True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get event by id.
Args:
eventId: event id
target: list of events' fields to receive instead of full events
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with events will be returned.
"""
return self._eventsClient.getEvent(
eventId=eventId,
accountId=self._accountId,
target=target,
raiseError=raiseError,
**kwargs,
)
[docs]
def headEvent(
self,
eventId: str,
raiseError: bool = True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Check event existence by id.
Args:
eventId: event id
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with events will be returned.
"""
return self._eventsClient.checkEvent(
eventId=eventId, accountId=self._accountId, raiseError=raiseError, **kwargs
)
[docs]
def getEventsStats(
self, query: StatQuery = StatQuery(), raiseError: bool = True, **kwargs
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get events statistics by aggregation (<aggregator> and <groupBy> on <target>) and filters.
Args:
query: query object or query dict, contained in StatQuery object
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with event statistics will be returned.
"""
query.addFilter("account_id", "eq", self._accountId)
return self._eventsClient.getEventsStats(
query=query,
raiseError=raiseError,
**kwargs,
)
[docs]
def deleteEvents(
self,
target: list[str] | None = None,
handlerId: str | None = None,
createTimeGte: str | None = None,
createTimeLt: str | None = None,
insertTimeGte: str | None = None,
insertTimeLt: str | None = None,
raiseError: bool = True,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Delete events by filters.
Args:
target: event deletion info targets
handlerId: event handler id
createTimeGte: lower including create time boundary
createTimeLt: upper excluding create time boundary
insertTimeGte: lower including insert time boundary
insertTimeLt: upper excluding insert time boundary
raiseError: whether to raise LunaApiException in case of failure
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with events will be returned.
"""
return self._eventsClient.deleteEvents(
target=target,
handlerId=handlerId,
createTimeGte=createTimeGte,
createTimeLt=createTimeLt,
insertTimeGte=insertTimeGte,
insertTimeLt=insertTimeLt,
raiseError=raiseError,
**kwargs,
)