Add hooks for frame complete and stream destroy

Author: rskew

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "aiko_services"
-version = "0.6.post22"
+version = "0.6.post25"
 readme = "ReadMe.md"
 description = "Distributed embedded service framework for A.I and robotics"
 requires-python = ">=3.9.0,<=3.13.7"

src/aiko_services/main/hook.py

@@ -105,7 +105,7 @@ def remove_hook_handler(self, hook_name, hook_function):
         pass
 
     @abstractmethod
-    def run_hook(self, hook_name):
+    def run_hook(self, hook_name, variables=None):
         pass
 
     @abstractmethod

src/aiko_services/main/pipeline.py

@@ -202,6 +202,10 @@
 _PIPELINE_HOOK_PROCESS_ELEMENT_POST = PIPELINE_HOOK_PROCESS_ELEMENT_POST+"0"
 PIPELINE_HOOK_PROCESS_FRAME = "pipeline.process_frame:"
 _PIPELINE_HOOK_PROCESS_FRAME = PIPELINE_HOOK_PROCESS_FRAME+"0"
+PIPELINE_HOOK_PROCESS_FRAME_COMPLETE = "pipeline.process_frame_complete:"
+_PIPELINE_HOOK_PROCESS_FRAME_COMPLETE = PIPELINE_HOOK_PROCESS_FRAME_COMPLETE+"0"
+PIPELINE_HOOK_DESTROY_STREAM = "pipeline.destroy_stream:"
+_PIPELINE_HOOK_DESTROY_STREAM = PIPELINE_HOOK_DESTROY_STREAM+"0"
 
 _GRACE_TIME = 60  # seconds
 _LOGGER = aiko.logger(__name__)
@@ -697,6 +701,8 @@ def __init__(self, context):
         self.add_hook(_PIPELINE_HOOK_PROCESS_ELEMENT)
         self.add_hook(_PIPELINE_HOOK_PROCESS_ELEMENT_POST)
         self.add_hook(_PIPELINE_HOOK_PROCESS_FRAME)
+        self.add_hook(_PIPELINE_HOOK_PROCESS_FRAME_COMPLETE)
+        self.add_hook(_PIPELINE_HOOK_DESTROY_STREAM)
 
         self.pipeline_graph = self._create_pipeline_graph(context.definition)
         self.share["element_count"] = self.pipeline_graph.element_count
@@ -1094,13 +1100,15 @@ def destroy_stream(self, stream_id,
                         stream.state = StreamState.ERROR
 
             # Notify listeners that the stream has stopped
-            stop_state = stream.state
-            if stop_state >= StreamState.RUN:
-                stop_state = StreamState.STOP
+            if stream.state >= StreamState.RUN:
+                stream.state = StreamState.STOP
+            self.run_hook(_PIPELINE_HOOK_DESTROY_STREAM,
+                lambda: {"stream": stream,
+                         "diagnostic": diagnostic})
             stream_info = {
                 "stream_id": stream.stream_id,
                 "frame_id": stream.frame_id,
-                "state": stop_state}
+                "state": stream.state}
             if stream.queue_response:
                 stream.queue_response.put((stream_info, diagnostic))
             if stream.topic_response:
@@ -1393,6 +1401,10 @@ def _process_frame_common(self, stream_dict, frame_data_in, new_frame)  \
                     "stream_id": stream.stream_id,
                     "frame_id": stream.frame_id,
                     "state": stream.state}
+                self.run_hook(_PIPELINE_HOOK_PROCESS_FRAME_COMPLETE,
+                    lambda: {
+                        "stream": stream,
+                        "frame_data_out": frame_data_out})
                 if stream.queue_response:
                     stream.queue_response.put((stream_info, frame_data_out))
                 elif stream.topic_response: