Skip to content

feat(scenario aggregation): introduce new feature scenario aggregation #90

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dusanparipovic wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat(scenario aggregation): introduce new feature scenario aggregation #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
de1f722
feat(scenario aggregation): introduce new feature scenario aggregation
dusanparipovic Jul 21, 2026
4dc63ca
fix(scenario aggregator): simplify aggregator and change everything o…
dusanparipovic Jul 21, 2026
9d3d748
refactor(tests): apply 3a pattern
dusanparipovic Jul 21, 2026
76b5251
fix(tests): update schema in temporal_metric_view for metric handling
dusanparipovic Jul 21, 2026
1162e86
chore(tests): update catalog schema by removing legacy metric definit…
dusanparipovic Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions gems_views_builder/aggregators/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
# Copyright (c) 2026, RTE (https://www.rte-france.com)
#
# See AUTHORS.txt
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.
94 changes: 94 additions & 0 deletions gems_views_builder/aggregators/scenario_aggregator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
# Copyright (c) 2026, RTE (https://www.rte-france.com)
#
# See AUTHORS.txt
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.

import logging
import os
import tempfile
from dataclasses import dataclass
from enum import Enum

import polars as pl

from gems_views_builder.common import PARQUET_COMPRESSION, PARQUET_COMPRESSION_LEVEL, PARQUET_ROW_GROUP_SIZE
from gems_views_builder.metric_view import MetricView


class ScenarioOperator(Enum):
EXP = "exp"
STD = "std"
MIN = "min"
MAX = "max"


# expectation (exp) == mean
SCENARIO_AGG_EXPRS = [
pl.col("metric_value").mean().alias(ScenarioOperator.EXP.value),
pl.col("metric_value").std(ddof=0).alias(ScenarioOperator.STD.value),
pl.col("metric_value").min().alias(ScenarioOperator.MIN.value),
pl.col("metric_value").max().alias(ScenarioOperator.MAX.value),
]


@dataclass
class ScenarioAggregator:

@guilpier-code guilpier-code Aug 7, 2026 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

According to this comment, this class would be turned into a function :

def to_scenario_view(temporal_metric_view: MetricView, scenario_op) -> None:
	file_descriptor, tmp_path = tempfile.mkstemp(suffix=".parquet")
	os.close(file_descriptor)

	scenario_op.run(temporal_metric_view, tmp_path)

	os.replace(tmp_path, temporal_metric_view.persistence_path)
	logging.info(f"Scenario aggregation written to {temporal_metric_view.persistence_path}")

Note that :

  • this function is much simpler than the class is replaces
  • code was split into 4 pieces :
    • a factory function makeScenarioOperator
    • this function to_scenario_view (which is a kind of orchestrator)
    • 2 scenario operator classes, both having a run(...) method : class ScenarioAggregation and class ScenarioColumnsAddition
  • these 4 entities could all be defined in the current source file
  • the current source file could now be renamed into into-scenario-view.py, for example, as it's no longer only about scenario aggregations

scenario_aggregation: bool

def run(self, temporal_metric_view: MetricView) -> None:
file_descriptor, tmp_path = tempfile.mkstemp(suffix=".parquet")
os.close(file_descriptor)

if not self.scenario_aggregation:
logging.info("Scenario aggregation disabled; preserving per-scenario rows")
(

@guilpier-code guilpier-code Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess we could remove the parenthesis containing the scenario transformations

pl.scan_parquet(temporal_metric_view.persistence_path)
.with_columns(
[
pl.lit(False, dtype=pl.Boolean).alias("scenario_aggregation"),
pl.lit(None, dtype=pl.Utf8).alias("scenario_stat"),
]
)
.sink_parquet(
tmp_path,
compression=PARQUET_COMPRESSION,
compression_level=PARQUET_COMPRESSION_LEVEL,
row_group_size=PARQUET_ROW_GROUP_SIZE,
)
)
else:
logging.info("Aggregating across scenarios (exp/std/min/max)")
index_columns = ["metric_id", "metric_location", "breakdown_properties", "view_date"]
(
pl.scan_parquet(temporal_metric_view.persistence_path)
.group_by(index_columns)
.agg(SCENARIO_AGG_EXPRS)
.unpivot(
on=[op.value for op in ScenarioOperator],
index=index_columns,
variable_name="scenario_stat",
value_name="metric_value",
)
.with_columns(
[
pl.lit(None, dtype=pl.Int64).alias("scenario_id"),
pl.lit(True, dtype=pl.Boolean).alias("scenario_aggregation"),
]
)
.sink_parquet(
tmp_path,
compression=PARQUET_COMPRESSION,
compression_level=PARQUET_COMPRESSION_LEVEL,
row_group_size=PARQUET_ROW_GROUP_SIZE,
)
)

