SIENTIAPDE-1095

Refactor predictions_batch to use server_id instead of server_name; update values.yaml to set GITHUB_BRANCH to main; add new laborious pipeline configuration in samples.json

orchestrator/utils/orchestrator_functions.py

@@ -65,18 +65,18 @@ def process_path_priority(path_priority: list[str]):
 def predictions_batch(config: dict[str, Any]):
     tags = {}
     for tag in config['write_tags']:
-        if tag['server_name'] not in tags:
-            tags[tag['server_name']] = {}
+        if tag['server_id'] not in tags:
+            tags[tag['server_id']] = {}
 
         tag_type = tag['type']
 
         if tag_type == 'prediction' or tag_type == 'confidence':
             tag_type_str = f"{tag_type}_tags"
 
-            if tag_type_str not in tags[tag['server_name']]:
-                tags[tag['server_name']][tag_type_str] = {}
+            if tag_type_str not in tags[tag['server_id']]:
+                tags[tag['server_id']][tag_type_str] = {}
 
-            tags[tag['server_name']][tag_type_str][tag['addr']] = {
+            tags[tag['server_id']][tag_type_str][tag['addr']] = {
                 "data_type": tag.get('data_type', 'float'),
             }
 

samples.json

@@ -57,6 +57,65 @@
               }
             ],
             "tag_retention_minutes": 60
+          },
+        "2": {
+            "schedule_name": "laborious-orchestrated-pipeline",
+            "model_id": "1",
+            "workflow_type": "predictions_batch",
+            "frequency": "30s",
+            "max_retry_policy": 1,
+            "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
+            "retention_time": 60,
+            "opc_output_config": [
+              {
+                "server_id": "1",
+                "type": "prediction",
+                "addr": "ns=2;i=2",
+                "data_type": "float"
+              },
+              {
+                "server_id": "1",
+                "type": "confidence",
+                "addr": "ns=2;i=2",
+                "data_type": "float"
+              }
+            ],
+            "input_filters": [
+              {
+                "filter_name": "EMPTY_DATA",
+                "policy": "STOP"
+              },
+              {
+                "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
+                "policy": "CONTINUE",
+                "config": {
+                  "VARIABLES": [
+                    "Counter"
+                  ]
+                }
+              }
+            ],
+            "mlflow_transform_filters": [
+              {
+                "filter_name": "API_ERROR",
+                "policy": "REPEAT"
+              },
+              {
+                "filter_name": "NAN_VALUES",
+                "policy": "STOP"
+              }
+            ],
+            "mlflow_predict_filters": [
+              {
+                "filter_name": "API_ERROR",
+                "policy": "CONTINUE"
+              }
+            ],
+            "path_priority": [
+              "STOP",
+              "CONTINUE",
+              "REPEAT"
+            ]
           }
     },
     "opc-servers": {

values.yaml

@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
+    value: "main"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"