Added in complete window type data to satisfy orca core on processor registration

Author: solvicode

examples/simpletrigger/simpletrigger/algorithms.py

@@ -33,5 +33,7 @@ class BadEnvVar(BaseOrcaException):
 class BadConfigFile(BaseOrcaException):
     """Raised when the orca.json config file is poorly defined"""
 
+
 class BrokenRemoteAlgorithmStubs(BaseOrcaException):
-    """Raised when remote algorithm stubs cannot be properly parsed and read"""
+    """Raised when remote algorithm stubs cannot be properly parsed and read"""
+

orca_python/exceptions.py

@@ -6,7 +6,6 @@
 which are managed by Orca-core.
 """
 
-from io import StringIO
 import re
 import sys
 import asyncio
@@ -52,10 +51,10 @@
 
 from orca_python import envs
 from orca_python.exceptions import (
-    BrokenRemoteAlgorithmStubs,
     InvalidDependency,
     InvalidWindowArgument,
     InvalidAlgorithmArgument,
+    BrokenRemoteAlgorithmStubs,
     InvalidAlgorithmReturnType,
     InvalidMetadataFieldArgument,
 )
@@ -207,13 +206,15 @@ def __call__(
         self, params: ExecutionParams, *args: Any, **kwargs: Any
     ) -> returnResult: ...
 
+
 @dataclass
-class RemoteAlgorithm():
+class RemoteAlgorithm:
     ProcessorName: str
     ProcessorRuntime: str
     Name: str
     Version: str
 
+
 T = TypeVar("T", bound=AlgorithmFn)
 
 
@@ -331,7 +332,9 @@ def _add_algorithm(self, name: str, algorithm: Algorithm) -> None:
         )
         self._algorithms[name] = algorithm
 
-    def _add_dependency(self, algorithm: str, dependency: AlgorithmFn, remote: bool = False) -> None:
+    def _add_dependency(
+        self, algorithm: str, dependency: AlgorithmFn, remote: bool = False
+    ) -> None:
         """
         Adds a dependency to an algorithm.
 