os.replace(tmp_path, temporal_metric_view.persistence_path)
logging.info(f"Scenario aggregation written to {temporal_metric_view.persistence_path}")
18 changes: 15 additions & 3 deletions gems_views_builder/terms_aggregator.py → ...s_builder/aggregators/terms_aggregator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,15 @@
# Copyright (c) 2026, RTE (https://www.rte-france.com)
#
# See AUTHORS.txt
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.

import logging
import tempfile
from pathlib import Path
Expand Down Expand Up @@ -31,14 +43,14 @@ def run(self, metric_structure_table: MetricStructureTable, metric: Metric) -> M
logging.info(f"[{metric.id}] Aggregating terms with operator {metric.terms_operator.value}")
value_agg = pl.col("value").sum() if metric.terms_operator == TermsOperator.SUM else pl.col("value").mean()
metric_view = (
structured_simulation_table.with_columns(pl.col("scenario_index").alias("scenario"))
structured_simulation_table.with_columns(pl.col("scenario_index").alias("scenario_id"))
.group_by(
[
"metric_id",
"metric_location",
"breakdown_properties",
"absolute_time_index",
"scenario",
"scenario_id",
]
)
.agg(
Expand All @@ -54,7 +66,7 @@ def run(self, metric_structure_table: MetricStructureTable, metric: Metric) -> M
"metric_location",
"breakdown_properties",
"absolute_time_index",
"scenario",
"scenario_id",
"granular_metric_value",
"granular_date",
]
Expand Down
16 changes: 14 additions & 2 deletions gems_views_builder/time_aggregator.py → ...ws_builder/aggregators/time_aggregator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,15 @@
# Copyright (c) 2026, RTE (https://www.rte-france.com)
#
# See AUTHORS.txt
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
# SPDX-License-Identifier: MPL-2.0
#
# This file is part of the Antares project.

import atexit
import logging
import tempfile
Expand Down Expand Up @@ -52,7 +64,7 @@ def run(self, metric_view: MetricView, metric: Metric) -> MetricView:
"metric_id",
"metric_location",
"breakdown_properties",
"scenario",
"scenario_id",
"view_date",
]
)
Expand All @@ -63,7 +75,7 @@ def run(self, metric_view: MetricView, metric: Metric) -> MetricView:
"metric_location",
"breakdown_properties",
"view_date",
"scenario",
"scenario_id",
pl.col("metric_value").cast(pl.Float64),
]
)
Expand Down
10 changes: 8 additions & 2 deletions gems_views_builder/input/view_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ class Scope(ViewBuilderBasedModel):

class Aggregation(ViewBuilderBasedModel):
time: TimeAggregation | None = None
scenario: bool | None = None


class CatalogId(ViewBuilderBasedModel):
Expand All @@ -53,7 +54,7 @@ class MetricId(ViewBuilderBasedModel):
class RawViewConfig(ViewBuilderBasedModel):
id: str
scope: list[Scope]
aggregation: list[Aggregation]
aggregation: Aggregation
catalog: list[CatalogId]
metrics: list[MetricId]

Expand All @@ -66,6 +67,7 @@ class ViewConfig:
location_taxonomy_category: str
catalog_ids: set[str] = field(default_factory=set)
time_aggregation: TimeAggregation | None = None
scenario_aggregation: bool = False
metric_ids: list[str] = field(default_factory=list)
metrics: list[Metric] = field(default_factory=list)

