This is an automated email from the ASF dual-hosted git repository. oehler pushed a commit to branch 1260-refactor-required-streams-for-python-functions in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 29df939d62caf7800768e2b25b66d1bf55a6553c Author: Sven Oehler <[email protected]> AuthorDate: Wed Apr 26 14:39:10 2023 +0200 Replace requiredStreamIds with consumed_streams --- ...ive-data-from-the-streampipes-data-stream.ipynb | 27 ++++++++------- .../streampipes/function_zoo/river_function.py | 19 ++-------- .../streampipes/functions/streampipes_function.py | 3 +- .../tests/functions/test_function_handler.py | 40 +++++++++++++--------- 4 files changed, 42 insertions(+), 47 deletions(-) diff --git a/streampipes-client-python/docs/tutorials/3-getting-live-data-from-the-streampipes-data-stream.ipynb b/streampipes-client-python/docs/tutorials/3-getting-live-data-from-the-streampipes-data-stream.ipynb index 5b206df5f..6f20e4f90 100644 --- a/streampipes-client-python/docs/tutorials/3-getting-live-data-from-the-streampipes-data-stream.ipynb +++ b/streampipes-client-python/docs/tutorials/3-getting-live-data-from-the-streampipes-data-stream.ipynb @@ -246,13 +246,14 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "Next we can create a StreamPipesFunction. For this we need to implement the 4 following methods:\n", - "- In `requiredStreamIds` you need to insert the `element_id` of the required streams.\n", + "Next we can create a StreamPipesFunction. For this we need to implement the 3 following methods:\n", "- `onServiceStarted` is called when the function gets started. There you can use the given meta information of the `FunctionContext` to initialize the function.\n", "- `onEvent` is called when ever a new event arrives. The `event` contains the live data and you can use the `streamId` to identify a stream if the function is connected to multiple data streams.\n", "- `onServiceStopped` is called when the function gets stopped.\n", "\n", - "For this tutorial we just create a function that saves every new event in a `pandas DataFrame` and plots the first column of the `DataFrame` when the function gets stopped." + "For this tutorial we just create a function that saves every new event in a `pandas DataFrame` and plots the first column of the `DataFrame` when the function gets stopped. \n", + " \n", + "(If you want to use the same structure as in Java you can overwrite the `getFunctionId` and `requiredStreamIds` methods instead of using the `FunctionDefinition`)" ] }, { @@ -261,7 +262,7 @@ "metadata": {}, "outputs": [], "source": [ - "from typing import List, Dict, Any\n", + "from typing import Dict, Any\n", "import pandas as pd\n", "from datetime import datetime\n", "import matplotlib.pyplot as plt\n", @@ -269,17 +270,14 @@ "from streampipes.functions.registration import Registration\n", "from streampipes.functions.streampipes_function import StreamPipesFunction\n", "from streampipes.functions.utils.function_context import FunctionContext\n", - "\n", + "from streampipes.model.resource.function_definition import FunctionDefinition, FunctionId\n", "\n", "class ExampleFunction(StreamPipesFunction):\n", - " def __init__(self) -> None:\n", - " super().__init__()\n", + " def __init__(self, function_definition: FunctionDefinition) -> None:\n", + " super().__init__(function_definition)\n", " # Create the Dataframe to save the live data\n", " self.df = pd.DataFrame()\n", "\n", - " def requiredStreamIds(self) -> List[str]:\n", - " return [\"urn:streampipes.apache.org:eventstream:uPDKLI\"]\n", - "\n", " def onServiceStarted(self, context: FunctionContext):\n", " # Get the name of the timestamp field\n", " for event_property in context.schema[context.streams[0]].event_schema.event_properties:\n", @@ -308,7 +306,7 @@ "cell_type": "markdown", "metadata": {}, "source": [ - "Now we can start the function. We have to register an instance of the `StreamPipesFunction` and them we can start the functions by initializing the `FunctionHandler`. (it's also possible to register multiple functions with `.register(...).register(...)`)" + "Now we can start the function. First we create an instance of the `ExampleFunction` and insert the `element_id` of the stream which data we want to consume. Then we have to register this function and we can start all functions by initializing the `FunctionHandler`. (it's also possible to register multiple functions with `.register(...).register(...)`)" ] }, { @@ -326,7 +324,12 @@ } ], "source": [ - "example_function = ExampleFunction()\n", + "example_function = ExampleFunction(\n", + " FunctionDefinition(\n", + " function_id=FunctionId(id=\"example-function\"),\n", + " consumed_streams=[\"urn:streampipes.apache.org:eventstream:uPDKLI\"]\n", + " )\n", + ")\n", "\n", "registration = Registration()\n", "registration.register(example_function)\n", diff --git a/streampipes-client-python/streampipes/function_zoo/river_function.py b/streampipes-client-python/streampipes/function_zoo/river_function.py index b42a7dea9..9bbcab3a7 100644 --- a/streampipes-client-python/streampipes/function_zoo/river_function.py +++ b/streampipes-client-python/streampipes/function_zoo/river_function.py @@ -39,8 +39,6 @@ class RiverFunction(StreamPipesFunction): ---------- function_definition: FunctionDefinition The function definition which contains the output stream. - stream_ids: List[str] - The ids of the data stream to train the model. model: Any The model to train. It meant to be a River model/pipeline, but can be every model with a 'learn_one' and 'predict_one' method. @@ -59,7 +57,6 @@ class RiverFunction(StreamPipesFunction): def __init__( self, function_definition: FunctionDefinition, - stream_ids: List[str], model: Any, supervised: bool, target_label: Optional[str], @@ -68,7 +65,6 @@ class RiverFunction(StreamPipesFunction): on_stop: Callable[[Any], None], ) -> None: super().__init__(function_definition) - self.stream_ids = stream_ids self.model = model self.supervised = supervised self.target_label = target_label @@ -78,17 +74,6 @@ class RiverFunction(StreamPipesFunction): self.learning = True - def requiredStreamIds(self) -> List[str]: - """Returns the stream ids required by this function. - - Returns - ------- - stream_ids: List[str] - List of stream ids required by the function - - """ - return self.stream_ids - def onServiceStarted(self, context: FunctionContext): """Executes the `on_start` method of the function. @@ -194,9 +179,9 @@ class OnlineML: attributes=attributes, broker=get_broker_description(client.dataStreamApi.get(stream_ids[0])), # type: ignore ) - function_definition = FunctionDefinition().add_output_data_stream(output_stream) + function_definition = FunctionDefinition(consumed_streams=stream_ids).add_output_data_stream(output_stream) self.sp_function = RiverFunction( - function_definition, stream_ids, model, supervised, target_label, on_start, on_event, on_stop + function_definition, model, supervised, target_label, on_start, on_event, on_stop ) def start(self): diff --git a/streampipes-client-python/streampipes/functions/streampipes_function.py b/streampipes-client-python/streampipes/functions/streampipes_function.py index bfb267c4a..6e93f1d2f 100644 --- a/streampipes-client-python/streampipes/functions/streampipes_function.py +++ b/streampipes-client-python/streampipes/functions/streampipes_function.py @@ -83,7 +83,6 @@ class StreamPipesFunction(ABC): collector.disconnect() self.onServiceStopped() - @abstractmethod def requiredStreamIds(self) -> List[str]: """Get the ids of the streams needed by the function. @@ -92,7 +91,7 @@ class StreamPipesFunction(ABC): stream_ids: List[str] List of the stream ids """ - raise NotImplementedError # pragma: no cover + return self.function_definition.consumed_streams @abstractmethod def onServiceStarted(self, context: FunctionContext) -> None: diff --git a/streampipes-client-python/tests/functions/test_function_handler.py b/streampipes-client-python/tests/functions/test_function_handler.py index fdb27ea46..7feeebc6a 100644 --- a/streampipes-client-python/tests/functions/test_function_handler.py +++ b/streampipes-client-python/tests/functions/test_function_handler.py @@ -39,9 +39,6 @@ from streampipes.model.resource.function_definition import FunctionDefinition class TestFunction(StreamPipesFunction): - def requiredStreamIds(self) -> List[str]: - return ["urn:streampipes.apache.org:eventstream:uPDKLI"] - def onServiceStarted(self, context: FunctionContext): self.context = context self.data: List[Dict[str, Any]] = [] @@ -54,9 +51,6 @@ class TestFunction(StreamPipesFunction): class TestFunctionTwoStreams(StreamPipesFunction): - def requiredStreamIds(self) -> List[str]: - return ["urn:streampipes.apache.org:eventstream:uPDKLI", "urn:streampipes.apache.org:eventstream:HHoidJ"] - def onServiceStarted(self, context: FunctionContext): self.context = context self.data1: List[Dict[str, Any]] = [] @@ -73,9 +67,6 @@ class TestFunctionTwoStreams(StreamPipesFunction): class TestFunctionOutput(StreamPipesFunction): - def requiredStreamIds(self) -> List[str]: - return ["urn:streampipes.apache.org:eventstream:uPDKLI"] - def onServiceStarted(self, context: FunctionContext): self.context = context self.i = 0 @@ -192,7 +183,9 @@ class TestFunctionHandler(TestCase): ) registration = Registration() - test_function = TestFunction() + test_function = TestFunction( + FunctionDefinition(consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]) + ) registration.register(test_function) function_handler = FunctionHandler(registration, client) function_handler.initializeFunctions() @@ -229,7 +222,9 @@ class TestFunctionHandler(TestCase): ) registration = Registration() - test_function = TestFunction() + test_function = TestFunction( + FunctionDefinition(consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]) + ) registration.register(test_function) function_handler = FunctionHandler(registration, client) function_handler.initializeFunctions() @@ -262,7 +257,9 @@ class TestFunctionHandler(TestCase): ) registration = Registration() - test_function = TestFunction() + test_function = TestFunction( + FunctionDefinition(consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]) + ) registration.register(test_function) function_handler = FunctionHandler(registration, client) with self.assertRaises(UnsupportedBrokerError): @@ -301,8 +298,17 @@ class TestFunctionHandler(TestCase): ) registration = Registration() - test_function1 = TestFunction() - test_function2 = TestFunctionTwoStreams() + test_function1 = TestFunction( + FunctionDefinition(consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]) + ) + test_function2 = TestFunctionTwoStreams( + FunctionDefinition( + consumed_streams=[ + "urn:streampipes.apache.org:eventstream:uPDKLI", + "urn:streampipes.apache.org:eventstream:HHoidJ", + ] + ) + ) registration.register(test_function1).register(test_function2) function_handler = FunctionHandler(registration, client) function_handler.initializeFunctions() @@ -363,7 +369,9 @@ class TestFunctionHandler(TestCase): output_stream = create_data_stream("test", attributes={"number": RuntimeType.INTEGER.value}) test_function = TestFunctionOutput( - function_definition=FunctionDefinition().add_output_data_stream(output_stream) + function_definition=FunctionDefinition( + consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"] + ).add_output_data_stream(output_stream) ) registration = Registration() registration.register(test_function) @@ -425,7 +433,7 @@ class TestFunctionHandler(TestCase): "test", attributes={"number": RuntimeType.INTEGER.value}, broker=SupportedBroker.KAFKA ) test_function = TestFunctionOutput( - function_definition=FunctionDefinition().add_output_data_stream(output_stream) + function_definition=FunctionDefinition(consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]).add_output_data_stream(output_stream) ) registration = Registration() registration.register(test_function)
