Skip to content

Commit

Permalink
refactor(utils): change SPECIFIC_MATRICES structure to include NamedT…
Browse files Browse the repository at this point in the history 
…uple
  • Loading branch information
@MartinBelthle
MartinBelthle committed Feb 6, 2024
1 parent 5228043 commit e5e93b7
Show file tree
Hide file tree
Showing 2 changed files with 108 additions and 115 deletions.
34 changes: 12 additions & 22 deletions antarest/study/service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,8 +114,9 @@
upgrade_study,
)
from antarest.study.storage.utils import (
SPECIFIC_MATRICES,
MatrixProfile,
assert_permission,
get_specific_matrices_according_to_version,
get_start_date,
is_managed,
remove_from_cache,
Expand Down Expand Up @@ -151,42 +152,31 @@ def get_disk_usage(path: t.Union[str, Path]) -> int:


def _handle_specific_matrices(
study_version: int,
df: pd.DataFrame,
specific_matrix_name: str,
matrix_profile: MatrixProfile,
matrix_path: str,
*,
with_index: bool,
with_columns: bool,
) -> pd.DataFrame:
json_matrix = SPECIFIC_MATRICES[specific_matrix_name]
if with_columns:
if json_matrix["alias"] == "bindingconstraints":
if study_version < 870:
cols: t.List[str] = json_matrix["cols_with_version"]["before_870"] # type: ignore
else:
cols: t.List[str] = json_matrix["cols_with_version"]["after_870"] # type: ignore
elif json_matrix["alias"] == "links":
cols = _handle_links_columns(study_version, matrix_path, json_matrix)
if Path(matrix_path).parts[1] == "links":
cols = _handle_links_columns(matrix_path, matrix_profile)
else:
cols: t.List[str] = json_matrix["cols"] # type: ignore
cols = matrix_profile.cols
if cols:
df.columns = pd.Index(cols)
rows: t.List[str] = json_matrix["rows"] # type: ignore
rows = matrix_profile.rows
if with_index and rows:
df.index = rows # type: ignore
return df


def _handle_links_columns(study_version: int, matrix_path: str, json_matrix: t.Mapping[str, t.Any]) -> t.List[str]:
def _handle_links_columns(matrix_path: str, matrix_profile: MatrixProfile) -> t.List[str]:
path_parts = Path(matrix_path).parts
area_id_1 = path_parts[2]
area_id_2 = path_parts[3]
result: t.List[str] = (
json_matrix["cols_with_version"]["before_820"]
if study_version < 820
else json_matrix["cols_with_version"]["after_820"]
)
result = matrix_profile.cols
for k, col in enumerate(result):
if col == "Hurdle costs direct":
result[k] = f"{col} ({area_id_1}->{area_id_2})"
Expand Down Expand Up @@ -2465,12 +2455,12 @@ def get_matrix_with_index_and_header(
)
df_matrix.index = time_column

for specific_matrix in SPECIFIC_MATRICES:
specific_matrices = get_specific_matrices_according_to_version(int(study.version))
for specific_matrix in specific_matrices:
if re.match(specific_matrix, path):
return _handle_specific_matrices(
int(study.version),
df_matrix,
specific_matrix,
specific_matrices[specific_matrix],
path,
with_index=with_index,
with_columns=with_columns,
Expand Down
189 changes: 96 additions & 93 deletions antarest/study/storage/utils.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import calendar
import copy
import logging
import math
import os
Expand Down Expand Up @@ -268,48 +269,44 @@ def _generate_columns(column_suffix: str) -> t.List[str]:
return [f"{i}{column_suffix}" for i in range(101)]


class MatrixProfile(t.NamedTuple):
"""
Matrix profile for time series or classic tables.
"""

cols: t.List[str]
rows: t.List[str]
stats: bool


SPECIFIC_MATRICES = {
"input/hydro/common/capacity/creditmodulations_*": {
"alias": "creditmodulations",
"cols": _generate_columns(""),
"rows": ["Generating Power", "Pumping Power"],
"stats": False,
},
"input/hydro/common/capacity/maxpower_*": {
"alias": "maxpower",
"cols": [
"input/hydro/common/capacity/creditmodulations_*": MatrixProfile(
cols=_generate_columns(""),
rows=["Generating Power", "Pumping Power"],
stats=False,
),
"input/hydro/common/capacity/maxpower_*": MatrixProfile(
cols=[
"Generating Max Power (MW)",
"Generating Max Energy (Hours at Pmax)",
"Pumping Max Power (MW)",
"Pumping Max Energy (Hours at Pmax)",
],
"rows": [],
"stats": False,
},
"input/hydro/common/capacity/reservoir_*": {
"alias": "reservoir",
"cols": ["Lev Low (p.u)", "Lev Avg (p.u)", "Lev High (p.u)"],
"rows": [],
"stats": False,
},
"input/hydro/common/capacity/waterValues_*": {
"alias": "waterValues",
"cols": _generate_columns("%"),
"rows": [],
"stats": False,
},
"input/hydro/series/*/mod": {"alias": "mod", "cols": [], "rows": [], "stats": True},
"input/hydro/series/*/ror": {"alias": "ror", "cols": [], "rows": [], "stats": True},
"input/hydro/common/capacity/inflowPattern_*": {
"alias": "inflowPattern",
"cols": ["Inflow Pattern (X)"],
"rows": [],
"stats": False,
},
"input/hydro/prepro/*/energy": {
"alias": "energy",
"cols": ["Expectation (MWh)", "Std Deviation (MWh)", "Min. (MWh)", "Max. (MWh)", "ROR Share"],
"rows": [
rows=[],
stats=False,
),
"input/hydro/common/capacity/reservoir_*": MatrixProfile(
cols=["Lev Low (p.u)", "Lev Avg (p.u)", "Lev High (p.u)"],
rows=[],
stats=False,
),
"input/hydro/common/capacity/waterValues_*": MatrixProfile(cols=_generate_columns("%"), rows=[], stats=False),
"input/hydro/series/*/mod": MatrixProfile(cols=[], rows=[], stats=True),
"input/hydro/series/*/ror": MatrixProfile(cols=[], rows=[], stats=True),
"input/hydro/common/capacity/inflowPattern_*": MatrixProfile(cols=["Inflow Pattern (X)"], rows=[], stats=False),
"input/hydro/prepro/*/energy": MatrixProfile(
cols=["Expectation (MWh)", "Std Deviation (MWh)", "Min. (MWh)", "Max. (MWh)", "ROR Share"],
rows=[
"January",
"February",
"March",
Expand All @@ -323,66 +320,72 @@ def _generate_columns(column_suffix: str) -> t.List[str]:
"November",
"December",
],
"stats": False,
},
"input/thermal/prepro/*/*/modulation": {
"alias": "modulation",
"cols": ["Marginal cost modulation", "Market bid modulation", "Capacity modulation", "Min gen modulation"],
"rows": [],
"stats": False,
},
"input/thermal/prepro/*/*/data": {
"alias": "data",
"cols": ["FO Duration", "PO Duration", "FO Rate", "PO Rate", "NPO Min", "NPO Max"],
"rows": [],
"status": False,
},
"input/reserves/*": {
"alias": "reserves",
"cols": ["Primary Res. (draft)", "Strategic Res. (draft)", "DSM", "Day Ahead"],
"rows": [],
"status": False,
},
"input/misc-gen/miscgen-*": {
"alias": "miscgen",
"cols": ["CHP", "Bio Mass", "Bio Gaz", "Waste", "GeoThermal", "Other", "PSP", "ROW Balance"],
"rows": [],
"status": False,
},
"input/bindingconstraints/*": {
"alias": "bindingconstraints",
"rows": [],
"stats": False,
"cols_with_version": {"before_870": ["<", ">", "="], "after_870": []},
},
"input/links/*/*": {
"alias": "links",
"rows": [],
"stats": False,
"cols_with_version": {
"after_820": [
"Hurdle costs direct",
"Hurdle costs indirect",
"Impedances",
"Loop flow",
"P.Shift Min",
"P.Shift Max",
],
"before_820": [
"Capacités de transmission directes",
"Capacités de transmission indirectes",
"Hurdle costs direct",
"Hurdle costs indirect",
"Impedances",
"Loop flow",
"P.Shift Min",
"P.Shift Max",
],
},
},
stats=False,
),
"input/thermal/prepro/*/*/modulation": MatrixProfile(
cols=["Marginal cost modulation", "Market bid modulation", "Capacity modulation", "Min gen modulation"],
rows=[],
stats=False,
),
"input/thermal/prepro/*/*/data": MatrixProfile(
cols=["FO Duration", "PO Duration", "FO Rate", "PO Rate", "NPO Min", "NPO Max"],
rows=[],
stats=False,
),
"input/reserves/*": MatrixProfile(
cols=["Primary Res. (draft)", "Strategic Res. (draft)", "DSM", "Day Ahead"],
rows=[],
stats=False,
),
"input/misc-gen/miscgen-*": MatrixProfile(
cols=["CHP", "Bio Mass", "Bio Gaz", "Waste", "GeoThermal", "Other", "PSP", "ROW Balance"],
rows=[],
stats=False,
),
"input/bindingconstraints/*": MatrixProfile(cols=["<", ">", "="], rows=[], stats=False),
"input/links/*/*": MatrixProfile(
cols=[
"Capacités de transmission directes",
"Capacités de transmission indirectes",
"Hurdle costs direct",
"Hurdle costs indirect",
"Impedances",
"Loop flow",
"P.Shift Min",
"P.Shift Max",
],
rows=[],
stats=False,
),
}


SPECIFIC_MATRICES_820 = copy.deepcopy(SPECIFIC_MATRICES)
SPECIFIC_MATRICES_820["input/links/*/*"] = MatrixProfile(
cols=[
"Hurdle costs direct",
"Hurdle costs indirect",
"Impedances",
"Loop flow",
"P.Shift Min",
"P.Shift Max",
],
rows=[],
stats=False,
)

SPECIFIC_MATRICES_870 = copy.deepcopy(SPECIFIC_MATRICES_820)
SPECIFIC_MATRICES_870["input/bindingconstraints/*"] = MatrixProfile(cols=[], rows=[], stats=False)


def get_specific_matrices_according_to_version(study_version: int) -> t.Dict[str, MatrixProfile]:
if study_version < 820:
return SPECIFIC_MATRICES
elif study_version < 870:
return SPECIFIC_MATRICES_820
return SPECIFIC_MATRICES_870


def get_start_date(
file_study: FileStudy,
output_id: t.Optional[str] = None,
Expand Down

0 comments on commit e5e93b7

Please sign in to comment.