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 1/2] 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 2/2] 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('')