from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from pydantic import ConfigDict, Field, PrivateAttr
from pyobs.interfaces import FitsHeaderEntry
from pyobs.robotic.instruments import InstrumentCapabilities
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.storage.observationarchive import ObservationArchive
from pyobs.robotic.storage.taskarchive import TaskArchive
from pyobs.robotic.scheduler.constraints import Constraint
from pyobs.robotic.scheduler.merits import Merit
from pyobs.utils.serialization import BaseModel, PolymorphicBaseModel
[docs]
@dataclass
class TaskData:
task: Task
observation_archive: ObservationArchive | None = None
task_archive: TaskArchive | None = None
target: Target | None = None
instrument_capabilities: InstrumentCapabilities | None = None
@property
def resolved_target(self) -> Target | None:
"""The target for this specific run -- the observation's own record if known,
otherwise the task's live (possibly shared/reused) target."""
return self.target if self.target is not None else self.task.target
[docs]
class Task(PolymorphicBaseModel):
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
updated_at: str | None = None
"""Last-modification timestamp as emitted by pyobs-portal (DB-derived ``updated_at``,
added in pyobs-portal#84). Carried for round-trip fidelity so the strict ``Task``
model (``extra="forbid"``) accepts the field the portal now sends; not used by pyobs itself."""
_resolved_target: Target | None = PrivateAttr(default=None)
_cant_run_reason: str | None = None
_running_script: Script | 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)
[docs]
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:
script = self.create_script()
can_run = await script.can_run(data)
self._cant_run_reason = script.cant_run_reason()
return can_run
return True
[docs]
def cant_run_reason(self) -> str | None:
"""Returns reason why task cannot run, or None if it can."""
return self._cant_run_reason
@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
[docs]
async def run(self, data: TaskData | None) -> None:
"""Run a task"""
if self.script is not None:
script = self.create_script()
self._running_script = script
await script.run(data)
[docs]
def is_finished(self) -> bool:
"""Whether task is finished."""
return False
def estimate_duration(
self, time: Time | None = None, instrument_capabilities: InstrumentCapabilities | None = None
) -> float:
if self.script:
return self.create_script().estimate_duration(
TaskData(task=self, instrument_capabilities=instrument_capabilities), time
)
return self.duration
[docs]
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
[docs]
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
[docs]
def reset_resolved_target(self) -> None:
"""Reset resolved target."""
self._resolved_target = None
@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
[docs]
class Project(BaseModel):
code: str
name: str = ""
priority: float | None = Field(ge=0.0, le=9999.0, default=1.0)
users: list[str] = Field(default_factory=list)
public: bool = False # ingest backend `public` flag; default keeps old backends valid
updated_at: str | None = None
"""Last-modification timestamp as emitted by pyobs-portal (DB-derived ``updated_at``,
added in pyobs-portal#134). Carried for round-trip fidelity so the strict ``Project`` model
(``extra="forbid"``) accepts the field the portal now sends; not used by pyobs itself."""
__all__ = ["Task", "TaskData", "Project"]