Source code for ska_sdp_config.entity.script

"""Script model."""

import ast
from typing import Annotated, Literal

from pydantic import Field, field_validator

from .base import EntityBaseModel, EntityKeyBaseModel, MultiEntityBaseModel
from .resource_management import SUPPORTED_KIND




class ScriptResources(EntityBaseModel):
    """The Resources estimated to be required by an SDP Script."""



    @field_validator("quantity")
    @classmethod
    def quantity_valid_syntax(cls, quantity_definition: str) -> str:
        """Ensure the quantity is specified in a python parsable expression."""
        try:
            ast.parse(quantity_definition.lstrip())
            return quantity_definition
        except SyntaxError as exc:
            raise ValueError(
                "Invalid syntax detected in script resources."
            ) from exc


    kind: Annotated[
        Literal[tuple(SUPPORTED_KIND)],
        Field(pattern=r"^[a-z-]{1,96}$"),
    ]
    """The kind of resource."""

    name: Annotated[str, Field(pattern=r"^[a-zA-Z0-9_-]+$")]
    """The name of the resource definition."""

    phases: list[str] = []
    """The phases that this applies to."""

    quantity: str
    """The quantity as a python parsable expression."""





class Script(MultiEntityBaseModel):
    """An SDP Script."""



    class Key(EntityKeyBaseModel):
        """An SDP Script primary key."""

        kind: Annotated[str, Field(pattern=r"^(realtime)|(batch)$")]
        """The kind of this script (realtime or batch)."""

        name: Annotated[str, Field(pattern=r"^[a-zA-Z0-9_-]+$")]
        """The name of this script."""

        version: Annotated[str, Field(pattern=r"^[a-zA-Z0-9_\.-]+$")]
        """The version of this script."""

        def __str__(self):
            return f"{self.kind}:{self.name}:{self.version}"


    key: Annotated[Key, Field(exclude=True)]

    image: Annotated[str, Field(pattern=r"^[a-zA-Z0-9_:\./-]+$")]
    """The OCI image used to launch this script."""

    parameters: Annotated[dict, Field(default_factory=dict)]
    """Parameters for the script."""

    sdp_version: Annotated[
        str | None,
        Field(
            pattern=(
                r"^(?:[><]=?|==)\d+\.\d+\.\d+"
                r"(?:,\s*(?:[><]=?|==)\d+\.\d+\.\d+)*$"
            )
        ),
    ] = None
    """The range of SDP versions this script is compatible with."""

    resources: list[ScriptResources] = []