feat(arcor_runtime): add callbacks for pause and resume events

README.md

@@ -84,7 +84,7 @@ The following video by [Kinali](https://www.kinali.cz/en/) shows the use case (o
 
 [README](src/python/arcor2_execution/README.md) | [CHANGELOG](src/python/arcor2_execution/CHANGELOG.md)
 
- - 2024-06-26: [1.4.2](https://github.com/robofit/arcor2/releases/tag/arcor2_execution%2F1.4.2) ([docker](https://hub.docker.com/r/arcor2/arcor2_execution/tags?page=1&ordering=last_updated&name=1.4.2), [pypi](https://pypi.org/project/arcor2-execution/1.4.2/)).
+ - 2024-08-14: [1.5.0](https://github.com/robofit/arcor2/releases/tag/arcor2_execution%2F1.5.0) ([docker](https://hub.docker.com/r/arcor2/arcor2_execution/tags?page=1&ordering=last_updated&name=1.5.0), [pypi](https://pypi.org/project/arcor2-execution/1.5.0/)).
 
 ### arcor2_execution_data
 
@@ -126,7 +126,7 @@ The following video by [Kinali](https://www.kinali.cz/en/) shows the use case (o
 
 [README](src/python/arcor2_runtime/README.md) | [CHANGELOG](src/python/arcor2_runtime/CHANGELOG.md)
 
- - 2024-06-14: [1.3.0](https://github.com/robofit/arcor2/releases/tag/arcor2_runtime%2F1.3.0) ([pypi](https://pypi.org/project/arcor2-runtime/1.3.0/)).
+ - 2024-08-14: [1.4.0](https://github.com/robofit/arcor2/releases/tag/arcor2_runtime%2F1.4.0) ([pypi](https://pypi.org/project/arcor2-runtime/1.4.0/)).
 
 ### arcor2_scene
 

compose-files/fit-demo/docker-compose.yml

@@ -55,7 +55,7 @@ services:
       - fit-demo-project-network
 
   fit-demo-execution:
-    image: arcor2/arcor2_execution:1.4.2
+    image: arcor2/arcor2_execution:1.5.0
     container_name: fit-demo-execution
     networks:
       - fit-demo-execution-network

src/docker/arcor2_execution/BUILD

@@ -1 +1 @@
-docker_image(name="arcor2_execution", repository="arcor2/arcor2_execution", image_tags=["1.4.2"])
+docker_image(name="arcor2_execution", repository="arcor2/arcor2_execution", image_tags=["1.5.0"])

src/python/arcor2_execution/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
+## [1.5.0] - 2024-08-14
+
+### Changed
+
+- Dependency on `arcor2_runtime~=1.4.0`.
+
 ## [1.4.2] - 2024-06-26
 
 ### Fixed

src/python/arcor2_execution/VERSION

@@ -1 +1 @@
-1.4.2
+1.5.0

src/python/arcor2_runtime/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
+## [1.4.0] - 2024-08-14
+
+### Added
+
+- There are optional callbacks for pause and resume events in the `action.Globals`.
+
 ## [1.3.0] - 2024-06-14
 
 ### Changed

src/python/arcor2_runtime/VERSION

@@ -1 +1 @@
-1.3.0
+1.4.0

src/python/arcor2_runtime/action.py

@@ -12,6 +12,7 @@
 from arcor2.object_types.abstract import Generic
 from arcor2.object_types.utils import iterate_over_actions
 from arcor2.parameter_plugins.utils import plugin_from_instance
+from arcor2_runtime.exceptions import print_exception
 
 
 class Commands(StrEnum):
@@ -23,6 +24,8 @@ class Commands(StrEnum):
 ACTION_NAME_ID_MAPPING_ATTR = "_action_name_id_mapping"
 AP_ID_ATTR = "_ap_id"
 
+CB_TYPE = Callable[[], None] | None
+
 
 @dataclass
 class Globals:
@@ -41,6 +44,13 @@ class Globals:
 
     lock: threading.Lock = field(default_factory=threading.Lock)
 
+    # can be used to react on pause/resume events
+    # callbacks are called before the actual event happens
+    pause_callback: CB_TYPE = None
+    resume_callback: CB_TYPE = None
+    # id of the thread within the one of the callbacks (only one can be called at the time)
+    callback_thread_id: int = 0
+
 
 g = Globals()
 
@@ -114,7 +124,7 @@ def print_event(event: Event) -> None:
     sys.stdout.flush()
 
 
-def _get_commands():
+def _get_stdin_commands():
     """Reads stdin and checks for commands from parent script (e.g. Execution
     unit). Prints events to stdout. State is signalled using events.
 
@@ -124,6 +134,16 @@ def _get_commands():
     :return:
     """
 
+    def callback(cb: CB_TYPE) -> None:
+        if cb:
+            g.callback_thread_id = threading.get_ident()
+            try:
+                cb()
+            except Exception as e:  # otherwise, the thread will stop silently
+                print_exception(e)
+                sys.exit(1)
+            g.callback_thread_id = 0
+
     while True:
         raw_cmd = read_stdin(0.1)
 
@@ -135,6 +155,7 @@ def _get_commands():
         with g.lock:
             if g.pause.is_set():
                 if cmd in (Commands.STEP, Commands.RESUME):
+                    callback(g.resume_callback)
                     g.pause.clear()
                     print_event(PackageState(PackageState.Data(PackageState.Data.StateEnum.RUNNING)))
 
@@ -144,26 +165,29 @@ def _get_commands():
                     g.resume.set()
             else:
                 if cmd == Commands.PAUSE:
+                    callback(g.pause_callback)
                     g.resume.clear()
                     g.pause.set()
                     print_event(PackageState(PackageState.Data(PackageState.Data.StateEnum.PAUSED)))
 
 
-_cmd_thread = threading.Thread(target=_get_commands)
+_cmd_thread = threading.Thread(target=_get_stdin_commands)
 _cmd_thread.daemon = True
 _cmd_thread.start()
 
 
-def handle_stdin_commands(*, before: bool, breakpoint: bool = False) -> None:
+def handle_commands(*, before: bool, breakpoint: bool = False) -> None:
     """Actual handling of commands in (potentially parallel) actions."""
 
     with g.lock:
+        # this is to handle pause caused by breakpoint
         if (breakpoint or (before and g.pause_on_next_action.is_set())) and not g.pause.is_set():
             g.pause_on_next_action.clear()
             g.resume.clear()
             g.pause.set()
             print_event(PackageState(PackageState.Data(PackageState.Data.StateEnum.PAUSED)))
 
+    # waiting - the same for both "external" and "internal" pause (command from the Execution service, or breakpoint)
     if g.pause.is_set():
         g.resume.wait()
 
@@ -215,6 +239,10 @@ def wrapper(obj: Generic, *action_args: Any, an: None | str = None, **kwargs: An
         if thread_id not in g.depth:
             g.depth[thread_id] = 0
 
+        # this allows calling actions in pause/resume callbacks by treating them as ordinary methods (no events, etc.)
+        if thread_id == g.callback_thread_id:
+            return f(obj, *action_args, an=an, **kwargs)
+
         try:
             action_id: None | str = None
             action_mapping_provided = hasattr(obj, ACTION_NAME_ID_MAPPING_ATTR)
@@ -273,7 +301,7 @@ def wrapper(obj: Generic, *action_args: Any, an: None | str = None, **kwargs: An
 
                 print_event(state_before)
 
-                handle_stdin_commands(before=True, breakpoint=make_a_break)
+                handle_commands(before=True, breakpoint=make_a_break)
 
             else:  # handle outermost actions
                 # if not set (e.g. when writing actions manually), do not attempt to get action IDs from names
@@ -311,7 +339,7 @@ def wrapper(obj: Generic, *action_args: Any, an: None | str = None, **kwargs: An
 
                 g.ea[thread_id] = None
 
-            handle_stdin_commands(before=False)
+            handle_commands(before=False)
 
             return res