Expand Down Expand Up @@ -114,13 +116,17 @@ def load_view_config(config_file_path: Path) -> ViewConfig:
raise ValueError(
f"view_config.yml '{raw_view_config.id}': no calendar configured in scope. One calendar must be configured in scope"
)

view_config = ViewConfig(
id=raw_view_config.id,
input_data_path=input_data_path,
calendar_id=calendar_id,
location_taxonomy_category=location_taxonomy_category,
catalog_ids={c.id for c in raw_view_config.catalog},
time_aggregation=raw_view_config.aggregation[0].time if raw_view_config.aggregation else None,
time_aggregation=raw_view_config.aggregation.time if raw_view_config.aggregation.time is not None else None,
scenario_aggregation=raw_view_config.aggregation.scenario

@guilpier-code guilpier-code Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

scenario_aggregation=bool(raw_view_config.aggregation.scenario)

is shorter than the if...else... instruction.

Same for time_aggregation

if raw_view_config.aggregation.scenario is not None
else False,
metric_ids=[metric.id for metric in raw_view_config.metrics],
)
logging.info(
Expand Down
8 changes: 6 additions & 2 deletions gems_views_builder/view/views_builder.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,12 @@

"""ViewBuilder."""

from gems_views_builder.aggregators.scenario_aggregator import ScenarioAggregator
from gems_views_builder.aggregators.terms_aggregator import TermsAggregator
from gems_views_builder.aggregators.time_aggregator import TimeAggregator
from gems_views_builder.input.input_data import InputData
from gems_views_builder.metric_view import MetricView
from gems_views_builder.metrics_structure_builder import MetricStructureTableBuilder
from gems_views_builder.terms_aggregator import TermsAggregator
from gems_views_builder.time_aggregator import TimeAggregator


class ViewBuilder:
Expand All @@ -30,12 +31,15 @@ def __init__(self, input_data: InputData, metric_structure_table_builder: Metric
# # Aggregator for step 2C
self.time_aggregator = TimeAggregator(self.input_data.view_config.time_aggregation)

self.scenario_aggregator = ScenarioAggregator(self.input_data.view_config.scenario_aggregation)

def build(self) -> list[MetricView]:
metric_views: list[MetricView] = []
for metric in self.input_data.view_config.metrics:
metric_structure_table = self.metric_structure_table_builder.build(metric)
metric_view = self.terms_aggregator.run(metric_structure_table, metric)
temporal_metric_view = self.time_aggregator.run(metric_view, metric)
self.scenario_aggregator.run(temporal_metric_view)

@aoustry aoustry Jul 22, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a specific reason why scenario_aggregator modify a MetricView, where a time_aggregator returns one ? This is does not look very homogeneous.

@dusanparipovic dusanparipovic Jul 23, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As you can see on line 43, we append temporal_metric_view to the metric_views list.

We don't have any strict requirements for the MetricView object. In practice, it only stores the path to the temporary view.

If you look at the run method of the ScenarioAggregator, you'll notice that it doesn't return anything. That's because I intentionally rely on Python's reference semantics: the temporal_metric_view object is passed by reference, and the aggregator updates its content (specifically, the path stored in the MetricView object).

From my perspective, there's no need to create a new object with identical data. We can simply update the existing object instead.

@tbittar tbittar Jul 24, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add a comment in the beginning of the function to make the behavior difference explicit :

"""Rewrites temporal_metric_view's parquet file in place (same path, new content)
and returns the same MetricView instance; it does not produce a separate artifact.""

@guilpier-code guilpier-code Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's a good remark : this step changes the input table whatever the value of view_config.scenario_aggregation, so it would make sense to make this step return another view.

@guilpier-code guilpier-code Aug 7, 2026 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

self.scenario_aggregator.run(temporal_metric_view)

What is a bit misleading is that this step doesn't necessarily aggregate the input table by scenario (although it says it does) : case where view_config.scenario_aggregation is false.
So this step is more a "scenario transformation" or "to scenario format conversion", where the operation performed is either an aggregation or a columns addition with no aggregation.

So I would suggest the following (in views_builder.py) :

class ViewBuilder:
    def __init__(self, input_data: InputData, ...) -> None:
        ...
        self.scenario_op = makeScenarioOperator(self.input_data.view_config.scenario_aggregation)

    def build(self) -> list[MetricView]:
        metric_views: list[MetricView] = []
        for metric in self.input_data.view_config.metrics:
            ...
            temporal_metric_view = self.time_aggregator.run(metric_view, metric)
            scenario_metric_view = to_scenario_view(temporal_metric_view, scenario_op)
            metric_views.append(scenario_metric_view)
        return metric_views

Note that :

  • factory function makeScenarioOperator would be defined in scenario_aggregator.py, and returns an object corresponding either to scenario aggregation or to a simple columns addition.
  • the scenario aggregator's run(...) method is replaced with a simple function to_scenario_view
  • code is broken into smaller pieces (with smaller responsibilities) : in particular, code inside scenario_aggregator.run(...) gets much smaller in function to_scenario_view (please see this comment)

@guilpier-code guilpier-code Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question :
Maybe I missing something, but I see that, for a given metric, at the time aggregation step, we print the resulting table on disc via pl.sink_parquet.
Then, in the scenario step, we read it from disk, transform it and then print it again.
Why don't we pass a table from time aggregation to scenario transformation without printing ?

More generally, can we avoid printing and scanning to/from disc multiple times ?

metric_views.append(temporal_metric_view)

return metric_views
2 changes: 1 addition & 1 deletion resources/tests_inputs/filtering_and_breakdown/catalogs/catalog.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

############################ Columns of a Business View #######################################

# metric_id | metric_location | breakdown_property | view_date | scenario_id | metric_value |
# metric_id | metric_location | breakdown_properties | view_date | scenario_id | scenario_aggregation | scenario_stat | metric_value

####################################################################################################

Expand Down
3 changes: 2 additions & 1 deletion resources/tests_inputs/filtering_and_breakdown/view_config.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@ view:
- calendar: calendar_file # calendar_file.csv

aggregation:
- time: hour
time: hour
scenario: false

@guilpier-code guilpier-code Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

scenario: false

Do we have to make this change ?
If we don't, does it change anything ?

Same question for every view-config.yml file in this PR


catalog: #Pointer to "catalogs" = "Business Metrics definitions files".
- id: catalog
Expand Down
3 changes: 2 additions & 1 deletion resources/tests_inputs/filtering_and_breakdown_property_order/view_config.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@ view:
- calendar: calendar_file # calendar_file.csv

aggregation:
- time: hour
time: hour
scenario: false

catalog: #Pointer to "catalogs" = "Business Metrics definitions files".
- id: catalog
Expand Down
3 changes: 2 additions & 1 deletion resources/tests_inputs/test_3/view_config.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@ view:
- calendar: calendar_file # change this in for test purposes

aggregation:
- time: hour
time: hour
scenario: false

catalog: #Pointer to "catalogs" = "Business Metrics definitions files".
- id: catalog
Expand Down
2 changes: 1 addition & 1 deletion tests/test_business_view_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ def test_raises_on_invalid_metric_id_format(tmp_path: Path) -> None:
- taxonomy-category: balance
- calendar: calendar_file
aggregation:
- time: hour
time: hour
catalog:
- id: catalog_1
metrics:
Expand Down
16 changes: 8 additions & 8 deletions tests/test_filtering_and_breakdown.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,11 +64,11 @@ def _view_generation_total_by_scenario(
df = _at(view, metric_id)
if breakdown_properties is not None:
df = df.filter(pl.col("breakdown_properties") == breakdown_properties)
return df.group_by("scenario").agg(pl.col("metric_value").sum().alias("view_total")).sort("scenario")
return df.group_by("scenario_id").agg(pl.col("metric_value").sum().alias("view_total")).sort("scenario_id")


def _assert_totals_close(got: pl.DataFrame, exp: pl.DataFrame, *, msg: str = "") -> None:
merged = got.join(exp, on="scenario", how="inner")
merged = got.join(exp, on="scenario_id", how="inner")
assert merged.height == exp.height, msg
raw_max = (merged["view_total"] - merged["expected_total"]).abs().max()
assert isinstance(raw_max, Real), f"{msg} unexpected max diff type: {type(raw_max)}"
Expand All @@ -93,8 +93,8 @@ def _expected_generation_total_by_scenario(
return (
base.group_by("scenario_index")
.agg(pl.col("value").sum().alias("expected_total"))
.rename({"scenario_index": "scenario"})
.sort("scenario")
.rename({"scenario_index": "scenario_id"})
.sort("scenario_id")
)


Expand Down Expand Up @@ -125,12 +125,12 @@ def test_filter_nuclear_production_is_subset_of_by_tech(fb_view_result: pl.DataF
PRODUCTION_BY_TECH slice at breakdown {(technology,nuclear)}.
"""
nuclear = _at(fb_view_result, "NUCLEAR_PRODUCTION").select(
["metric_location", "view_date", "scenario", "metric_value"]
["metric_location", "view_date", "scenario_id", "metric_value"]
)
by_tech_nuclear = (
_at(fb_view_result, "PRODUCTION_BY_TECH")
.filter(pl.col("breakdown_properties") == "{(technology,nuclear)}")
.select(["metric_location", "view_date", "scenario", "metric_value"])
.select(["metric_location", "view_date", "scenario_id", "metric_value"])
)
assert nuclear.sort(nuclear.columns).to_dicts() == by_tech_nuclear.sort(by_tech_nuclear.columns).to_dicts()

Expand All @@ -140,7 +140,7 @@ def test_production_equals_sum_by_tech_and_company_partitions(fb_view_result: pl
For each (location, date, scenario), PRODUCTION must equal the sum across the
breakdown partitions (tech, company, and tech+company).
"""
keys = ["metric_location", "view_date", "scenario"]
keys = ["metric_location", "view_date", "scenario_id"]

prod = _at(fb_view_result, "PRODUCTION").group_by(keys).agg(pl.col("metric_value").sum().alias("v"))

Expand Down Expand Up @@ -229,7 +229,7 @@ def test_single_generator_slice_matches_component_simulation_sum(
_at(view, "PRODUCTION_BY_TECH_AND_COMPANY")
.filter(
(pl.col("breakdown_properties") == "{(technology,nuclear),(company,rhonepower)}")
& (pl.col("scenario") == scenario)
& (pl.col("scenario_id") == scenario)
)
.select(pl.col("metric_value").sum())
.item()
Expand Down
10 changes: 5 additions & 5 deletions tests/test_filtering_and_breakdown_property_order.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ def property_order_workspace(test_files_root: Path, tmp_path: Path) -> tuple[Pat


def _assert_totals_close(got: pl.DataFrame, exp: pl.DataFrame, *, msg: str = "") -> None:
merged = got.join(exp, on="scenario", how="inner")
merged = got.join(exp, on="scenario_id", how="inner")
assert merged.height == exp.height, msg
raw_max = (merged["view_total"] - merged["expected_total"]).abs().max()
assert isinstance(raw_max, Real), f"{msg} unexpected max diff type: {type(raw_max)}"
Expand Down Expand Up @@ -93,18 +93,18 @@ def test_same_breakdown_group_sums_all_matching_generators(property_order_worksp
sim.filter((pl.col("output") == "generation") & pl.col("component").is_in(_GAS_RHONEPOWER_GENERATORS))
.group_by("scenario_index")
.agg(pl.col("value").sum().alias("expected_total"))
.rename({"scenario_index": "scenario"})
.sort("scenario")
.rename({"scenario_index": "scenario_id"})
.sort("scenario_id")
)

got = (
view.filter(
(pl.col("metric_id") == "PRODUCTION_BY_TECH_AND_COMPANY")
& (pl.col("breakdown_properties") == _GAS_RHONEPOWER_BREAKDOWN)
)
.group_by("scenario")
.group_by("scenario_id")
.agg(pl.col("metric_value").sum().alias("view_total"))
.sort("scenario")
.sort("scenario_id")
)

_assert_totals_close(got, expected, msg="PRODUCTION_BY_TECH_AND_COMPANY (gas, rhonepower)")
Expand Down
Loading
Loading