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