Source code for luna_handlers.classes.schemas.attribute_upgrade
"""
Module contains schemas for attribute upgrade handler
"""
from typing import Literal
from pydantic import Field, constr, model_validator
from vlutils.regexps import UUID_REGEXP_STR
from vlutils.structures.pydantic import UpdatableList
from classes.schemas import types
from classes.schemas.base_schema import BaseSchema, HandlerSettings
from classes.schemas.types import ALLOWED_BODY_DESCRIPTOR_VERSIONS, ALLOWED_FACE_DESCRIPTOR_VERSIONS, EventId, FaceId
from crutches_on_wheels.cow.errors.errors import Error
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.errors.pydantic_errors import PydanticError
from crutches_on_wheels.cow.pydantic.types import OptionalNotNullable
SamplesIDs = UpdatableList(
constr(pattern=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.StrictFloat01 = 0.0
# whether to extract face descriptor
extractDescriptor: types.Int01 = 1
# descriptor version
descriptorVersion: Literal[tuple(ALLOWED_FACE_DESCRIPTOR_VERSIONS)] = Field(
default_factory=lambda: HandlerSettings.faceDescriptorVersion
)
[docs]
@model_validator(mode="after")
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: Literal[tuple(ALLOWED_BODY_DESCRIPTOR_VERSIONS)] = 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: FaceId = OptionalNotNullable()
# face upgrade data
upgradeData: FaceUpgradeData = OptionalNotNullable()
# event id
eventId: EventId = OptionalNotNullable()
# face upgrade data
faceUpgradeData: FaceUpgradeData = OptionalNotNullable()
# body upgrade data
bodyUpgradeData: BodyUpgradeData = OptionalNotNullable()
[docs]
@model_validator(mode="before")
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 == {"face_id", "upgrade_data"}
isValidForEvent = (
"event_id" in existingKeys
and ("face_upgrade_data" in existingKeys or "body_upgrade_data" in existingKeys)
and not (existingKeys.intersection({"face_id", "samples"}))
)
if isValidForFace == isValidForEvent:
raise PydanticError.PydanticValidationError.format(
"Required one of: `face_id` with `upgrade_data` or "
"`event_id` with `face_upgrade_data` and/or `body_upgrade_data` in request",
)()
return values