Refactor robotic part to allow other backends, add backend for using simple files

pyobs/modules/robotic/mastermind.py

@@ -7,9 +7,9 @@
 from pyobs.events.taskfinished import TaskFinishedEvent
 from pyobs.events.taskstarted import TaskStartedEvent
 from pyobs.interfaces import IFitsHeaderBefore, IAutonomous
-from pyobs.robotic.task import Task, ScheduledTask
+from pyobs.robotic import Task, Observation
 from pyobs.utils.time import Time
-from pyobs.robotic import TaskRunner, TaskSchedule
+from pyobs.robotic import TaskRunner, ObservationArchive
 
 log = logging.getLogger(__name__)
 
@@ -21,7 +21,7 @@ class Mastermind(Module, IAutonomous, IFitsHeaderBefore):
 
     def __init__(
         self,
-        schedule: TaskSchedule | dict[str, Any],
+        schedule: ObservationArchive | dict[str, Any],
         runner: TaskRunner | dict[str, Any],
         allowed_late_start: int = 300,
         allowed_overrun: int = 300,
@@ -47,8 +47,8 @@ def __init__(
         self.add_background_task(self._run_thread, True)
 
         # get schedule and runner
-        self._task_schedule = self.add_child_object(schedule, TaskSchedule)
-        self._task_runner = self.add_child_object(runner, TaskRunner)
+        self._observation_archive = self.add_child_object(schedule, ObservationArchive)
+        self._task_runner = self.add_child_object(runner, TaskRunner, observation_archive=self._observation_archive)
 
         # observation name and exposure number
         self._task: Task | None = None
@@ -98,7 +98,7 @@ async def _run_thread(self) -> None:
             now = Time.now()
 
             # find task that we want to run now
-            scheduled_task: ScheduledTask | None = await self._task_schedule.get_task(now)
+            scheduled_task: Observation | None = await self._observation_archive.get_task(now)
             if scheduled_task is None or not await self._task_runner.can_run(scheduled_task.task):
                 # no task found
                 await asyncio.sleep(10)
@@ -128,15 +128,15 @@ async def _run_thread(self) -> None:
             self._task = scheduled_task.task
 
             # ETA
-            eta = now + self._task.duration
+            eta = now + self._task.duration * u.second
 
             # send event
             await self.comm.send_event(TaskStartedEvent(name=self._task.name, id=self._task.id, eta=eta))
 
             # run task in thread
             log.info("Running task %s...", self._task.name)
             try:
-                await self._task_runner.run_task(self._task, task_schedule=self._task_schedule)
+                await self._task_runner.run_task(self._task)
             except:
                 # something went wrong
                 log.warning("Task %s failed.", self._task.name)

pyobs/modules/robotic/scheduler.py

@@ -14,7 +14,7 @@
 from pyobs.utils.time import Time
 from pyobs.interfaces import IStartStop, IRunnable
 from pyobs.modules import Module
-from pyobs.robotic import TaskArchive, TaskSchedule, ScheduledTask, Task
+from pyobs.robotic import TaskArchive, ObservationArchive, Task, ObservationList
 
 log = logging.getLogger(__name__)
 
@@ -28,7 +28,7 @@ def __init__(
         self,
         scheduler: dict[str, Any] | TaskScheduler,
         tasks: Union[Dict[str, Any], TaskArchive],
-        schedule: Union[Dict[str, Any], TaskSchedule],
+        schedule: Union[Dict[str, Any], ObservationArchive],
         trigger_on_task_started: bool = False,
         trigger_on_task_finished: bool = False,
         schedule_range: float = 24.0,
@@ -52,8 +52,8 @@ def __init__(
 
         # get scheduler
         self._scheduler = self.add_child_object(scheduler, TaskScheduler)
-        self._task_archive = self.add_child_object(tasks, TaskArchive)
-        self._schedule = self.add_child_object(schedule, TaskSchedule)
+        self._task_archive = self.add_child_object(tasks, TaskArchive, on_tasks_changed=self._update_schedule)
+        self._schedule = self.add_child_object(schedule, ObservationArchive)
 
         # store
         self._running = True
@@ -76,7 +76,6 @@ def __init__(
 
         # update threads
         self.add_background_task(self._schedule_worker)
-        self.add_background_task(self._update_worker)
 
     async def open(self) -> None:
         """Open module."""
@@ -88,6 +87,9 @@ async def open(self) -> None:
             await self.comm.register_event(TaskFinishedEvent, self._on_task_finished)
             await self.comm.register_event(GoodWeatherEvent, self._on_good_weather)
 
+        # schedule an update run
+        asyncio.create_task(self._update_schedule())
+
     async def start(self, **kwargs: Any) -> None:
         """Start scheduler."""
         self._running = True
@@ -100,32 +102,6 @@ async def is_running(self, **kwargs: Any) -> bool:
         """Whether scheduler is running."""
         return self._running
 
-    async def _update_worker(self) -> None:
-        # time of last change in blocks
-        last_change = None
-
-        # run forever
-        while True:
-            # not running?
-            if not self._running:
-                await asyncio.sleep(1)
-                return
-
-            # got new time of last change?
-            t = await self._task_archive.last_changed()
-            more_1day = (Time.now() - t) > TimeDelta(1 * u.day)
-            if last_change is None or last_change < t and not more_1day:
-                try:
-                    last_change = t
-                    await self._update_schedule()
-                except asyncio.CancelledError:
-                    return
-                except:
-                    log.exception("Something went wrong when updating schedule.")
-
-            # sleep a little
-            await asyncio.sleep(5)
-
     async def _update_schedule(self) -> None:
         # get schedulable tasks and sort them
         log.info("Found update in schedulable block, downloading them...")
@@ -228,7 +204,7 @@ async def _schedule_worker(self) -> None:
                     end = start + TimeDelta(self._schedule_range)
 
                     # schedule
-                    scheduled_tasks: list[ScheduledTask] = []
+                    scheduled_tasks = ObservationList()
                     first = True
                     async for scheduled_task in self._scheduler.schedule(self._tasks, start, end):
                         # remember for later
@@ -242,13 +218,13 @@ async def _schedule_worker(self) -> None:
                         if first:
                             first = False
                             log.info("Finished calculating next task:")
-                            self._log_scheduled_task([scheduled_task])
+                            self._log_scheduled_task(ObservationList([scheduled_task]))
 
                             # set new safety_time as duration + 20%
                             self._safety_time = (time.time() - start_time) * 1.2 * u.second
 
                             # submit it
-                            await self._schedule.add_schedule([scheduled_task])
+                            await self._schedule.add_schedule(ObservationList([scheduled_task]))
 
                     if self._need_update:
                         log.info("Not using scheduler results, since update was requested.")
@@ -273,8 +249,9 @@ async def _schedule_worker(self) -> None:
             # sleep a little
             await asyncio.sleep(1)
 
-    def _log_scheduled_task(self, scheduled_tasks: list[ScheduledTask]) -> None:
+    def _log_scheduled_task(self, scheduled_tasks: ObservationList) -> None:
         for scheduled_task in scheduled_tasks:
+            print(scheduled_task)
             log.info(
                 "  - %s to %s: %s (%d)",
                 scheduled_task.start.strftime("%H:%M:%S"),

pyobs/object.py

@@ -475,6 +475,13 @@ def add_child_object(
         **kwargs: Any,
     ) -> ObjectClass: ...
 
+    @overload
+    def add_child_object(
+        self,
+        config_or_object: ObjectClass,
+        **kwargs: Any,
+    ) -> ObjectClass: ...
+
     @overload
     def add_child_object(
         self,

pyobs/robotic/__init__.py

@@ -1,5 +1,5 @@
-from .taskschedule import TaskSchedule
-from .task import Task, ScheduledTask
-from .observation import Observation
+from .observationarchive import ObservationArchive
+from .task import Task
+from .observation import Observation, ObservationList, ObservationState
 from .taskarchive import TaskArchive
 from .taskrunner import TaskRunner

pyobs/robotic/filesystem/__init__.py

@@ -0,0 +1,2 @@
+from .taskarchive import YamlTaskArchive
+from .observationarchive import YamlObservationArchive

pyobs/robotic/filesystem/observationarchive.py

@@ -0,0 +1,173 @@
+from __future__ import annotations
+import datetime
+import os
+from typing import Any, Literal
+import abc
+
+from pyobs.utils.time import Time
+from .. import ObservationArchive
+from .. import Task
+from ..observation import ObservationList, Observation
+from ...vfs import VirtualFileSystem
+
+
+class FileSystemObservationArchive(ObservationArchive, metaclass=abc.ABCMeta):
+    def __init__(self, path: str, extension: str, mode: Literal["day", "night"], **kwargs: Any):
+        ObservationArchive.__init__(self, **kwargs)
+        self._path = path
+        self._extension = extension
+        self._mode = mode
+
+    def _get_filename(self, time: Time) -> str:
+        """Returns the filename associated with the given time. If mode==night, the last sunrise is used,
+        otherwise the last sunset.
+
+        Args:
+            time: Time to get filename for.
+
+        Returns:
+            Filename for schedule file.
+        """
+        if self.observer is None:
+            raise ValueError("Observer is not set.")
+        day = (
+            self.observer.sun_rise_time(time, "previous")
+            if self._mode == "night"
+            else self.observer.sun_set_time(time, "previous")
+        )
+        return f"{day.isot[:10]}.{self._extension}"
+
+    async def _load_observations(self, time: Time) -> ObservationList:
+        """Loads observations from file for given time.
+
+        Args:
+            time: Time defines the night/day to load observations for.
+
+        Returns:
+            List of observations.
+        """
+        filename = self._get_filename(time)
+        full_path = os.path.join(self._path, filename)
+        try:
+            return await self._load_observations_from_file(full_path, self.vfs)
+        except FileNotFoundError:
+            return ObservationList()
+
+    async def _save_observations(self, time: Time, observations: ObservationList) -> None:
+        """Saves observations to file.
+
+        Args:
+            time: Time defines the night/day to save observations for.
+            observations: List of observations to save.
+        """
+
+        filename = self._get_filename(time)
+        full_path = os.path.join(self._path, filename)
+        await self._save_observations_to_file(full_path, observations, self.vfs)
+
+    @classmethod
+    @abc.abstractmethod
+    async def _load_observations_from_file(cls, path: str, vfs: VirtualFileSystem) -> ObservationList: ...
+
+    @classmethod
+    @abc.abstractmethod
+    async def _save_observations_to_file(
+        cls, path: str, observations: ObservationList, vfs: VirtualFileSystem
+    ) -> None: ...
+
+    async def add_schedule(self, observations: ObservationList) -> None:
+        """Add the list of scheduled tasks to the schedule.
+
+        Args:
+            observations: Scheduled tasks.
+        """
+        if len(observations) == 0:
+            return
+        time = observations[0].start
+        schedule = await self._load_observations(time)
+        schedule += observations
+        await self._save_observations(time, schedule)
+
+    async def clear_schedule(self, start_time: Time) -> None:
+        """Clear schedule after given start time.
+
+        Args:
+            start_time: Start time to clear from.
+        """
+        schedule = await self._load_observations(start_time)
+        cleared = ObservationList([obs for obs in schedule if obs.end <= start_time])
+        await self._save_observations(start_time, cleared)
+
+    async def get_schedule(self) -> ObservationList:
+        """Fetch schedule from portal.
+
+        Returns:
+            Dictionary with tasks.
+
+        Raises:
+            Timeout: If request timed out.
+            ValueError: If something goes wrong.
+        """
+        return await self._load_observations(Time.now())
+
+    async def get_task(self, time: Time) -> Observation | None:
+        """Returns the active scheduled task at the given time.
+
+        Args:
+            time: Time to return task for.
+
+        Returns:
+            Scheduled task at the given time.
+        """
+
+        # get schedule
+        schedule = await self._load_observations(time)
+
+        # loop all tasks
+        for obs in schedule:
+            # running now?
+            if obs.start <= time < obs.end and not obs.task.is_finished():
+                return obs
+
+        # nothing found
+        return None
+
+    async def observations_for_task(self, task: Task) -> ObservationList:
+        """Returns list of observations for the given task.
+
+        Args:
+            task: Task to get observations for.
+
+        Returns:
+            List of observations for the given task.
+        """
+        return ObservationList()
+
+    async def observations_for_night(self, date: datetime.date) -> ObservationList:
+        """Returns list of observations for the given task.
+
+        Args:
+            date: Date of night to get observations for.
+
+        Returns:
+            List of observations for the given task.
+        """
+        return ObservationList()
+
+
+class YamlObservationArchive(FileSystemObservationArchive):
+    def __init__(self, path: str, **kwargs: Any):
+        FileSystemObservationArchive.__init__(self, path, "yaml", **kwargs)
+
+    @classmethod
+    async def _load_observations_from_file(cls, path: str, vfs: VirtualFileSystem) -> ObservationList:
+        observations = await vfs.read_yaml(path)
+        return ObservationList([Observation.model_validate(obs) for obs in observations])
+
+    @classmethod
+    async def _save_observations_to_file(cls, path: str, observations: ObservationList, vfs: VirtualFileSystem) -> None:
+        data = [obs.model_dump() for obs in observations]
+        await vfs.write_yaml(path, data)
+
+
+__all__ = ["FileSystemObservationArchive", "YamlObservationArchive"]