Source code for luna_handlers.classes.schemas.types

"""
Module contains custom basics for validation schemas
It highly recommended to use represented classes instead of pydantic defaults!
"""
from uuid import UUID

import fastjsonschema
from pydantic import BeforeValidator, model_validator
from typing_extensions import Annotated, Literal
from vlutils.descriptors.data import DescriptorsEnum, DescriptorType
from vlutils.regexps import UUID_REGEXP_STR

from classes.schemas.base_schema import BaseSchema, SchemaUpdaterMixin
from classes.schemas.validators import isNumber, uuidRegexValidator
from crutches_on_wheels.cow.maps.vl_maps import DEEPFAKE_MAP, EMOTION_MAP, ETHNIC_MAP, LIVENESS_MAP, MASK_MAP
from crutches_on_wheels.cow.pydantic.types import CustomFloat, CustomInt, CustomString
from crutches_on_wheels.cow.web.query_getters import timeFilterGetter

MAX_POLICY_LIST_LENGTH = 30
MAX_MATCH_FILTER_LIST_LENGTH = 30
MAX_ITEMS_LIST_LENGTH = 1000
DEFAULT_MATCH_LIMIT = 3
_SDK_INT_MAGIC_NUMBER = 2**30
MAX_TAG_LENGTH = 36
DATETIME_SCHEMA = fastjsonschema.compile({"type": "string", "format": "date-time"})
DATETIME_NOW_SCHEMA = fastjsonschema.compile({"type": "string", "pattern": r"^now-(\d+)[smhdwMy]$"})


UUIDObject = Annotated[UUID, BeforeValidator(uuidRegexValidator)]


[docs] class ObjectId(CustomString): """ID type, string uuid""" pattern = rf"^{UUID_REGEXP_STR}"
[docs] class IntNonNegative(CustomInt): """Int non negative""" ge = 0
# Int in range 0 - 1 Int01 = Literal[0, 1] # Int in range 0 - 2 Int02 = Literal[0, 1, 2]
[docs] class SDKInt(CustomInt): """Int in range from min to max SDK magic numbers""" ge = -_SDK_INT_MAGIC_NUMBER le = _SDK_INT_MAGIC_NUMBER
[docs] class IntEthnicities(CustomInt): """Int in range of available ethnicities values""" ge = 1 le = len(ETHNIC_MAP)
[docs] class IntEmotions(CustomInt): """Int in range of available emotions values""" ge = 1 le = len(EMOTION_MAP)
[docs] class IntMasks(CustomInt): """Int in range of available masks values""" ge = 1 le = len(MASK_MAP)
[docs] class IntLiveness(CustomInt): """Int in range of available liveness state values""" ge = min(LIVENESS_MAP.values()) le = max(LIVENESS_MAP.values())
[docs] class IntDeepfake(CustomInt): """Int in range of available deepfake state values""" ge = min(DEEPFAKE_MAP.values()) le = max(DEEPFAKE_MAP.values())
[docs] class IntAge(CustomInt): """Int in range 0 - 100""" ge = 0 le = 100
[docs] class IntAngle180(CustomInt): """Int for angles - in range 0 - 180""" ge = 0 le = 180
[docs] class IntMatchingLimit(CustomInt, SchemaUpdaterMixin): """Int in range 0 - RESULT_CANDIDATE_LIMIT""" ge = 1 le = SchemaUpdaterMixin.settings.resultCandidateLimit
[docs] @classmethod def update(cls): cls.le = cls.settings.resultCandidateLimit
[docs] class IntAttributeTTL(CustomInt, SchemaUpdaterMixin): """Int in range 0 - MAX_ATTRIBUTE_TTL""" ge = 1 le = SchemaUpdaterMixin.settings.maxAttributeTTL
[docs] @classmethod def update(cls): cls.le = cls.settings.maxAttributeTTL
[docs] class StrictFloat01(CustomFloat): """Strict float in range 0.0 - 1.0""" ge = 0.0 le = 1.0
[docs] class StrictFloatNonNegative(CustomFloat): """Strict float non negative""" ge = 0.0
[docs] class DetectTs(CustomFloat): """User-defined timestamp relative to something, such as the start of a video""" ge = 0.0 le = 158731466399.999
[docs] class Float180(CustomFloat): """Strict float in range -180 to 180""" ge = -180.0 le = 180.0
class _ConstrainedFloat(CustomFloat): """Custom type for float constrained by size and type limits, and multiply result number""" @classmethod def __get_validators__(cls) -> "CallableGenerator": """ Validator generator """ yield isNumber yield cls.numberSizeValidator yield cls.numberMultipleValidator
[docs] class FloatLongitude(_ConstrainedFloat): """Non-strict float for longitude in range -180.0 - 180.0""" ge = -180.0 le = 180.0
[docs] class FloatLatitude(_ConstrainedFloat): """Non-strict float for latitude in range -90.0 - 90.0""" ge = -90.0 le = 90.0
[docs] class FloatGeoDelta(_ConstrainedFloat): """Non-strict float for latitude in range 0.0 - 9.0""" ge = 0.0 le = 90.0
[docs] class Str36(CustomString): """Strict string 36 len max""" max_length = 36
[docs] class Str128(CustomString): """Strict string 128 len max""" max_length = 128
[docs] class Str256(CustomString): """Strict string 128 len max""" max_length = 256
[docs] class GeoPosition(BaseSchema): # (Optional[str]): latitude latitude: FloatLatitude # (Optional[str]): longitude longitude: FloatLongitude
ALLOWED_FACE_DESCRIPTOR_VERSIONS = [d.value.version for d in DescriptorsEnum if d.value.type == DescriptorType.face] ALLOWED_BODY_DESCRIPTOR_VERSIONS = [d.value.version for d in DescriptorsEnum if d.value.type == DescriptorType.body]
[docs] class LocationName(CustomString): """ Location name type """ # max name length max_length = 36
[docs] class UserData(CustomString): """ Event user_data type """ # max length max_length = 128
[docs] class ExternalId(CustomString): """External id type""" # max length max_length = 36
[docs] class FaceId(ObjectId): """Face ID type"""
[docs] class EventId(ObjectId): """Event ID type"""
[docs] class StreamId(ObjectId): """Stream ID type"""
[docs] class DateTimeString(CustomString): """ Datetime string type/ rfc3339 """
[docs] @model_validator(mode="after") def validateTime(self, value): """ Validate that value can be parsed as time Args: value: time in rfc3339 Returns: value """ return str(timeFilterGetter(value))