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