Orc 48

examples/simpletrigger/processor.py

@@ -1,16 +1,32 @@
 import time
 
-from orca_python import Processor, WindowType, StructResult, ExecutionParams
+from orca_python import (
+    Processor,
+    WindowType,
+    StructResult,
+    MetadataField,
+    ExecutionParams,
+)
 
 proc = Processor("ml")
 
+trip_id = MetadataField(name="trip_id", description="The unique ID of the trip")
+bus_id = MetadataField(name="bus_id", description="The unique ID of the bus")
+
 Every30Second = WindowType(
-    name="Every30Second", version="1.0.0", description="Triggers every 30 seconds"
+    name="Every30Second",
+    version="1.0.0",
+    description="Triggers every 30 seconds",
+    metadataFields=[trip_id, bus_id],
 )
 
 
 @proc.algorithm("MyAlgo", "1.0.0", Every30Second)
 def my_algorithm(params: ExecutionParams) -> StructResult:
+    trip_id = params.window.metadata.get("trip_id", None)
+    bus_id = params.window.metadata.get("bus_id", None)
+    print(trip_id, bus_id)
+
     time.sleep(5)
     return StructResult({"result": 42})
 

examples/simpletrigger/window.py

@@ -2,6 +2,7 @@
 import datetime as dt
 
 import schedule
+from processor import bus_id, trip_id
 
 from orca_python import Window, EmitWindow
 
@@ -14,13 +15,15 @@ def emitWindow() -> None:
         name="Every30Second",
         version="1.0.0",
         origin="Example",
+        metadata={bus_id.name: 1, trip_id.name: 2},
     )
     EmitWindow(window)
 
 
 schedule.every(30).seconds.do(emitWindow)
 
 if __name__ == "__main__":
+    emitWindow()
     while True:
         schedule.run_pending()
         time.sleep(1)

orca_python/__init__.py

@@ -7,13 +7,15 @@
     ArrayResult,
     ValueResult,
     StructResult,
+    MetadataField,
     ExecutionParams,
 )
 
 __all__ = [
     "Processor",
     "EmitWindow",
     "Window",
+    "MetadataField",
     "WindowType",
     "StructResult",
     "ValueResult",

orca_python/envs.py

@@ -27,4 +27,4 @@ def getenvs() -> Tuple[bool, str, str, str]:
     return is_production, orcaserver, port, host
 
 
-is_production, ORCASERVER, PORT, HOST = getenvs()
+is_production, ORCACORE, PORT, HOST = getenvs()

orca_python/exceptions.py

@@ -14,6 +14,10 @@ class InvalidWindowArgument(BaseOrcaException):
     """Raised when an argument to the Window class is not valid"""
 
 
+class InvalidMetadataFieldArgument(BaseOrcaException):
+    """Raised when an argument to a metadata field is not valid"""
+
+
 class InvalidDependency(BaseOrcaException):
     """Raised when a dependency is invalid"""
 

orca_python/main.py

@@ -55,6 +55,7 @@
     InvalidWindowArgument,
     InvalidAlgorithmArgument,
     InvalidAlgorithmReturnType,
+    InvalidMetadataFieldArgument,
 )
 
 # Regex patterns for validation
@@ -66,24 +67,49 @@
 LOGGER = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class MetadataField:
+    name: str
+    description: str
+
+    def __post_init__(self) -> None:
+        if self.name == "":
+            raise InvalidMetadataFieldArgument("Metadata field name cannot be empty")
+
+        if self.description == "":
+            raise InvalidMetadataFieldArgument(
+                "Metadata field description cannot be empty"
+            )
+
+
 @dataclass
 class WindowType:
     name: str
     version: str
     description: str
+    metadataFields: List[MetadataField] = field(default_factory=list)
 
-    def __post__init__(self) -> None:
+    def __post_init__(self) -> None:
         if not re.match(WINDOW_NAME, self.name):
