feat(parquet): variables metadata and download API for parquet outputs

antarest/core/serde/parquet_writer.py

@@ -55,7 +55,7 @@ def __enter__(self) -> "BatchParquetWriter":
     def __exit__(self, *args: Any, **kwargs: Any) -> None:
         self.close()
 
-    def add_table(self, table: pa.Table) -> None:
+    def append_table(self, table: pa.Table) -> None:
         if self._closed:
             raise ValueError("Writer is closed")
         self._current_batch.append(table)
@@ -115,7 +115,7 @@ def write_dataframes_in_parquet_format_by_column_sets(
         first_df = _adapt_polars_schema(first_df)
         table = first_df.to_arrow()
         current_writer = BatchParquetWriter(file_path, table.schema)
-        current_writer.add_table(table)
+        current_writer.append_table(table)
 
         while True:
             try:
@@ -143,7 +143,7 @@ def write_dataframes_in_parquet_format_by_column_sets(
 
                     current_writer = BatchParquetWriter(file_path, table.schema)
 
-                current_writer.add_table(table)
+                current_writer.append_table(table)
 
             except StopIteration:
                 return file_paths, new_index
@@ -181,7 +181,7 @@ def write_dataframes_stream_parquet(path: Path, dataframes: Iterator[pd.DataFram
         raise ValueError("No dataframe provided")
 
     with BatchParquetWriter(path, schema) as writer:
-        writer.add_table(first_table)
+        writer.append_table(first_table)
         for df in dataframes:
             table = pa.Table.from_pandas(df)
-            writer.add_table(table)
+            writer.append_table(table)

antarest/output/filestudy/model.py

@@ -11,12 +11,14 @@
 # This file is part of the Antares project.
 from dataclasses import dataclass
 from enum import Enum, StrEnum
+from functools import cached_property
 from pathlib import Path
-from typing import Callable, Generic, Literal, Sequence, TypeAlias, TypeVar
+from typing import Callable, Generic, Iterable, Literal, Sequence, TypeAlias, TypeVar
 
 import polars as pl
 
 from antarest.core.exceptions import OutputSubFolderNotFound
+from antarest.study.model import MatrixFrequency
 
 """Column name for the Monte Carlo year."""
 MCYEAR_COL = "mcYear"
@@ -163,3 +165,117 @@ def find_mode_dir(output_dir: Path) -> Path:
         if mode_dir.exists():
             return mode_dir
     raise OutputSubFolderNotFound(output_dir.name, "economy|adequacy")
+
+
+class FileOutput:
+    """
+    Provides a collection of methods to inspect the content of an output directory.
+
+    Caches properties that may take some time to build (typically going through files), for efficiency.
+
+    Attributes:
+        output_dir (Path): The path to the output directory typically (<study_dir>/outputs/<output_id>).
+    """
+
+    def __init__(self, output_dir: Path):
+        self.output_dir = output_dir
+
+    @cached_property
+    def mc_years(self) -> list[int]:
+        mode_dir = find_mode_dir(self.output_dir)
+        mc_ind_dir = mode_dir / "mc-ind"
+        if not mc_ind_dir.exists():
+            return []
+        return sorted(int(d.name) for d in mc_ind_dir.iterdir())
+
+    @property
+    def first_mc_year(self) -> int:
+        return self.mc_years[0]
+
+    @property
+    def mode(self) -> str:
+        return self.mode_dir.name
+
+    @cached_property
+    def mode_dir(self) -> Path:
+        return find_mode_dir(self.output_dir)
+
+    @property
+    def mc_all_dir(self) -> Path:
+        return self.mode_dir / "mc-all"
+
+    @property
+    def mc_ind_dir(self) -> Path:
+        return self.mode_dir / "mc-ind"
+
+    def get_mc_year_dir(self, year: int) -> Path:
+        return self.mc_ind_dir / f"{year:05d}"
+
+    @cached_property
+    def mc_ind_link_ids(self) -> tuple[str, ...]:
+        """
+        IDs of links that have data in mc-ind, sorted.
+        """
+        return tuple(sorted(d.name for d in self.iter_links_dir(self.first_mc_year)))
+
+    @cached_property
+    def mc_ind_area_ids(self) -> tuple[str, ...]:
+        """
+        IDs of areas that have data in mc-ind, sorted.
+        """
+        return tuple(sorted(d.name for d in self.iter_areas_dir(self.first_mc_year)))
+
+    @cached_property
+    def mc_all_link_ids(self) -> tuple[str, ...]:
+        """
+        IDs of links that have data in mc-all, sorted.
+        """
+        links_dir = self.mc_all_dir / "links"
+        return tuple(sorted(d.name for d in links_dir.iterdir()))
+
+    @cached_property
+    def mc_all_area_ids(self) -> tuple[str, ...]:
+        """
+        IDs of areas that have data in mc-all, sorted.
+        """
+        areas_dir = self.mc_all_dir / "areas"
+        return tuple(sorted(d.name for d in areas_dir.iterdir()))
+
+    def iter_areas_dir(self, mc_year: int) -> Iterable[Path]:
+        """
+        No ordering guarantee.
+        """
+        return (self.get_mc_year_dir(mc_year) / "areas").iterdir()
+
+    def iter_links_dir(self, mc_year: int) -> Iterable[Path]:
+        """
+        No ordering guarantee.
+        """
+        return (self.get_mc_year_dir(mc_year) / "links").iterdir()
+
+    def get_mc_all_file(
+        self,
+        file_type: MCAllAreasQueryFile | MCAllLinksQueryFile,
+        area_id: str,
+        frequency: MatrixFrequency,
+    ) -> Path | None:
+        """
+        Returns the path corresponding to the specified data, if it exists.
+        """
+        element_type = "areas" if isinstance(file_type, MCAllAreasQueryFile) else "links"
+        file_path = self.mc_all_dir / element_type / area_id / f"{file_type}-{frequency}.txt"
+        return file_path if file_path.exists() else None
+
+    def get_mc_ind_file(
+        self,
+        mc_year: int,
+        file_type: MCIndAreasQueryFile | MCIndLinksQueryFile,
+        area_id: str,
+        frequency: MatrixFrequency,
+    ) -> Path | None:
+        """
+        Returns the path corresponding to the specified data, if it exists.
+        """
+        element_type = "areas" if isinstance(file_type, MCIndAreasQueryFile) else "links"
+        file_path = self.get_mc_year_dir(mc_year) / element_type / area_id / f"{file_type}-{frequency}.txt"
+        return file_path if file_path.exists() else None

antarest/output/storage/v2/dbmodel.py

@@ -0,0 +1,108 @@
+# 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.
+from typing import Any, Literal, TypeAlias
+
+from sqlalchemy import BigInteger, Dialect, ForeignKeyConstraint, SmallInteger, String, types
+from sqlalchemy.orm import Mapped, mapped_column
+from typing_extensions import override
+
+from antarest.dbmodel import Base
+
+ElementType: TypeAlias = Literal[
+    "area",
+    "link",
+    "binding_constraint",
+    "thermal_cluster",
+    "renewable_cluster",
+    "short_term_storage",
+]
+
+ScenarioAggregation: TypeAlias = Literal["mc-ind", "mc-all"]
+
+
+class IntList(types.TypeDecorator[list[int]]):
+    """
+    Stores a list of integers as a comma separated string.
+
+    Can avoid many to many relationships which would not be useful.
+    """
+
+    impl = String
+    cache_ok = True
+
+    @override
+    def process_bind_param(self, value: list[int] | None, dialect: Dialect) -> str:
+        if not isinstance(value, list):
+            raise ValueError("Expected a list of int")
+        return ",".join(str(c) for c in value)
+
+    @override
+    def process_result_value(self, value: Any | None, dialect: Dialect) -> list[int]:
+        if not isinstance(value, str):
+            raise ValueError("Expected a string.")
+        return [int(c) for c in value.split(",")]
+
+
+class DbParquetOutput(Base):
+    # TODO: we should merge the existing v2_output_metadata tables into this one
+    #       the integer identifier will be easier and more efficient to use than the couple of strings
+    #       study_id / output_id
+
+    __tablename__ = "parquet_output"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    mc_years: Mapped[list[int]] = mapped_column(IntList)
+
+
+class DbParquetVariable(Base):
+    """
+    Represents one of the variables referenced in an output.
+
+    Those variables are then referenced by elements of the system (areas, links ...), that contain
+    actual data for them.
+
+    Attributes:
+        column: the column offset in the actual parquet file, compared to index columns (starts at 0).
+    """
+
+    __tablename__ = "parquet_variable"
+
+    __table_args__ = (ForeignKeyConstraint(["output_id"], ["parquet_output.id"]),)  # TODO
+
+    output_id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    scenario_aggregation: Mapped[ScenarioAggregation] = mapped_column(primary_key=True)
+    element_type: Mapped[ElementType] = mapped_column(primary_key=True)
+    column: Mapped[int] = mapped_column(SmallInteger, primary_key=True)
+    name: Mapped[str]
+    unit: Mapped[str | None]
+    statistic_type: Mapped[str | None]
+
+
+class DbParquetArea(Base):
+    """
+    Information related to an area of an output, in particular which variables it has data for,
+    in mc-ind and in mc-all (they may differ).
+
+    The variables are reference through their column index.
+    """
+
+    __tablename__ = "parquet_area"
+
+    __table_args__ = (ForeignKeyConstraint(["output_id"], ["parquet_output.id"]),)  # TODO
+
+    output_id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    area_id: Mapped[str] = mapped_column(primary_key=True)
+    mc_all_vars: Mapped[list[int]] = mapped_column(IntList)
+    mc_ind_vars: Mapped[list[int]] = mapped_column(IntList)
+
+
+# TODO: add tables for other element types: links, thermal clusters, etc

antarest/output/storage/v2/download.py

@@ -0,0 +1,60 @@
+# 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.
+
+
+"""
+Support for the "download" API
+"""
+
+from pathlib import Path
+
+from antarest.output.model import (
+    MatrixAggregationResultDTO,
+    StudyDownloadDTO,
+    StudyDownloadType,
+    TimeSerie,
+    TimeSeriesData,
+)
+from antarest.output.storage.v2.iteration import iterate_areas_df
+from antarest.output.storage.v2.metadata import IParquetOutputMetadata
+
+
+def build_matrix_aggregation_result(
+    output_metadata: IParquetOutputMetadata, parquet_dir: Path, data_selection: StudyDownloadDTO
+) -> MatrixAggregationResultDTO:
+
+    element_results: dict[str, TimeSeriesData] = {}  # one TimeSeriesData for each element of the system
+
+    if data_selection.type in {StudyDownloadType.AREA, StudyDownloadType.DISTRICT}:
+        area_dfs = iterate_areas_df(
+            output_metadata,
+            parquet_dir,
+            data_selection.level,
+            data_selection.years,
+            data_selection.filter,
+            data_selection.columns,
+        )
+        for area_df in area_dfs:
+            year, area_id, df, vars = area_df.year, area_df.area_id, area_df.data, area_df.variables
+            ts_data = element_results.setdefault(
+                area_id, TimeSeriesData(type=data_selection.type, name=area_id, data={})
+            )
+            for var_index, var in enumerate(vars):
+                numerical_data = df.to_series(var_index).cast(float).to_list()
+                ts_data.data.setdefault(str(year), []).append(
+                    TimeSerie(name=var.name, unit=var.unit_repr(), data=numerical_data)
+                )
+
+    return MatrixAggregationResultDTO(
+        index=output_metadata.get_time_index(data_selection.level),
+        data=list(element_results.values()),
+    )