Source code for luna_handlers.classes.schemas.verifier

"""
Module contains schemas for verifier
"""

import asyncio
from itertools import chain
from typing import Any

from luna3.client import Client
from luna3.remote_sdk import http_objs as sdk
from pydantic import Field

from classes.event import HandlerEvent as Event
from classes.schemas import types
from classes.schemas.base_schema import BaseSchema
from classes.schemas.detect_policies import BaseDetectPolicy
from classes.schemas.extract_policies import BaseExtractPolicy
from classes.schemas.handler import FaceInputEstimationsModel
from classes.schemas.match_policy import BaseMatchPolicy
from classes.schemas.storage_policy import AttributeStorePolicy, FaceSamplePolicy
from crutches_on_wheels.cow.monitoring.points import DataForMonitoring, monitorTime


[docs] class VerifierMatchPolicy(BaseMatchPolicy): """Verifier match policy schema"""
[docs] class VerifierDetectPolicy(BaseDetectPolicy): """Verifier detect policy"""
[docs] class VerifierExtractPolicy(BaseExtractPolicy): """Verifier extract policy"""
[docs] class VerifierAttributeStorePolicy(BaseSchema): """Verifier attribute storage policy""" # whether to store attribute storeAttribute: types.Int01 = 0
[docs] async def execute(self, events: list[Event], accountId: str, luna3Client: Client) -> None: """ Save attributes. Args: events: events accountId: account id luna3Client: client """ await AttributeStorePolicy(store_attribute=self.storeAttribute).execute(events, accountId, luna3Client)
[docs] class VerifierFaceSampleStorePolicy(BaseSchema): """Verifier face sample storage policy""" # whether to store face sample storeSample: types.Int01 = 0 ttl: types.IntTTL | None = None
[docs] def asDict(self) -> dict: """Get ImageOriginPolicy as dict with ttl""" result = super().asDict() result["ttl"] = self.ttl return result
[docs] async def execute(self, events: list[Event], accountId, bucket: str, luna3Client: Client): """ Save face samples. Args: events: events accountId: account id bucket: bucket name luna3Client: client """ await FaceSamplePolicy(store_sample=self.storeSample).execute(events, accountId, bucket, luna3Client)
[docs] class VerifierStoragePolicy(BaseSchema): """Verifier storage policy""" # attribute storage policy attributePolicy: VerifierAttributeStorePolicy | None = VerifierAttributeStorePolicy() # face sample storage policy faceSamplePolicy: VerifierFaceSampleStorePolicy | None = VerifierFaceSampleStorePolicy()
[docs] async def execute( self, events: list[Event], accountId: str, monitoring: DataForMonitoring, facesBucket: str, luna3Client: Client, ): """ Execute storage policy - save objects. Args: events: events accountId: account id monitoring: monitoring data facesBucket: faces bucket luna3Client: luna3 client """ async def _faceSample() -> None: with monitorTime(monitoring, "face_sample_storage_policy_time"): await self.faceSamplePolicy.execute(events, accountId, facesBucket, luna3Client) async def _attribute() -> None: with monitorTime(monitoring, "face_attribute_storage_policy_time"): await self.attributePolicy.execute(events, accountId, luna3Client) await _faceSample() # save attribute and face only after executing previous policies (^^^ samples are updated here ^^^) await _attribute()
[docs] class VerifierPoliciesModel(BaseSchema): """Verifier policies""" __slots__ = ("matchPolicy",) # detect policy detectPolicy: VerifierDetectPolicy = Field(default_factory=lambda: VerifierDetectPolicy()) # extract policy extractPolicy: VerifierExtractPolicy = VerifierExtractPolicy() # storage policy storagePolicy: VerifierStoragePolicy = VerifierStoragePolicy() # verification threshold verificationThreshold: types.StrictFloat01 = 0.9
[docs] def model_post_init(self, __context: Any) -> None: object.__setattr__(self, "matchPolicy", [])
@property def estimator(self): targets = sdk.Targets( exif=self.detectPolicy.extractExif, faceDetection=1, faceLandmarks5=1, faceLandmarks68=self.detectPolicy.detectLandmarks68, faceWarp=1 * self.storagePolicy.faceSamplePolicy.storeSample, gaze=self.detectPolicy.estimateGaze, headPose=self.detectPolicy.estimateHeadPose, eyes=self.detectPolicy.estimateEyesAttributes, mouthAttributes=self.detectPolicy.estimateMouthAttributes, faceWarpQuality=self.detectPolicy.estimateQuality, emotions=self.detectPolicy.estimateEmotions, mask=self.detectPolicy.estimateMask, glasses=self.detectPolicy.estimateGlasses, liveness=self.detectPolicy.estimateLiveness.estimate, deepfake=self.detectPolicy.estimateDeepfake.estimate, faceQuality=self.detectPolicy.faceQuality if self.detectPolicy.faceQuality.estimate else None, faceDescriptor=1, basicAttributes=self.extractPolicy.extractBasicAttributes, ) filters = sdk.Filters( faceDetectionFilters=sdk.FaceDetectionFilters( rollThreshold=self.detectPolicy.rollThreshold, yawThreshold=self.detectPolicy.yawThreshold, pitchThreshold=self.detectPolicy.pitchThreshold, livenessStates=self.detectPolicy.livenessStates, deepfakeStates=self.detectPolicy.deepfakeStates, maskStates=self.detectPolicy.maskStates, scoreThreshold=self.extractPolicy.fdScoreThreshold, ) ) estimationConfig = sdk.EstimatorsParams( livenessParams=sdk.LivenessParams( scoreThreshold=self.detectPolicy.estimateLiveness.livenessThreshold, qualityThreshold=self.detectPolicy.estimateLiveness.qualityThreshold, ), deepfakeParams=sdk.DeepfakeParams( realThreshold=self.detectPolicy.estimateDeepfake.realThreshold, mode=self.detectPolicy.estimateDeepfake.mode, ), ) params = sdk.Params( targets=targets, filters=filters, estimatorsParams=estimationConfig, multifacePolicy=self.detectPolicy.multifacePolicy, aggregate=0, ) return sdk.Estimator(params, images=...)
[docs] async def execute( self, events: list[Event], accountId: str, monitoring: DataForMonitoring, facesBucket: str, luna3Client: Client, ): """ Executes given policies against provided data. Args: events: events accountId: account id monitoring: monitoring data facesBucket: faces bucket luna3Client: luna platform client """ # storage await self.storagePolicy.execute( events, accountId=accountId, monitoring=monitoring, facesBucket=facesBucket, luna3Client=luna3Client, ) # matching with monitorTime(monitoring, "match_policy_time"): await asyncio.gather( *[matchByListPolicy.execute(events, luna3Client) for matchByListPolicy in self.matchPolicy] ) for event in events: event.raw["verifications"] = [] for verification in chain(*(match["candidates"] for match in event.raw["matches"])): verification["status"] = verification["similarity"] >= self.verificationThreshold event.raw["verifications"].append(verification)
[docs] class VerifierModel(BaseSchema): """Verifier""" # verifier description description: types.Str128 = "" # verifier policies policies: VerifierPoliciesModel = Field(default_factory=lambda: VerifierPoliciesModel())
[docs] class VerifierInputEstimationsModel(FaceInputEstimationsModel): """Verifier model for incoming estimations: urls, image (only face bounding boxes available) or samples"""