Merge pull request #12 from kornia/user_hooks

User hooks

examples/default_cmps.py

@@ -1,7 +1,6 @@
 """Basic example with predefined cmps."""
 import asyncio
 from sys import version_info
-import copy
 
 import torch
 
@@ -35,6 +34,8 @@
     pipeline = Pipeline()
     pipeline.add_nodes([c1, t1, t2, stack, show])
     pipeline.run(1)
+    # You can rerun the pipeline as many times as you want and will continue from the last iteration
+    pipeline.run(1)
 else:
     # run 1 iteration using the asyncio loop
     print("Run with loop:")

limbus/core/__init__.py

@@ -1,4 +1,4 @@
-from limbus.core.component import Component, iterations_manager
+from limbus.core.component import Component, executions_manager
 from limbus.core.states import ComponentState, PipelineState, VerboseMode
 from limbus.core.param import NoValue, Param, Reference, InputParam, OutputParam
 from limbus.core.params import Params, InputParams, OutputParams
@@ -12,7 +12,7 @@
     "PipelineState",
     "VerboseMode",
     "Component",
-    "iterations_manager",
+    "executions_manager",
     "ComponentState",
     "Params",
     "Reference",

limbus/core/app.py

@@ -1,5 +1,5 @@
 """High level template to create apps."""
-from typing import List
+from __future__ import annotations
 from abc import abstractmethod
 
 from limbus.core import Pipeline, VerboseMode, Component
@@ -15,7 +15,7 @@ def __init__(self):
         self._pipeline.add_nodes(self._get_component_attrs())
         self._pipeline.set_verbose_mode(VerboseMode.DISABLED)
 
-    def _get_component_attrs(self) -> List[Component]:
+    def _get_component_attrs(self) -> list[Component]:
         """Get the component attribute by name."""
         return [getattr(self, attr) for attr in dir(self) if isinstance(getattr(self, attr), Component)]
 

limbus/core/async_utils.py

@@ -2,12 +2,22 @@
 from __future__ import annotations
 import asyncio
 import inspect
-from typing import Coroutine, TYPE_CHECKING, Optional
+from typing import Coroutine, TYPE_CHECKING
 from sys import version_info
 
 if TYPE_CHECKING:
     from limbus.core.component import Component
 
+# Get the loop that is going to run the pipeline. Doing it in this way allows to rerun the pipeline.
+loop = asyncio.new_event_loop()
+
+
+def reset_loop() -> asyncio.AbstractEventLoop:
+    """Reset the loop."""
+    global loop
+    loop = asyncio.new_event_loop()
+    return loop
+
 
 def run_coroutine(coro: Coroutine) -> None:
     """Run a coroutine in an event loop.
@@ -16,25 +26,20 @@ def run_coroutine(coro: Coroutine) -> None:
         coro: coroutine to run.
 
     """
-    if version_info.major != 3:
-        raise ValueError("Only python 3 is supported.")
-    if version_info.minor < 10:
-        # for python <3.10 the loop must be run in this way to avoid creating a new loop.
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(coro)
-    elif version_info.minor >= 10:
-        # for python >=3.10 the loop should be run in this way.
-        asyncio.run(coro)
-
-
-def get_task_if_exists(component: Component) -> Optional[asyncio.Task]:
+    global loop
+    if loop.is_closed():
+        loop = reset_loop()
+    loop.run_until_complete(coro)
+
+
+def get_task_if_exists(component: Component) -> None | asyncio.Task:
     """Get the task associated to a given component if it exists.
 
     Args:
         component: component to check.
 
     Returns:
-        Optional[asyncio.Task]: task associated to the component if it exists, None otherwise.
+        None | asyncio.Task: task associated to the component if it exists, None otherwise.
 
     """
     task: asyncio.Task