v1.10.0

version 1.10.0

CHANGELOG.rst

@@ -1,3 +1,8 @@
+v1.10.0 (2023-12-23)
+*******************
+* Added CallModule script.
+* Changed ScriptRunner module so that it can run a script multiple times.
+
 v1.9.0 (2023-12-23)
 *******************
 * Added getters and safe getters for Image class.

pyobs/modules/robotic/scriptrunner.py

@@ -2,20 +2,21 @@
 from typing import Any, Dict
 
 from pyobs.modules import Module
-from pyobs.interfaces import IAutonomous
+from pyobs.interfaces import IRunnable
 from pyobs.robotic.scripts import Script
 
 log = logging.getLogger(__name__)
 
 
-class ScriptRunner(Module, IAutonomous):
+class ScriptRunner(Module, IRunnable):
     """Module for running a script."""
 
     __module__ = "pyobs.modules.robotic"
 
     def __init__(
         self,
         script: Dict[str, Any],
+        run_once: bool = False,
         **kwargs: Any,
     ):
         """Initialize a new script runner.
@@ -27,29 +28,24 @@ def __init__(
 
         # store
         self.script = script
-        copy_comm = "comm" not in script.keys()
-        self._script = self.add_child_object(script, Script, copy_comm=copy_comm)
+        self._script = self.add_child_object(script, Script)
 
         # add thread func
-        self.add_background_task(self._run_thread, False)
+        if run_once:
+            self.add_background_task(self._run_thread, False)
 
-    async def start(self, **kwargs: Any) -> None:
-        """Starts a service."""
-        pass
+    async def run(self, **kwargs: Any) -> None:
+        """Run script."""
+        script = self.get_object(self.script, Script)
+        await script.run(None)
 
-    async def stop(self, **kwargs: Any) -> None:
-        """Stops a service."""
+    async def abort(self, **kwargs: Any) -> None:
+        """Abort current actions."""
         pass
 
-    async def is_running(self, **kwargs: Any) -> bool:
-        """Whether a service is running."""
-        return True
-
     async def _run_thread(self) -> None:
         """Run the script."""
-
-        # run script
-        await self._script.run(None)
+        await self.run()
 
 
 __all__ = ["ScriptRunner"]

pyobs/robotic/scripts/__init__.py

@@ -5,3 +5,4 @@
 from .parallel import ParallelRunner
 from .selector import SelectorScript
 from .conditional import ConditionalRunner
+from .callmodule import CallModule

pyobs/robotic/scripts/callmodule.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+import logging
+from typing import Any, Dict, Optional, List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pyobs.robotic import TaskRunner, TaskSchedule, TaskArchive
+from pyobs.robotic.scripts import Script
+
+log = logging.getLogger(__name__)
+
+
+class CallModule(Script):
+    """Script for calling method on a module."""
+
+    __module__ = "pyobs.modules.robotic"
+
+    def __init__(
+        self,
+        module: str,
+        method: str,
+        params: Optional[List[Any]],
+        **kwargs: Any,
+    ):
+        """Initialize a new SequentialRunner.
+
+        Args:
+            script: list or dict of scripts to run in a sequence.
+        """
+        Script.__init__(self, **kwargs)
+        self.module = module
+        self.method = method
+        self.params = params or []
+
+    async def can_run(self) -> bool:
+        try:
+            self.comm.proxy(self.module)
+            return True
+        except ValueError:
+            return False
+
+    async def run(
+        self,
+        task_runner: TaskRunner,
+        task_schedule: Optional[TaskSchedule] = None,
+        task_archive: Optional[TaskArchive] = None,
+    ) -> None:
+        proxy = self.comm.proxy(self.module)
+        await proxy.execute(self.method, *self.params)
+
+
+__all__ = ["CallModule"]

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "pyobs-core"
 packages = [{ include = "pyobs" }]
-version = "1.9.0"
+version = "1.10.0"
 description = "robotic telescope software"
 authors = ["Tim-Oliver Husser <[email protected]>"]
 license = "MIT"