Source code for pyobs.robotic.task

from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from pydantic import Field, PrivateAttr, ConfigDict

from pyobs.robotic.scheduler import DataProvider
from pyobs.robotic.scheduler.targets import Target
from pyobs.robotic.scripts import Script
from pyobs.utils.time import Time

if TYPE_CHECKING:
    from pyobs.robotic.observationarchive import ObservationArchive
    from pyobs.robotic.taskarchive import TaskArchive

from pyobs.robotic.scheduler.constraints import Constraint
from pyobs.robotic.scheduler.merits import Merit
from pyobs.utils.serialization import BaseModel




@dataclass
class TaskData:
    task: Task
    observation_archive: ObservationArchive | None = None
    task_archive: TaskArchive | None = None





class Task(BaseModel):
    id: Any | None = None
    name: str = Field(default="")
    project: str = Field(default="")
    duration: float = Field(ge=0.0, le=84000.0, default=0.0)
    priority: float | None = Field(ge=0.0, le=9999.0, default=1.0)
    constraints: list[Constraint] = Field(default_factory=list)
    merits: list[Merit] = Field(default_factory=list)
    static_target: Target | None = Field(default=None, alias="target")
    script: dict[str, Any] = Field(default_factory=dict)
    active: bool = True

    _resolved_target: Target | None = PrivateAttr(default=None)

    model_config = ConfigDict(populate_by_name=True)

    def __str__(self) -> str:
        s = f"Task {self.id}: {self.name} (duration: {self.duration}s"
        if self.priority is not None:
            s += f", priority: {self.priority}"
        if self.target is not None:
            s += f", target: {self.target.name}"
        s += ")"
        return s

    def create_script(self) -> Script:
        return self.pyobs_model_validate(Script, self.script, by_alias=True)



    async def can_run(self, data: TaskData | None) -> bool:
        """Checks whether this task could run now.

        Returns:
            True, if the task can run now.
        """
        if self.script is not None:
            return await self.create_script().can_run(data)
        return True


    @property
    def can_start_late(self) -> bool:
        """Whether this tasks is allowed to start later than the user-set time, e.g. for flatfields.

        Returns:
            True, if task can start late.
        """
        return False



    async def run(self, data: TaskData | None) -> None:
        """Run a task"""
        if self.script is not None:
            await self.create_script().run(data)




    def is_finished(self) -> bool:
        """Whether task is finished."""
        return False




    def get_fits_headers(self, namespaces: list[str] | None = None) -> dict[str, tuple[Any, str]]:
        """Returns FITS header for the current status of this module.

        Args:
            namespaces: If given, only return FITS headers for the given namespaces.

        Returns:
            Dictionary containing FITS headers.
        """
        return {}


    def estimate_duration(self) -> float:
        if self.script:
            return self.create_script().estimate_duration()
        return self.duration



    async def resolve_target(self, time: Time, task: Task, data: DataProvider) -> bool:
        """Resolve dynamic target. Returns False if no valid target found."""
        if self.static_target is None:
            return True
        if self._resolved_target is not None:
            return True
        self._resolved_target = await self.static_target.resolve(time, self, data)
        return self._resolved_target is not None




    def set_resolved_target(self, target: Target | None) -> None:
        """Set the resolved target if not already set, e.g. when restoring from an observation."""
        if self._resolved_target is None:
            self._resolved_target = target


    @property
    def target(self) -> Target | None:
        """The resolved target, or the static target if not dynamic."""
        if self._resolved_target is not None:
            return self._resolved_target
        return self.static_target





class Project(BaseModel):
    code: str
    name: str = ""
    priority: float | None = Field(ge=0.0, le=9999.0, default=1.0)



__all__ = ["Task", "TaskData", "Project"]