From 36b0c75893fe9833f77cb4ccb60951c5605ed692 Mon Sep 17 00:00:00 2001 From: VeikoAunapuu Date: Wed, 21 Jan 2026 10:28:59 +0200 Subject: [PATCH 01/13] feature: to be able to in need also quality from EU CGM --- config/model_quality/model_quality.properties | 4 +++- emf/model_quality/quality_functions.py | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/config/model_quality/model_quality.properties b/config/model_quality/model_quality.properties index 0bd1eab5..c95f4cfa 100644 --- a/config/model_quality/model_quality.properties +++ b/config/model_quality/model_quality.properties @@ -5,4 +5,6 @@ ELK_STATISTICS_INDEX = emfos-model-statistics BORDER_LIMIT = 250 LINE_LIMIT_TEMPERATURE = 25 C IGM_RULE_SET = impedance,line_rating -CGM_RULE_SET = kruonis,rtec,outage,lt_pl_xborder \ No newline at end of file +CGM_RULE_SET = kruonis,rtec,outage,lt_pl_xborder +#either a BA or EU +CGM_TYPE = BA \ No newline at end of file diff --git a/emf/model_quality/quality_functions.py b/emf/model_quality/quality_functions.py index 24bfb617..e1c154aa 100644 --- a/emf/model_quality/quality_functions.py +++ b/emf/model_quality/quality_functions.py @@ -13,7 +13,7 @@ def generate_quality_report(handler, network, object_type, model_metadata, rule_ report = {} - if object_type == "CGM" and model_metadata['pmd:Area'] == 'BA': + if object_type == "CGM" and model_metadata['pmd:Area'] in CGM_TYPE: report = check_generator_quality(report, network) report = check_lt_pl_crossborder(report, network, tieflow_data=tieflow_data, border_limit=BORDER_LIMIT) From 7af5540f11b4ff1f803d1a5ba7e63ea1de932444 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Sat, 14 Mar 2026 22:08:25 +0200 Subject: [PATCH 02/13] feat: #330 #394 #446 force replacement added, separate configuration per time horizon and replacement logic consolidated into single codeblock --- config/cgm_worker/replacement_conf.json | 853 +++++++++--------- .../manual_task_generator.properties | 1 + config/task_generator/process_conf.json | 14 + emf/model_merger/model_merger.py | 80 +- emf/model_merger/replacement.py | 535 ++++++++--- emf/task_generator/manual_worker.py | 1 + 6 files changed, 898 insertions(+), 586 deletions(-) diff --git a/config/cgm_worker/replacement_conf.json b/config/cgm_worker/replacement_conf.json index ecdddc02..fdd0cb75 100644 --- a/config/cgm_worker/replacement_conf.json +++ b/config/cgm_worker/replacement_conf.json @@ -1,413 +1,444 @@ { - "configuration": { - "validity_start": "2024-02-22T00:00:00Z", - "validity_end": "2024-02-22T23:59:59Z", - "time_zone": "CET" - }, - "time_horizon": { - "request_list": ["ID", "1D", "2D", "WK", "1D", "MO", "YR"] - }, - "replacement_length": { - "request_list": ["2w", "2w", "2w", "2w", "6d", "1M", "4M"] - }, - "month_ahead": { - "hours": ["09:30", "08:30", "10:30", "11:30", "07:30", "12:30"], - "business_type": ["1D","2D"] - }, - "hours": [ - { - "hour": "00:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "PT3H"}, - {"5": "PT4H"}, - {"6": "PT5H"} - ] - }, - { - "hour": "01:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "PT3H"}, - {"5": "-PT1H"}, - {"6": "PT4H"} - ] - }, - { - "hour": "02:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "-PT1H"}, - {"4": "PT2H"}, - {"5": "-PT2H"}, - {"6": "PT3H"} - ] - }, - { - "hour": "03:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "-PT2H"}, - {"5": "PT2H"}, - {"6": "-PT3H"} - ] - }, - { - "hour": "04:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "-PT2H"}, - {"4": "-PT3H"}, - {"5": "PT1H"}, - {"6": "-PT4H"} - ] - }, - { - "hour": "05:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "-PT2H"}, - {"5": "-PT3H"} - ] - }, - { - "hour": "06:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "PT3H"}, - {"5": "PT4H"}, - {"6": "PT5H"}, - {"7": "PT6H"} - ] - }, - { - "hour": "07:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "PT3H"}, - {"5": "PT4H"}, - {"6": "PT5H"} - ] - }, - { - "hour": "08:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "PT3H"}, - {"5": "PT4H"} - ] - }, - { - "hour": "09:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "-PT1H"}, - {"4": "PT2H"}, - {"5": "PT3H"} - ] - }, - { - "hour": "10:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "-PT2H"}, - {"4": "PT1H"}, - {"5": "PT2H"} - ] - }, - { - "hour": "11:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "PT2H"}, - {"5": "-PT2H"} - ] - }, - { - "hour": "12:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "-PT2H"}, - {"4": "PT1H"}, - {"5": "-PT3H"} - ] - }, - { - "hour": "13:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "-PT2H"}, - {"5": "-PT3H"} - ] - }, - { - "hour": "14:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "-PT1H"}, - {"5": "-PT2H"}, - {"6": "-PT3H"}, - {"7": "-PT4H"} - ] - }, - { - "hour": "15:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "PT2H"}, - {"4": "-PT1H"}, - {"5": "-PT2H"}, - {"6": "-PT3H"} - ] - }, - { - "hour": "16:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "-PT1H"}, - {"4": "-PT2H"}, - {"5": "-PT3H"}, - {"6": "-PT4H"} - ] - }, - { - "hour": "17:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "-PT1H"}, - {"4": "-PT2H"}, - {"5": "-PT3H"}, - {"6": "-PT4H"}, - {"7": "-PT5H"} - ] - }, - { - "hour": "18:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "-PT2H"}, - {"4": "-PT3H"}, - {"5": "-PT4H"}, - {"6": "-PT5H"}, - {"7": "-PT6H"}, - {"8": "PT1H"} - - ] - }, - { - "hour": "19:30", - "priority": [ - {"1": "PT0H"}, - {"2": "PT1H"}, - {"3": "-PT1H"}, - {"4": "PT2H"}, - {"5": "PT3H"} - ] - }, - { - "hour": "20:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "PT2H"}, - {"5": "PT3H"} - ] - }, - { - "hour": "21:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "-PT2H"}, - {"5": "PT2H"} - ] - }, - { - "hour": "22:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "PT1H"}, - {"4": "-PT2H"}, - {"5": "-PT3H"} - ] - }, - { - "hour": "23:30", - "priority": [ - {"1": "PT0H"}, - {"2": "-PT1H"}, - {"3": "-PT2H"}, - {"4": "-PT3H"}, - {"5": "-PT4H"} - ] - } - ], - "days": [ - { - "day": 0, - "priority": [ - {"1": "P0D"}, - {"2": "P1D"}, - {"3": "-P7D"}, - {"4": "-P6D"}, - {"5": "-P5D"}, - {"6": "-P4D"}, - {"7": "-P3D"}, - {"8": "-P14D"}, - {"9": "-P13D"}, - {"10": "-P12D"}, - {"11": "-P11D"}, - {"12": "-P10D"}, - {"13": "-P2D"}, - {"14": "-P1D"} - ] - }, - { - "day": 1, - "priority": [ - {"1": "P0D"}, - {"2": "P1D"}, - {"3": "-P1D"}, - {"4": "-P7D"}, - {"5": "-P8D"}, - {"6": "-P6D"}, - {"7": "-P5D"}, - {"8": "-P4D"}, - {"9": "-P14D"}, - {"10": "-P13D"}, - {"11": "-P12D"}, - {"12": "-P11D"}, - {"13": "-P3D"}, - {"14": "-P2D"} - ] - }, - { - "day": 2, - "priority": [ - {"1": "P0D"}, - {"2": "P1D"}, - {"3": "-P1D"}, - {"4": "-P2D"}, - {"5": "-P7D"}, - {"6": "-P6D"}, - {"7": "-P5D"}, - {"8": "-P8D"}, - {"9": "-P9D"}, - {"10": "-P14D"}, - {"11": "-P13D"}, - {"12": "-P12D"}, - {"13": "-P4D"}, - {"14": "-P3D"} - ] - }, - { - "day": 3, - "priority": [ - {"1": "P0D"}, - {"2": "-P1D"}, - {"3": "P1D"}, - {"4": "-P2D"}, - {"5": "-P3D"}, - {"6": "-P7D"}, - {"7": "-P6D"}, - {"8": "-P8D"}, - {"9": "-P9D"}, - {"10": "-P10D"}, - {"11": "-P14D"}, - {"12": "-P5D"}, - {"13": "-P6D"}, - {"14": "-P13D"} - ] - }, - { - "day": 4, - "priority": [ - {"1": "P0D"}, - {"2": "-P1D"}, - {"3": "-P2D"}, - {"4": "-P3D"}, - {"5": "-P4D"}, - {"5": "-P7D"}, - {"7": "-P8D"}, - {"8": "-P9D"}, - {"9": "-P10D"}, - {"10": "-P11D"}, - {"11": "-P14D"}, - {"12": "-P6D"}, - {"13": "-P13D"}, - {"14": "-P5D"} - ] - }, - { - "day": 5, - "priority": [ - {"1": "P0D"}, - {"2": "-P7D"}, - {"3": "-P14D"}, - {"4": "-P6D"}, - {"5": "-P13D"}, - {"6": "-P1D"}, - {"7": "-P2D"}, - {"8": "-P3D"}, - {"9": "-P4D"}, - {"10": "-P5D"}, - {"11": "-P8D"}, - {"12": "-P9D"}, - {"13": "-P10D"}, - {"14": "-P11D"} - ] - }, - { - "day": 6, - "priority": [ - {"1": "P0D"}, - {"2": "-P7D"}, - {"3": "-P14D"}, - {"4": "-P1D"}, - {"5": "-P8D"}, - {"6": "-P2D"}, - {"7": "-P3D"}, - {"8": "-P4D"}, - {"9": "-P5D"}, - {"10": "-P6D"}, - {"11": "-P9D"}, - {"12": "-P10D"}, - {"13": "-P11D"}, - {"14": "-P12D"}, - {"15": "-P13D"} - ] + "configuration": { + "validity_start": "2024-02-22T00:00:00Z", + "validity_end": "2024-02-22T23:59:59Z", + "time_zone": "CET" + }, + "time_horizons": { + "ID": { + "replacement_length": "2w", + "priority_order": 1, + "request_list": ["ID", "1D", "2D", "WK", "1D", "MO", "YR"], + "description": "Intraday time horizon" + }, + "1D": { + "replacement_length": "2w", + "priority_order": 2, + "request_list": ["1D", "2D", "WK", "MO", "YR"], + "description": "Day-ahead time horizon" + }, + "2D": { + "replacement_length": "2w", + "priority_order": 3, + "request_list": ["2D", "WK", "MO", "YR"], + "description": "Two-day ahead time horizon" + }, + "WK": { + "replacement_length": "2w", + "priority_order": 4, + "request_list": ["WK", "MO", "YR"], + "description": "Week-ahead time horizon" + }, + "MO": { + "replacement_length": "1M", + "priority_order": 5, + "request_list": ["MO", "YR"], + "description": "Month-ahead time horizon", + "special_handling": { + "hours": ["09:30", "08:30", "10:30", "11:30", "07:30", "12:30"], + "business_type": ["1D", "2D"] } - ] - } + }, + "YR": { + "replacement_length": "4M", + "priority_order": 6, + "request_list": ["YR"], + "description": "Year-ahead time horizon" + } + }, + "hours": [ + { + "hour": "00:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "PT3H"}, + {"5": "PT4H"}, + {"6": "PT5H"} + ] + }, + { + "hour": "01:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "PT3H"}, + {"5": "-PT1H"}, + {"6": "PT4H"} + ] + }, + { + "hour": "02:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "-PT1H"}, + {"4": "PT2H"}, + {"5": "-PT2H"}, + {"6": "PT3H"} + ] + }, + { + "hour": "03:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "-PT2H"}, + {"5": "PT2H"}, + {"6": "-PT3H"} + ] + }, + { + "hour": "04:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "-PT2H"}, + {"4": "-PT3H"}, + {"5": "PT1H"}, + {"6": "-PT4H"} + ] + }, + { + "hour": "05:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "-PT2H"}, + {"5": "-PT3H"} + ] + }, + { + "hour": "06:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "PT3H"}, + {"5": "PT4H"}, + {"6": "PT5H"}, + {"7": "PT6H"} + ] + }, + { + "hour": "07:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "PT3H"}, + {"5": "PT4H"}, + {"6": "PT5H"} + ] + }, + { + "hour": "08:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "PT3H"}, + {"5": "PT4H"} + ] + }, + { + "hour": "09:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "-PT1H"}, + {"4": "PT2H"}, + {"5": "PT3H"} + ] + }, + { + "hour": "10:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "-PT2H"}, + {"4": "PT1H"}, + {"5": "PT2H"} + ] + }, + { + "hour": "11:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "PT2H"}, + {"5": "-PT2H"} + ] + }, + { + "hour": "12:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "-PT2H"}, + {"4": "PT1H"}, + {"5": "-PT3H"} + ] + }, + { + "hour": "13:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "-PT2H"}, + {"5": "-PT3H"} + ] + }, + { + "hour": "14:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "-PT1H"}, + {"5": "-PT2H"}, + {"6": "-PT3H"}, + {"7": "-PT4H"} + ] + }, + { + "hour": "15:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "PT2H"}, + {"4": "-PT1H"}, + {"5": "-PT2H"}, + {"6": "-PT3H"} + ] + }, + { + "hour": "16:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "-PT1H"}, + {"4": "-PT2H"}, + {"5": "-PT3H"}, + {"6": "-PT4H"} + ] + }, + { + "hour": "17:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "-PT1H"}, + {"4": "-PT2H"}, + {"5": "-PT3H"}, + {"6": "-PT4H"}, + {"7": "-PT5H"} + ] + }, + { + "hour": "18:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "-PT2H"}, + {"4": "-PT3H"}, + {"5": "-PT4H"}, + {"6": "-PT5H"}, + {"7": "-PT6H"}, + {"8": "PT1H"} + ] + }, + { + "hour": "19:30", + "priority": [ + {"1": "PT0H"}, + {"2": "PT1H"}, + {"3": "-PT1H"}, + {"4": "PT2H"}, + {"5": "PT3H"} + ] + }, + { + "hour": "20:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "PT2H"}, + {"5": "PT3H"} + ] + }, + { + "hour": "21:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "-PT2H"}, + {"5": "PT2H"} + ] + }, + { + "hour": "22:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "PT1H"}, + {"4": "-PT2H"}, + {"5": "-PT3H"} + ] + }, + { + "hour": "23:30", + "priority": [ + {"1": "PT0H"}, + {"2": "-PT1H"}, + {"3": "-PT2H"}, + {"4": "-PT3H"}, + {"5": "-PT4H"} + ] + } + ], + "days": [ + { + "day": 0, + "priority": [ + {"1": "P0D"}, + {"2": "P1D"}, + {"3": "-P7D"}, + {"4": "-P6D"}, + {"5": "-P5D"}, + {"6": "-P4D"}, + {"7": "-P3D"}, + {"8": "-P14D"}, + {"9": "-P13D"}, + {"10": "-P12D"}, + {"11": "-P11D"}, + {"12": "-P10D"}, + {"13": "-P2D"}, + {"14": "-P1D"} + ] + }, + { + "day": 1, + "priority": [ + {"1": "P0D"}, + {"2": "P1D"}, + {"3": "-P1D"}, + {"4": "-P7D"}, + {"5": "-P8D"}, + {"6": "-P6D"}, + {"7": "-P5D"}, + {"8": "-P4D"}, + {"9": "-P14D"}, + {"10": "-P13D"}, + {"11": "-P12D"}, + {"12": "-P11D"}, + {"13": "-P3D"}, + {"14": "-P2D"} + ] + }, + { + "day": 2, + "priority": [ + {"1": "P0D"}, + {"2": "P1D"}, + {"3": "-P1D"}, + {"4": "-P2D"}, + {"5": "-P7D"}, + {"6": "-P6D"}, + {"7": "-P5D"}, + {"8": "-P8D"}, + {"9": "-P9D"}, + {"10": "-P14D"}, + {"11": "-P13D"}, + {"12": "-P12D"}, + {"13": "-P4D"}, + {"14": "-P3D"} + ] + }, + { + "day": 3, + "priority": [ + {"1": "P0D"}, + {"2": "-P1D"}, + {"3": "P1D"}, + {"4": "-P2D"}, + {"5": "-P3D"}, + {"6": "-P7D"}, + {"7": "-P6D"}, + {"8": "-P8D"}, + {"9": "-P9D"}, + {"10": "-P10D"}, + {"11": "-P14D"}, + {"12": "-P5D"}, + {"13": "-P6D"}, + {"14": "-P13D"} + ] + }, + { + "day": 4, + "priority": [ + {"1": "P0D"}, + {"2": "-P1D"}, + {"3": "-P2D"}, + {"4": "-P3D"}, + {"5": "-P4D"}, + {"5": "-P7D"}, + {"7": "-P8D"}, + {"8": "-P9D"}, + {"9": "-P10D"}, + {"10": "-P11D"}, + {"11": "-P14D"}, + {"12": "-P6D"}, + {"13": "-P13D"}, + {"14": "-P5D"} + ] + }, + { + "day": 5, + "priority": [ + {"1": "P0D"}, + {"2": "-P7D"}, + {"3": "-P14D"}, + {"4": "-P6D"}, + {"5": "-P13D"}, + {"6": "-P1D"}, + {"7": "-P2D"}, + {"8": "-P3D"}, + {"9": "-P4D"}, + {"10": "-P5D"}, + {"11": "-P8D"}, + {"12": "-P9D"}, + {"13": "-P10D"}, + {"14": "-P11D"} + ] + }, + { + "day": 6, + "priority": [ + {"1": "P0D"}, + {"2": "-P7D"}, + {"3": "-P14D"}, + {"4": "-P1D"}, + {"5": "-P8D"}, + {"6": "-P2D"}, + {"7": "-P3D"}, + {"8": "-P4D"}, + {"9": "-P5D"}, + {"10": "-P6D"}, + {"11": "-P9D"}, + {"12": "-P10D"}, + {"13": "-P11D"}, + {"14": "-P12D"}, + {"15": "-P13D"} + ] + } + ] +} diff --git a/config/task_generator/manual_task_generator.properties b/config/task_generator/manual_task_generator.properties index 999b9e43..268a2995 100644 --- a/config/task_generator/manual_task_generator.properties +++ b/config/task_generator/manual_task_generator.properties @@ -9,6 +9,7 @@ RUN_TYPE = INCLUDED_TSO = EXCLUDED_TSO = LOCAL_IMPORT = +REPLACE_TSO = PROCESS_TIME_SHIFT = TASK_REFERENCE_TIME = TASK_VERSION = diff --git a/config/task_generator/process_conf.json b/config/task_generator/process_conf.json index 24efebb8..e802ea3a 100644 --- a/config/task_generator/process_conf.json +++ b/config/task_generator/process_conf.json @@ -28,6 +28,7 @@ "mas": "http://www.baltic-rsc.eu/OperationalPlanning", "included": [], "excluded": [], + "replace_tso": [], "time_horizon": "ID", "version": "001", "replacement": "False", @@ -63,6 +64,7 @@ "mas": "http://www.baltic-rsc.eu/OperationalPlanning", "included": [], "excluded": [], + "replace_tso": [], "time_horizon": "ID", "version": "001", "replacement": "False", @@ -98,6 +100,7 @@ "mas": "http://www.baltic-rsc.eu/OperationalPlanning", "included": [], "excluded": [], + "replace_tso": [], "time_horizon": "ID", "version": "001", "replacement": "False", @@ -133,6 +136,7 @@ "mas": "http://www.baltic-rsc.eu/OperationalPlanning", "included": [], "excluded": [], + "replace_tso": [], "time_horizon": "1D", "version": "001", "replacement": "False", @@ -168,6 +172,7 @@ "mas": "http://www.baltic-rsc.eu/OperationalPlanning", "included": [], "excluded": [], + "replace_tso": [], "time_horizon": "2D", "version": "001", "replacement": "False", @@ -220,6 +225,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "ID", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -262,6 +268,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "ID", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -304,6 +311,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "ID", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -345,6 +353,7 @@ ], "excluded": [], "local_import": [], + "replace_tso": [], "time_horizon": "ID", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -387,6 +396,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "1D", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -429,6 +439,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "2D", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -471,6 +482,7 @@ "local_import": [ "LITGRID" ], + "replace_tso": [], "time_horizon": "YR", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -517,6 +529,7 @@ "AST", "ELERING" ], + "replace_tso": [], "time_horizon": "WK", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", @@ -560,6 +573,7 @@ "AST", "ELERING" ], + "replace_tso": [], "time_horizon": "MO", "version": "001", "mas": "http://www.baltic-rsc.eu/OperationalPlanning/RMM", diff --git a/emf/model_merger/model_merger.py b/emf/model_merger/model_merger.py index 5372a0c0..a8df7955 100644 --- a/emf/model_merger/model_merger.py +++ b/emf/model_merger/model_merger.py @@ -179,6 +179,7 @@ def handle(self, task_object: dict, properties: dict, **kwargs): included_models = task_properties.get('included', []) excluded_models = task_properties.get('excluded', []) local_import_models = task_properties.get('local_import', []) + replace_tso = task_properties.get('replace_tso', []) time_horizon = task_properties["time_horizon"] scenario_datetime = task_properties["timestamp_utc"] schedule_start = task_properties.get("reference_schedule_start_utc") @@ -245,29 +246,9 @@ def handle(self, task_object: dict, properties: dict, **kwargs): if acnp_dict: additional_models = filter_models_by_acnp(additional_models, merged_model, acnp_dict, ACNP_THRESHOLD, CONFORM_LOAD_FACTOR) missing_local_import = [tso for tso in local_import_models if tso not in [model['pmd:TSO'] for model in additional_models]] - - # Perform local replacement if configured - if model_replacement and missing_local_import: - try: - logger.info(f"Running replacement for local storage missing models: {missing_local_import}") - replacement_models_local = run_replacement(tso_list=missing_local_import, - time_horizon=time_horizon, - scenario_date=scenario_datetime, - data_source='PDN', - acnp_dict=acnp_dict, - acnp_threshold=ACNP_THRESHOLD, - conform_load_factor=CONFORM_LOAD_FACTOR) - - logger.info( - f"Local storage replacement model(s) found: {[model['pmd:fileName'] for model in replacement_models_local]}") - replaced_entities_local = [ModelEntity(data_source='PDN', quality_indicator='Substituted', **model).__dict__ for model in - replacement_models_local] - merged_model.replaced_entity.extend(replaced_entities_local) - additional_models.extend(replacement_models_local) - except Exception as error: - logger.error(f"Failed to run replacement: {error} {error.with_traceback()}") else: additional_models = [] + missing_local_import = [] # Check missing models for replacement if included_models: @@ -295,29 +276,22 @@ def handle(self, task_object: dict, properties: dict, **kwargs): excluded_incorrect = [model for model in valid_model_tsos if model not in [model['pmd:TSO'] for model in models] if model not in missing_models] missing_models = missing_models + excluded_incorrect - # Run replacement on missing models - if model_replacement and missing_models: - try: - logger.info(f"Running replacement for missing models: {missing_models}") - replacement_models = run_replacement(missing_models, - time_horizon, - scenario_datetime, - acnp_dict=acnp_dict, - acnp_threshold=ACNP_THRESHOLD, - conform_load_factor=CONFORM_LOAD_FACTOR) - if replacement_models: - logger.info( - f"Replacement model(s) found: {[model['pmd:fileName'] for model in replacement_models]}") - replaced_entities = [ModelEntity(data_source='OPDM', quality_indicator='Substituted', **model).__dict__ for model in - replacement_models] - merged_model.replaced_entity.extend(replaced_entities) - models.extend(replacement_models) - merged_model.replaced = True - else: - merged_model.replaced = False - except Exception as error: - logger.error(f"Failed to run replacement: {error}") - merged_model.replaced = False + # Execute consolidated model replacement logic + models, additional_models = run_replacement( + models=models, + additional_models=additional_models, + model_replacement=model_replacement, + local_import_models=local_import_models, + missing_local_import=missing_local_import, + missing_models=missing_models, + replace_tso=replace_tso, + time_horizon=time_horizon, + scenario_datetime=scenario_datetime, + merged_model=merged_model, + acnp_dict=acnp_dict, + acnp_threshold=ACNP_THRESHOLD, + conform_load_factor=CONFORM_LOAD_FACTOR + ) # Store models together with boundary set and check whether there are enough models to merge input_models = models + additional_models + [latest_boundary] @@ -340,7 +314,7 @@ def handle(self, task_object: dict, properties: dict, **kwargs): if force_outage_fix: # force outage fix on all models if set tso_list = merged_model.included elif outage_update and merging_area == 'BA' and any(tso in ['LITGRID', 'AST', 'ELERING'] for tso in - replaced_tso_list): # by default do it on Baltic merge replaced models + replaced_tso_list): # by default do it on Baltic merge replaced models tso_list = replaced_tso_list if tso_list: # if not set force and not replaced BA then nothing to fix merged_model = merge_functions.update_model_outages(merged_model=merged_model, @@ -431,24 +405,25 @@ def handle(self, task_object: dict, properties: dict, **kwargs): logger.debug(f"Post processing took: {(post_p_end - post_p_start).total_seconds()} seconds") logger.debug(f"Merging took: {(merge_end - merge_start).total_seconds()} seconds") - # Upload to OPDM + # Upload to OPDM if model_upload_to_opdm: - if merged_model.loadflow[0]['status'] == 'CONVERGED' and merged_model.scaled: # Only upload if the model LF is solved and scaled = true + if merged_model.loadflow[0][ + 'status'] == 'CONVERGED' and merged_model.scaled: # Only upload if the model LF is solved and scaled = true try: self.opdm_service = opdm.OPDM() if SEND_TYPE == 'SOAP': for item in serialized_data: logger.info(f"Uploading to OPDM: {item.name}") - #time.sleep(1) tested before with 2 and 4. Currently removing as FS gonna be default. + # time.sleep(1) tested before with 2 and 4. Currently removing as FS gonna be default. async_call(function=self.opdm_service.publication_request, - callback=log_opdm_response, - file_path_or_file_object=item) + callback=log_opdm_response, + file_path_or_file_object=item) elif SEND_TYPE == 'FS': for item in serialized_data: logger.info(f"Uploading to OPDM: {item.name}") self.opdm_service.put_file(file_id=item.name, - file_content=item) + file_content=item) merged_model.uploaded_to_opde = True except Exception as error: @@ -584,6 +559,7 @@ def handle(self, task_object: dict, properties: dict, **kwargs): "included": [], "excluded": [], "local_import": [], + "replace_tso": [], "time_horizon": "ID", "version": "000", "mas": "http://www.baltic-rsc.eu/OperationalPlanning", @@ -600,4 +576,4 @@ def handle(self, task_object: dict, properties: dict, **kwargs): } worker = HandlerMergeModels() - finished_task = worker.handle(sample_task, {}) + finished_task = worker.handle(sample_task, {}) \ No newline at end of file diff --git a/emf/model_merger/replacement.py b/emf/model_merger/replacement.py index 20e5a3c7..0e9f1ee2 100644 --- a/emf/model_merger/replacement.py +++ b/emf/model_merger/replacement.py @@ -1,5 +1,5 @@ import pandas as pd -from datetime import datetime +from datetime import datetime, timedelta from isodate import parse_duration import logging import config @@ -17,127 +17,410 @@ replacement_config = json.load(config.paths.cgm_worker.replacement_conf) -def run_replacement(tso_list: list, - time_horizon: str, - scenario_date: str, - config: list = replacement_config, - data_source: str = 'OPDM', - acnp_dict: dict = None, - acnp_threshold: str = 200, - conform_load_factor: str = 0.2 - ): +def run_replacement(models, additional_models, model_replacement, local_import_models, + missing_local_import, missing_models, replace_tso, time_horizon, + scenario_datetime, merged_model, acnp_dict=None, acnp_threshold=None, + conform_load_factor=None): """ - Args: - tso_list: a list of tso's which models are missing models - time_horizon: time_horizon of the merging process - scenario_date: scenario_date of the merging process - config: model replacement logic configuration - data_source: model provision source type + Execute consolidated model replacement logic with priority order: FORCED → OPDM → PDN + Forced replacement intelligently detects which source (OPDM or PDN) each TSO should be replaced in, + regardless of whether models currently exist (it's FORCED replacement). - Returns: from configuration a list of replaced models + Args: + models: List of OPDM models + additional_models: List of PDN models + model_replacement: Boolean flag for normal replacement + local_import_models: List of TSOs to import from local (PDN) + missing_local_import: List of missing TSOs in local import + missing_models: List of missing TSOs in main models + replace_tso: List of TSOs to force replacement for (highest priority, ignores existing models) + time_horizon: Time horizon for replacement query + scenario_datetime: Scenario datetime for replacement query + merged_model: MergedModel instance to update + acnp_dict: AC Net Position dictionary (optional) + acnp_threshold: ACNP threshold (optional) + conform_load_factor: Conform load factor (optional) + + Returns: + Tuple of (models, additional_models) with replacements applied """ - replacement_models = [] - replacements = pd.DataFrame() - # TODO time horizon exclusion logic + exclude available models from query - # TODO put in query object type if CGM metadata objects will be stored - # Get replacement length by time horizon - query_filter = 'now-' + config["replacement_length"]["request_list"][config["time_horizon"]["request_list"].index(time_horizon)] - # Query for available replacement models - query = {"pmd:TSO.keyword": tso_list, "valid": True, "data-source": data_source} - model_df = pd.DataFrame(query_data(query, query_filter)) + from emf.model_merger.model_merger import ModelEntity + + # Define all replacement scenarios upfront (in priority order: FORCED → OPDM → PDN) + replacement_scenarios = [] + replaced_tsos_tracker = set() # Track TSOs already replaced to avoid duplicates + + # Scenario 1: Forced Replacement for specific TSOs - HIGHEST PRIORITY (overrides replacement flag) + # FORCED replacement happens regardless of whether models currently exist + if replace_tso: + # For FORCED replacement, we replace TSOs based on their assignment: + # - If TSO is in local_import_models, force replace from PDN + # - Otherwise, force replace from OPDM (default) + + forced_replace_in_opdm = [tso for tso in replace_tso if tso not in local_import_models] + forced_replace_in_pdn = [tso for tso in replace_tso if tso in local_import_models] + + # Create scenarios for forced replacements from OPDM + if forced_replace_in_opdm: + logger.info( + f"Forced replacement requested for OPDM TSO(s) (ignoring current models): {forced_replace_in_opdm}") + replacement_scenarios.append({ + 'name': 'FORCED-OPDM', + 'tso_list': forced_replace_in_opdm, + 'model_list': models, # Add to OPDM models only + 'data_source': 'OPDM', + }) + + # Create scenarios for forced replacements from PDN + if forced_replace_in_pdn: + logger.info( + f"Forced replacement requested for PDN TSO(s) (ignoring current models): {forced_replace_in_pdn}") + replacement_scenarios.append({ + 'name': 'FORCED-PDN', + 'tso_list': forced_replace_in_pdn, + 'model_list': additional_models, # Add to PDN models only + 'data_source': 'PDN', + }) + + # Scenario 2: OPDM (Main) Replacement - MEDIUM PRIORITY (OPDM models only) + if model_replacement and missing_models: + replacement_scenarios.append({ + 'name': 'OPDM', + 'tso_list': missing_models, + 'model_list': models, # Add ONLY to OPDM models + 'data_source': 'OPDM', + }) + + # Scenario 3: PDN (Local Import) Replacement - LOWEST PRIORITY (PDN models only) + if local_import_models and model_replacement and missing_local_import: + replacement_scenarios.append({ + 'name': 'PDN', + 'tso_list': missing_local_import, + 'model_list': additional_models, # Add ONLY to PDN models + 'data_source': 'PDN', + }) + + # Execute all replacement scenarios with unified logic (priority order maintained) + for scenario in replacement_scenarios: + # Filter out TSOs already replaced in previous scenarios (respects priority) + tsos_to_replace = [tso for tso in scenario['tso_list'] if tso not in replaced_tsos_tracker] + + if not tsos_to_replace: + logger.info( + f"{scenario['name']} replacement skipped - all TSOs already replaced in higher priority scenarios: {scenario['tso_list']}") + continue + + logger.info(f"Attempting {scenario['name']} replacement for: {tsos_to_replace}") + + try: + # Get existing models for this scenario to avoid duplicates + # For forced replacement, we only exclude models from the same TSO to avoid exact duplicates + if scenario['name'] in ['FORCED-OPDM', 'FORCED-PDN']: + existing_models_for_scenario = [ + model for model in scenario['model_list'] + if model.get('pmd:TSO') in tsos_to_replace + ] + else: + existing_models_for_scenario = scenario['model_list'] + + replacement_models = find_replacement_models( + tso_list=tsos_to_replace, + time_horizon=time_horizon, + scenario_date=scenario_datetime, + data_source=scenario['data_source'], + acnp_dict=acnp_dict, + acnp_threshold=acnp_threshold, + conform_load_factor=conform_load_factor, + existing_models=existing_models_for_scenario + ) or [] - # Set scenario dat to UTC - if not model_df.empty: - scenario_date = parser.parse(scenario_date).strftime("%Y-%m-%dT%H:%M:%SZ") - replacement_df = create_replacement_table(scenario_date, time_horizon, model_df, config) - # Exclude models from replacement that fall outside of set schedule deadbands + if replacement_models: + replaced_tsos_list = [m['pmd:TSO'] for m in replacement_models] + logger.info( + f"{scenario['name']} replacement succeeded for TSO(s): {replaced_tsos_list} " + f"({[m['pmd:fileName'] for m in replacement_models]})" + ) + + # Track replaced TSOs + replaced_tsos_tracker.update(replaced_tsos_list) + + # Create entity records + replaced_entities = [ + ModelEntity( + data_source=scenario['data_source'], + quality_indicator='Substituted', + **model + ).__dict__ + for model in replacement_models + ] + merged_model.replaced_entity.extend(replaced_entities) + + # Add to appropriate model list based on scenario type + # For FORCED replacements, we might need to remove old models first + if scenario['name'] in ['FORCED-OPDM', 'FORCED-PDN']: + # Remove old models for these TSOs from the appropriate list + model_list = scenario['model_list'] + old_tsos_to_remove = replaced_tsos_list + model_list[:] = [m for m in model_list if m.get('pmd:TSO') not in old_tsos_to_remove] + logger.debug(f"{scenario['name']}: Removed old models for TSO(s): {old_tsos_to_remove}") + + scenario['model_list'].extend(replacement_models) + + # Set replaced flag for main scenarios (OPDM-based) + if scenario['name'] in ['OPDM', 'FORCED-OPDM']: + merged_model.replaced = True + else: + logger.warning(f"No {scenario['name']} replacements available for: {tsos_to_replace}") + # Only set replaced=False if this is OPDM/FORCED-OPDM and no replacements found + # and no other main scenario has succeeded yet + if scenario['name'] in ['OPDM', 'FORCED-OPDM'] and scenario['name'] not in [s['name'] for s in + replacement_scenarios[ + :replacement_scenarios.index( + scenario)] if + s['name'] in ['OPDM', + 'FORCED-OPDM']]: + merged_model.replaced = False + + except Exception as error: + logger.error(f"{scenario['name']} replacement failed for TSO(s) {tsos_to_replace}: {error}") + if scenario['name'] in ['OPDM', 'FORCED-OPDM'] and scenario['name'] not in [s['name'] for s in + replacement_scenarios[ + :replacement_scenarios.index( + scenario)] if + s['name'] in ['OPDM', + 'FORCED-OPDM']]: + merged_model.replaced = False + + return models, additional_models + + +def _build_replacement_query(tso_list: list, data_source: str, config: dict, time_horizon: str) -> tuple: + """Build query filter and query dict for replacement models.""" + # Get replacement length directly from time horizon configuration + time_horizon_config = config["time_horizons"][time_horizon] + query_filter = 'now-' + time_horizon_config["replacement_length"] + query = {"pmd:TSO.keyword": tso_list, "valid": True, "data-source": data_source} + return query, query_filter + + +def _select_best_replacement_models(replacement_df: pd.DataFrame, target_time_horizon, target_date, config) -> pd.DataFrame: + """Select the best replacement model for each TSO based on 4-step priority rules.""" + if replacement_df.empty: + return replacement_df + + def apply_priority_selection(candidates): + """Apply priority-based selection to candidate models.""" + return ( + candidates[candidates["priority_business"] == candidates["priority_business"].min()] + .pipe(lambda df: df[df["priority_hour"] == df["priority_hour"].min()]) + .pipe(lambda df: df[df["priority_day"] == df["priority_day"].min()]) + .pipe(lambda df: df[df["pmd:versionNumber"] == df["pmd:versionNumber"].max()]) + .pipe(lambda df: df[df["pmd:creationDate"] == df["pmd:creationDate"].max()]) + ) + + # Convert target_date to datetime if it's a string + if isinstance(target_date, str): + target_date = parser.parse(target_date) + target_date_only = target_date.date() + + selected_models = [] + + for tso in replacement_df["pmd:TSO"].unique(): + tso_models = replacement_df[replacement_df["pmd:TSO"] == tso].copy() + best_model = None + + # STEP 1: Use an IGM of the same time horizon of the same energy delivery day + step1_candidates = tso_models[ + (tso_models["pmd:timeHorizon"] == target_time_horizon) & + (tso_models["pmd:scenarioDate"].apply(lambda x: parser.parse(x).date() == target_date_only))] + if not step1_candidates.empty: + best_model = apply_priority_selection(step1_candidates) + logger.debug(f"STEP 1: Found same time horizon and same day replacement for {tso}") + + # STEP 2: If not available, use an IGM from the same energy delivery day (other time horizon) + if best_model is None or best_model.empty: + step2_candidates = tso_models[ + (tso_models["pmd:scenarioDate"].apply(lambda x: parser.parse(x).date() == target_date_only)) & + (tso_models["pmd:timeHorizon"] != target_time_horizon) + ] + if not step2_candidates.empty: + # Sort by time horizon priority based on config + if target_time_horizon in config["time_horizons"]: + priority_list = config["time_horizons"][target_time_horizon]["request_list"] + + step2_candidates["horizon_priority"] = step2_candidates["pmd:timeHorizon"].apply( + lambda horizon: priority_list.index(horizon) if horizon in priority_list else len(priority_list) + ) + step2_candidates = step2_candidates.sort_values("horizon_priority") + step2_candidates = step2_candidates.drop("horizon_priority", axis=1) + + best_model = apply_priority_selection(step2_candidates) + logger.debug(f"STEP 2: Found same day (different time horizon) replacement for {tso}") + + # STEP 3: If not available, use an IGM from the same time horizon of older models of the same day type + if best_model is None or best_model.empty: + target_weekday = target_date.weekday() + step3_candidates = tso_models[ + (tso_models["pmd:timeHorizon"] == target_time_horizon) & + (tso_models["pmd:scenarioDate"].apply(lambda x: + parser.parse(x).date().weekday() == target_weekday + )) + ] + + if not step3_candidates.empty: + # Sort by date (most recent first) and then by priority + step3_candidates = step3_candidates.sort_values( + by=["pmd:scenarioDate", "priority_hour", "priority_day", "priority_business"], + ascending=[False, True, True, True] + ) + best_model = step3_candidates.iloc[0:1] + logger.debug(f"STEP 3: Found same time horizon, same day type replacement for {tso}") + + # STEP 4: If not available, use older files of a different day type (original logic) + if best_model is None or best_model.empty: + best_model = apply_priority_selection(tso_models) + logger.debug(f"STEP 4: Used priority-based selection for {tso}") + + if len(best_model) > 1: + logger.warning(f"Replacement filtering unreliable for: '{tso}'") + best_model = best_model.iloc[:1] + + selected_models.append(best_model) + + return pd.concat(selected_models, ignore_index=True) if selected_models else pd.DataFrame() + + +def _exclude_existing_models(replacement_df: pd.DataFrame, existing_models: list) -> pd.DataFrame: + """Exclude replacement models that match existing models exactly (same TSO, timestamp, time horizon).""" + if replacement_df.empty or not existing_models: + return replacement_df + + # Create a set of existing model identifiers (TSO, scenarioDate, timeHorizon) + existing_identifiers = set() + for model in existing_models: + tso = model.get('pmd:TSO') + scenario_date = model.get('pmd:scenarioDate') + time_horizon = model.get('pmd:timeHorizon') + + if tso and scenario_date and time_horizon: + # Normalize scenario_date format for comparison + if isinstance(scenario_date, str): + try: + scenario_date = parser.parse(scenario_date).strftime("%Y-%m-%dT%H:%M:%SZ") + except: + pass + existing_identifiers.add((tso, scenario_date, time_horizon)) + + if not existing_identifiers: + return replacement_df + + # Filter out replacement models that match existing ones + mask = ~replacement_df.apply( + lambda row: ( + row['pmd:TSO'], + row['pmd:scenarioDate'], + row['pmd:timeHorizon'] + ) in existing_identifiers, + axis=1 + ) + + filtered_df = replacement_df[mask] + + # Log if any models were excluded + excluded_count = len(replacement_df) - len(filtered_df) + if excluded_count > 0: + logger.info(f"Excluded {excluded_count} replacement models that match existing models") + + return filtered_df + + +def _log_replacement_results(replacement_df: pd.DataFrame, original_tso_list: list) -> None: + """Log warnings for missing and unreplaced TSOs.""" + if replacement_df.empty: + return + + unique_tsos = replacement_df["pmd:TSO"].unique().tolist() + + tso_missing = [tso for tso in original_tso_list if tso not in unique_tsos] + if tso_missing: + logger.warning(f"No replacement models found for TSO(s): {tso_missing}") + + +def find_replacement_models(tso_list: list[str], + time_horizon: str, + scenario_date: str, + config: dict = replacement_config, + data_source: str = 'OPDM', + acnp_dict: dict | None = None, + acnp_threshold: str = 200, + conform_load_factor: str = 0.2, + existing_models: list | None = None + ) -> list[dict]: + """ + Find replacement models for missing TSOs based on priority rules. + + Args: + tso_list: List of TSOs which are missing models + time_horizon: Time horizon of the merging process + scenario_date: Scenario date of the merging process + config: Model replacement logic configuration + data_source: Model provision source type + acnp_dict: AC Net Position dictionary (optional) + acnp_threshold: ACNP threshold (optional) + conform_load_factor: Conform load factor (optional) + existing_models: List of existing models to exclude from replacement (optional) + + Returns: + List of replacement models with content loaded + """ + if not tso_list: + return [] + + try: + # Build and execute query + query, query_filter = _build_replacement_query(tso_list, data_source, config, time_horizon) + model_df = pd.DataFrame(query_data(query, query_filter)) + + if model_df.empty: + logger.warning(f"No replacement models found in Elastic for TSO(s): {tso_list}") + return [] + + # Process replacement candidates + scenario_date_utc = parser.parse(scenario_date).strftime("%Y-%m-%dT%H:%M:%SZ") + replacement_df = create_replacement_table(scenario_date_utc, time_horizon, model_df, config) + + # Apply ACNP filtering if provided if acnp_dict: replacement_df = filter_replacements_by_acnp(replacement_df, acnp_dict, acnp_threshold, conform_load_factor) - if not replacement_df.empty: - unique_tsos_list = replacement_df["pmd:TSO"].unique().tolist() - for unique_tso in unique_tsos_list: - sample_tso = replacement_df.loc[(replacement_df["pmd:TSO"] == unique_tso)] - sample_tso = sample_tso.loc[(sample_tso["priority_day"] == sample_tso["priority_day"].min())] - sample_tso = sample_tso.loc[(sample_tso["priority_business"] == sample_tso["priority_business"].min())] - sample_tso = sample_tso.loc[(sample_tso["priority_hour"] == sample_tso["priority_hour"].min())] - sample_tso = sample_tso.loc[(sample_tso["pmd:versionNumber"] == sample_tso["pmd:versionNumber"].max())] - sample_tso_min = sample_tso.loc[(sample_tso["pmd:creationDate"] == sample_tso["pmd:creationDate"].max())] - if len(sample_tso_min) > 1: - logger.warning(f"Replacement filtering unreliable for: '{unique_tso}'") - sample_tso_min = sample_tso_min.iloc[:1] - replacements = pd.concat([replacements, sample_tso_min]) - - replacement_models = replacements.to_dict(orient='records') if not replacements.empty else None - for num, model in enumerate(replacement_models): - replacement_models[num] = get_content(model) - - replaced_tso = replacements['pmd:TSO'].unique().tolist() - not_replaced = [model for model in unique_tsos_list if model not in replaced_tso] - if not_replaced: - logger.warning(f"Unable to find replacements within given replacement logic for TSO's: {not_replaced}") - - tso_missing = [model for model in tso_list if model not in unique_tsos_list] - if tso_missing: - logger.warning(f"No replacement models found for TSO(s): {tso_missing}") - else: - logger.error(f"No replacement models found, replacement list is empty, possibly due to incorrect schedules") - else: - logger.warning(f"No replacement models found in Elastic for TSO(s): {tso_list}") - - return replacement_models - - -# TODO deprecated, move to backlog -def run_replacement_local(tso_list: list, - time_horizon: str, - scenario_date: str, - config: list = json.load(config.paths.task_generator.timeframe_conf), - ): - """ - Args: - tso_list: a list of tso's which models are missing models - time_horizon: time_horizon of the merging process - scenario_date: scenario_date of the merging process - config: model replacement logic configuration - - Returns: from configuration a list of replaced models - """ - replacement_models = [] - replacements = pd.DataFrame() - # TODO time horizon exclusion logic + exclude available models from query - client = ObjectStorage() - list_elements = client.get_all_objects_name(bucket_name='opde-confidential-models', prefix='IGM') - model_df=pd.DataFrame([item.split('-') for item in list_elements], columns=["pmd:scenarioDate","pmd:timeHorizon", "pmd:TSO", "pmd:versionNumber" ]) - model_df["pmd:creationDate"] = datetime.now() - model_df["pmd:fileName"] = ['IGM/' + item for item in list_elements] - model_df["pmd:TSO"] = model_df["pmd:TSO"] - # print(model_df) - # Set scenario dat to UTC - if not model_df.empty: - model_df["pmd:versionNumber"] = model_df["pmd:versionNumber"].apply(lambda x: x.split('.')[0]) - scenario_date = parser.parse(scenario_date).strftime("%Y-%m-%dT%H:%M:%SZ") - replacement_df = create_replacement_table(scenario_date, time_horizon, model_df, config) - if not replacement_df.empty: - unique_tsos_list = tso_list - - for unique_tso in unique_tsos_list: - sample_tso = replacement_df.loc[(replacement_df["pmd:TSO"] == unique_tso)] - sample_tso = sample_tso.loc[(sample_tso["priority_day"] == sample_tso["priority_day"].min())] - sample_tso = sample_tso.loc[(sample_tso["priority_business"] == sample_tso["priority_business"].min())] - sample_tso = sample_tso.loc[(sample_tso["priority_hour"] == sample_tso["priority_hour"].min())] - sample_tso = sample_tso.loc[(sample_tso["pmd:versionNumber"] == sample_tso["pmd:versionNumber"].max())] - sample_tso_min = sample_tso.loc[(sample_tso["pmd:creationDate"] == sample_tso["pmd:creationDate"].max())] - if len(sample_tso_min) > 1: - logger.warning(f"Replacement filtering unreliable for: '{unique_tso}'") - sample_tso_min = sample_tso_min.iloc[:1] - replacements = pd.concat([replacements, sample_tso_min]) - - replacement_models = replacements.to_dict(orient='records') if not replacements.empty else None - if replacement_models: - for num, model in enumerate(replacement_models): - replacement_models[num] = model - - return replacement_models + + # Exclude models that already exist + if existing_models: + replacement_df = _exclude_existing_models(replacement_df, existing_models) + + if replacement_df.empty: + logger.error("No replacement models found, replacement list is empty, possibly due to incorrect schedules") + return [] + + # Select best models and load content + selected_models = _select_best_replacement_models(replacement_df, + target_time_horizon=time_horizon, + target_date=scenario_date_utc, + config=config) + _log_replacement_results(selected_models, tso_list) + + if selected_models.empty: + return [] + + # Load content for each model + replacement_models = selected_models.to_dict(orient='records') + for i, model in enumerate(replacement_models): + replacement_models[i] = get_content(model) + + return replacement_models + + except Exception as e: + logger.error(f"Error finding replacement models for TSOs {tso_list}: {e}") + return [] def make_lists_priority(timestamp, target_timehorizon, conf): @@ -164,14 +447,16 @@ def make_lists_priority(timestamp, target_timehorizon, conf): hour_list_final = list(map(lambda x: (date_time + parse_duration(x)).strftime("%H:%M"), hour_list)) day_list_final = list(map(lambda x: (date_time + parse_duration(x)).strftime("%Y-%m-%d"), day_list)) - business_list = conf["time_horizon"]["request_list"] - business_list_final = business_list[business_list.index(target_timehorizon):] # make list of relevant businesstypes + business_list = conf["time_horizons"][target_timehorizon]["request_list"] + business_list_final = business_list # Month ahead requires separate replacement logic if target_timehorizon == 'MO': - hour_list_final = [hour for hour in conf["month_ahead"]["hours"]] - day_list_final = [get_first_monday_of_last_month(timestamp).strftime("%Y-%m-%d")] - business_list_final = conf["month_ahead"]['business_type'] + time_horizon_config = conf["time_horizons"][target_timehorizon] + if "special_handling" in time_horizon_config: + hour_list_final = [hour for hour in time_horizon_config["special_handling"]["hours"]] + day_list_final = [get_first_monday_of_last_month(timestamp).strftime("%Y-%m-%d")] + business_list_final = time_horizon_config["special_handling"]["business_type"] return hour_list_final, day_list_final, business_list_final @@ -192,6 +477,9 @@ def create_replacement_table(target_timestamp, target_timehorizon, valid_models_ # Change ID naming for simpler replacement logic valid_models_df['pmd:timeHorizon'] = valid_models_df['pmd:timeHorizon'].apply(lambda x: 'ID' if x in [f'{i:02}' for i in range(1, 25)] else x) + + # Add target time horizon for reference in selection logic + valid_models_df['target_time_horizon'] = target_timehorizon valid_models_df["priority_business"] = valid_models_df["pmd:timeHorizon"].apply(lambda x: list_business_priority.index(x) if x in list_business_priority else None) valid_models_df["pmd:scenarioDate"] = valid_models_df["pmd:scenarioDate"].apply(lambda x: parser.parse(x).strftime("%Y-%m-%dT%H:%M:%SZ")) @@ -208,8 +496,9 @@ def create_replacement_table(target_timestamp, target_timehorizon, valid_models_ def get_tsos_available_in_storage(time_horizon: str): metadata = {"opde:Object-Type": "IGM", "valid": True} - # Get query length by time horizon from configuration - query_filter = 'now-' + replacement_config["replacement_length"]["request_list"][replacement_config["time_horizon"]["request_list"].index(time_horizon)] + # Get query length directly from time horizon configuration + time_horizon_config = replacement_config["time_horizons"][time_horizon] + query_filter = 'now-' + time_horizon_config["replacement_length"] unique_tsos = fetch_unique_values(metadata_query=metadata, field="pmd:TSO.keyword", query_filter=query_filter) return unique_tsos diff --git a/emf/task_generator/manual_worker.py b/emf/task_generator/manual_worker.py index 00b8d1df..3318b867 100644 --- a/emf/task_generator/manual_worker.py +++ b/emf/task_generator/manual_worker.py @@ -51,6 +51,7 @@ process_config_json[0]['runs'][0]['properties']['included'] = [tso.strip() for tso in INCLUDED_TSO.split(',')] if INCLUDED_TSO else [] process_config_json[0]['runs'][0]['properties']['excluded'] = [tso.strip() for tso in EXCLUDED_TSO.split(',')] if EXCLUDED_TSO else [] process_config_json[0]['runs'][0]['properties']['local_import'] = [tso.strip() for tso in LOCAL_IMPORT.split(',')] if LOCAL_IMPORT else [] +process_config_json[0]['runs'][0]['properties']['replace_tso'] = [tso.strip() for tso in REPLACE_TSO.split(',')] if REPLACE_TSO else [] process_config_json[0]['runs'][0]['properties']['fix_net_interchange2'] = FIX_NET_INTERCHANGE2 process_config_json[0]['runs'][0]['properties']['version'] = TASK_VERSION process_config_json[0]['runs'][0]['properties']['replacement'] = RUN_REPLACEMENT From 883b7d0705e2c05036b111b22a9ed2bfa3505899 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Wed, 18 Mar 2026 09:30:08 +0200 Subject: [PATCH 03/13] replacement conf updated --- config/cgm_worker/replacement_conf.json | 6 ------ emf/model_merger/replacement.py | 11 ----------- 2 files changed, 17 deletions(-) diff --git a/config/cgm_worker/replacement_conf.json b/config/cgm_worker/replacement_conf.json index fdd0cb75..c8186de2 100644 --- a/config/cgm_worker/replacement_conf.json +++ b/config/cgm_worker/replacement_conf.json @@ -7,31 +7,26 @@ "time_horizons": { "ID": { "replacement_length": "2w", - "priority_order": 1, "request_list": ["ID", "1D", "2D", "WK", "1D", "MO", "YR"], "description": "Intraday time horizon" }, "1D": { "replacement_length": "2w", - "priority_order": 2, "request_list": ["1D", "2D", "WK", "MO", "YR"], "description": "Day-ahead time horizon" }, "2D": { "replacement_length": "2w", - "priority_order": 3, "request_list": ["2D", "WK", "MO", "YR"], "description": "Two-day ahead time horizon" }, "WK": { "replacement_length": "2w", - "priority_order": 4, "request_list": ["WK", "MO", "YR"], "description": "Week-ahead time horizon" }, "MO": { "replacement_length": "1M", - "priority_order": 5, "request_list": ["MO", "YR"], "description": "Month-ahead time horizon", "special_handling": { @@ -41,7 +36,6 @@ }, "YR": { "replacement_length": "4M", - "priority_order": 6, "request_list": ["YR"], "description": "Year-ahead time horizon" } diff --git a/emf/model_merger/replacement.py b/emf/model_merger/replacement.py index 0e9f1ee2..44641703 100644 --- a/emf/model_merger/replacement.py +++ b/emf/model_merger/replacement.py @@ -524,14 +524,3 @@ def get_first_monday_of_last_month(timestamp): first_monday = first_day_of_month + timedelta(days=days_to_add) return first_monday - - -if __name__ == "__main__": - - missing_tso = ['PSE', 'LITGRID', 'AST'] - - test_time_horizon = "MO" - test_scenario_date = "2025-03-12T09:30:00Z" - # print('hello') - response_list = run_replacement(missing_tso, test_time_horizon, test_scenario_date) - print('') From c3930e673fd1f69d5e84f3070bb9b1f5d764fdda Mon Sep 17 00:00:00 2001 From: "martynas.karobcikas" Date: Wed, 1 Apr 2026 09:56:57 +0300 Subject: [PATCH 04/13] #453: lf settings retrieving from eleastic api key --- emf/common/loadflow_tool/settings_manager.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/emf/common/loadflow_tool/settings_manager.py b/emf/common/loadflow_tool/settings_manager.py index 7a18ef64..10d61fd3 100644 --- a/emf/common/loadflow_tool/settings_manager.py +++ b/emf/common/loadflow_tool/settings_manager.py @@ -45,16 +45,14 @@ class LoadflowSettingsManager: def __init__(self, elastic_server: str = ELK_SERVER, - elastic_username: str | None = None, - elastic_password: str | None = None, + elastic_api_key: str | None = None, elastic_index: str = 'config-lf-parameters', settings_keyword: str = 'EU_DEFAULT', override_path: str | None = None, ): self.elastic_server = elastic_server - self.elastic_username = elastic_username - self.elastic_password = elastic_password + self.elastic_api_key = elastic_api_key self.elastic_index = elastic_index self.settings_keyword = settings_keyword @@ -82,7 +80,7 @@ def __init__(self, # ----------------- I/O ----------------- def _get_defaults_from_elastic(self) -> dict: - client = Elasticsearch(self.elastic_server) + client = Elasticsearch(self.elastic_server, api_key=self.elastic_api_key) logger.info(f"Retrieving base loadflow settings fromm Elasticsearch with key: {self.settings_keyword}") response = client.get(index=self.elastic_index, id=self.settings_keyword) From 416965c0d6b8bd891a9301de5938b79cf31896dc Mon Sep 17 00:00:00 2001 From: "martynas.karobcikas" Date: Wed, 1 Apr 2026 09:58:44 +0300 Subject: [PATCH 05/13] #453: lf settings retrieving from eleastic api key --- emf/common/loadflow_tool/settings_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/emf/common/loadflow_tool/settings_manager.py b/emf/common/loadflow_tool/settings_manager.py index 10d61fd3..2055acb1 100644 --- a/emf/common/loadflow_tool/settings_manager.py +++ b/emf/common/loadflow_tool/settings_manager.py @@ -45,7 +45,7 @@ class LoadflowSettingsManager: def __init__(self, elastic_server: str = ELK_SERVER, - elastic_api_key: str | None = None, + elastic_api_key: str = ELK_TOKEN, elastic_index: str = 'config-lf-parameters', settings_keyword: str = 'EU_DEFAULT', override_path: str | None = None, From 7c1189ecfd7e1824ded0c47fcac54c9907f601ac Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Thu, 2 Apr 2026 15:02:04 +0300 Subject: [PATCH 06/13] #450 replacement conf updated --- config/cgm_worker/replacement_conf.json | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/config/cgm_worker/replacement_conf.json b/config/cgm_worker/replacement_conf.json index c8186de2..c808d702 100644 --- a/config/cgm_worker/replacement_conf.json +++ b/config/cgm_worker/replacement_conf.json @@ -7,22 +7,22 @@ "time_horizons": { "ID": { "replacement_length": "2w", - "request_list": ["ID", "1D", "2D", "WK", "1D", "MO", "YR"], + "request_list": ["ID", "1D", "2D"], "description": "Intraday time horizon" }, "1D": { "replacement_length": "2w", - "request_list": ["1D", "2D", "WK", "MO", "YR"], + "request_list": ["1D", "2D"], "description": "Day-ahead time horizon" }, "2D": { "replacement_length": "2w", - "request_list": ["2D", "WK", "MO", "YR"], + "request_list": ["2D", "1D", "ID"], "description": "Two-day ahead time horizon" }, "WK": { "replacement_length": "2w", - "request_list": ["WK", "MO", "YR"], + "request_list": ["WK", "1D"], "description": "Week-ahead time horizon" }, "MO": { From 25e647fc60e553c7bd0e13a7724ca5a1bbc2abe6 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Thu, 2 Apr 2026 15:23:10 +0300 Subject: [PATCH 07/13] #453 task generator AUTO bugfix --- emf/task_generator/manual_worker.py | 1 - emf/task_generator/task_generator.py | 5 ++++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/emf/task_generator/manual_worker.py b/emf/task_generator/manual_worker.py index 3318b867..19277bfa 100644 --- a/emf/task_generator/manual_worker.py +++ b/emf/task_generator/manual_worker.py @@ -52,7 +52,6 @@ process_config_json[0]['runs'][0]['properties']['excluded'] = [tso.strip() for tso in EXCLUDED_TSO.split(',')] if EXCLUDED_TSO else [] process_config_json[0]['runs'][0]['properties']['local_import'] = [tso.strip() for tso in LOCAL_IMPORT.split(',')] if LOCAL_IMPORT else [] process_config_json[0]['runs'][0]['properties']['replace_tso'] = [tso.strip() for tso in REPLACE_TSO.split(',')] if REPLACE_TSO else [] -process_config_json[0]['runs'][0]['properties']['fix_net_interchange2'] = FIX_NET_INTERCHANGE2 process_config_json[0]['runs'][0]['properties']['version'] = TASK_VERSION process_config_json[0]['runs'][0]['properties']['replacement'] = RUN_REPLACEMENT process_config_json[0]['runs'][0]['properties']['scaling'] = RUN_SCALING diff --git a/emf/task_generator/task_generator.py b/emf/task_generator/task_generator.py index b092bd06..b67db751 100644 --- a/emf/task_generator/task_generator.py +++ b/emf/task_generator/task_generator.py @@ -192,7 +192,7 @@ def set_task_version(task: dict): # Check versioning mode auto_versioning_enabled = False - if task['task_properties']['version'] == 'AUTO': + if task['task_properties']['version'].strip().lower() == 'auto': logger.debug("Task versioning set to AUTO mode") auto_versioning_enabled = True @@ -232,6 +232,9 @@ def set_task_version(task: dict): task['task_properties']['version'] = updated_version logger.info(f"Version set to: {updated_version}") + if task['task_properties']['version'].strip().lower() == 'auto': + raise ValueError() + except Exception as e: logger.warning("Elastic query for task versioning unsuccessful, version not updated") logger.warning(f"Exception traceback: {e}") From 284fe5d6b98edceb2fbf222f9049982e5db51e83 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Thu, 2 Apr 2026 15:56:07 +0300 Subject: [PATCH 08/13] #453 task generator AUTO bugfix --- emf/task_generator/task_generator.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/emf/task_generator/task_generator.py b/emf/task_generator/task_generator.py index b67db751..4fa44996 100644 --- a/emf/task_generator/task_generator.py +++ b/emf/task_generator/task_generator.py @@ -6,6 +6,7 @@ import json import logging import os +import pandas as pd from emf.common.helpers.time import parse_duration, convert_to_utc, convert_to_timezone, timezone, reference_times, utcnow from emf.common.helpers.tasks import update_task_status from emf.common.integrations.elastic import Elastic @@ -217,6 +218,10 @@ def set_task_version(task: dict): logger.info("Task versioning mode 'AUTO', defaulting to: '001'") updated_version = '001' else: + # filter out non integer version values + num = pd.to_numeric(tasks_df["col"], errors="coerce") + tasks_df = tasks_df[num.notna() & (num % 1 == 0)] + latest_version = tasks_df['task_properties.version'].max() logger.info(f"Latest available task version: {latest_version}") if auto_versioning_enabled: From 8e27b3f4f7406923a7693d648f148cc0fd29f253 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Thu, 2 Apr 2026 16:18:32 +0300 Subject: [PATCH 09/13] #453 task generator AUTO bugfix --- emf/task_generator/task_generator.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/emf/task_generator/task_generator.py b/emf/task_generator/task_generator.py index 4fa44996..87ee35d1 100644 --- a/emf/task_generator/task_generator.py +++ b/emf/task_generator/task_generator.py @@ -219,10 +219,10 @@ def set_task_version(task: dict): updated_version = '001' else: # filter out non integer version values - num = pd.to_numeric(tasks_df["col"], errors="coerce") + num = pd.to_numeric(tasks_df["task_properties.version"], errors="coerce") tasks_df = tasks_df[num.notna() & (num % 1 == 0)] - latest_version = tasks_df['task_properties.version'].max() + latest_version = pd.to_numeric(tasks_df['task_properties.version']).max() logger.info(f"Latest available task version: {latest_version}") if auto_versioning_enabled: updated_version = str(int(latest_version) + 1).zfill(3) @@ -231,7 +231,7 @@ def set_task_version(task: dict): updated_version = str(int(latest_version) + 1).zfill(3) else: logger.info("Using version for task configuration") - updated_version = task['task_properties']['version'] + updated_version = str(int(task['task_properties']['version']) + 1).zfill(3) if updated_version: task['task_properties']['version'] = updated_version From 6364d2ca924838f61d455ad5c417e28ce83f4f6f Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Fri, 3 Apr 2026 15:34:50 +0300 Subject: [PATCH 10/13] #452 Sanitized task versioning query and made auto versioning logic more robust with better exception handling --- emf/common/helpers/tasks.py | 4 ++ emf/task_generator/task_generator.py | 81 +++++++++++++++------------- 2 files changed, 48 insertions(+), 37 deletions(-) diff --git a/emf/common/helpers/tasks.py b/emf/common/helpers/tasks.py index 0044de32..58ac4bf9 100644 --- a/emf/common/helpers/tasks.py +++ b/emf/common/helpers/tasks.py @@ -26,6 +26,10 @@ def update_task_status(task: dict, status_text: str, publish: bool = True): "timestamp": utc_now }) + # Fix boolean values + for key, value in task['task_properties'].items(): + task['task_properties'][key] = value.strip().lower() == "true" if isinstance(value, str) and value.strip().lower() in {"true", "false"} else value + # TODO - better handling if elk is not available, possibly set elk connection timeout really small or refactor the sending to happen via rabbit if publish: try: diff --git a/emf/task_generator/task_generator.py b/emf/task_generator/task_generator.py index 87ee35d1..6cf82118 100644 --- a/emf/task_generator/task_generator.py +++ b/emf/task_generator/task_generator.py @@ -191,58 +191,65 @@ def generate_tasks(task_window_duration: str, def set_task_version(task: dict): - # Check versioning mode - auto_versioning_enabled = False - if task['task_properties']['version'].strip().lower() == 'auto': - logger.debug("Task versioning set to AUTO mode") - auto_versioning_enabled = True + task_version = task['task_properties']['version'] - query = { - "bool": { - "must": [ - {"match": {"task_properties.timestamp_utc": task['task_properties']['timestamp_utc']}}, - {"term": {"task_properties.time_horizon.keyword": task['task_properties']['time_horizon']}}, - {"term": {"task_properties.merge_type.keyword": task['task_properties']['merge_type']}}, - ] - } - } + # Set versioning mode + if task_version.strip().lower() == 'auto': + logger.debug("Task versioning set to automatic") + auto_versioning_enabled = True + else: + auto_versioning_enabled = False - service = Elastic() try: - updated_version = None + service = Elastic() + + query = { + "bool": { + "must": [ + {"match": {"task_properties.timestamp_utc": task['task_properties']['timestamp_utc']}}, + {"term": {"task_properties.time_horizon.keyword": task['task_properties']['time_horizon']}}, + {"term": {"task_properties.merge_type.keyword": task['task_properties']['merge_type']}}, + ] + } + } tasks_df = service.get_docs_by_query(index=TASK_ELK_INDEX, query=query) + # Filter out non integer values + num = pd.to_numeric(tasks_df["task_properties.version"], errors="coerce") + tasks_df = tasks_df[num.notna() & (num % 1 == 0)] + + # If no previous tasks found, set version to 001 if tasks_df.empty: - logger.info(f"No previous runs found for task, using version from configuration: {task['task_properties']['version']}") + logger.info(f"No previous runs found for this task") if auto_versioning_enabled: - logger.info("Task versioning mode 'AUTO', defaulting to: '001'") - updated_version = '001' - else: - # filter out non integer version values - num = pd.to_numeric(tasks_df["task_properties.version"], errors="coerce") - tasks_df = tasks_df[num.notna() & (num % 1 == 0)] + set_version = '001' + else: + set_version = str(int(task_version)).zfill(3) + else: + # Get latest task available version from ELK latest_version = pd.to_numeric(tasks_df['task_properties.version']).max() logger.info(f"Latest available task version: {latest_version}") + if auto_versioning_enabled: - updated_version = str(int(latest_version) + 1).zfill(3) - elif int(latest_version) >= int(task['task_properties']['version']): - logger.warning("Latest available version is equal or higher than defined in task, increasing from latest") - updated_version = str(int(latest_version) + 1).zfill(3) + set_version = str(int(latest_version) + 1).zfill(3) + elif int(latest_version) >= int(task_version): + logger.warning("Latest version is equal or lower than task config, incrementing from latest") + set_version = str(int(latest_version) + 1).zfill(3) else: - logger.info("Using version for task configuration") - updated_version = str(int(task['task_properties']['version']) + 1).zfill(3) - - if updated_version: - task['task_properties']['version'] = updated_version - logger.info(f"Version set to: {updated_version}") - - if task['task_properties']['version'].strip().lower() == 'auto': - raise ValueError() + logger.info("Using version from task config") + set_version = str(int(task_version)).zfill(3) + + task['task_properties']['version'] = set_version + logger.info(f"Version set to: '{set_version}'") except Exception as e: - logger.warning("Elastic query for task versioning unsuccessful, version not updated") logger.warning(f"Exception traceback: {e}") + if auto_versioning_enabled: + task['task_properties']['version'] = None + logger.error("Elastic query for task versioning unsuccessful, version not set") + else: + logger.warning("Elastic query for task versioning unsuccessful, using provided value") if __name__ == "__main__": From bf7f1da6f3a30d7018c3348b87dc4e269dcd68cb Mon Sep 17 00:00:00 2001 From: "martynas.karobcikas" Date: Mon, 13 Apr 2026 16:11:45 +0300 Subject: [PATCH 11/13] feature: Elastic api support index rollover flag bugfix: for logs disabled monthly indexes --- emf/common/integrations/elastic.py | 12 +++++++++--- emf/common/logging/custom_logger.py | 5 ++++- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/emf/common/integrations/elastic.py b/emf/common/integrations/elastic.py index 2da449e8..d13ddcfd 100644 --- a/emf/common/integrations/elastic.py +++ b/emf/common/integrations/elastic.py @@ -33,6 +33,7 @@ def send_to_elastic(index: str, server: str = ELK_SERVER, api_key: str = ELK_TOKEN, iso_timestamp: str = None, + index_rollover: bool = True, debug: bool = False): """ Method to send single message to ELK @@ -41,6 +42,7 @@ def send_to_elastic(index: str, :param id: :param server: url of ELK server :param iso_timestamp: message timestamp + :param index_rollover: if true, adds monthly indication to index name :param debug: flag for debug mode :return: """ @@ -53,8 +55,9 @@ def send_to_elastic(index: str, json_message["@timestamp"] = iso_timestamp # Create server url with relevant index pattern - _index = f"{index}-{datetime.datetime.today():%Y%m}" - url = f"{server}/{_index}/_doc" + if index_rollover: + index = f"{index}-{datetime.datetime.today():%Y%m}" + url = f"{server}/{index}/_doc" if id: url = url + f"/{id}" @@ -84,6 +87,7 @@ def send_to_elastic_bulk(index: str, api_key: str = ELK_TOKEN, batch_size: int = int(BATCH_SIZE), iso_timestamp: str | None = None, + index_rollover: bool = True, debug: bool = False): """ Method to send bulk message to ELK @@ -95,6 +99,7 @@ def send_to_elastic_bulk(index: str, :param server: url of ELK server :param batch_size: maximum size of batch :param iso_timestamp: timestamp to be included in documents + :param index_rollover: if true, adds monthly indication to index name :param debug: flag for debug mode :return: """ @@ -116,7 +121,8 @@ def __generate_id(element): json_message_list = [{**element, '@timestamp': iso_timestamp} for element in json_message_list] # Define server url with relevant index pattern (monthly indication is added) - index = f"{index}-{datetime.datetime.today():%Y%m}" + if index_rollover: + index = f"{index}-{datetime.datetime.today():%Y%m}" url = f"{server}/{index}/_bulk" if id_from_metadata: diff --git a/emf/common/logging/custom_logger.py b/emf/common/logging/custom_logger.py index 771cd743..793aec72 100644 --- a/emf/common/logging/custom_logger.py +++ b/emf/common/logging/custom_logger.py @@ -133,7 +133,10 @@ def emit(self, record): elk_record.update(self.extra) # Send to Elk - elastic.Elastic.send_to_elastic(index=self.index, json_message=elk_record, server=self.server) + elastic.Elastic.send_to_elastic(index=self.index, + json_message=elk_record, + server=self.server, + index_rollover=False) # TODO - Move tracing to seperate class, that on destroy will stop tracing? def start_trace(self, trace_parameters: dict): From 0a2679fa5a627d8e16133bf62137ea3a27824ac1 Mon Sep 17 00:00:00 2001 From: Gintautas Poderys <91978404+gpoderys@users.noreply.github.com> Date: Fri, 17 Apr 2026 15:32:45 +0300 Subject: [PATCH 12/13] #394 Simplified replacement logic steps --- emf/model_merger/replacement.py | 30 +++++------------------------- 1 file changed, 5 insertions(+), 25 deletions(-) diff --git a/emf/model_merger/replacement.py b/emf/model_merger/replacement.py index 44641703..8cb6afb9 100644 --- a/emf/model_merger/replacement.py +++ b/emf/model_merger/replacement.py @@ -236,44 +236,24 @@ def apply_priority_selection(candidates): if not step1_candidates.empty: best_model = apply_priority_selection(step1_candidates) logger.debug(f"STEP 1: Found same time horizon and same day replacement for {tso}") - + # STEP 2: If not available, use an IGM from the same energy delivery day (other time horizon) if best_model is None or best_model.empty: step2_candidates = tso_models[ (tso_models["pmd:scenarioDate"].apply(lambda x: parser.parse(x).date() == target_date_only)) & (tso_models["pmd:timeHorizon"] != target_time_horizon) - ] + ] if not step2_candidates.empty: - # Sort by time horizon priority based on config - if target_time_horizon in config["time_horizons"]: - priority_list = config["time_horizons"][target_time_horizon]["request_list"] - - step2_candidates["horizon_priority"] = step2_candidates["pmd:timeHorizon"].apply( - lambda horizon: priority_list.index(horizon) if horizon in priority_list else len(priority_list) - ) - step2_candidates = step2_candidates.sort_values("horizon_priority") - step2_candidates = step2_candidates.drop("horizon_priority", axis=1) - best_model = apply_priority_selection(step2_candidates) logger.debug(f"STEP 2: Found same day (different time horizon) replacement for {tso}") - + # STEP 3: If not available, use an IGM from the same time horizon of older models of the same day type if best_model is None or best_model.empty: - target_weekday = target_date.weekday() step3_candidates = tso_models[ - (tso_models["pmd:timeHorizon"] == target_time_horizon) & - (tso_models["pmd:scenarioDate"].apply(lambda x: - parser.parse(x).date().weekday() == target_weekday - )) + (tso_models["pmd:timeHorizon"] == target_time_horizon) ] - if not step3_candidates.empty: - # Sort by date (most recent first) and then by priority - step3_candidates = step3_candidates.sort_values( - by=["pmd:scenarioDate", "priority_hour", "priority_day", "priority_business"], - ascending=[False, True, True, True] - ) - best_model = step3_candidates.iloc[0:1] + best_model = apply_priority_selection(step3_candidates) logger.debug(f"STEP 3: Found same time horizon, same day type replacement for {tso}") # STEP 4: If not available, use older files of a different day type (original logic) From 03aeb8456414b1196d711b8d7c01f9d6045fd7a1 Mon Sep 17 00:00:00 2001 From: "martynas.karobcikas" Date: Wed, 22 Apr 2026 08:36:37 +0300 Subject: [PATCH 13/13] bugfix: Elastic support for SSL certs --- config/integrations/elastic.properties | 1 + emf/common/integrations/elastic.py | 38 +++++++++++++++++++------- emf/common/logging/custom_logger.py | 3 +- 3 files changed, 31 insertions(+), 11 deletions(-) diff --git a/config/integrations/elastic.properties b/config/integrations/elastic.properties index c8cc31c4..c71527df 100644 --- a/config/integrations/elastic.properties +++ b/config/integrations/elastic.properties @@ -1,4 +1,5 @@ [MAIN] ELK_SERVER = access_url ELK_TOKEN = access_token +ELK_SSL_VERIFY = True BATCH_SIZE = 20000 \ No newline at end of file diff --git a/emf/common/integrations/elastic.py b/emf/common/integrations/elastic.py index d13ddcfd..64271211 100644 --- a/emf/common/integrations/elastic.py +++ b/emf/common/integrations/elastic.py @@ -5,15 +5,12 @@ import pandas as pd import json import uuid +import os from typing import List, Dict from elasticsearch import Elasticsearch import config from emf.common.config_parser import parse_app_properties -import warnings -from elasticsearch.exceptions import ElasticsearchWarning -warnings.simplefilter('ignore', ElasticsearchWarning) - logger = logging.getLogger(__name__) parse_app_properties(caller_globals=globals(), path=config.paths.integrations.elastic) @@ -21,10 +18,28 @@ class Elastic: - def __init__(self, server: str = ELK_SERVER, api_key: str = ELK_TOKEN, debug: bool = False): + def __init__(self, + server: str = ELK_SERVER, + api_key: str = ELK_TOKEN, + ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()), + debug: bool = False): + self.server = server + self.api_key = api_key + self.ssl_verify = ssl_verify self.debug = debug - self.client = Elasticsearch(self.server, api_key=api_key) + + # Override SSL CERT FILE from ENV variables if defined + ssl_cert_file = None + if self.ssl_verify: + ssl_cert_file = os.getenv("SSL_CERT_FILE", None) + if not ssl_cert_file: + raise Exception("SSL verification is enabled but SSL_CERT_FILE environment variable is not set.") + else: + logger.debug(f"Using SSL certificate file: {ssl_cert_file}") + + # Create client + self.client = Elasticsearch(self.server, api_key=self.api_key, verify_certs=self.ssl_verify, ca_certs=ssl_cert_file) @staticmethod def send_to_elastic(index: str, @@ -32,6 +47,7 @@ def send_to_elastic(index: str, id: str = None, server: str = ELK_SERVER, api_key: str = ELK_TOKEN, + ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()), iso_timestamp: str = None, index_rollover: bool = True, debug: bool = False): @@ -69,7 +85,7 @@ def send_to_elastic(index: str, json_message.pop('args') json_data = json.dumps(json_message, default=str, ensure_ascii=True, skipkeys=True) headers = {"Authorization": f"ApiKey {api_key}", "Content-Type": "application/json"} - response = requests.post(url=url, data=json_data.encode(), headers=headers, verify=False) + response = requests.post(url=url, data=json_data.encode(), headers=headers, verify=ssl_verify) if json.loads(response.content).get('error'): logger.error(f"Send to Elasticsearch responded with error: {response.text}") if debug: @@ -85,6 +101,7 @@ def send_to_elastic_bulk(index: str, hashing: bool = False, server: str = ELK_SERVER, api_key: str = ELK_TOKEN, + ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()), batch_size: int = int(BATCH_SIZE), iso_timestamp: str | None = None, index_rollover: bool = True, @@ -141,7 +158,7 @@ def __generate_id(element): data=(ndjson.dumps(json_message_list[batch:batch + batch_size])+"\n").encode(), timeout=None, headers=headers, - verify=False) + verify=ssl_verify) if json.loads(response.content).get('errors'): logger.error(f"Send to Elasticsearch responded with errors: {response.text}") if debug: @@ -232,7 +249,7 @@ def __init__(self, hashing: bool = False, headers: Dict | None = None, auth: object | None = None, - verify: bool = False, + ssl_verify: bool = json.loads(ELK_SSL_VERIFY.lower()), debug: bool = False): self.index = index @@ -240,13 +257,14 @@ def __init__(self, self.id_from_metadata = id_from_metadata self.id_metadata_list = id_metadata_list self.hashing = hashing + self.ssl_verify = ssl_verify self.debug = debug if not headers: headers = {'Content-Type': 'text/json'} self.session = requests.Session() - self.session.verify = verify + self.session.verify = self.ssl_verify self.session.headers.update(headers) self.session.auth = auth diff --git a/emf/common/logging/custom_logger.py b/emf/common/logging/custom_logger.py index 793aec72..9652f2a3 100644 --- a/emf/common/logging/custom_logger.py +++ b/emf/common/logging/custom_logger.py @@ -1,4 +1,5 @@ import sys +import os import logging import requests from emf.common.integrations import elastic @@ -107,7 +108,7 @@ def __init__(self, def elk_connection(self): try: headers = {"Authorization": f"ApiKey {self.api_key}"} - response = requests.get(self.server, timeout=5, headers=headers, verify=False) + response = requests.get(self.server, timeout=5, headers=headers, verify=os.getenv("ELK_SSL_VERIFY", "false").lower() == "true") if response.status_code == 200: logger.info(f"Connection to {self.server} successful") return True