Task

task

__semver__ module-attribute

__semver__ = pep440_to_semver(__version__)

TaskParameters

Bases: TaskParameters

Base class for storing parameters for the task.

rng_seed class-attribute instance-attribute

rng_seed = Field(
    default=None,
    description="Seed of the random number generator",
)

aind_behavior_services_pkg_version class-attribute instance-attribute

aind_behavior_services_pkg_version = Field(
    default=__semver__,
    pattern=SEMVER_REGEX,
    title="aind_behavior_services package version",
    frozen=True,
)

coerce_version classmethod

coerce_version(v, ctx)

Source code in src/aind_behavior_services/task/__init__.py

@field_validator("aind_behavior_services_pkg_version", mode="before", check_fields=False)
@classmethod
def coerce_version(cls, v: str, ctx) -> str:
    return coerce_schema_version(cls, v, ctx.field_name)

Task

Bases: Task

Base class for task schemas.

task_parameters class-attribute instance-attribute

task_parameters = Field(
    description="Parameters of the task",
    validate_default=True,
)

version class-attribute instance-attribute

version = Field(
    pattern=curriculum_task.SEMVER_REGEX,
    description="task schema version",
)

coerce_version classmethod

coerce_version(v)

Source code in src/aind_behavior_services/task/__init__.py

@field_validator("version", mode="before", check_fields=False)
@classmethod
def coerce_version(cls, v: str) -> str:
    return coerce_schema_version(cls, v)

coerce_schema_version

coerce_schema_version(cls, v, version_string='version')

Source code in src/aind_behavior_services/base.py

def coerce_schema_version(cls: type[BaseModel], v: str, version_string: str = "version") -> str:
    semver = Version.parse(v)

    _default_schema_version: Version | None = None

    try:  # Get the default schema version from the model literal field
        annotation = cls.model_fields[version_string].annotation
        if get_origin(annotation) is Literal:
            _default_schema_version = Version.parse(get_args(annotation)[0])
    except IndexError:  # This handles the case where the base class does not define a literal schema_version value
        return v

    if _default_schema_version is None:
        return v

    if semver != _default_schema_version:
        logger.warning(
            "Deserialized versioned field %s, expected %s. Will attempt to coerce. "
            "This will be considered a best-effort operation.",
            semver,
            _default_schema_version,
        )
    return str(_default_schema_version)