From 16e76f11eadf3cdd4be34c0a3d27688907586c3f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 28 Oct 2026 15:52:35 -0300 Subject: [PATCH] SIENTIAPDE-1974 SIENTIAPDE-1974 Extract PI tag resolution and name Mongo stage constants - Move PI Web API read/output tag resolution out of the schedule loop into _resolve_pi_pipeline, so the loop only handles the held case. - Replace repeated '$match'/'$project' literals in the default connector queries with MATCH/PROJECT constants. --- orchestrator/activities/formatters.py | 67 +++++++++++++++++--------- orchestrator/workflows/orchestrator.py | 20 +++++--- 2 files changed, 57 insertions(+), 30 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index d59a0a1..2d3868f 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -99,6 +99,46 @@ def _merge_issues( issues.setdefault(kind, set()).update(f'{schedule_name}/{value}' for value in values) +def _resolve_pi_pipeline( + pipeline: dict[str, Any], + pi_tags: dict[str, dict[str, Any]], + pi_connectors: dict[str, dict[str, Any]], + issues: dict[str, set[str]], +) -> dict[str, Any] | None: + """ + The pipeline with its PI Web API tags resolved through the tag database. + + pi_web_api_scouter read tags get their WebIDs (and the pipeline its connector); + predictions_batch output tags get {tag_name: WebID}. Other pipelines come back as + they are. What cannot be resolved is added to issues. None means the pipeline must + be held: its read tags span several connectors. + """ + workflow_type = pipeline['workflow_type'] + schedule_name = pipeline['schedule_name'] + + if workflow_type == 'pi_web_api_scouter' and isinstance(pipeline.get('read_tags'), dict): + read_tags, connector, read_issues = resolve_pi_read_tags( + pipeline['read_tags'], pi_tags, pi_connectors + ) + _merge_issues(issues, read_issues, schedule_name) + if read_issues['mixed']: + return None + pipeline = {**pipeline, 'read_tags': read_tags} + if connector is not None: + pipeline['pi_web_api_connector'] = connector + + if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance( + pipeline.get('pi_web_api_output_config'), dict + ): + output_config, output_issues = resolve_pi_output_tags( + pipeline['pi_web_api_output_config'], pi_tags, pi_connectors + ) + _merge_issues(issues, output_issues, schedule_name) + pipeline = {**pipeline, 'pi_web_api_output_config': output_config} + + return pipeline + + class ScheduleType(TypedDict): """ Type definition for schedule type configuration entries. @@ -292,28 +332,11 @@ class Formatters(SientiaMonitoring): 'read_tags': resolve_tag_names(pipeline['read_tags'], tags_by_id), } - if workflow_type == 'pi_web_api_scouter' and isinstance( - pipeline.get('read_tags'), dict - ): - read_tags, connector, issues = resolve_pi_read_tags( - pipeline['read_tags'], pi_tags, pi_connectors - ) - _merge_issues(pi_issues, issues, schedule_name) - if issues['mixed']: - schedule_config[namespace][schedule_name] = {'held': True} - continue - pipeline = {**pipeline, 'read_tags': read_tags} - if connector is not None: - pipeline['pi_web_api_connector'] = connector - - if workflow_type in PI_OUTPUT_WORKFLOW_TYPES and isinstance( - pipeline.get('pi_web_api_output_config'), dict - ): - output_config, issues = resolve_pi_output_tags( - pipeline['pi_web_api_output_config'], pi_tags, pi_connectors - ) - _merge_issues(pi_issues, issues, schedule_name) - pipeline = {**pipeline, 'pi_web_api_output_config': output_config} + resolved = _resolve_pi_pipeline(pipeline, pi_tags, pi_connectors, pi_issues) + if resolved is None: + schedule_config[namespace][schedule_name] = {'held': True} + continue + pipeline = resolved schedule_config[namespace][schedule_name] = { **function(pipeline), diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index e5288bf..de412ba 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -14,6 +14,10 @@ with workflow.unsafe.imports_passed_through(): referenced_tag_ids, ) +# MongoDB aggregation stages +MATCH = '$match' +PROJECT = '$project' + # Tag database (SIENTIAPDE-1569) read by the slot v2 (SIENTIAPDE-1974). The workflow # input may override these aggregations (connectors_query / connector_tags_query); the # workflow puts a tag_id $match in front of connector_tags_query. Only the fields the @@ -22,9 +26,9 @@ with workflow.unsafe.imports_passed_through(): DEFAULT_CONNECTORS_QUERY = { 'collection': 'connectors', 'aggregation': [ - {'$match': {'type': 'opc_ua', 'status': 'Active'}}, + {MATCH: {'type': 'opc_ua', 'status': 'Active'}}, { - '$project': { + PROJECT: { '_id': 0, 'connector_id': 1, 'name': 1, @@ -39,8 +43,8 @@ DEFAULT_CONNECTORS_QUERY = { DEFAULT_CONNECTOR_TAGS_QUERY = { 'collection': 'connector_tags', 'aggregation': [ - {'$match': {'category': 'Read', 'status': 'Active'}}, - {'$project': {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}}, + {MATCH: {'category': 'Read', 'status': 'Active'}}, + {PROJECT: {'_id': 0, 'tag_id': 1, 'connector_id': 1, 'address': 1, 'tag_name': 1}}, ], } # Runs started before the tag database loads existed replay without them @@ -53,16 +57,16 @@ TAG_DATABASE_PATCH = 'SIENTIAPDE-1974-tag-database' DEFAULT_PI_CONNECTORS_QUERY = { 'collection': 'connectors', 'aggregation': [ - {'$match': {'type': 'pi_web_api', 'status': 'Active'}}, - {'$project': {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}}, + {MATCH: {'type': 'pi_web_api', 'status': 'Active'}}, + {PROJECT: {'_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, 'pi_web_api': 1}}, ], } DEFAULT_PI_CONNECTOR_TAGS_QUERY = { 'collection': 'connector_tags', 'aggregation': [ - {'$match': {'status': 'Active'}}, + {MATCH: {'status': 'Active'}}, { - '$project': { + PROJECT: { '_id': 0, 'tag_id': 1, 'connector_id': 1,