Add test with suggested inputs

Author: Carlos-Marques

nortech/derivers/services/schema.py

@@ -116,8 +116,9 @@ def check_create_deriver_schema_imports(
     # Retrieve the source code of the function
     source_code = dedent(getsource(create_deriver_schema))
 
-    pattern = r"suggestedInputs=\[.*?\],?\s*"
+    pattern = r"suggested_inputs=\[.*?\],?\s*"
     source_code_without_suggested_inputs = sub(pattern, "", source_code, flags=DOTALL)
+    print(source_code_without_suggested_inputs)
 
     # Define a clean global environment for execution
     clean_env: dict[str, Any] = {}

nortech/derivers/services/visualize.py

@@ -11,6 +11,7 @@ def create_deriver_schema_subgraph(deriver_schema_dag: DeriverSchemaDAG):
     for deriver_input in deriver_schema_dag.inputs:
         if deriver_input.name != "timestamp":
             node_id = sha256(deriver_schema_dag.name.encode()).hexdigest()[:8]
+
             if deriver_input.physical_quantity:
                 mermaid += f"""
                     {node_id}_{deriver_input.name}["{deriver_input.name}<br/>[{deriver_input.physical_quantity.si_unit_symbol.replace(" ", "")}]"] --> transform_stream_{node_id}["transform_stream"]
@@ -23,6 +24,7 @@ def create_deriver_schema_subgraph(deriver_schema_dag: DeriverSchemaDAG):
     for deriver_output in deriver_schema_dag.outputs:
         if deriver_output.name != "timestamp":
             node_id = sha256(deriver_schema_dag.name.encode()).hexdigest()[:8]
+
             if deriver_output.physical_quantity:
                 mermaid += f"""
                     transform_stream_{node_id} --> {node_id}_{deriver_output.name}["{deriver_output.name}<br/>[{deriver_output.physical_quantity.si_unit_symbol.replace(" ", "")}]"]
@@ -50,6 +52,7 @@ def create_deriver_schema_dag_mermaid(mermaid: str, deriver_schema_dag: DeriverS
 
                 source_id = sha256(suggested_input.deriver_schema_dag.name.encode()).hexdigest()[:8]
                 target_id = sha256(deriver_schema_dag.name.encode()).hexdigest()[:8]
+
                 mermaid += f"""
                     {source_id}_{suggested_input.name} -.->|suggestedInput| {target_id}_{deriver_input.name}
                 """

tests/integration/derivers/snapshots/test_deriver/test_visualize_deriver/markdown_obj.txt

@@ -3,17 +3,17 @@
 flowchart LR
     subgraph "Deriver (Test Deriver)"
 
-        subgraph "DeriverSchema (Test Schema)"
+        subgraph "DeriverSchema (Test Schema with suggested inputs)"
 
-                    4a9b921b_input_signal["input_signal<br/>[K]"] --> transform_stream_4a9b921b["transform_stream"]
+                    91ea505a_input_signal["input_signal<br/>[K]"] --> transform_stream_91ea505a["transform_stream"]
 
-                    transform_stream_4a9b921b --> 4a9b921b_output_signal["output_signal<br/>[K]"]
+                    transform_stream_91ea505a --> 91ea505a_output_signal["output_signal<br/>[K]"]
 
         end
 
-            847cb464_Signal["Signal<br/>[°C]"] --> 4a9b921b_input_signal
+            847cb464_Signal["Signal<br/>[°C]"] --> 91ea505a_input_signal
 
-            4a9b921b_output_signal --> 847cb464_output_signal["output_signal<br/>[°C]"]
+            91ea505a_output_signal --> 847cb464_output_signal["output_signal<br/>[°C]"]
 
 end
 ```

tests/integration/derivers/snapshots/test_deriver/test_visualize_deriver_schema/markdown_obj.txt

@@ -8,6 +8,16 @@ flowchart LR
 
                     transform_stream_4a9b921b --> 4a9b921b_output_signal["output_signal<br/>[K]"]
 
+        end
+    
+                    4a9b921b_output_signal -.->|suggestedInput| 91ea505a_input_signal
+                
+        subgraph "DeriverSchema (Test Schema with suggested inputs)"
+    
+                    91ea505a_input_signal["input_signal<br/>[K]"] --> transform_stream_91ea505a["transform_stream"]
+                
+                    transform_stream_91ea505a --> 91ea505a_output_signal["output_signal<br/>[K]"]
+                
         end
 
 ```

tests/integration/derivers/test_deriver.py

@@ -74,7 +74,66 @@ def transform_stream(
     )
 
 
-deriver_schema = create_test_schema()
+deriver_schema_without_suggested_inputs = create_test_schema()
+
+
+def create_test_schema_with_suggested_inputs():
+    import bytewax.operators as op
+    from bytewax.dataflow import Stream
+    from pydantic import BaseModel
+
+    from nortech.derivers import (
+        DeriverInputSchema,
+        DeriverOutputSchema,
+        DeriverSchema,
+        InputField,
+        OutputField,
+        physical_units,
+    )
+
+    class Input(DeriverInputSchema):
+        input_signal: float | None = InputField(
+            description="Input signal description",
+            physical_quantity=physical_units.temperature,
+            suggested_inputs=[deriver_schema_without_suggested_inputs.outputs.output_signal],
+        )
+
+    class Output(DeriverOutputSchema):
+        output_signal: float | None = OutputField(
+            description="Output signal description",
+            physical_quantity=physical_units.temperature,
+            create_deriver_schema=create_test_schema_with_suggested_inputs,
+        )
+
+    class Configurations(BaseModel):
+        pass
+
+    def transform_stream(
+        stream: Stream[Input],
+        config: Configurations,
+    ) -> Stream[Output]:
+        output_stream = op.map(
+            step_id="map_output",
+            up=stream,
+            mapper=lambda input_message: Output(
+                timestamp=input_message.timestamp,
+                output_signal=input_message.input_signal,
+            ),
+        )
+
+        return output_stream
+
+    return DeriverSchema(
+        name="Test Schema with suggested inputs",
+        description="Test Schema description with suggested inputs",
+        inputs=Input,
+        outputs=Output,
+        configurations=Configurations,
+        transform_stream=transform_stream,
+    )
+
+
+deriver_schema = create_test_schema_with_suggested_inputs()
 
 inputs = {
     deriver_schema.inputs.input_signal: DeriverInput(
@@ -107,7 +166,7 @@ def transform_stream(
     outputs=outputs,
     configurations=configurations,
     start_at=datetime(2022, 1, 1, 0, 0, 0),
-    create_deriver_schema=create_test_schema,
+    create_deriver_schema=create_test_schema_with_suggested_inputs,
 )