"""
Module contains luna-handlers client adapted for usage in lambda
"""
from typing import Any, Awaitable
from luna3.common.http_objs import BinaryImage, Policies, SampleForDetection, UrlForDetection, VerifierPolicies
from luna3.common.luna_response import LunaResponse
from luna3.handlers.handlers import HandlersApi
from luna3.handlers.http_objs import EventAttributeUpgrade, FaceAttributeUpgrade, RawEvent, StreamEventsSource
from luna3.public.common import BinaryDescriptorData
from luna3.public.matcher import BinaryReference
[docs]
class Handlers:
"""Luna-handlers client"""
[docs]
def __init__(self, handlersClient: HandlersApi, accountId: str):
self._accountId = accountId
self._handlersClient = handlersClient
[docs]
def getAddress(self) -> str:
"""Get luna-handlers address"""
return self._handlersClient.baseUri
[docs]
def detectFaces(
self,
inputData: BinaryImage | list[BinaryImage] | list[UrlForDetection] | list[str],
multifacePolicy: int = 1,
estimateHeadPose: int = 0,
detectLandmarks68: int = 0,
extractExif: int = 0,
estimateQuality: int = 0,
estimateGaze: int = 0,
estimateEyesAttributes: int = 0,
estimateEmotions: int = 0,
estimateMask: int = 0,
estimateMouthAttributes: int = 0,
pitchThreshold: int | None = None,
rollThreshold: int | None = None,
yawThreshold: int | None = None,
warpedImage: int = 0,
useExifInfo: int | None = None,
estimateFaceQuality: int = 0,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Detect faces on input images.
Args:
inputData: one of:
- Single image or images list for extraction (supported formats "jpeg, png, tif, ppm, bmp").
- list of urls with images
- list of images' ids
multifacePolicy: multiple face detection policy:
0 - multiple face detection not allowed,
1 - multiple face detection allowed,
2 - get best detection from the image
estimateHeadPose: Available values : 0, 1
detectLandmarks68: Available values : 0, 1
extractExif: Available values : 0, 1
estimateQuality: Available values : 0, 1
estimateGaze: Available values : 0, 1
estimateEyesAttributes: Available values : 0, 1
estimateMouthAttributes: Available values : 0, 1
estimateEmotions: Available values : 0, 1
estimateMask: Available values : 0, 1
pitchThreshold: maximum deviation pitch angle from 0
rollThreshold: maximum deviation roll angle from 0
yawThreshold: maximum deviation yaw angle from 0
warpedImage: Whether input image is a warped or arbitrary image.
useExifInfo: whether to use exif info for auto orientation
estimateFaceQuality: whether to estimate face quality
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with samples, exif, filename for succeeded images and
errors for failed images will be returned.
"""
return self._handlersClient.detectFaces(
accountId=self._accountId,
inputData=inputData,
multifacePolicy=multifacePolicy,
estimateHeadPose=estimateHeadPose,
detectLandmarks68=detectLandmarks68,
extractExif=extractExif,
estimateQuality=estimateQuality,
estimateGaze=estimateGaze,
estimateEyesAttributes=estimateEyesAttributes,
estimateEmotions=estimateEmotions,
estimateMask=estimateMask,
estimateMouthAttributes=estimateMouthAttributes,
pitchThreshold=pitchThreshold,
rollThreshold=rollThreshold,
yawThreshold=yawThreshold,
warpedImage=warpedImage,
useExifInfo=useExifInfo,
estimateFaceQuality=estimateFaceQuality,
**kwargs,
)
[docs]
def extractAttrFromSample(
self,
sampleIds: list[str],
scoreThreshold: float = 0,
extractDescriptor: int = 1,
extractBasicAttributes: int = 0,
aggregateAttributes: int = 0,
ttl: int | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Extract attributes from samples.
Args:
sampleIds: list of samples
scoreThreshold: number [ 0 .. 1 ]; example: 0.99; descriptor quality score threshold.
All the attributes with quality score below the threshold will be ignored (and not stored in the DB).
The function will proceed as usual with all the remaining descriptors (if left).
extractDescriptor: Available values : 0, 1; whether to extract face descriptor(s)
extractBasicAttributes: Available values : 0, 1;
whether to extract basic attributes (gender, age, ethnicity)
aggregateAttributes: Available values : 0, 1; whether to aggregate face descriptor(s);
If true, all extracted attributes will be aggregated and stored in as a single attribute.
Otherwise, all attributes will be stored for every sample.
ttl: temporary attribute ttl
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with attributes info
"""
return self._handlersClient.extractAttrFromSample(
accountId=self._accountId,
sampleIds=sampleIds,
scoreThreshold=scoreThreshold,
extractDescriptor=extractDescriptor,
extractBasicAttributes=extractBasicAttributes,
aggregateAttributes=aggregateAttributes,
ttl=ttl,
**kwargs,
)
[docs]
def upgradeObjectFromSample(
self,
data: FaceAttributeUpgrade | EventAttributeUpgrade,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Upgrade basic attributes and/or descriptor in a object using samples.
Args:
data: face or attribute upgrade object
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with attributes info
"""
return self._handlersClient.upgradeObjectFromSample(
accountId=self._accountId,
data=data,
**kwargs,
)
[docs]
def createHandler(
self,
policies: Policies | None = None,
handlerType: int | None = None,
lambdaId: str | None = None,
description: str | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Create new handler. Handler determinate a list of rules for proccessing input images.
You can set parameters for detecting faces, extracting attributes.
You can set rules for matching attributes and creating faces from them.
Args:
policies: object with policies information
handlerType: handler type (0 - static, 1 - dynamic, 2 - lambda)
lambdaId: lambda id
description: user description
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` will return json new handler id and url
"""
return self._handlersClient.createHandler(
accountId=self._accountId,
policies=policies,
handlerType=handlerType,
lambdaId=lambdaId,
description=description,
**kwargs,
)
[docs]
def getHandlers(
self,
description: str | None = None,
handlerType: int | None = None,
page: int = 1,
pageSize: int = 10,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get handlers by filters
Args:
description: Find all handlers with description which "like" of this parameter
handlerType: handler type (0 - static, 1 - dynamic, 2 - lambda)
page: The number of page
pageSize: The numbers of items on page
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with handlers info
"""
return self._handlersClient.getHandlers(
accountId=self._accountId,
description=description,
handlerType=handlerType,
page=page,
pageSize=pageSize,
**kwargs,
)
[docs]
def getHandlersCount(
self,
description: str | None = None,
handlerType: int | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get handlers count by filters
Args:
description: Find all handlers with description which "like" of this parameter
handlerType: handler type (0 - static, 1 - dynamic, 2 - lambda)
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: handlers count
"""
return self._handlersClient.getHandlersCount(
accountId=self._accountId,
description=description,
handlerType=handlerType,
**kwargs,
)
[docs]
def getHandlerById(self, handlerId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get handler by its id
Args:
handlerId: handler id (uuid4)
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with handler info
"""
return self._handlersClient.getHandlerById(
accountId=self._accountId,
handlerId=handlerId,
**kwargs,
)
[docs]
def updateHandler(
self,
handlerId: str,
policies: Policies | None = None,
handlerType: int | None = None,
lambdaId: str | None = None,
description: str | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Update a handler. You can not update a part of handler.
Args:
handlerId: handler id
policies: object with policies information
handlerType: handler type (0 - static, 1 - dynamic, 2 - lambda)
lambdaId: lambda id
description: user description
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` will return nothing in text field
"""
return self._handlersClient.updateHandler(
accountId=self._accountId,
handlerId=handlerId,
policies=policies,
handlerType=handlerType,
lambdaId=lambdaId,
description=description,
**kwargs,
)
[docs]
def validateHandler(
self,
policies: Policies | dict,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Validate handler policies.
Args:
policies: object with policies information
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
"""
return self._handlersClient.validateHandler(
accountId=self._accountId,
policies=policies,
**kwargs,
)
[docs]
def checkHandlerById(self, handlerId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Check whether handler exists by its id
Args:
handlerId: handler id (uuid4)
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: no text returns
"""
return self._handlersClient.checkHandlerById(
accountId=self._accountId,
handlerId=handlerId,
**kwargs,
)
[docs]
def deleteHandlerById(self, handlerId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Remove handler by id
Args:
handlerId: handler id (uuid4)
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: no text returns
"""
return self._handlersClient.deleteHandlerById(
accountId=self._accountId,
handlerId=handlerId,
**kwargs,
)
[docs]
def emitEvents(
self,
handlerId: str,
inputData: BinaryImage | list[BinaryImage] | list[UrlForDetection] | list[str] | list[SampleForDetection],
policies: Policies | dict | None = None,
externalId: str | None = None,
userData: str | None = None,
aggregateAttributes: int | None = None,
source: str | None = None,
streamId: str | None = None,
city: str | None = None,
street: str | None = None,
district: str | None = None,
area: str | None = None,
houseNumber: str | None = None,
tags: str | None = None,
latitude: float | None = None,
longitude: float | None = None,
imageType: int = None,
lunaEventTime: str | None = None,
lunaEventEndTime: str | None = None,
trackId: str | None = None,
useExifInfo: int | None = None,
meta: dict | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Emit input events.
Args:
handlerId: handler id
inputData: one of:
- Single image or images list for extraction (supported formats "jpeg, png, tif, ppm, bmp").
- list of urls with images
- list of images' ids
policies: object with policies information for dynamic handler
externalId: external id
userData: user data
aggregateAttributes: aggregate all attributes (descriptot, age, gender) or not (0 or 1)
source: source
streamId: stream id
city: event city
street: event street
area: event area
district: event district
houseNumber: event house number
tags: tags
latitude: event latitude in degrees
longitude: event longitude in degrees
imageType: image type (0 - raw image, 1 - face warped image, 2 - body warped image)
lunaEventTime: user defined event create time
lunaEventEndTime: user defined event end time
trackId: (str) event track id
useExifInfo: whether to use exif info for auto orientation
meta: user-defined event metadata
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` will return json new handler id and url
"""
return self._handlersClient.emitEvents(
accountId=self._accountId,
handlerId=handlerId,
inputData=inputData,
policies=policies,
externalId=externalId,
userData=userData,
aggregateAttributes=aggregateAttributes,
source=source,
streamId=streamId,
city=city,
street=street,
district=district,
area=area,
houseNumber=houseNumber,
tags=tags,
latitude=latitude,
longitude=longitude,
imageType=imageType,
lunaEventTime=lunaEventTime,
lunaEventEndTime=lunaEventEndTime,
trackId=trackId,
useExifInfo=useExifInfo,
meta=meta,
**kwargs,
)
[docs]
def emitStreamEvents(
self,
handlerId: str,
inputData: StreamEventsSource,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Emit stream events.
Args:
handlerId: handler id
inputData: structure with stream event(s) source
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` will return json new handler id and url
"""
return self._handlersClient.emitStreamEvents(
accountId=self._accountId,
handlerId=handlerId,
inputData=inputData,
**kwargs,
)
[docs]
def saveEvent(
self,
handlerId: str,
event: RawEvent | dict[str, Any],
waitSaving: int | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Save raw event.
Args:
handlerId: handler id
event: raw event
waitSaving: whether to wait for event saving
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
"""
return self._handlersClient.saveEvent(
accountId=self._accountId,
handlerId=handlerId,
event=event,
waitSaving=waitSaving,
**kwargs,
)
[docs]
def createVerifier(
self, policies: VerifierPolicies | None = None, description: str | None = None, **kwargs
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Create new verifier. Verifier determinate a list of rules for processing & verification of input images.
You can set parameters for detecting faces, extracting attributes.
You can set rules for matching attributes and creating faces from them.
Args:
policies: object with policies information
description: user description
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` will return json new verifier id and url
"""
return self._handlersClient.createVerifier(
accountId=self._accountId,
policies=policies,
description=description,
**kwargs,
)
[docs]
def getVerifier(self, verifierId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get verifier
Args:
verifierId: verifier id
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with verifier info
"""
return self._handlersClient.getVerifier(
accountId=self._accountId,
verifierId=verifierId,
**kwargs,
)
[docs]
def putVerifier(
self,
verifierId: str,
policies: VerifierPolicies | None = None,
description: str | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Replace verifier
Args:
verifierId: verifier id
policies: object with policies information
description: user description
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: no text returns
"""
return self._handlersClient.putVerifier(
accountId=self._accountId,
verifierId=verifierId,
policies=policies,
description=description,
**kwargs,
)
[docs]
def checkVerifier(self, verifierId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Check verifier existence
Args:
verifierId: verifier id
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: no text returns
"""
return self._handlersClient.checkVerifier(
accountId=self._accountId,
verifierId=verifierId,
**kwargs,
)
[docs]
def deleteVerifier(self, verifierId: str, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Remove verifier
Args:
verifierId: verifier id
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: no text returns
"""
return self._handlersClient.deleteVerifier(
accountId=self._accountId,
verifierId=verifierId,
**kwargs,
)
[docs]
def getVerifiersCount(self, description: str | None = None, **kwargs) -> Awaitable[LunaResponse] | LunaResponse:
"""
Count verifiers that satisfy the filters
Args:
description: verifier description ("like" parameter)
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: verifier count
"""
return self._handlersClient.getVerifiersCount(
accountId=self._accountId,
description=description,
**kwargs,
)
[docs]
def getVerifiers(
self,
description: str | None = None,
page: int = 1,
pageSize: int = 10,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Get verifiers by filters
Args:
description: verifier description ("like" parameter)
page: The number of page
pageSize: The numbers of items on page
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with verifiers info
"""
return self._handlersClient.getVerifiers(
accountId=self._accountId,
description=description,
page=page,
pageSize=pageSize,
**kwargs,
)
[docs]
def verifyRaw(
self,
candidates: list[BinaryReference],
references: list[BinaryReference],
verifierId: str,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Verify raw API allows to match each candidate with each reference descriptors.
Args:
verifierId: verifier id
candidates: list of BinaryReference objects for candidates
references: list of BinaryReference objects for references
verifierId: verifier id
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: results of matching
"""
return self._handlersClient.verifyRaw(
accountId=self._accountId,
candidates=candidates,
references=references,
verifierId=verifierId,
**kwargs,
)
[docs]
def verifyData(
self,
verifierId: str,
inputData: BinaryImage | BinaryDescriptorData | list[BinaryImage | BinaryDescriptorData],
eventIds: list[str] | None = None,
externalIds: list[str] | None = None,
faceIds: list[str] | None = None,
attributeIds: list[str] | None = None,
imageType: int | None = None,
useExifInfo: int | None = None,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Perform data verification against specified objects.
Only one objects set is allowed.
Args:
verifierId: verifier id
inputData: one of:
- Single image
- Single descriptor
- List of images and descriptors
faceIds: faces ids to verify against
eventIds: event ids to verify against
externalIds: external ids to verify against
attributeIds: temporary attribute ids to verify against
imageType: image type (0 - raw image, 1 - face warped image)
useExifInfo: whether to use exif info for auto orientation
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse`: json with verifications info
"""
return self._handlersClient.verifyData(
accountId=self._accountId,
verifierId=verifierId,
inputData=inputData,
eventIds=eventIds,
externalIds=externalIds,
faceIds=faceIds,
attributeIds=attributeIds,
imageType=imageType,
useExifInfo=useExifInfo,
**kwargs,
)