Changes to scheduler structure to allow for different ones and new merit scheduler

pyobs/robotic/__init__.py

@@ -1,4 +1,4 @@
 from .taskschedule import TaskSchedule
-from .task import Task
+from .task import Task, ScheduledTask
 from .taskarchive import TaskArchive
 from .taskrunner import TaskRunner

pyobs/robotic/lco/task.py

@@ -1,8 +1,20 @@
 from __future__ import annotations
 import logging
-from typing import Union, Dict, Tuple, Optional, List, Any, TYPE_CHECKING, cast
+from typing import Any, TYPE_CHECKING, cast
+from astropy.coordinates import SkyCoord
+import astropy.units as u
 
 from pyobs.object import get_object
+from pyobs.robotic.scheduler.constraints import (
+    TimeConstraint,
+    Constraint,
+    AirmassConstraint,
+    MoonSeparationConstraint,
+    MoonIlluminationConstraint,
+    SolarElevationConstraint,
+)
+from pyobs.robotic.scheduler.merits import Merit
+from pyobs.robotic.scheduler.targets import Target, SiderealTarget
 from pyobs.robotic.scripts import Script
 from pyobs.robotic.task import Task
 from pyobs.utils.logger import DuplicateFilter
@@ -25,13 +37,13 @@ class ConfigStatus:
     def __init__(self, state: str = "ATTEMPTED", reason: str = ""):
         """Initializes a new Status with an ATTEMPTED."""
         self.start: Time = Time.now()
-        self.end: Optional[Time] = None
+        self.end: Time | None = None
         self.state: str = state
         self.reason: str = reason
         self.time_completed: float = 0.0
 
     def finish(
-        self, state: Optional[str] = None, reason: Optional[str] = None, time_completed: float = 0.0
+        self, state: str | None = None, reason: str | None = None, time_completed: float = 0.0
     ) -> "ConfigStatus":
         """Finish this status with the given values and the current time.
 
@@ -48,7 +60,7 @@ def finish(
         self.end = Time.now()
         return self
 
-    def to_json(self) -> Dict[str, Any]:
+    def to_json(self) -> dict[str, Any]:
         """Convert status to JSON for sending to portal."""
         return {
             "state": self.state,
@@ -65,45 +77,89 @@ def to_json(self) -> Dict[str, Any]:
 class LcoTask(Task):
     """A task from the LCO portal."""
 
-    def __init__(self, config: Dict[str, Any], **kwargs: Any):
+    def __init__(
+        self,
+        config: dict[str, Any],
+        id: Any | None = None,
+        name: str | None = None,
+        duration: float | None = None,
+        **kwargs: Any,
+    ):
         """Init LCO task (called request there).
 
         Args:
             config: Configuration for task
         """
-        Task.__init__(self, **kwargs)
-
-        # store stuff
-        self.config = config
-        self.cur_script: Optional[Script] = None
 
-    @property
-    def id(self) -> Any:
-        """ID of task."""
-        if "request" in self.config and "id" in self.config["request"]:
-            return self.config["request"]["id"]
-        else:
-            raise ValueError("No id found in request.")
-
-    @property
-    def name(self) -> str:
-        """Returns name of task."""
-        if "name" in self.config and isinstance(self.config["name"], str):
-            return self.config["name"]
-        else:
-            raise ValueError("No name found in request group.")
+        req = config["request"]
+        if id is None:
+            id = req["id"]
+        if name is None:
+            name = req["id"]
+        if duration is None:
+            duration = float(req["duration"])
+
+        if "constraints" not in kwargs:
+            kwargs["constraints"] = self._create_constraints(req)
+        if "merits" not in kwargs:
+            kwargs["merits"] = self._create_merits(req)
+        if "target" not in kwargs:
+            kwargs["target"] = self._create_target(req)
+
+        Task.__init__(
+            self,
+            id=id,
+            name=name,
+            duration=duration,
+            config=config,
+            **kwargs,
+        )
 
-    @property
-    def duration(self) -> float:
-        """Returns estimated duration of task in seconds."""
-        if (
-            "request" in self.config
-            and "duration" in self.config["request"]
-            and isinstance(self.config["request"]["duration"], int)
-        ):
-            return float(self.config["request"]["duration"])
+        # store stuff
+        self.cur_script: Script | None = None
+
+    @staticmethod
+    def _create_constraints(req: dict[str, Any]) -> list[Constraint]:
+        # get constraints
+        constraints: list[Constraint] = []
+
+        # time constraints?
+        if "windows" in req:
+            constraints.extend([TimeConstraint(Time(wnd["start"]), Time(wnd["end"])) for wnd in req["windows"]])
+
+        # take first config
+        cfg = req["configurations"][0]
+
+        # constraints
+        if "constraints" in cfg:
+            c = cfg["constraints"]
+            if "max_airmass" in c and c["max_airmass"] is not None:
+                constraints.append(AirmassConstraint(c["max_airmass"]))
+            if "min_lunar_distance" in c and c["min_lunar_distance"] is not None:
+                constraints.append(MoonSeparationConstraint(c["min_lunar_distance"]))
+            if "max_lunar_phase" in c and c["max_lunar_phase"] is not None:
+                constraints.append(MoonIlluminationConstraint(c["max_lunar_phase"]))
+                # if max lunar phase <= 0.4 (which would be DARK), we also enforce the sun to be <-18 degrees
+                if c["max_lunar_phase"] <= 0.4:
+                    constraints.append(SolarElevationConstraint(-18.0))
+
+        return constraints
+
+    def _create_merits(self, req: dict[str, Any]) -> list[Merit]:
+        # take merits from first config
+        cfg = req["configurations"][0]
+        return [self.get_object(m) for m in cfg["merits"]] if "merits" in cfg else []
+
+    def _create_target(self, req: dict[str, Any]) -> Target | None:
+        # target
+        target = req["configurations"][0]["target"]
+        if "ra" in target and "dec" in target:
+            coord = SkyCoord(target["ra"] * u.deg, target["dec"] * u.deg, frame=target["type"].lower())
+            name = target["name"]
+            return SiderealTarget(name, coord)
         else:
-            raise ValueError("No duration found in request.")
+            log.warning("Unsupported coordinate type.")
+            return None
 
     def __eq__(self, other: object) -> bool:
         """Compares to tasks."""
@@ -112,22 +168,6 @@ def __eq__(self, other: object) -> bool:
         else:
             return False
 
-    @property
-    def start(self) -> Time:
-        """Start time for task"""
-        if "start" in self.config and isinstance(self.config["start"], Time):
-            return self.config["start"]
-        else:
-            raise ValueError("No start time found in request group.")
-
-    @property
-    def end(self) -> Time:
-        """End time for task"""
-        if "end" in self.config and isinstance(self.config["end"], Time):
-            return self.config["end"]
-        else:
-            raise ValueError("No end time found in request group.")
-
     @property
     def observation_type(self) -> str:
         """Returns observation_type of this task.
