From d6272247a957134a6fbb8767c7922e1c05ad1547 Mon Sep 17 00:00:00 2001 From: Frederick Mannings Date: Mon, 22 Sep 2025 10:47:36 +0100 Subject: [PATCH 1/5] Added window type metadata fields & fixed bugs --- examples/simpletrigger/processor.py | 11 +++++++- orca_python/__init__.py | 2 ++ orca_python/envs.py | 2 +- orca_python/exceptions.py | 4 +++ orca_python/main.py | 40 ++++++++++++++++++++++++----- pyproject.toml | 5 ++++ tests/test_windows.py | 38 +++++++++++++++++++++++++++ 7 files changed, 93 insertions(+), 9 deletions(-) create mode 100644 tests/test_windows.py diff --git a/examples/simpletrigger/processor.py b/examples/simpletrigger/processor.py index 630fcdd..da33084 100644 --- a/examples/simpletrigger/processor.py +++ b/examples/simpletrigger/processor.py @@ -4,13 +4,22 @@ 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.trip_id + bus_id = params.window.metadata.bus_id + time.sleep(5) return StructResult({"result": 42}) diff --git a/orca_python/__init__.py b/orca_python/__init__.py index e861125..78fb3a5 100644 --- a/orca_python/__init__.py +++ b/orca_python/__init__.py @@ -7,6 +7,7 @@ ArrayResult, ValueResult, StructResult, + MetadataField, ExecutionParams, ) @@ -14,6 +15,7 @@ "Processor", "EmitWindow", "Window", + "MetadataField", "WindowType", "StructResult", "ValueResult", diff --git a/orca_python/envs.py b/orca_python/envs.py index 957ea89..133f636 100644 --- a/orca_python/envs.py +++ b/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() diff --git a/orca_python/exceptions.py b/orca_python/exceptions.py index a9b1544..898f612 100644 --- a/orca_python/exceptions.py +++ b/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""" diff --git a/orca_python/main.py b/orca_python/main.py index de8f7b9..b75d182 100644 --- a/orca_python/main.py +++ b/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}") @@ -657,14 +683,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}") diff --git a/pyproject.toml b/pyproject.toml index f286a73..0bc11a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,11 @@ markers = [ "live: marks tests as requireing access to live services" ] +[tool.poe.env] +ORCA_CORE="" +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 " diff --git a/tests/test_windows.py b/tests/test_windows.py new file mode 100644 index 0000000..9000896 --- /dev/null +++ b/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") + ], + ) From 734f6bc40cadf0907282ee20123a65a85e98118d Mon Sep 17 00:00:00 2001 From: Frederick Mannings Date: Tue, 23 Sep 2025 14:59:06 +0100 Subject: [PATCH 2/5] Collapsed the window type definition to simply be the python dataclass --- examples/simpletrigger/processor.py | 13 ++++++++++--- orca | 2 +- orca_python/main.py | 28 +++++++++++++++------------- 3 files changed, 26 insertions(+), 17 deletions(-) diff --git a/examples/simpletrigger/processor.py b/examples/simpletrigger/processor.py index da33084..65f9793 100644 --- a/examples/simpletrigger/processor.py +++ b/examples/simpletrigger/processor.py @@ -1,6 +1,12 @@ import time -from orca_python import Processor, WindowType, StructResult, ExecutionParams +from orca_python import ( + Processor, + WindowType, + StructResult, + MetadataField, + ExecutionParams, +) proc = Processor("ml") @@ -17,8 +23,9 @@ @proc.algorithm("MyAlgo", "1.0.0", Every30Second) def my_algorithm(params: ExecutionParams) -> StructResult: - trip_id = params.window.metadata.trip_id - bus_id = params.window.metadata.bus_id + 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}) diff --git a/orca b/orca index 33369bd..afcc89e 160000 --- a/orca +++ b/orca @@ -1 +1 @@ -Subproject commit 33369bd4e916b3583d9e5f8fbaa1f15330a53522 +Subproject commit afcc89ee386bb1da93eb0c544c0d44c41bf01ba4 diff --git a/orca_python/main.py b/orca_python/main.py index b75d182..3a30f63 100644 --- a/orca_python/main.py +++ b/orca_python/main.py @@ -260,8 +260,7 @@ 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. @@ -269,9 +268,7 @@ class Algorithm: name: str version: str - window_name: str - window_version: str - window_description: str + window_type: WindowType exec_fn: AlgorithmFn processor: str runtime: str @@ -285,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: @@ -319,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 @@ -667,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: @@ -830,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, From 2e1be440bfcaf5a0305119e53ccda9022e00eaaf Mon Sep 17 00:00:00 2001 From: Frederick Mannings Date: Tue, 23 Sep 2025 15:19:31 +0100 Subject: [PATCH 3/5] Converted to using the new orca core version where the metadata is more explicit and garuanteed --- examples/simpletrigger/window.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/simpletrigger/window.py b/examples/simpletrigger/window.py index eb208fb..818eb47 100644 --- a/examples/simpletrigger/window.py +++ b/examples/simpletrigger/window.py @@ -1,7 +1,7 @@ -import time import datetime as dt import schedule +from processor import bus_id, trip_id from orca_python import Window, EmitWindow @@ -14,6 +14,7 @@ def emitWindow() -> None: name="Every30Second", version="1.0.0", origin="Example", + metadata={bus_id.name: 1, trip_id.name: 2}, ) EmitWindow(window) @@ -21,6 +22,7 @@ def emitWindow() -> None: schedule.every(30).seconds.do(emitWindow) if __name__ == "__main__": + emitWindow() while True: schedule.run_pending() time.sleep(1) From 402823ae3ac592ffd94115185f4e8c2c116c4fd7 Mon Sep 17 00:00:00 2001 From: Frederick Mannings Date: Tue, 23 Sep 2025 15:23:27 +0100 Subject: [PATCH 4/5] Bumped orca version --- orca | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/orca b/orca index afcc89e..16b6306 160000 --- a/orca +++ b/orca @@ -1 +1 @@ -Subproject commit afcc89ee386bb1da93eb0c544c0d44c41bf01ba4 +Subproject commit 16b63068e7011df9622a5fe935eb854ccbe6baac From 4b8d80ff4512b50ac48ecb1f307e6f748c7a9c11 Mon Sep 17 00:00:00 2001 From: Frederick Mannings Date: Tue, 23 Sep 2025 15:32:20 +0100 Subject: [PATCH 5/5] missing import --- examples/simpletrigger/window.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/simpletrigger/window.py b/examples/simpletrigger/window.py index 818eb47..d510c55 100644 --- a/examples/simpletrigger/window.py +++ b/examples/simpletrigger/window.py @@ -1,3 +1,4 @@ +import time import datetime as dt import schedule