Source code for luna_handlers.classes.schemas.attribute_upgrade
"""
Module contains schemas for attribute upgrade handler
"""
from pydantic import Field, constr, root_validator
from vlutils.regexps import UUID_REGEXP_STR
from classes.schemas import types
from classes.schemas.base_schema import BaseSchema, HandlerSettings
from classes.schemas.types import OptionalNotNullable
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
SamplesIDs = types.userConstrainedList(
constr(regex=UUID_REGEXP_STR), minItems=1, maxItemsGetter=lambda: HandlerSettings.receivedImagesLimit
)
[docs]class FaceUpgradePolicy(BaseSchema):
"""Face attribute upgrade policy"""
_descriptorVersion: int
# whether to extract basic attributes
extractBasicAttributes: types.Int01 = 0
# descriptor quality score threshold
fdScoreThreshold: types.Float01 = 0.0
# whether to extract face descriptor
extractDescriptor: types.Int01 = 1
# descriptor version
descriptorVersion: types.FaceDescriptorVersion = Field(
default_factory=lambda: HandlerSettings.faceDescriptorVersion
)
[docs] @root_validator(pre=False, skip_on_failure=True)
def validatePolicy(cls, values):
"""
Validate attributes are select for extract
Raises:
VLException(Error.NotSelectedAttributesForExtract, 400, False) if no attributes select for extract
"""
isExtractBasicAttributes = values["extractBasicAttributes"]
isExtractDescriptor = values["extractDescriptor"]
if not isExtractDescriptor and not isExtractBasicAttributes:
raise VLException(Error.NotSelectedAttributesForExtract, 400, False)
return values
[docs]class BodyUpgradePolicy(BaseSchema):
"""Body attribute upgrade policy"""
# descriptor version
descriptorVersion: types.BodyDescriptorVersion = Field(
default_factory=lambda: HandlerSettings.bodyDescriptorVersion
)
[docs]class FaceUpgradeData(BaseSchema):
"""Face upgrade data schema"""
# sample ids
samples: SamplesIDs
# face attribute upgrade policy
policy: FaceUpgradePolicy = FaceUpgradePolicy()
[docs]class BodyUpgradeData(BaseSchema):
"""Body upgrade data schema"""
# sample ids
samples: SamplesIDs
# body attribute upgrade policy
policy: BodyUpgradePolicy = BodyUpgradePolicy()
[docs]class SamplesForUpgrade(BaseSchema):
"""Samples for upgrade schema"""
# face id
faceId: constr(regex=UUID_REGEXP_STR) = OptionalNotNullable()
# face upgrade data
upgradeData: FaceUpgradeData = OptionalNotNullable()
# event id
eventId: constr(regex=UUID_REGEXP_STR) = OptionalNotNullable()
# face upgrade data
faceUpgradeData: FaceUpgradeData = OptionalNotNullable()
# body upgrade data
bodyUpgradeData: BodyUpgradeData = OptionalNotNullable()
[docs] @root_validator(pre=False, skip_on_failure=True)
def validateOneOf(cls, values):
"""
Validate one of request value:
'face_id' and 'upgrade_data' or
'event_id' and 'face_upgrade_data' and/or 'body_upgrade_data'
"""
existingKeys = {key for key, value in values.items() if value}
isValidForFace = existingKeys == {"faceId", "upgradeData"}
isValidForEvent = (
"eventId" in existingKeys
and ("faceUpgradeData" in existingKeys or "bodyUpgradeData" in existingKeys)
and not (existingKeys.intersection({"faceId", "samples"}))
)
if isValidForFace == isValidForEvent:
raise ValueError(
"Required one of: `face_id` with `upgrade_data` or "
"`event_id` with `face_upgrade_data` and/or `body_upgrade_data` in request"
)
return values