@@ -149,7 +189,7 @@ def can_start_late(self) -> bool:
         """
         return self.observation_type == "DIRECT"
 
-    def _get_config_script(self, config: Dict[str, Any], scripts: Optional[Dict[str, Script]] = None) -> Script:
+    def _get_config_script(self, config: dict[str, Any], scripts: dict[str, Script] | None = None) -> Script:
         """Get config script for given configuration.
 
         Args:
@@ -176,7 +216,7 @@ def _get_config_script(self, config: Dict[str, Any], scripts: Optional[Dict[str,
             observer=self.observer,
         )
 
-    async def can_run(self, scripts: Optional[Dict[str, Script]] = None) -> bool:
+    async def can_run(self, scripts: dict[str, Script] | None = None) -> bool:
         """Checks, whether this task could run now.
 
         Returns:
@@ -206,9 +246,9 @@ async def can_run(self, scripts: Optional[Dict[str, Script]] = None) -> bool:
     async def run(
         self,
         task_runner: TaskRunner,
-        task_schedule: Optional[TaskSchedule] = None,
-        task_archive: Optional[TaskArchive] = None,
-        scripts: Optional[Dict[str, Script]] = None,
+        task_schedule: TaskSchedule | None = None,
+        task_archive: TaskArchive | None = None,
+        scripts: dict[str, Script] | None = None,
     ) -> None:
         """Run a task"""
         from pyobs.robotic.lco import LcoTaskSchedule
@@ -217,7 +257,7 @@ async def run(
         req = self.config["request"]
 
         # loop configurations
-        status: Optional[ConfigStatus]
+        status: ConfigStatus | None
         for config in req["configurations"]:
             # send an ATTEMPTED status
             if isinstance(task_schedule, LcoTaskSchedule):
@@ -253,9 +293,9 @@ async def _run_script(
         self,
         script: Script,
         task_runner: TaskRunner,
-        task_schedule: Optional[TaskSchedule] = None,
-        task_archive: Optional[TaskArchive] = None,
-    ) -> Union[ConfigStatus, None]:
+        task_schedule: TaskSchedule | None = None,
+        task_archive: TaskArchive | None = None,
+    ) -> ConfigStatus | None:
         """Run a config
 
         Args:
@@ -308,7 +348,7 @@ def is_finished(self) -> bool:
         else:
             return False
 
-    def get_fits_headers(self, namespaces: Optional[List[str]] = None) -> Dict[str, Tuple[Any, str]]:
+    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:

pyobs/robotic/lco/taskarchive.py

@@ -1,21 +1,11 @@
 import logging
-from typing import List, Dict, Optional, Any
-from astroplan import (
-    TimeConstraint,
-    AirmassConstraint,
-    ObservingBlock,
-    FixedTarget,
-    MoonSeparationConstraint,
-    MoonIlluminationConstraint,
-    AtNightConstraint,
-)
-from astropy.coordinates import SkyCoord
-import astropy.units as u
+from typing import Dict, Optional, Any
 
 from pyobs.utils.time import Time
 from pyobs.robotic.taskarchive import TaskArchive
 from .portal import Portal
 from .task import LcoTask
+from .. import Task
 
 log = logging.getLogger(__name__)
 
@@ -67,11 +57,11 @@ async def last_changed(self) -> Optional[Time]:
             # even in case of errors, return last time
             return self._last_changed
 
-    async def get_schedulable_blocks(self) -> List[ObservingBlock]:
-        """Returns list of schedulable blocks.
+    async def get_schedulable_tasks(self) -> list[Task]:
+        """Returns list of schedulable tasks.
 
         Returns:
-            List of schedulable blocks
+            List of schedulable tasks
         """
 
         # get data
@@ -82,7 +72,7 @@ async def get_schedulable_blocks(self) -> List[ObservingBlock]:
         tac_priorities = {p["id"]: p["tac_priority"] for p in proposals}
 
         # loop all request groups
-        blocks = []
+        tasks: list[Task] = []
         for group in schedulable:
             # get base priority, which is tac_priority * ipp_value
             proposal = group["proposal"]
@@ -97,57 +87,29 @@ async def get_schedulable_blocks(self) -> List[ObservingBlock]:
                 if req["state"] != "PENDING":
                     continue
 
-                # duration
-                duration = req["duration"] * u.second
-
-                # time constraints
-                time_constraints = [TimeConstraint(Time(wnd["start"]), Time(wnd["end"])) for wnd in req["windows"]]
-
-                # loop configs
-                for cfg in req["configurations"]:
-                    # get instrument and check, whether we schedule it
-                    instrument = cfg["instrument_type"]
-                    if instrument.lower() not in self._instrument_type:
-                        continue
-
-                    # target
-                    t = cfg["target"]
-                    if "ra" in t and "dec" in t:
-                        target = SkyCoord(t["ra"] * u.deg, t["dec"] * u.deg, frame=t["type"].lower())
-                    else:
-                        log.warning("Unsupported coordinate type.")
-                        continue
-
-                    # constraints
-                    c = cfg["constraints"]
-                    constraints = []
-                    if "max_airmass" in c and c["max_airmass"] is not None:
-                        constraints.append(AirmassConstraint(max=c["max_airmass"], boolean_constraint=False))
-                    if "min_lunar_distance" in c and c["min_lunar_distance"] is not None:
-                        constraints.append(MoonSeparationConstraint(min=c["min_lunar_distance"] * u.deg))
-                    if "max_lunar_phase" in c and c["max_lunar_phase"] is not None:
-                        constraints.append(MoonIlluminationConstraint(max=c["max_lunar_phase"]))
-                        # if max lunar phase <= 0.4 (which would be DARK), we also enforce the sun to be <-18 degrees
-                        if c["max_lunar_phase"] <= 0.4:
-                            constraints.append(AtNightConstraint.twilight_astronomical())
-
-                    # priority is base_priority times duration in minutes
-                    # priority = base_priority * duration.value / 60.
-                    priority = base_priority
-
-                    # create block
-                    block = ObservingBlock(
-                        FixedTarget(target, name=req["id"]),
-                        duration,
-                        priority,
-                        constraints=[*constraints, *time_constraints],
-                        configuration={"request": req},
-                        name=group["name"],
-                    )
-                    blocks.append(block)
+                # just take first config and ignore the rest
+                cfg = req["configurations"][0]
+
+                # get instrument and check, whether we schedule it
+                instrument = cfg["instrument_type"]
+                if instrument.lower() not in self._instrument_type:
+                    continue
+
+                # priority is base_priority times duration in minutes
+                # priority = base_priority * duration.value / 60.
+                priority = base_priority
+
+                # create task
+                task = LcoTask(
+                    id=req["id"],
+                    name=group["name"],
+                    priority=priority,
+                    config={"request": req},
+                )
+                tasks.append(task)
 
         # return blocks
-        return blocks
+        return tasks
 
 
 __all__ = ["LcoTaskArchive"]