added a few default scripts

pyobs/robotic/scripts/__init__.py

@@ -1,8 +1,11 @@
 from .script import Script
-from .skyflats import SkyFlats
+from .cases import CasesRunner
+from .callmodule import CallModule
+from .conditional import ConditionalRunner
 from .darkbias import DarkBias
-from .sequential import SequentialRunner
+from .debugtrigger import DebugTriggerRunner
 from .parallel import ParallelRunner
+from .log import LogRunner
 from .selector import SelectorScript
-from .conditional import ConditionalRunner
-from .callmodule import CallModule
+from .sequential import SequentialRunner
+from .skyflats import SkyFlats

pyobs/robotic/scripts/cases.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import logging
+from typing import Any, Dict, Optional, List, TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    from pyobs.robotic import TaskRunner, TaskSchedule, TaskArchive
+from pyobs.robotic.scripts import Script
+
+log = logging.getLogger(__name__)
+
+
+class CasesRunner(Script):
+    """Script for distinguishing cases."""
+
+    __module__ = "pyobs.modules.robotic"
+
+    def __init__(
+        self,
+        expression: str,
+        cases: Dict[Union[str, int, float], Any],
+        **kwargs: Any,
+    ):
+        """Initialize a new CasesRunner.
+
+        Args:
+            expression: expression to check
+            cases: dictionary with cases
+        """
+        Script.__init__(self, **kwargs)
+        self.expression = expression
+        self.cases = cases
+
+    async def can_run(self) -> bool:
+        return True
+
+    async def run(
+        self,
+        task_runner: TaskRunner,
+        task_schedule: Optional[TaskSchedule] = None,
+        task_archive: Optional[TaskArchive] = None,
+    ) -> None:
+        # evaluate condition
+        value = eval(self.expression, {"now": datetime.now(timezone.utc), "config": self.configuration})
+
+        # check in cases
+        if value in self.cases:
+            await self.get_object(self.cases[value], Script).run(task_runner, task_schedule, task_archive)
+        elif "else" in self.cases:
+            await self.get_object(self.cases["else"], Script).run(task_runner, task_schedule, task_archive)
+
+
+__all__ = ["CasesRunner"]

pyobs/robotic/scripts/debugtrigger.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import logging
+from typing import Any, Optional, List, TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    from pyobs.robotic import TaskRunner, TaskSchedule, TaskArchive
+from pyobs.robotic.scripts import Script
+
+log = logging.getLogger(__name__)
+
+
+class DebugTriggerRunner(Script):
+    """Script for a debug trigger."""
+
+    __module__ = "pyobs.modules.robotic"
+
+    def __init__(self, **kwargs: Any):
+        """Initialize a new DebugTriggerRunner."""
+        Script.__init__(self, **kwargs)
+        self.triggered = False
+
+    async def can_run(self) -> bool:
+        return True
+
+    async def run(
+        self,
+        task_runner: TaskRunner,
+        task_schedule: Optional[TaskSchedule] = None,
+        task_archive: Optional[TaskArchive] = None,
+    ) -> None:
+        self.triggered = True
+
+
+__all__ = ["DebugTriggerRunner"]

pyobs/robotic/scripts/log.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import logging
+from typing import Any, Optional, List, TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    from pyobs.robotic import TaskRunner, TaskSchedule, TaskArchive
+from pyobs.robotic.scripts import Script
+
+log = logging.getLogger(__name__)
+
+
+class LogRunner(Script):
+    """Script for logging something."""
+
+    __module__ = "pyobs.modules.robotic"
+
+    def __init__(
+        self,
+        expression: str,
+        **kwargs: Any,
+    ):
+        """Initialize a new LogRunner.
+
+        Args:
+            expression: expression to check
+        """
+        Script.__init__(self, **kwargs)
+        self.expression = expression
+
+    async def can_run(self) -> bool:
+        return True
+
+    async def run(
+        self,
+        task_runner: TaskRunner,
+        task_schedule: Optional[TaskSchedule] = None,
+        task_archive: Optional[TaskArchive] = None,
+    ) -> None:
+        # evaluate condition
+        value = eval(self.expression, {"now": datetime.now(timezone.utc), "config": self.configuration})
+
+        # log it
+        log.info(value)
+
+
+__all__ = ["LogRunner"]

tests/robotic/scripts/test_cases.py

@@ -0,0 +1,40 @@
+import pytest
+
+from pyobs.robotic.scripts.cases import CasesRunner
+from pyobs.robotic.scripts.debugtrigger import DebugTriggerRunner
+
+
+@pytest.mark.asyncio
+async def test_int_cases():
+    cases = {1: DebugTriggerRunner(), 2: DebugTriggerRunner(), 10: DebugTriggerRunner()}
+    cr = CasesRunner("10", cases)
+    await cr.run(None, None, None)
+    assert cases[1].triggered is False
+    assert cases[2].triggered is False
+    assert cases[10].triggered is True
+
+
+@pytest.mark.asyncio
+async def test_float_cases():
+    cases = {3.14: DebugTriggerRunner(), 42.0: DebugTriggerRunner(), 2.7: DebugTriggerRunner()}
+    cr = CasesRunner("3.14", cases)
+    await cr.run(None, None, None)
+    assert cases[3.14].triggered is True
+    assert cases[42.0].triggered is False
+    assert cases[2.7].triggered is False
+
+
+@pytest.mark.asyncio
+async def test_else():
+    cases = {
+        "1": DebugTriggerRunner(),
+        "2": DebugTriggerRunner(),
+        "3": DebugTriggerRunner(),
+        "else": DebugTriggerRunner(),
+    }
+    cr = CasesRunner("4", cases)
+    await cr.run(None, None, None)
+    assert cases["1"].triggered is False
+    assert cases["2"].triggered is False
+    assert cases["3"].triggered is False
+    assert cases["else"].triggered is True

tests/robotic/scripts/test_debugtrigger.py

@@ -0,0 +1,11 @@
+import logging
+import pytest
+
+from pyobs.robotic.scripts.debugtrigger import DebugTriggerRunner
+
+
+@pytest.mark.asyncio
+async def test_trigger():
+    runner = DebugTriggerRunner()
+    await runner.run(None, None, None)
+    assert runner.triggered is True