-            raise InvalidAlgorithmArgument(
+            raise InvalidWindowArgument(
                 f"Window name '{self.name}' must be in PascalCase"
             )
 
         if not re.match(SEMVER_PATTERN, self.version):
-            raise InvalidAlgorithmArgument(
+            raise InvalidWindowArgument(
                 f"Window version '{self.version}' must follow basic semantic "
                 "versioning (e.g., '1.0.0') without release portions"
             )
 
+        _seenFields = set()
+        for field in self.metadataFields:
+            if field in _seenFields:
+                raise InvalidWindowArgument(
+                    f"Two or more metadata fields provided with the same name:'{field.name}' and description@ '{field.description}"
+                )
+            else:
+                _seenFields.add(field)
+
 
 @dataclass
 class StructResult:
@@ -213,14 +239,14 @@ def EmitWindow(window: Window) -> None:
     if envs.is_production:
         # secure channel with TLS
         with grpc.secure_channel(
-            envs.ORCASERVER, grpc.ssl_channel_credentials()
+            envs.ORCACORE, grpc.ssl_channel_credentials()
         ) as channel:
             stub = service_pb2_grpc.OrcaCoreStub(channel)
             response = stub.EmitWindow(window_pb)
             LOGGER.info(f"Window emitted: {response}")
     else:
         # insecure channel for local development
-        with grpc.insecure_channel(envs.ORCASERVER) as channel:
+        with grpc.insecure_channel(envs.ORCACORE) as channel:
             stub = service_pb2_grpc.OrcaCoreStub(channel)
             response = stub.EmitWindow(window_pb)
             LOGGER.info(f"Window emitted: {response}")
@@ -234,18 +260,15 @@ class Algorithm:
     Attributes:
         name (str): The name of the algorithm (PascalCase).
         version (str): Semantic version of the algorithm (e.g., "1.0.0").
-        window_name (str): The window type name that triggers the algorithm.
-        window_version (str): The version of the window type.
+        window_type (WindowType): The window type triggers the algorithm.
         exec_fn (AlgorithmFn): The execution function for the algorithm.
         processor (str): Name of the processor where it's registered.
         runtime (str): Python runtime used for execution.
     """
 
     name: str
     version: str
-    window_name: str
-    window_version: str
-    window_description: str
+    window_type: WindowType
     exec_fn: AlgorithmFn
     processor: str
     runtime: str
@@ -259,7 +282,7 @@ def full_name(self) -> str:
     @property
     def full_window_name(self) -> str:
         """Returns the full window name as `window_name_window_version`."""
-        return f"{self.window_name}_{self.window_version}"
+        return f"{self.window_type.name}_{self.window_type.version}"
 
 
 class Algorithms:
@@ -293,7 +316,7 @@ def _add_algorithm(self, name: str, algorithm: Algorithm) -> None:
             LOGGER.error(f"Attempted to register duplicate algorithm: {name}")
             raise ValueError(f"Algorithm {name} already exists")
         LOGGER.info(
-            f"Registering algorithm: {name} (window: {algorithm.window_name}_{algorithm.window_version})"
+            f"Registering algorithm: {name} (window: {algorithm.window_type.name}_{algorithm.window_type.version})"
         )
         self._algorithms[name] = algorithm
 
@@ -641,9 +664,16 @@ def Register(self) -> None:
             algo_msg.result_type = result_type_pb
 
             # Add window type
-            algo_msg.window_type.name = algorithm.window_name
-            algo_msg.window_type.version = algorithm.window_version
-            algo_msg.window_type.description = algorithm.window_description
+            algo_msg.window_type.name = algorithm.window_type.name
+            algo_msg.window_type.version = algorithm.window_type.version
+            algo_msg.window_type.description = algorithm.window_type.description
+
+            # 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()
+                    metadata_fields_msg.name = metadataField.name
+                    metadata_fields_msg.description = metadataField.description
 
             # Add dependencies if they exist
             if algorithm.full_name in self._algorithmsSingleton._dependencies:
@@ -657,14 +687,14 @@ def Register(self) -> None:
         if envs.is_production:
             # secure channel with TLS
             with grpc.secure_channel(
-                envs.ORCASERVER, grpc.ssl_channel_credentials()
+                envs.ORCACORE, grpc.ssl_channel_credentials()
             ) as channel:
                 stub = service_pb2_grpc.OrcaCoreStub(channel)
                 response = stub.RegisterProcessor(registration_request)
                 LOGGER.info(f"Algorithm registration response received: {response}")
         else:
             # insecure channel for local development
-            with grpc.insecure_channel(envs.ORCASERVER) as channel:
+            with grpc.insecure_channel(envs.ORCACORE) as channel:
                 stub = service_pb2_grpc.OrcaCoreStub(channel)
                 response = stub.RegisterProcessor(registration_request)
                 LOGGER.info(f"Algorithm registration response received: {response}")
@@ -804,9 +834,7 @@ def wrapper(
             algorithm = Algorithm(
                 name=name,
                 version=version,
-                window_name=window_type.name,
-                window_version=window_type.version,
-                window_description=window_type.description,
+                window_type=window_type,
                 exec_fn=wrapper,
                 processor=self._name,
                 runtime=sys.version,

pyproject.toml

@@ -58,6 +58,11 @@ markers = [
   "live: marks tests as requireing access to live services"
 ]
 
+[tool.poe.env]
+ORCA_CORE="<dummy_grpc_url>"
+PROCESSOR_PORT="5051"
+PROCESSOR_ADDRESS="[::]"
+
 [tool.poe.tasks]
 _lint_check = "ruff check orca_python tests examples"
 _lint_fix = "ruff check orca_python tests examples --fix "

tests/test_windows.py

@@ -0,0 +1,38 @@
+import pytest
+
+from orca_python import WindowType, MetadataField
+from orca_python.exceptions import InvalidWindowArgument, InvalidMetadataFieldArgument
+
+
+def test_metadata_fields():
+    with pytest.raises(InvalidMetadataFieldArgument):
+        MetadataField(name="", description="test description")
+
+    with pytest.raises(InvalidMetadataFieldArgument):
+        MetadataField(name="test name", description="")
+
+    with pytest.raises(InvalidMetadataFieldArgument):
+        MetadataField(name="", description="")
+
+    MetadataField(name="test name", description="test description")
+
+
+def test_window_type_definition():
+    with pytest.raises(InvalidWindowArgument):
+        WindowType(
+            name="TestWindow",
+            version="1.0.0",
+            description="test description",
+            metadataFields=[
+                MetadataField(name="testName", description="test description"),
+                MetadataField(name="testName", description="test description"),
+            ],
+        )
+    WindowType(
+        name="TestWindow",
+        version="1.0.0",
+        description="test description",
+        metadataFields=[
+            MetadataField(name="test name", description="test description")
+        ],
+    )