@@ -345,22 +348,23 @@ def _add_dependency(self, algorithm: str, dependency: AlgorithmFn, remote: bool
         """
         LOGGER.debug(f"Adding dependency for algorithm: {algorithm}")
         if remote:
-            # TODO: how do we resolve this - we need information about a remote
-            # algorithm but do not have it. We don't want to go further than stubs as
-            # it makes deployment messy.
             remoteDepMetadata = getattr(dependency, "__orca_metadata__", None)
             if remoteDepMetadata is None:
-                raise BrokenRemoteAlgorithmStubs("Could not parse metadata from Orca stubs. Rerun stub generation: `orca sync`")
+                raise BrokenRemoteAlgorithmStubs(
+                    "Could not parse metadata from Orca stubs. Rerun stub generation: `orca sync`"
+                )
             try:
                 remoteAlgo = RemoteAlgorithm(**remoteDepMetadata)
             except Exception as e:
-                raise BrokenRemoteAlgorithmStubs(f"Could not parse metadata from Orca stubs: {e} Rerun stub generation: `orca sync`")
+                raise BrokenRemoteAlgorithmStubs(
+                    f"Could not parse metadata from Orca stubs: {e} Rerun stub generation: `orca sync`"
+                )
 
             if algorithm not in self._remoteDependencies:
                 self._remoteDependencies[algorithm] = [remoteAlgo]
             else:
                 self._remoteDependencies[algorithm].append(remoteAlgo)
-                
+
             return
 
         dependencyAlgo = None
@@ -694,7 +698,7 @@ def Register(self) -> None:
             algo_msg.window_type.version = algorithm.window_type.version
             algo_msg.window_type.description = algorithm.window_type.description
 
-            # Fill in metadata fields if present
+            # fill in metadata fields if present
             if len(algorithm.window_type.metadataFields) > 0:
                 for metadataField in algorithm.window_type.metadataFields:
                     metadata_fields_msg = algo_msg.window_type.metadataFields.add()
@@ -709,10 +713,12 @@ def Register(self) -> None:
                     dep_msg.version = dep.version
                     dep_msg.processor_name = dep.processor
                     dep_msg.processor_runtime = dep.runtime
-            
+
             # Add remote dependencies if they exist
             if algorithm.full_name in self._algorithmsSingleton._remoteDependencies:
-                for dep in self._algorithmsSingleton._remoteDependencies[algorithm.full_name]:
+                for dep in self._algorithmsSingleton._remoteDependencies[
+                    algorithm.full_name
+                ]:
                     dep_msg = algo_msg.dependencies.add()
                     dep_msg.name = dep.Name
                     dep_msg.version = dep.Version
@@ -840,7 +846,7 @@ def algorithm(
                 f"Version '{version}' must follow basic semantic "
                 "versioning (e.g., '1.0.0') without release portions"
             )
-            
+
         def inner(algo: T) -> T:
             def wrapper(
                 params: ExecutionParams,
@@ -892,14 +898,18 @@ def wrapper(
             )
 
             for dependency in depends_on:
-                if not getattr(dependency, "__orca_is_remote__", False) and not self._algorithmsSingleton._has_algorithm_fn(dependency):
+                if not getattr(
+                    dependency, "__orca_is_remote__", False
+                ) and not self._algorithmsSingleton._has_algorithm_fn(dependency):
                     message = (
                         f"Cannot add function `{dependency.__name__}` to dependency stack. All dependencies must "
                         "be decorated with `@algorithm` before they can be used as dependencies."
                     )
                     raise InvalidDependency(message)
                 self._algorithmsSingleton._add_dependency(
-                    algorithm.full_name, dependency, getattr(dependency, "__orca_is_remote__", False)
+                    algorithm.full_name,
+                    dependency,
+                    getattr(dependency, "__orca_is_remote__", False),
                 )
 
             # TODO: check for circular dependencies. It's not easy to create one in python as the function

orca_python/main.py

@@ -38,7 +38,7 @@ def _load_stub_metadata(func_name: str) -> dict[str, Any]:
                 # look at previous lines for metadata comment
                 for i in range(func_line - 1, max(0, func_line - 5), -1):
                     line = lines[i].strip()
-                    if match := re.search(r"#\s*ALGO_METADATA:\s*(\{.*\})", line):
+                    if match := re.search(r"#\s*METADATA:\s*(\{.*\})", line):
                         result["metadata"] = json.loads(match.group(1))
                         break
 
@@ -66,6 +66,7 @@ def __init__(self, func_name: str):
             self.__orca_metadata__ = stub_info["metadata"]
 
         def __call__(self, *args, **kwargs):
+            _, _ = args, kwargs
             raise NotImplementedError(
                 f"{self.__name__} is only available as a remote algorithm. "
                 f"Metadata: {self.__orca_metadata__}"

orca_python/registry/algorithms.py

@@ -0,0 +1,78 @@
+import re
+import ast
+import json
+from typing import Any
+from pathlib import Path
+
+from orca_python.exceptions import BrokenRemoteAlgorithmStubs
+
+
+def _load_stub_metadata(obj_name: str) -> dict[str, Any]:
+    """Load annotations and metadata from the .pyi stub file."""
+    stub_file = (
+        Path.cwd() / ".orca" / "orca_python" / "registry" / "metadata_fields.pyi"
+    )
+
+    if not stub_file.exists():
+        return {"annotations": {}, "metadata": {}}
+
+    try:
+        content = stub_file.read_text()
+        tree = ast.parse(content)
+
+        result = {"annotations": {}, "metadata": {}}
+
+        # get the annotated assignment node (e.g., asset_id: MetadataField)
+        for node in ast.walk(tree):
+            if isinstance(node, ast.AnnAssign):
+                # check if this is the target we're looking for
+                if isinstance(node.target, ast.Name) and node.target.id == obj_name:
+                    # get annotation
+                    if node.annotation:
+                        result["annotations"]["type"] = ast.unparse(node.annotation)
+
+                    # get metadata from comment above assignment
+                    lines = content.split("\n")
+                    assign_line = node.lineno - 1  # 0-indexed
+
+                    # look at previous lines for metadata comment
+                    for i in range(assign_line - 1, max(0, assign_line - 5), -1):
+                        line = lines[i].strip()
+                        if match := re.search(r"#\s*METADATA:\s*(\{.*\})", line):
+                            result["metadata"] = json.loads(match.group(1))
+                            break
+
+                    break
+
+        return result
+    except Exception as e:
+        print(f"Error parsing stub: {e}")
+
+    return {"annotations": {}, "metadata": {}}
+
+
+def __getattr__(name):
+    if name.startswith("_"):
+        raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
+    stub_info = _load_stub_metadata(name)
+
+    class StubMetadataField:
+        __orca_is_remote__ = True
+        name: str
+        description: str
+
+        def __init__(self, field_name: str):
+            self.__name__ = field_name
+            self.__annotations__ = stub_info["annotations"]
+            self.__orca_metadata__ = stub_info["metadata"]
+            name = stub_info["metadata"].get("Name", None)
+            description = stub_info["metadata"].get("Description", None)
+            if name is None or description is None:
+                raise BrokenRemoteAlgorithmStubs(
+                    "Stubs appear broken. Regenerate with `orca sync`."
+                )
+            self.name = name
+            self.description = description
+
+    return StubMetadataField(name)

orca_python/registry/metadata_fields.py

@@ -0,0 +1,109 @@
+import re
+import ast
+import json
+from typing import Any, List
+from pathlib import Path
+from dataclasses import dataclass
+
+from orca_python.main import BrokenRemoteAlgorithmStubs
+
+
+def _load_stub_metadata(obj_name: str) -> dict[str, Any]:
+    """Load annotations and metadata from the .pyi stub file."""
+    stub_file = Path.cwd() / ".orca" / "orca_python" / "registry" / "window_types.pyi"
+
+    if not stub_file.exists():
+        return {"annotations": {}, "metadata": {}}
+
+    try:
+        content = stub_file.read_text()
+        tree = ast.parse(content)
+
+        result = {"annotations": {}, "metadata": {}}
+
+        # get the annotated assignment node
+        for node in ast.walk(tree):
+            if isinstance(node, ast.AnnAssign):
+                # check if this is the target we're looking for
+                if isinstance(node.target, ast.Name) and node.target.id == obj_name:
+                    # get annotation
+                    if node.annotation:
+                        result["annotations"]["type"] = ast.unparse(node.annotation)
+
+                    # get metadata from comment above assignment
+                    lines = content.split("\n")
+                    assign_line = node.lineno - 1  # 0-indexed
+
+                    # look at previous lines for metadata comment
+                    for i in range(assign_line - 1, max(0, assign_line - 5), -1):
+                        line = lines[i].strip()
+                        if match := re.search(r"#\s*METADATA:\s*(\{.*\})", line):
+                            result["metadata"] = json.loads(match.group(1))
+                            break
+
+                    break
+
+        return result
+    except Exception as e:
+        print(f"Error parsing stub: {e}")
+
+    return {"annotations": {}, "metadata": {}}
+
+
+def __getattr__(name):
+    if name.startswith("_"):
+        raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
+
+    stub_info = _load_stub_metadata(name)
+
+    @dataclass
+    class StubMetadataField:
+        __orca_is_remote__ = True
+        name: str
+        description: str
+
+    class StubWindowType:
+        __orca_is_remote__ = True
+        name: str
+        version: str
+        description: str
+        metadataFields: List[StubMetadataField]
+
+        def __init__(self, window_name: str):
+            self.__annotations__ = stub_info["annotations"]
+            self.__orca_metadata__ = stub_info["metadata"]
+
+            # assign the values that the main package expects from a window
+            name = stub_info["metadata"].get("Name", None)
+            version = stub_info["metadata"].get("Version", None)
+            description = stub_info["metadata"].get("Description", None)
+            metadataFields = stub_info["metadata"].get("MetadataFields", None)
+
+            if (
+                name is None
+                or version is None
+                or description is None
+                or metadataFields is None
+            ):
+                raise BrokenRemoteAlgorithmStubs(
+                    "Stubs appear broken. Regenerate them with `orca sync`."
+                )
+
+            self.name = name
+            self.version = version
+            self.description = description
+            try:
+                self.metadataFields = [
+                    StubMetadataField(
+                        name=v.get("Name"), description=v.get("Description")
+                    )
+                    for v in metadataFields
+                ]
+            except Exception as e:
+                raise BrokenRemoteAlgorithmStubs(
+                    f"Stubs appear broken: {e}. Regenerate with `orca sync`"
+                )
+
+            self.__name__ = window_name
+
+    return StubWindowType(name)