diff --git a/PyPSA/PypsaReader.py b/PyPSA/PypsaReader.py
new file mode 100644
index 0000000..487c747
--- /dev/null
+++ b/PyPSA/PypsaReader.py
@@ -0,0 +1,579 @@
+import inspect
+import os
+import sys
+
+import numpy as np
+import pandas as pd
+import pypsa
+import yaml
+
+PYPSA_DIR: str = os.path.dirname(os.path.abspath(__file__))
+PARENT_DIR: str = os.path.dirname(PYPSA_DIR)
+sys.path.insert(0, os.path.dirname(PARENT_DIR))
+sys.path.insert(0, PARENT_DIR)
+
+import pypsa_helper as h
+from ExcelWriter import ExcelWriter
+from InOutModule.printer import Printer
+
+printer = Printer.getInstance()
+
+
+class Conversions:
+ """Registry of conversion functions for unit transformations."""
+
+ @staticmethod
+ def EUR_to_MEUR(val: pd.Series, df: pd.DataFrame = None) -> pd.Series:
+ """Converts € to Mio.€"""
+ return val * 1e-6
+
+ @staticmethod
+ def MEUR_to_EUR(val: pd.Series, df: pd.DataFrame = None) -> pd.Series:
+ """Converts Mio.€ to €"""
+ return val * 1e6
+
+ @staticmethod
+ def MW_to_kW(val: pd.Series, df: pd.DataFrame = None) -> pd.Series:
+ """Converts MW to kW."""
+ return val * 1e3
+
+ @staticmethod
+ def V_to_kV(val, df: pd.DataFrame = None) -> pd.Series:
+ """Converts V to kV."""
+ return val * 1e-3
+
+ @staticmethod
+ def EUR_per_MVA_to_MEUR(val, df: pd.DataFrame = None) -> pd.Series:
+ """Calculates total cost in MEUR: (EUR/unit) * capacity * 1e-6."""
+ capacity = df.s_nom
+ return val * capacity * 1e-6
+
+ @staticmethod
+ def pu_to_absolute(val, df: pd.DataFrame) -> pd.Series:
+ """Converts per-unit values to absolute values using the base value from the DataFrame."""
+ return val * df.p_nom
+
+ @staticmethod
+ def bool_to_binary(val, df: pd.DataFrame = None) -> pd.Series:
+ """Converts boolean values to binary (0/1) integers."""
+ if val.isnull().all():
+ return 0
+ else:
+ # Use infer_objects to explicitly handle the type conversion from object to bool
+ val = val.fillna(False).infer_objects(copy=False)
+ return val.astype(int)
+
+ @staticmethod
+ def year_and_lifetime_to_year_decom(val, df: pd.DataFrame) -> pd.Series:
+ """Calculates decommissioning year based on commissioning year and lifetime."""
+ # Only return a decom year if build_year and lifetime is available and build_year is not 0; otherwise return NaN
+ mask = df.build_year.notnull() & (df.build_year != 0) & df.lifetime.notnull()
+ res = pd.Series(np.nan, index=df.index)
+ res[mask] = df.loc[mask, "build_year"] + df.loc[mask, "lifetime"]
+ return res
+
+ @staticmethod
+ def is_ldes(val: pd.Series, df: pd.DataFrame, metadata: dict) -> pd.Series:
+ """Sets binary to 1 if energy-to-power ratio (max_hours) is greater than the threshold in metadata."""
+ threshold = metadata.get("Metadata", {}).get("LDES_threshold", 2160)
+ return (val > threshold).astype(int)
+
+ @staticmethod
+ def line_carrier_to_tec_repr(val, df: pd.DataFrame) -> pd.Series:
+ """Maps line carriers to LEGO's technology representations (e.g., DC-OPF for AC lines)."""
+ if df.carrier.isnull().all():
+ return "DC-OPF"
+ else:
+ return df.carrier.map({"AC": "DC-OPF", "DC": "TP"})
+
+ @staticmethod
+ def total_capacity_to_number_of_units(val, df: pd.DataFrame) -> pd.Series:
+ """Calculates the number of units based on total capacity and nominal capacity per unit."""
+ # Check if the input value is all NaN or all infinite, and return a default of 100 in that case
+ if val.isnull().all() or np.isinf(val).all():
+ return 100
+ else:
+ val = val.fillna(0).replace(
+ np.inf, 1000
+ ) # Treat NaN as 0 for investment calculation
+ p_nom = (
+ df.p_nom.fillna(1).replace(np.inf, 100).replace(0, 1)
+ ) # Avoid division by zero and treat NaN as 1 for unit calculation
+ return (np.ceil(val / p_nom)).astype(int)
+
+ @staticmethod
+ def _get_fuel_costs(df: pd.DataFrame, metadata: dict) -> pd.Series:
+ """Helper to get fuel costs for each row in the DataFrame based on carrier."""
+ fuel_mapping = {}
+ meta_config = metadata.get("Metadata", {})
+ for key, fuel_info in meta_config.items():
+ if (
+ isinstance(fuel_info, dict)
+ and "filter" in fuel_info
+ and "cost" in fuel_info
+ ):
+ for carrier in fuel_info["filter"]:
+ fuel_mapping[carrier] = fuel_info["cost"]
+
+ # Map carriers to costs; default to NaN if not found
+ fuel_costs = df["carrier"].map(fuel_mapping)
+
+ # Performance check: only check for missing carriers if DataFrame isn't empty
+ if not fuel_costs.empty:
+ missing_carriers = df.loc[fuel_costs.isnull(), "carrier"].unique()
+ if len(missing_carriers) > 0:
+ printer.warning(
+ f"No fuel cost defined in Metadata for carrier(s): {missing_carriers.tolist()}"
+ )
+
+ return fuel_costs
+
+ @staticmethod
+ def get_fuel_cost_from_metadata(
+ val: pd.Series, df: pd.DataFrame, metadata: dict
+ ) -> pd.Series:
+ """Retrieves fuel costs based on the carrier and Metadata configuration."""
+ return Conversions._get_fuel_costs(df, metadata)
+
+ @staticmethod
+ def _get_storage_capacity_costs(df: pd.DataFrame, metadata: dict) -> pd.Series:
+ """Helper to get storage capacity costs for each row in the DataFrame based on carrier."""
+ cost_mapping = {}
+ meta_config = metadata.get("Metadata", {})
+ for key, info in meta_config.items():
+ if (
+ isinstance(info, dict)
+ and "filter" in info
+ and "default_storage_capacity_cost" in info
+ ):
+ for carrier in info["filter"]:
+ cost_mapping[carrier] = info["default_storage_capacity_cost"]
+
+ # Map carriers to costs; default to NaN if not found
+ costs = df["carrier"].map(cost_mapping)
+
+ return costs
+
+ @staticmethod
+ def get_storage_capacity_cost_from_metadata(
+ val: pd.Series, df: pd.DataFrame, metadata: dict
+ ) -> pd.Series:
+ """Retrieves storage capacity costs based on the carrier and Metadata configuration."""
+ return Conversions._get_storage_capacity_costs(df, metadata)
+
+ @staticmethod
+ def EUR_per_hour_to_MWh_per_hour(
+ val: pd.Series, df: pd.DataFrame, metadata: dict
+ ) -> pd.Series:
+ """Converts costs per hour of thermal generation (e.g. stand_by_cost) to costs per MWh based on the fuel cost specified in the metadata."""
+ fuel_costs = Conversions._get_fuel_costs(df, metadata)
+ # Result is MWh/h = (EUR/h) / (EUR/MWh)
+ # Avoid division by zero, handle NaN/Inf
+ with np.errstate(divide="ignore", invalid="ignore"):
+ res = val / fuel_costs
+ return res.replace([np.inf, -np.inf], 0).fillna(0)
+
+ @staticmethod
+ def EUR_to_MWh(val: pd.Series, df: pd.DataFrame, metadata: dict) -> pd.Series:
+ """Converts costs to costs per MWh based on the fuel cost specified in the metadata."""
+ fuel_costs = Conversions._get_fuel_costs(df, metadata)
+ # Result is MWh = EUR / (EUR/MWh)
+ with np.errstate(divide="ignore", invalid="ignore"):
+ res = val / fuel_costs
+ return res.replace([np.inf, -np.inf], 0).fillna(0)
+
+ @staticmethod
+ def greater_or_equal_to_one(val: pd.Series) -> pd.Series:
+ """Ensures that all values are greater or equal to 1, replacing values less than 1 with 1."""
+ return val.apply(lambda x: max(x, 1) if pd.notnull(x) else x)
+
+ @staticmethod
+ def pu_to_absolute_and_p_nom_if_nan_or_zero(
+ val: pd.Series, df: pd.DataFrame
+ ) -> pd.Series:
+ val = val * df.p_nom
+ return val.where((val.notnull() & (val != 0)), df.p_nom)
+
+ @staticmethod
+ def one_if_nan_or_zero(
+ val: pd.Series,
+ ) -> pd.Series:
+ """Replaces NaN or zero values with 1, keeping other values unchanged."""
+ return val.where((val.notnull() & (val != 0)), 1)
+
+ @staticmethod
+ def bool_to_binary_zero_if_p_nom_is_zero(
+ val: pd.Series, df: pd.DataFrame
+ ) -> pd.Series:
+ """Converts boolean values to binary (0/1) integers and sets to 0, if p_nom is zero."""
+ if val.isnull().all():
+ return 0
+ else:
+ # Use infer_objects to explicitly handle the type conversion from object to bool
+ val = val.fillna(False).infer_objects(copy=False)
+ val = val.where(df.p_nom != 0, 0) # set to 0 if p_nom is zero
+ return val.astype(int)
+
+
+class NetworkDataExtractor:
+ def __init__(
+ self,
+ network: pypsa.Network,
+ config_path: str = None,
+ table_definitions_path: str = None,
+ allow_placeholders: bool = False,
+ ):
+ """
+ Initializes the extractor with a PyPSA network and configuration files.
+
+ :param network: The PyPSA network instance to extract data from.
+ :param config_path: Path to the mapping configuration YAML file.
+ :param table_definitions_path: Path to the TableDefinitions XML file.
+ :param allow_placeholders: If the placeholder value in the config file is allowed.
+ """
+ self.network = network
+ self._conv_params_cache = {} # Performance: Cache function signatures
+
+ if config_path is None:
+ config_path = os.path.join(PYPSA_DIR, "pypsa_lego_mapping_config.yaml")
+
+ if table_definitions_path is None:
+ table_definitions_path = os.path.join(
+ PARENT_DIR, "TableDefinitions.xml"
+ )
+
+ with open(config_path, "r") as f:
+ config_text = f.read()
+
+ if not allow_placeholders and "to be filled out" in config_text:
+ printer.error(f"Placeholder value 'to be filled out' is still present in the config file {os.path.basename(config_path)}. "
+ f"Please fill in all placeholder values before running the converter.")
+ exit(1)
+
+ self.config = yaml.safe_load(config_text)
+
+ # Initialize ExcelWriter to get definitions for checking/normalization
+ self.writer = ExcelWriter(table_definitions_path)
+ self.excel_definitions = self.writer.excel_definitions
+
+ self.dataframes = self._extract_dataframes()
+ # Normalize and check dataframes (moving checking functionality from testing script)
+ self.dataframes = self._normalize_dataframes()
+
+ def _normalize_dataframes(self):
+ """Checks and normalizes dataframes to ensure they are ready for LEGO output."""
+ normalized_dfs = {}
+ for name, df in self.dataframes.items():
+ # Only write tables that are defined in TableDefinitions.xml
+ table_id = name[1:] if name.startswith("d") else name
+
+ if table_id not in self.excel_definitions:
+ printer.warning(f"Skipping {name} (not defined in TableDefinitions.xml)")
+ continue
+
+ # Add mandatory 'scenario' column for LEGO format if missing
+ if "scenario" not in df.columns:
+ df["scenario"] = "ScenarioA"
+
+ # Add 'id' column if missing
+ if "id" not in df.columns:
+ df["id"] = np.nan
+
+ # Ensure all columns from TableDefinition are present (at least as NaN)
+ definition = self.excel_definitions[table_id]
+ for col_def in definition.columns:
+ if col_def.db_name not in df.columns and col_def.db_name != "NOEXCL":
+ df[col_def.db_name] = np.nan
+
+ normalized_dfs[name] = df
+
+ return normalized_dfs
+
+ def _extract_dataframes(self):
+ """Extracts and transforms data from the PyPSA network into LEGO DataFrames."""
+ df_dict = {}
+
+ for table_name, cfg in self.config.items():
+ if table_name == "Metadata":
+ continue
+
+ # 1. Resolve Filters from Category
+ self._resolve_category_filters(cfg)
+
+ # 2. Get Source Data
+ source_df = self._get_source_df(cfg)
+ if source_df is None:
+ continue
+
+ # 3. Process Column Mapping
+ df = self._map_columns(source_df, cfg)
+
+ # 4. Handle Indexing
+ if "index" in cfg:
+ self._apply_indexing(df, source_df, cfg["index"])
+
+ # 5. Normalize for LEGO Format
+ # Drop columns that are now in the index to avoid "already exists" error on reset_index
+ for idx_name in df.index.names:
+ if idx_name in df.columns:
+ df = df.drop(columns=[idx_name])
+
+ # Reduces length of index to less than 450 characters
+ # This is only implemented because the LEGO database currently cannot handle more than 450 characters in the index!
+ if isinstance(df.index, pd.MultiIndex):
+ df.index = pd.MultiIndex.from_frame(
+ df.index.to_frame().astype(str).apply(lambda x: x.str[:449])
+ )
+ else:
+ df.index = df.index.astype(str).str[:449]
+
+ df = df.reset_index()
+ df = self._add_scenario_columns(df)
+
+ df_dict[table_name] = df
+
+ return df_dict
+
+ def _resolve_category_filters(self, cfg):
+ """Resolves technology filters from Metadata if a category is defined."""
+ if "category" not in cfg:
+ return
+
+ category = cfg["category"]
+ meta_data = self.config.get("Metadata", {})
+ meta_cat = meta_data.get(category, {})
+
+ # Combine filters from listed technologies or use direct filter
+ cat_filter = meta_cat.get("filter", [])
+ if isinstance(cat_filter, (str, int, float)):
+ cat_filter = [cat_filter]
+ else:
+ cat_filter = list(cat_filter)
+
+ for tech in meta_cat.get("technologies", []):
+ tech_filter = meta_data.get(tech, {}).get("filter", [])
+ if isinstance(tech_filter, list):
+ cat_filter.extend(tech_filter)
+ else:
+ cat_filter.append(tech_filter)
+
+ if cat_filter:
+ if "source" not in cfg:
+ cfg["source"] = {}
+ # Ensure uniqueness and format as query string
+ unique_filter = list(set(cat_filter))
+ cfg["source"]["filter"] = f"carrier in {unique_filter}"
+
+ def _get_source_df(self, cfg):
+ """Retrieves the source DataFrame based on the configuration."""
+ src = cfg.get("source")
+ if not src:
+ return None
+
+ if src["type"] == "attribute":
+ source_df = getattr(self.network, src["name"])
+ if "filter" in src:
+ source_df = source_df.query(src["filter"])
+ return source_df
+ elif src["type"] == "helper":
+ helper = getattr(h, src["name"])
+ if "metadata_arg" not in src:
+ return helper(self.network, cfg)
+ meta = self.config.get("Metadata", {})
+ return helper(self.network, cfg, meta.get(src["metadata_arg"]))
+ return None
+
+ def _map_columns(self, source_df, cfg):
+ """Maps PyPSA attributes to LEGO columns using the mapping configuration."""
+ column_data = {}
+ for lego_col, mapping in cfg.get("mapping", {}).items():
+ if isinstance(mapping, dict):
+ # Attribute mapping with potential unit conversion or transformation function
+ attr = mapping.get("attr")
+ if attr and attr in source_df.columns:
+ val = source_df[attr]
+ else:
+ # Try to get value as series of NaNs or fixed value
+ val = pd.Series(mapping.get("value", np.nan), index=source_df.index)
+
+ # Priority 1: Named conversion function
+ if "conversion" in mapping:
+ conv_name = mapping["conversion"].replace("()", "")
+ if hasattr(Conversions, conv_name):
+ conv_func = getattr(Conversions, conv_name)
+
+ # Performance Improvement: Cache function parameter counts
+ if conv_name not in self._conv_params_cache:
+ sig = inspect.signature(conv_func)
+ self._conv_params_cache[conv_name] = len(sig.parameters)
+
+ num_params = self._conv_params_cache[conv_name]
+
+ try:
+ if num_params == 3:
+ # Vectorized call: (Series, DataFrame, config)
+ val = conv_func(val, source_df, self.config)
+ elif num_params == 2:
+ # Vectorized call: (Series, DataFrame)
+ val = conv_func(val, source_df)
+ else:
+ # Vectorized call: (Series)
+ val = conv_func(val)
+ except Exception as e:
+ raise ValueError(
+ f"Error applying conversion '{conv_name}' to column '{lego_col}': {e}"
+ )
+ else:
+ printer.warning(
+ f"Conversion function '{conv_name}' not found in Conversions class."
+ )
+
+ # Priority 3: Simple multiplier factor
+ elif "factor" in mapping:
+ val = val * mapping["factor"]
+
+ column_data[lego_col] = val
+ elif isinstance(mapping, (int, float)):
+ # Static value
+ column_data[lego_col] = mapping
+ else:
+ # Direct attribute string mapping
+ if mapping in source_df.columns:
+ column_data[lego_col] = source_df[mapping]
+ else:
+ column_data[lego_col] = np.nan
+ return pd.DataFrame(column_data)
+
+ @staticmethod
+ def _apply_indexing(df, source_df, idx_cfg):
+ """Applies the indexing logic to the DataFrame."""
+ if isinstance(idx_cfg, dict):
+ # MultiIndex from specified columns/attributes
+ index_data = {}
+ for lego_idx, pypsa_source in idx_cfg.items():
+ if pypsa_source in source_df.columns:
+ index_data[lego_idx] = source_df[pypsa_source]
+ elif pypsa_source == "index":
+ index_data[lego_idx] = source_df.index
+ else:
+ index_data[lego_idx] = np.nan
+ df.index = pd.MultiIndex.from_frame(
+ pd.DataFrame(index_data, index=source_df.index)
+ )
+ elif isinstance(idx_cfg, str):
+ # Simple index renaming
+ df.index = source_df.index.rename(idx_cfg)
+ elif isinstance(idx_cfg, list):
+ # Fallback for current list style: assumes columns match LEGO names
+ df.index = pd.MultiIndex.from_frame(source_df[idx_cfg]).set_names(idx_cfg)
+
+ def _add_scenario_columns(self, df) -> pd.DataFrame:
+ """Adds dataPackage and dataSource columns based on config or defaults."""
+ meta = self.config.get("Metadata", {})
+ df["dataPackage"] = meta.get("dataPackage", "default-package")
+ df["dataSource"] = meta.get("dataSource", "default-source")
+ return df
+
+ def get_dataframes(self):
+ """Returns the dictionary of extracted and normalized DataFrames."""
+ return self.dataframes
+
+
+def translate_pypsa_to_lego(
+ input_directory: str = r"./input_data",
+ input_file: str = r"pypsa-model.nc",
+ output_directory: str = r"./output_data",
+ output_folder_name: str = "LEGO-Model",
+ allow_placeholders: bool = False,
+):
+ """
+ Main execution block for converting a PyPSA network to LEGO-formatted Excel files.
+
+ To use this:
+ 1. Point 'input_directory' and 'input_file' to your .nc PyPSA network.
+ 2. Set 'output_directory' and 'output_folder_name' for the resulting Excel files.
+ 3. Run the script with command-line arguments. See: `python PypsaReader.py --help`
+ """
+ # Define the path to the PyPSA network (similar than implemented in PyPSA-LEGO-Translator_testing.py)
+ filepath = os.path.join(input_directory, input_file)
+
+ if not os.path.exists(filepath):
+ printer.error(f"File not found at {filepath}")
+ else:
+ # Load the network
+ printer.information(f"Loading PyPSA network from {filepath}...")
+ try:
+ net = pypsa.Network(filepath)
+ except Exception as e:
+ printer.error(f"Could not load network: {e}")
+ exit(1)
+
+ # Extract data using NetworkDataExtractor
+ printer.information("Extracting data into LEGO format...")
+ extractor = NetworkDataExtractor(net, allow_placeholders=allow_placeholders)
+ dfs = extractor.get_dataframes()
+
+ # Initialize ExcelWriter
+ writer = ExcelWriter()
+
+ # Define output directory
+ output_dir = os.path.join(output_directory, output_folder_name)
+ if not os.path.exists(output_dir):
+ os.makedirs(output_dir, exist_ok=True)
+
+ # Writing Excel files (filtering/checking functionality is now inside NetworkDataExtractor)
+ printer.information("Writing Excel files...")
+ for name, df in dfs.items():
+ table_id = name[1:] if name.startswith("d") else name
+
+ printer.information(f"Writing {name} to {output_dir}...")
+ try:
+ writer._write_Excel_from_definition(df, output_dir, table_id)
+ except Exception as e:
+ printer.error(f"Error writing {name}: {e}")
+
+ printer.information(f"Conversion complete. Output files are in: {output_dir}")
+
+
+if __name__ == "__main__":
+ import argparse
+ from rich_argparse import RichHelpFormatter
+
+ parser = argparse.ArgumentParser(
+ description="Convert a PyPSA network file to LEGO formatted Excel files.",
+ formatter_class=RichHelpFormatter,
+ )
+ parser.add_argument(
+ "inputDirectory",
+ type=str,
+ help="Path to the folder containing the PyPSA .nc network file.",
+ )
+ parser.add_argument(
+ "inputFile",
+ type=str,
+ help="Name of the PyPSA .nc network file.",
+ )
+ parser.add_argument(
+ "outputDirectory",
+ type=str,
+ help="Path to the folder where the LEGO output folder should be created.",
+ )
+ parser.add_argument(
+ "outputFolderName",
+ type=str,
+ help="Name of the output folder for the generated LEGO Excel files.",
+ )
+ parser.add_argument(
+ "--allow-placeholders",
+ action="store_true",
+ help="Allow the converter to be run with the placeholder values in the mapping config.",
+ )
+ args = parser.parse_args()
+
+ translate_pypsa_to_lego(
+ input_directory=args.inputDirectory,
+ input_file=args.inputFile,
+ output_directory=args.outputDirectory,
+ output_folder_name=args.outputFolderName,
+ allow_placeholders=args.allow_placeholders,
+ )
diff --git a/PyPSA/README.md b/PyPSA/README.md
new file mode 100644
index 0000000..dd2e4bf
--- /dev/null
+++ b/PyPSA/README.md
@@ -0,0 +1,190 @@
+# How to translate a PyPSA-Eur dataset to LEGO
+
+This guide explains how to convert an existing PyPSA-Eur network file (`.nc`) to LEGO Excel files.
+
+To build the PyPSA-Eur dataset first, follow
+[How to create a PyPSA-Eur dataset](https://github.com/IEE-TUGraz/pypsa-eur-multivoltage/blob/fix/corine-bool-and-offshore-keyerror/README_pypsa_dataset.md)
+in the `pypsa-eur-multivoltage` repository.
+
+Since the conversion can require a lot of memory and may also take some time, it is recommended to use a powerful
+machine.
+
+## 1. Set the mapping metadata
+
+In `pypsa_lego_mapping_config.yaml`, update the dataset fields that are marked with the placeholder value
+`to be filled out`.
+
+Example for a dataset with the target year 2025 and weather year 2013:
+
+```yaml
+Metadata:
+ dataSource: "PyPSA-Eur_2025_CY2013"
+
+dData_Packages:
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions."
+
+dData_Sources:
+ personResponsible: "Jane Doe"
+ lastEdited: "21.07.2026"
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions."
+
+dGlobal_Scenarios:
+ comments: "PyPSA-Eur 2025 CY2013"
+```
+
+What these fields mean:
+
+- `dataSource`: Name of the source dataset.
+- `dData_Packages.comments`: Comment written to `Data_Packages.xlsx`.
+- `dData_Sources.personResponsible`: The person responsible for the source entry (your name).
+- `dData_Sources.lastEdited`: Date when the source entry was last edited.
+- `dData_Sources.comments`: Comment written to `Data_Sources.xlsx`.
+- `dGlobal_Scenarios.comments`: Comment written to `Global_Scenarios.xlsx`.
+
+> **Note:** The fuel costs of thermal technologies defined in the `pypsa_lego_mapping_config.yaml` are only dummy values and should get adapted for your specific case study!
+
+## 2. Translate to LEGO
+
+From the `PyPSA/` folder, run the converter with:
+
+```
+python PypsaReader.py
+```
+
+Example:
+
+```
+python PypsaReader.py "../../pypsa-eur-multivoltage/resources/PyPSA-Eur_2025_CY2013/networks" "base_s_all_elec.nc" "../../" "LEGO_PyPSA-Eur_2025_CY2013"
+```
+
+This will write the LEGO Excel files to:
+
+```
+../../LEGO_PyPSA-Eur_2025_CY2013
+```
+
+## About the generated files
+
+The converter generates the following LEGO files:
+
+- `Data_Packages.xlsx`
+- `Data_Sources.xlsx`
+- `Global_Scenarios.xlsx`
+- `Power_BusInfo.xlsx`
+- `Power_Demand.xlsx`
+- `Power_Hindex.xlsx`
+- `Power_Inflows.xlsx`
+- `Power_Network.xlsx`
+- `Power_Storage.xlsx`
+- `Power_ThermalGen.xlsx`
+- `Power_VRES.xlsx`
+- `Power_VRESProfiles.xlsx`
+- `Power_WeightsK.xlsx`
+- `Power_WeightsRP.xlsx`
+
+The following files are not generated and have to be added and adapted manually:
+
+- `Global_Parameters.xlsx`
+- `Power_Parameters.xlsx`
+
+# Reader Documentation
+
+The `PypsaReader` module provides a specialized pipeline for converting PyPSA (Python for Power System Analysis)
+networks into the Excel-based data
+format required by the LEGO model.
+
+## Purpose & limitations
+
+- This reader only converts PyPSA networks to LEGO input files (EXCEL). It does not return a LEGO CaseStudy object.
+- For running the converted case study in LEGO, the files `Global_Parameters.xlsx` and `Power_Parameters.xlsx` still
+ have to be provided manually.
+- Currently, only the power sector is supported by the converter.
+- The generated Excel files include all the required information for importing into the LEGO database.
+
+## General Workflow
+
+The conversion process follows a structured workflow:
+
+1. **Network Loading**: A PyPSA network is loaded from a NetCDF (`.nc`) file.
+2. **Configuration Parsing**: The `NetworkDataExtractor` reads `pypsa_lego_mapping_config.yaml` to determine how PyPSA
+ components (buses, lines,
+ generators,
+ etc.) map to LEGO tables.
+3. **Data Extraction & Transformation**:
+ - **Source Retrieval**: Data is pulled either directly from PyPSA network attributes (e.g., `net.buses`) or via
+ complex processing in
+ `pypsa_helper.py`.
+ - **Filtering**: Technologies are filtered based on their `carrier` attribute using definitions in the `Metadata`
+ section of the config.
+ - **Column Mapping**: PyPSA attributes are mapped to LEGO column names.
+ - **Unit Conversion**: The `Conversions` class applies transformations (e.g., MW to kW, EUR to MEUR, or calculating
+ decommissioning years).
+4. **Normalization**: DataFrames are augmented with mandatory LEGO columns (`scenario`, `id`, `dataPackage`,
+ `dataSource`) and aligned with
+ definitions in `TableDefinitions.xml`.
+5. **Excel Export**: The `ExcelWriter` saves each processed DataFrame into individual `.xlsx` files. **Warning:**
+ This can take up to several
+ hours for very
+ large networks!
+
+## Configuration File (`pypsa_lego_mapping_config.yaml`)
+
+The YAML configuration acts as the "translation map" between the two models.
+
+### Metadata Section
+
+* **Technology Filters**: Defines list of `carrier` strings that identify specific technologies (e.g.,
+ `solar: filter: ['solar', 'Solar']`).
+* **Fuel Costs**: Specifies fuel prices in EUR/MWh for thermal technologies.
+* **Categories**: Groups individual technologies into broader LEGO categories like `ThermalGen` or `VRES` for batch
+ processing.
+* **Global Settings**: Defines `dataPackage`, `dataSource`, and thresholds like `LDES_threshold`.
+
+### Table Mapping Section
+
+Each entry (e.g., `dPower_ThermalGen`) defines:
+
+* **source**: The data origin (`type: attribute` for direct PyPSA access or `type: helper` for function calls).
+* **category**: (Optional) References a metadata category to automatically filter the source data.
+* **index**: Defines the LEGO index columns (e.g., `i` for bus, `g` for generator).
+* **mapping**: A dictionary where keys are LEGO columns and values are either:
+ - A direct PyPSA attribute name.
+ - A static value (int/float).
+ - A dictionary specifying an `attr`, a `conversion` function, or a `factor`.
+* **metadata_arg**: (Optional) Passes a single value from the `Metadata` section into the helper
+ function.
+
+## Helper Functions (`pypsa_helper.py`)
+
+When simple attribute mapping is insufficient, helper functions handle complex data aggregation and profile generation:
+
+* **Network Topology**:
+ - `prepare_ac_lines_and_dc_links`: Combines PyPSA `lines`, `links` (filtered for DC), and `transformers` into a
+ single LEGO `Power_Network` table.
+ - `prepare_ac_lines` / `prepare_transformers`: Calculate missing electrical parameters ($r, x, b$) based on standard
+ type definitions if they are
+ not explicitly set.
+* **Time-Series Profiles**:
+ - `prepare_renewable_profiles`: Extracts `p_max_pu` for VRES generators and reshapes it into a long-form LEGO
+ format.
+ - `prepare_inflow_profiles`: Aggregates inflow data from both `storage_units` (hydro) and `generators` (
+ Run-of-River) into a unified time-series.
+ - `prepare_demand_profiles`: Sums PyPSA load data per bus and formats it for the LEGO demand table.
+* **Package, Scenario and Time Data**:
+ - `prepare_data_packages`: Creates the data package information.
+ - `prepare_data_sources`: Creates the data source information.
+ - `prepare_global_scenarios`: Creates the scenario information.
+ - `prepare_power_hindex`: Creates the relation between periods and representative periods.
+ - `prepare_power_weights_k`: Creates the representative time step weights.
+ - `prepare_power_weights_rp`: Creates the representative period weights.
+
+## Conversions & Logic
+
+The `Conversions` class in `PypsaReader.py` contains static methods for specialized logic:
+
+* **Unit Scaling**: `EUR_to_MEUR`, `MW_to_kW`, `V_to_kV`.
+* **Calculated Values**: `year_and_lifetime_to_year_decom` (derives decommissioning date) and
+ `total_capacity_to_number_of_units`.
+* **LEGO Specifics**: `is_ldes` (Long Duration Energy Storage detection) and `line_carrier_to_tec_repr` (mapping
+ carriers to LEGO technical
+ representations like `DC-OPF` or `TP`).
diff --git a/PyPSA/pypsa_helper.py b/PyPSA/pypsa_helper.py
new file mode 100644
index 0000000..0b326d7
--- /dev/null
+++ b/PyPSA/pypsa_helper.py
@@ -0,0 +1,401 @@
+import numpy as np
+import pandas as pd
+
+from InOutModule.printer import Printer
+
+printer = Printer.getInstance()
+
+
+def prepare_data_packages(net, config: dict, data_package: str):
+ """
+ Prepares data package information for the LEGO output.
+ """
+ data_packages = pd.DataFrame({
+ "dataPackage": [data_package],
+ "comments": [config.get("comments", "")],
+ })
+ return data_packages
+
+
+def prepare_data_sources(net, config: dict, data_source: str):
+ """
+ Prepares data source information for the LEGO output.
+ """
+ data_sources = pd.DataFrame({
+ "dataSource": [data_source],
+ "sources": [config.get("sources", "")],
+ "licenseInfo": [config.get("licenseInfo", "")],
+ "personResponsible": [config.get("personResponsible", "")],
+ "lastEdited": [config.get("lastEdited", "")],
+ "comments": [config.get("comments", "")],
+ })
+ return data_sources
+
+
+def prepare_global_scenarios(net, config: dict):
+ """
+ Prepares global scenario information for the LEGO output.
+ """
+ global_scenarios = pd.DataFrame({
+ "scenarioID": [config.get("scenarioID", "ScenarioA")],
+ "relativeWeight": [config.get("relativeWeight", 1.0)],
+ "comments": [config.get("comments", "")],
+ })
+ return global_scenarios
+
+
+def prepare_power_hindex(net, config: dict):
+ """
+ Prepares the relation between periods and representative periods for the LEGO output.
+ """
+ num_timesteps = len(net.snapshots)
+ power_hindex = pd.DataFrame({
+ "p": [f"h{i + 1:04d}" for i in range(num_timesteps)],
+ "rp": [config.get("rp", "rp01")] * num_timesteps,
+ "k": [f"k{i + 1:04d}" for i in range(num_timesteps)],
+ })
+ return power_hindex
+
+
+def prepare_power_weights_k(net, config: dict):
+ """
+ Prepares the representative time step weights for the LEGO output.
+ """
+ num_timesteps = len(net.snapshots)
+ power_weights_k = pd.DataFrame({
+ "k": [f"k{i + 1:04d}" for i in range(num_timesteps)],
+ "pWeight_k": [config.get("pWeight_k", 1)] * num_timesteps,
+ })
+ return power_weights_k
+
+
+def prepare_power_weights_rp(net, config: dict):
+ """
+ Prepares the representative period weights for the LEGO output.
+ """
+ power_weights_rp = pd.DataFrame({
+ "rp": [config.get("rp", "rp01")],
+ "pWeight_rp": [config.get("pWeight_rp", 1)],
+ })
+ return power_weights_rp
+
+
+def prepare_ac_lines(net, config: dict):
+ """
+ Prepares AC line data by calculating missing parameters (r, x, b) from line types,
+ converting them to per-unit values using BasePower and v_nom,
+ and ensuring consistent naming and carrier definitions.
+ """
+ lines = net.lines.copy()
+ types = net.line_types
+ s_base = config.get("BasePower", 100)
+
+ # Define the line parameters r, x, b if only line type is specified
+ r_per_len = lines["type"].map(types["r_per_length"])
+ x_per_len = lines["type"].map(types["x_per_length"])
+
+ # Determine b_per_length, assuming 50 Hz
+ # b_per_len [uS/km] = 2 * pi * 50 * C [nF/km] * 1e-3
+ b_per_len = lines["type"].map(types["c_per_length"]) * 2 * np.pi * 50 * 1e-3
+
+ # Vectorized calculation: replace 0 values with type-based defaults (SI values)
+ # Note: b_per_length in PyPSA line_types is typically in uS/km, so we multiply by 1e-6 to get Siemens
+ lines["r"] = lines["r"].where(lines["r"] != 0, r_per_len * lines["length"])
+ lines["x"] = lines["x"].where(lines["x"] != 0, x_per_len * lines["length"])
+ lines["b"] = lines["b"].where(lines["b"] != 0, b_per_len * lines["length"] * 1e-6)
+
+ # Get v_nom from buses (kV) for each line (using bus0 as reference)
+ v_nom = lines.bus0.map(net.buses.v_nom)
+
+ # Calculate Z_base = V_nom^2 / S_base [Ohm]
+ # Since V_nom is in kV and S_base is in MW, (kV^2 / MW) results in Ohms.
+ z_base = (v_nom ** 2) / s_base
+
+ # Convert electrical parameters to per-unit values
+ lines["r"] = lines["r"] / z_base
+ lines["x"] = lines["x"] / z_base
+ lines["b"] = lines["b"] * z_base
+
+ # Add tap ratios and phase shifts with default values (if not already present)
+ lines["tap_ratio"] = 1
+ lines["phase_shift"] = 0
+
+ ### INFO: Currently can the LEGO Database not handle similar line names (e.g. c1, c1, c2, c2, ...)
+ # 11-05-2026
+ # Ensure every line has an individual circuit identifier (c1, c2, ...) for parallel lines
+ # lines["name"] = "c" + (lines.groupby(["bus0", "bus1"]).cumcount() + 1).astype(str)
+
+ # Define individual line names until LEGO database can handle similar line names
+ lines["name"] = "AC_line_" + pd.Series(range(len(lines)), index=lines.index).astype(
+ str
+ )
+
+ # if carrier is nan or empty string (''), set to AC for all lines to get defined as DC-OPF
+ lines.carrier = lines.carrier.fillna("AC").where(lines.carrier != "", "AC")
+
+ return lines
+
+
+def prepare_dc_links(net, config: dict):
+ """
+ Extracts and prepares DC link data, initializing parameters for DC-OPF compatibility
+ and generating unique names.
+ """
+ links = net.links[net.links["carrier"] == "DC"].copy()
+
+ # Define line parameters as nan for DC links, as they are not relevant for DC-OPF
+ links["r"] = np.nan
+ links["x"] = np.nan
+ links["b"] = np.nan
+
+ # Define s_nom and s_nom_extendable to be consistent with lines and transformers
+ links["s_nom"] = links["p_nom"]
+ links["s_nom_extendable"] = links["p_nom_extendable"]
+ links["id"] = links.index
+ # Vectorized name generation
+ ### INFO: Currently can the LEGO Database not handle similar line names (e.g. c1, c1, c2, c2, ...)
+ # 11-05-2026
+ # links["name"] = "DC_Link_" + pd.Series(range(len(links)), index=links.index).astype(
+ # str
+ # )
+
+ links["name"] = "DC_link_" + pd.Series(range(len(links)), index=links.index).astype(
+ str
+ )
+
+ # if carrier is nan or empty string (''), set to DC for all links to get defined as transport problem (TP)
+ links.carrier = links.carrier.fillna("DC").where(links.carrier != "", "DC")
+
+ # Add tap ratios and phase shifts with default values (if not already present)
+ links["tap_ratio"] = 1
+ links["phase_shift"] = 0
+
+ return links
+
+
+def prepare_transformers(net, config: dict) -> pd.DataFrame:
+ """
+ Calculates transformer electrical parameters (r, x, b) based on their types
+ if they are not already defined, converts them to per-unit values based
+ on the system BasePower and v_nom, and sets default values for LEGO compatibility.
+ """
+ transformers = net.transformers.copy()
+ types = net.transformer_types
+ s_base = config.get("BasePower", 100)
+
+ # Define which values are considered "not defined" (0 or NaN)
+ r_missing = (transformers["r"] == 0) | transformers["r"].isna()
+ x_missing = (transformers["x"] == 0) | transformers["x"].isna()
+ b_missing = (transformers["b"] == 0) | transformers["b"].isna()
+
+ # Only perform type-based calculation if there are missing values
+ if r_missing.any() or x_missing.any() or b_missing.any():
+ vsc = transformers["type"].map(types["vsc"])
+ nlc = transformers["type"].map(types["i0"])
+ pfe = transformers["type"].map(types["pfe"])
+
+ # Avoid division by zero for s_nom
+ s_nom_safe = transformers.s_nom.where(transformers.s_nom != 0, 1)
+
+ if r_missing.any():
+ transformers.loc[r_missing, "r"] = vsc.loc[r_missing] / 100
+
+ if x_missing.any():
+ r_val = transformers["r"]
+ transformers.loc[x_missing, "x"] = np.sqrt(
+ np.maximum(
+ (vsc.loc[x_missing] / 100) ** 2 - r_val.loc[x_missing] ** 2, 0
+ )
+ )
+
+ if b_missing.any():
+ g = pfe / (1000 * s_nom_safe)
+ transformers.loc[b_missing, "b"] = -np.sqrt(
+ np.maximum((nlc.loc[b_missing] / 100) ** 2 - g.loc[b_missing] ** 2, 0)
+ )
+
+ # Get v_nom from buses (kV) for each transformer (using bus0 as reference)
+ v_nom = transformers.bus0.map(net.buses.v_nom)
+
+ # Calculate Z_base = V_nom^2 / S_base [Ohm]
+ z_base = (v_nom ** 2) / s_base
+
+ # Convert electrical parameters to per-unit values
+ # Transformers r, x, b in PyPSA are usually p.u. on transformer base (s_nom)
+ # To convert to system base (s_base):
+ # Z_pu_sys = Z_pu_trans * (S_base / S_trans)
+ # Y_pu_sys = Y_pu_trans * (S_trans / S_base)
+ s_nom = transformers.s_nom.where(transformers.s_nom != 0, 1)
+ scaling_factor_z = s_base / s_nom
+ scaling_factor_y = s_nom / s_base
+
+ transformers["r"] = transformers["r"] * scaling_factor_z
+ transformers["x"] = transformers["x"] * scaling_factor_z
+ transformers["b"] = transformers["b"] * scaling_factor_y
+
+ # Set carrier to AC for all transformers to get defined as DC-OPF
+ transformers["carrier"] = "AC"
+
+ ### INFO: Currently can the LEGO Database not handle similar line names (e.g. c1, c1, c2, c2, ...)
+ # 11-05-2026
+ # Ensure every transformer has an individual circuit identifier (c1, c2, ...) for parallel transformers
+ # transformers["name"] = "c" + (
+ # transformers.groupby(["bus0", "bus1"]).cumcount() + 1
+ # ).astype(str)
+
+ transformers["name"] = "Transformer_" + pd.Series(
+ range(len(transformers)), index=transformers.index
+ ).astype(str)
+
+ return transformers
+
+
+def prepare_ac_lines_and_dc_links(net, config: dict):
+ """
+ Combines AC lines, DC links, and transformers into a single DataFrame
+ for comprehensive network mapping.
+ """
+ ac_lines = prepare_ac_lines(net, config)
+ dc_links = prepare_dc_links(net, config)
+ transformers = prepare_transformers(net, config)
+ return pd.concat([ac_lines, dc_links, transformers], ignore_index=True)
+
+
+def prepare_renewable_profiles(net, config: dict):
+ """
+ Extracts renewable generation profiles (p_max_pu) for specified carriers
+ and formats them for LEGO input.
+ """
+ gens = net.generators.copy()
+ vres_gens = gens.query(config["source"]["filter"])
+ vres_ids = vres_gens.index.to_list()
+
+ profiles = net.generators_t.p_max_pu[vres_ids].copy()
+
+ # Change the index to k0001, k0002, ...
+ num_timesteps = len(profiles)
+ profiles.index = [f"k{i + 1:04d}" for i in range(num_timesteps)]
+
+ # Ensure the index (snapshots) has a known name before resetting
+ profiles = (
+ profiles.rename_axis("k")
+ .reset_index()
+ .melt(id_vars="k", var_name="generator_id", value_name="Capacity")
+ )
+ profiles["rp"] = "rp01" # add a dummy column for compatibility
+
+ return profiles # flat, column-based, no index set yet
+
+
+def prepare_inflow_profiles(net, config: dict):
+ """
+ Aggregates inflow data from hydro storage units and Run-of-River generators
+ into a unified profile format.
+ """
+ # Get hydro storage inflows
+ hydro_ids = net.storage_units.query(config["source"]["filter"]).index.to_list()
+ set_hydro_ids = set(hydro_ids)
+ set_inflow_columns = set(net.storage_units_t.inflow.columns.to_list())
+
+ # check if inflows are specified for hydro storage units
+ existing_inflows = list(set_hydro_ids & set_inflow_columns)
+ missing_inflows = list(set_hydro_ids - set_inflow_columns)
+
+ if len(existing_inflows) == 0:
+ printer.warning(
+ "No hydro storage units have inflow data. Storage inflow profiles will be empty."
+ )
+ inflow_storage = net.storage_units_t.inflow.copy()
+ else:
+ inflow_storage = net.storage_units_t.inflow[existing_inflows].copy()
+ if len(missing_inflows) > 0:
+ printer.warning(
+ f"The following hydro storage units are missing inflow data and therefore inflows will not be defined: {missing_inflows}"
+ )
+
+ # Get RoR generator inflows
+ ror_ids = net.generators.query(config["source"]["filter"]).index.to_list()
+ set_ror_ids = set(ror_ids)
+ set_ror_inflow_columns = set(net.generators_t.p_max_pu.columns.to_list())
+
+ # check if inflows are specified for RoR generators
+ existing_ror_inflows = list(set_ror_ids & set_ror_inflow_columns)
+ missing_ror_inflows = list(set_ror_ids - set_ror_inflow_columns)
+
+ if len(existing_ror_inflows) == 0:
+ printer.warning(
+ "No RoR generators have inflow data. RoR inflow profiles will be empty."
+ )
+ inflow_ror = pd.DataFrame()
+ else:
+ ror = net.generators.loc[existing_ror_inflows].copy()
+ inflow_ror = net.generators_t.p_max_pu[existing_ror_inflows].copy()
+ inflow_ror = inflow_ror.mul(ror["p_nom"], axis=1)
+ if len(missing_ror_inflows) > 0:
+ printer.warning(
+ f"The following RoR generators are missing inflow data and will not be defined: {missing_ror_inflows}"
+ )
+
+ # Concatenate both: hydro + RoR inflows → [time, generator]
+ combined = pd.concat([inflow_storage, inflow_ror], axis=1)
+
+ # sanitize combined inflow profiles for LEGO model
+ combined.fillna(
+ 0, inplace=True
+ ) # fill missing inflows with 0 (if any) to avoid NaNs in the final profiles
+
+ # Change the index to k0001, k0002, ...
+ num_timesteps = len(combined)
+ combined.index = [f"k{i + 1:04d}" for i in range(num_timesteps)]
+
+ combined = combined.T # index: generator_id, columns: time
+
+ # Convert to long format by renaming index to 'g' before resetting
+ inflow_long = (
+ combined.rename_axis("g")
+ .reset_index()
+ .melt(id_vars="g", var_name="k", value_name="Inflow")
+ )
+ inflow_long["rp"] = "rp01" # add a dummy column for compatibility
+
+ return inflow_long[["rp", "g", "k", "Inflow"]]
+
+
+def prepare_demand_profiles(net, config: dict):
+ """
+ Extracts load demand profiles and formats them into a long-form DataFrame
+ for LEGO representation.
+ """
+ df = net.loads_t.p_set.copy() # shape: [time, load_id]
+
+ # Map load IDs to bus IDs
+ df.columns = df.columns.map(net.loads.bus)
+
+ # Filter buses based on the filter defined in the config (e.g. from dPower_BusInfo)
+ bus_filter = config.get("source", {}).get("filter")
+ if bus_filter:
+ valid_buses = net.buses.query(bus_filter).index
+ # Only keep demands at valid buses
+ df = df[df.columns.intersection(valid_buses)]
+ else:
+ valid_buses = net.buses.index
+
+ # Aggregate demand per bus
+ if not df.empty:
+ df = df.groupby(level=0, axis=1).sum()
+
+ # Ensure all valid buses are present in the demand DataFrame
+ missing_buses = valid_buses.difference(df.columns)
+ if not missing_buses.empty:
+ index = df.index if not df.empty else net.snapshots
+ zero_demand = pd.DataFrame(0.0, index=index, columns=missing_buses)
+ df = pd.concat([df, zero_demand], axis=1)
+
+ # Change the index to k0001, k0002, ...
+ num_timesteps = len(df)
+ df.index = [f"k{i + 1:04d}" for i in range(num_timesteps)]
+ df = df.rename_axis("k").reset_index()
+
+ demand_long = df.melt(id_vars="k", var_name="n", value_name="Demand")
+ demand_long["rp"] = "rp01"
+ return demand_long[["rp", "n", "k", "Demand"]]
diff --git a/PyPSA/pypsa_lego_mapping_config.yaml b/PyPSA/pypsa_lego_mapping_config.yaml
new file mode 100644
index 0000000..f4a5935
--- /dev/null
+++ b/PyPSA/pypsa_lego_mapping_config.yaml
@@ -0,0 +1,378 @@
+Metadata:
+ source:
+ type: "metadata"
+ dataPackage: "PyPSA-Import" # Name of data package [text]
+ dataSource: "to be filled out" # Name of data source [text]
+ LDES_threshold: 2160 # defines the threshold in hours for storage units to be classified as LDES (Long Duration Energy Storage)
+
+ # Define names of carriers and their corresponding fuel costs (if thermal generator) for every technology
+ # [technology]:
+ # filter: list of strings to filter the carrier name in the PyPSA network
+ # cost: fuel cost in EUR/MWh (only needed for thermal generators)
+ naturalGas:
+ filter: [ 'Gas', 'gas' ]
+ cost: 100 # EUR/MWh
+ openCycleGasTurbine:
+ filter: [ 'OCGT' ]
+ cost: 100 # EUR/MWh
+ closedCycleGasTurbine:
+ filter: [ 'CCGT' ]
+ cost: 100 # EUR/MWh
+ coal:
+ filter: [ 'coal' ]
+ cost: 50 # EUR/MWh
+ hardCoal:
+ filter: [ 'Hard coal' ]
+ cost: 15 # EUR/MWh
+ brownCoal_lignite:
+ filter: [ 'Brown coal', 'lignite' ]
+ cost: 22 # EUR/MWh
+ oil:
+ filter: [ 'oil', 'Oil' ]
+ cost: 70 # EUR/MWh
+ nuclear:
+ filter: [ 'Nuclear', 'nuclear' ]
+ cost: 50 # EUR/MWh
+ biomass:
+ filter: [ 'biomass' ]
+ cost: 70 # EUR/MWh
+ geothermal:
+ filter: [ 'Geothermal', 'geothermal' ]
+ cost: 60 # EUR/MWh
+ waste:
+ filter: [ 'waste', 'Waste' ]
+ cost: 40 # EUR/MWh
+ solar:
+ filter: [ 'solar', 'Solar', 'solar-hsat' ]
+ onshoreWind:
+ filter: [ 'Wind Onshore', 'onwind', 'onshore-wind' ]
+ offshoreWind:
+ filter: [ 'Wind Offshore', 'offwind-float', 'offwind-dc', 'offwind-ac' ]
+ ror:
+ filter: [ 'ror', 'Run of River' ]
+ hydro:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'Storage Hydro', 'hydro' ]
+ phs:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'PHS', 'Pumped Hydro' ]
+
+ # Define categories of technologies and their corresponding technologies.
+ # The technologies must have the same name as specified in the mapping above.
+ ThermalGen:
+ technologies: [ 'naturalGas', 'openCycleGasTurbine', 'closedCycleGasTurbine', 'coal', 'hardCoal', 'brownCoal_lignite', 'oil', 'nuclear', 'biomass', 'geothermal', 'waste' ]
+ VRES:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind', 'ror' ]
+ Storage:
+ technologies: [ 'phs', 'hydro' ]
+ VRESProfiles:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind' ]
+ Inflows:
+ technologies: [ 'phs', 'hydro', 'ror' ]
+
+
+dData_Packages:
+ source:
+ type: "helper"
+ name: "prepare_data_packages"
+ metadata_arg: "dataPackage"
+ mapping:
+ dataPackage:
+ attr: "dataPackage"
+ comments:
+ attr: "comments"
+ comments: "to be filled out" # Additional comments, manual adjustments, etc. [text]
+
+
+dData_Sources:
+ source:
+ type: "helper"
+ name: "prepare_data_sources"
+ metadata_arg: "dataSource"
+ mapping:
+ dataSource:
+ attr: "dataSource"
+ sources:
+ attr: "sources"
+ licenseInfo:
+ attr: "licenseInfo"
+ personResponsible:
+ attr: "personResponsible"
+ lastEdited:
+ attr: "lastEdited"
+ comments:
+ attr: "comments"
+ sources: "PyPSA-Eur" # List of used sources (detailing which values are from where) [text]
+ licenseInfo: "Public" # Information about re-usability ['Public', 'Non-Public']
+ personResponsible: "to be filled out" # Person responsible that added this source and the corresponding entries [text]
+ lastEdited: "to be filled out" # Date of last edit (most likely when it was created) [DD.MM.YYYY]
+ comments: "to be filled out" # Additional comments, manual adjustments, etc. [text]
+
+
+dGlobal_Scenarios:
+ source:
+ type: "helper"
+ name: "prepare_global_scenarios"
+ mapping:
+ scenarioID:
+ attr: "scenarioID"
+ relativeWeight:
+ attr: "relativeWeight"
+ comments:
+ attr: "comments"
+ scenarioID: "ScenarioA" # Scenario name [text]
+ relativeWeight: 1.0 # Relative weight of scenario [float]
+ comments: "to be filled out" # Additional comments, manual adjustments, etc. [text]
+
+
+dPower_Hindex:
+ source:
+ type: "helper"
+ name: "prepare_power_hindex"
+ mapping:
+ p:
+ attr: "p"
+ rp:
+ attr: "rp"
+ k:
+ attr: "k"
+ rp: "rp01" # Representative period [rp]
+
+
+dPower_WeightsK:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_k"
+ mapping:
+ k:
+ attr: "k"
+ pWeight_k:
+ attr: "pWeight_k"
+ pWeight_k: 1 # Weight of representative hour within representative period [h]
+
+
+dPower_WeightsRP:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_rp"
+ mapping:
+ rp:
+ attr: "rp"
+ pWeight_rp:
+ attr: "pWeight_rp"
+ rp: "rp01" # Representative period [rp]
+ pWeight_rp: 1 # Weight of representative period within full timespan [h]
+
+
+dPower_BusInfo:
+ source:
+ type: "attribute"
+ name: "buses"
+ filter: "carrier in ['AC', 'DC']"
+ index: "i"
+ mapping:
+ z: "country"
+ pBusBaseV: "v_nom"
+ pBusMaxV: 1.1 # Fixed value used, since v_mag_pu_max is currently not used in PyPSA
+ pBusMinV: 0.9 # Fixed value used, since v_mag_pu_min is currently not used in PyPSA
+ lat: "y"
+ lon: "x"
+
+
+dPower_Network:
+ BasePower: 100
+ source:
+ type: "helper"
+ name: "prepare_ac_lines_and_dc_links"
+ index:
+ i: "bus0"
+ j: "bus1"
+ c: "name"
+ mapping:
+ c: "c1"
+ pRline: "r"
+ pXline: "x"
+ pBcline: "b"
+ pPmax: "s_nom"
+ pFOMCost:
+ attr: "fom_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pEnableInvest:
+ attr: "s_nom_extendable"
+ conversion: "bool_to_binary"
+ pInvestCost:
+ attr: "capital_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pTecRepr:
+ attr: "carrier"
+ conversion: "line_carrier_to_tec_repr" # if no carrier is specified, the technical representation is set to DC-OPF
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+ pAngle: 0
+ pRatio: 1
+
+
+dPower_ThermalGen:
+ category: "ThermalGen"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd:
+ attr: "p_min_pu"
+ conversion: "pu_to_absolute"
+ RampUp:
+ attr: "ramp_limit_up"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ RampDw:
+ attr: "ramp_limit_down"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ MinUpTime:
+ attr: "min_up_time"
+ conversion: "greater_or_equal_to_one"
+ MinDownTime:
+ attr: "min_down_time"
+ conversion: "greater_or_equal_to_one"
+ Efficiency: "efficiency"
+ FuelCost:
+ attr: "carrier"
+ conversion: "get_fuel_cost_from_metadata"
+ CommitConsumption:
+ attr: "stand_by_cost"
+ conversion: "EUR_per_hour_to_MWh_per_hour"
+ OMVarCost: "marginal_cost"
+ StartupConsumption:
+ attr: "start_up_consumption"
+ conversion: "EUR_to_MWh"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ InvestCost: "capital_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_VRES: # including ROR
+ category: "VRES"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCost: "capital_cost"
+ OMVarCost: "marginal_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Storage:
+ category: "Storage"
+ source:
+ type: "attribute"
+ name: "storage_units"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd: 0
+ MaxCons:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ DisEffic: "efficiency_dispatch"
+ ChEffic: "efficiency_store"
+ Self Discharge: "standing_loss"
+ MinReserve: 0
+ IniReserve: "state_of_charge_initial"
+ IsLDES:
+ attr: "max_hours"
+ conversion: "is_ldes"
+ OMVarCost: "marginal_cost"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCostPerMW: "capital_cost"
+ InvestCostPerMWh:
+ attr: "carrier"
+ conversion: "get_storage_capacity_cost_from_metadata"
+ Ene2PowRatio: "max_hours"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Inflows:
+ category: "Inflows"
+ source:
+ type: "helper"
+ name: "prepare_inflow_profiles"
+ index:
+ rp: "rp"
+ k: "k"
+ g: "g"
+ mapping:
+ value: "Inflow"
+
+
+dPower_Demand:
+ source:
+ type: "helper"
+ name: "prepare_demand_profiles"
+ filter: "carrier in ['AC', 'DC']" # Should be the same filter as for the buses in dPower_BusInfo
+ index:
+ rp: "rp"
+ k: "k"
+ i: "n"
+ mapping:
+ value: "Demand"
+
+
+dPower_VRESProfiles:
+ category: "VRESProfiles"
+ source:
+ type: "helper"
+ name: "prepare_renewable_profiles"
+ index:
+ rp: "rp"
+ g: "generator_id"
+ k: "k"
+ mapping:
+ value: "Capacity"
diff --git a/PyPSA/pypsa_lego_mapping_config_2023-cy2023.yaml b/PyPSA/pypsa_lego_mapping_config_2023-cy2023.yaml
new file mode 100644
index 0000000..ab04684
--- /dev/null
+++ b/PyPSA/pypsa_lego_mapping_config_2023-cy2023.yaml
@@ -0,0 +1,378 @@
+Metadata:
+ source:
+ type: "metadata"
+ dataPackage: "PyPSA-Import" # Name of data package [text]
+ dataSource: "PyPSA-Eur_2023_CY2023" # Name of data source [text]
+ LDES_threshold: 2160 # defines the threshold in hours for storage units to be classified as LDES (Long Duration Energy Storage)
+
+ # Define names of carriers and their corresponding fuel costs (if thermal generator) for every technology
+ # [technology]:
+ # filter: list of strings to filter the carrier name in the PyPSA network
+ # cost: fuel cost in EUR/MWh (only needed for thermal generators)
+ naturalGas:
+ filter: [ 'Gas', 'gas' ]
+ cost: 100 # EUR/MWh
+ openCycleGasTurbine:
+ filter: [ 'OCGT' ]
+ cost: 100 # EUR/MWh
+ closedCycleGasTurbine:
+ filter: [ 'CCGT' ]
+ cost: 100 # EUR/MWh
+ coal:
+ filter: [ 'coal' ]
+ cost: 50 # EUR/MWh
+ hardCoal:
+ filter: [ 'Hard coal' ]
+ cost: 15 # EUR/MWh
+ brownCoal_lignite:
+ filter: [ 'Brown coal', 'lignite' ]
+ cost: 22 # EUR/MWh
+ oil:
+ filter: [ 'oil', 'Oil' ]
+ cost: 70 # EUR/MWh
+ nuclear:
+ filter: [ 'Nuclear', 'nuclear' ]
+ cost: 50 # EUR/MWh
+ biomass:
+ filter: [ 'biomass' ]
+ cost: 70 # EUR/MWh
+ geothermal:
+ filter: [ 'Geothermal', 'geothermal' ]
+ cost: 60 # EUR/MWh
+ waste:
+ filter: [ 'waste', 'Waste' ]
+ cost: 40 # EUR/MWh
+ solar:
+ filter: [ 'solar', 'Solar', 'solar-hsat' ]
+ onshoreWind:
+ filter: [ 'Wind Onshore', 'onwind', 'onshore-wind' ]
+ offshoreWind:
+ filter: [ 'Wind Offshore', 'offwind-float', 'offwind-dc', 'offwind-ac' ]
+ ror:
+ filter: [ 'ror', 'Run of River' ]
+ hydro:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'Storage Hydro', 'hydro' ]
+ phs:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'PHS', 'Pumped Hydro' ]
+
+ # Define categories of technologies and their corresponding technologies.
+ # The technologies must have the same name as specified in the mapping above.
+ ThermalGen:
+ technologies: [ 'naturalGas', 'openCycleGasTurbine', 'closedCycleGasTurbine', 'coal', 'hardCoal', 'brownCoal_lignite', 'oil', 'nuclear', 'biomass', 'geothermal', 'waste' ]
+ VRES:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind', 'ror' ]
+ Storage:
+ technologies: [ 'phs', 'hydro' ]
+ VRESProfiles:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind' ]
+ Inflows:
+ technologies: [ 'phs', 'hydro', 'ror' ]
+
+
+dData_Packages:
+ source:
+ type: "helper"
+ name: "prepare_data_packages"
+ metadata_arg: "dataPackage"
+ mapping:
+ dataPackage:
+ attr: "dataPackage"
+ comments:
+ attr: "comments"
+ comments: "PyPSA-Eur dataset for target year 2023 and weather year 2023. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dData_Sources:
+ source:
+ type: "helper"
+ name: "prepare_data_sources"
+ metadata_arg: "dataSource"
+ mapping:
+ dataSource:
+ attr: "dataSource"
+ sources:
+ attr: "sources"
+ licenseInfo:
+ attr: "licenseInfo"
+ personResponsible:
+ attr: "personResponsible"
+ lastEdited:
+ attr: "lastEdited"
+ comments:
+ attr: "comments"
+ sources: "PyPSA-Eur" # List of used sources (detailing which values are from where) [text]
+ licenseInfo: "Public" # Information about re-usability ['Public', 'Non-Public']
+ personResponsible: "Orhidea Shatri" # Person responsible that added this source and the corresponding entries [text]
+ lastEdited: "31.07.2026" # Date of last edit (most likely when it was created) [DD.MM.YYYY]
+ comments: "PyPSA-Eur dataset for target year 2023 and weather year 2023. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dGlobal_Scenarios:
+ source:
+ type: "helper"
+ name: "prepare_global_scenarios"
+ mapping:
+ scenarioID:
+ attr: "scenarioID"
+ relativeWeight:
+ attr: "relativeWeight"
+ comments:
+ attr: "comments"
+ scenarioID: "ScenarioA" # Scenario name [text]
+ relativeWeight: 1.0 # Relative weight of scenario [float]
+ comments: "PyPSA-Eur 2023 CY2023" # Additional comments, manual adjustments, etc. [text]
+
+
+dPower_Hindex:
+ source:
+ type: "helper"
+ name: "prepare_power_hindex"
+ mapping:
+ p:
+ attr: "p"
+ rp:
+ attr: "rp"
+ k:
+ attr: "k"
+ rp: "rp01" # Representative period [rp]
+
+
+dPower_WeightsK:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_k"
+ mapping:
+ k:
+ attr: "k"
+ pWeight_k:
+ attr: "pWeight_k"
+ pWeight_k: 1 # Weight of representative hour within representative period [h]
+
+
+dPower_WeightsRP:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_rp"
+ mapping:
+ rp:
+ attr: "rp"
+ pWeight_rp:
+ attr: "pWeight_rp"
+ rp: "rp01" # Representative period [rp]
+ pWeight_rp: 1 # Weight of representative period within full timespan [h]
+
+
+dPower_BusInfo:
+ source:
+ type: "attribute"
+ name: "buses"
+ filter: "carrier in ['AC', 'DC']"
+ index: "i"
+ mapping:
+ z: "country"
+ pBusBaseV: "v_nom"
+ pBusMaxV: 1.1 # Fixed value used, since v_mag_pu_max is currently not used in PyPSA
+ pBusMinV: 0.9 # Fixed value used, since v_mag_pu_min is currently not used in PyPSA
+ lat: "y"
+ lon: "x"
+
+
+dPower_Network:
+ BasePower: 100
+ source:
+ type: "helper"
+ name: "prepare_ac_lines_and_dc_links"
+ index:
+ i: "bus0"
+ j: "bus1"
+ c: "name"
+ mapping:
+ c: "c1"
+ pRline: "r"
+ pXline: "x"
+ pBcline: "b"
+ pPmax: "s_nom"
+ pFOMCost:
+ attr: "fom_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pEnableInvest:
+ attr: "s_nom_extendable"
+ conversion: "bool_to_binary"
+ pInvestCost:
+ attr: "capital_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pTecRepr:
+ attr: "carrier"
+ conversion: "line_carrier_to_tec_repr" # if no carrier is specified, the technical representation is set to DC-OPF
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+ pAngle: 0
+ pRatio: 1
+
+
+dPower_ThermalGen:
+ category: "ThermalGen"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd:
+ attr: "p_min_pu"
+ conversion: "pu_to_absolute"
+ RampUp:
+ attr: "ramp_limit_up"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ RampDw:
+ attr: "ramp_limit_down"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ MinUpTime:
+ attr: "min_up_time"
+ conversion: "greater_or_equal_to_one"
+ MinDownTime:
+ attr: "min_down_time"
+ conversion: "greater_or_equal_to_one"
+ Efficiency: "efficiency"
+ FuelCost:
+ attr: "carrier"
+ conversion: "get_fuel_cost_from_metadata"
+ CommitConsumption:
+ attr: "stand_by_cost"
+ conversion: "EUR_per_hour_to_MWh_per_hour"
+ OMVarCost: "marginal_cost"
+ StartupConsumption:
+ attr: "start_up_consumption"
+ conversion: "EUR_to_MWh"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ InvestCost: "capital_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_VRES: # including ROR
+ category: "VRES"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCost: "capital_cost"
+ OMVarCost: "marginal_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Storage:
+ category: "Storage"
+ source:
+ type: "attribute"
+ name: "storage_units"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd: 0
+ MaxCons:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ DisEffic: "efficiency_dispatch"
+ ChEffic: "efficiency_store"
+ Self Discharge: "standing_loss"
+ MinReserve: 0
+ IniReserve: "state_of_charge_initial"
+ IsLDES:
+ attr: "max_hours"
+ conversion: "is_ldes"
+ OMVarCost: "marginal_cost"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCostPerMW: "capital_cost"
+ InvestCostPerMWh:
+ attr: "carrier"
+ conversion: "get_storage_capacity_cost_from_metadata"
+ Ene2PowRatio: "max_hours"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Inflows:
+ category: "Inflows"
+ source:
+ type: "helper"
+ name: "prepare_inflow_profiles"
+ index:
+ rp: "rp"
+ k: "k"
+ g: "g"
+ mapping:
+ value: "Inflow"
+
+
+dPower_Demand:
+ source:
+ type: "helper"
+ name: "prepare_demand_profiles"
+ filter: "carrier in ['AC', 'DC']" # Should be the same filter as for the buses in dPower_BusInfo
+ index:
+ rp: "rp"
+ k: "k"
+ i: "n"
+ mapping:
+ value: "Demand"
+
+
+dPower_VRESProfiles:
+ category: "VRESProfiles"
+ source:
+ type: "helper"
+ name: "prepare_renewable_profiles"
+ index:
+ rp: "rp"
+ g: "generator_id"
+ k: "k"
+ mapping:
+ value: "Capacity"
diff --git a/PyPSA/pypsa_lego_mapping_config_2025-cy2013.yaml b/PyPSA/pypsa_lego_mapping_config_2025-cy2013.yaml
new file mode 100644
index 0000000..d3d2a0a
--- /dev/null
+++ b/PyPSA/pypsa_lego_mapping_config_2025-cy2013.yaml
@@ -0,0 +1,378 @@
+Metadata:
+ source:
+ type: "metadata"
+ dataPackage: "PyPSA-Import" # Name of data package [text]
+ dataSource: "PyPSA-Eur_2025_CY2013" # Name of data source [text]
+ LDES_threshold: 2160 # defines the threshold in hours for storage units to be classified as LDES (Long Duration Energy Storage)
+
+ # Define names of carriers and their corresponding fuel costs (if thermal generator) for every technology
+ # [technology]:
+ # filter: list of strings to filter the carrier name in the PyPSA network
+ # cost: fuel cost in EUR/MWh (only needed for thermal generators)
+ naturalGas:
+ filter: [ 'Gas', 'gas' ]
+ cost: 100 # EUR/MWh
+ openCycleGasTurbine:
+ filter: [ 'OCGT' ]
+ cost: 100 # EUR/MWh
+ closedCycleGasTurbine:
+ filter: [ 'CCGT' ]
+ cost: 100 # EUR/MWh
+ coal:
+ filter: [ 'coal' ]
+ cost: 50 # EUR/MWh
+ hardCoal:
+ filter: [ 'Hard coal' ]
+ cost: 15 # EUR/MWh
+ brownCoal_lignite:
+ filter: [ 'Brown coal', 'lignite' ]
+ cost: 22 # EUR/MWh
+ oil:
+ filter: [ 'oil', 'Oil' ]
+ cost: 70 # EUR/MWh
+ nuclear:
+ filter: [ 'Nuclear', 'nuclear' ]
+ cost: 50 # EUR/MWh
+ biomass:
+ filter: [ 'biomass' ]
+ cost: 70 # EUR/MWh
+ geothermal:
+ filter: [ 'Geothermal', 'geothermal' ]
+ cost: 60 # EUR/MWh
+ waste:
+ filter: [ 'waste', 'Waste' ]
+ cost: 40 # EUR/MWh
+ solar:
+ filter: [ 'solar', 'Solar', 'solar-hsat' ]
+ onshoreWind:
+ filter: [ 'Wind Onshore', 'onwind', 'onshore-wind' ]
+ offshoreWind:
+ filter: [ 'Wind Offshore', 'offwind-float', 'offwind-dc', 'offwind-ac' ]
+ ror:
+ filter: [ 'ror', 'Run of River' ]
+ hydro:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'Storage Hydro', 'hydro' ]
+ phs:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'PHS', 'Pumped Hydro' ]
+
+ # Define categories of technologies and their corresponding technologies.
+ # The technologies must have the same name as specified in the mapping above.
+ ThermalGen:
+ technologies: [ 'naturalGas', 'openCycleGasTurbine', 'closedCycleGasTurbine', 'coal', 'hardCoal', 'brownCoal_lignite', 'oil', 'nuclear', 'biomass', 'geothermal', 'waste' ]
+ VRES:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind', 'ror' ]
+ Storage:
+ technologies: [ 'phs', 'hydro' ]
+ VRESProfiles:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind' ]
+ Inflows:
+ technologies: [ 'phs', 'hydro', 'ror' ]
+
+
+dData_Packages:
+ source:
+ type: "helper"
+ name: "prepare_data_packages"
+ metadata_arg: "dataPackage"
+ mapping:
+ dataPackage:
+ attr: "dataPackage"
+ comments:
+ attr: "comments"
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dData_Sources:
+ source:
+ type: "helper"
+ name: "prepare_data_sources"
+ metadata_arg: "dataSource"
+ mapping:
+ dataSource:
+ attr: "dataSource"
+ sources:
+ attr: "sources"
+ licenseInfo:
+ attr: "licenseInfo"
+ personResponsible:
+ attr: "personResponsible"
+ lastEdited:
+ attr: "lastEdited"
+ comments:
+ attr: "comments"
+ sources: "PyPSA-Eur" # List of used sources (detailing which values are from where) [text]
+ licenseInfo: "Public" # Information about re-usability ['Public', 'Non-Public']
+ personResponsible: "Orhidea Shatri" # Person responsible that added this source and the corresponding entries [text]
+ lastEdited: "21.07.2026" # Date of last edit (most likely when it was created) [DD.MM.YYYY]
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dGlobal_Scenarios:
+ source:
+ type: "helper"
+ name: "prepare_global_scenarios"
+ mapping:
+ scenarioID:
+ attr: "scenarioID"
+ relativeWeight:
+ attr: "relativeWeight"
+ comments:
+ attr: "comments"
+ scenarioID: "ScenarioA" # Scenario name [text]
+ relativeWeight: 1.0 # Relative weight of scenario [float]
+ comments: "PyPSA-Eur 2025 CY2013" # Additional comments, manual adjustments, etc. [text]
+
+
+dPower_Hindex:
+ source:
+ type: "helper"
+ name: "prepare_power_hindex"
+ mapping:
+ p:
+ attr: "p"
+ rp:
+ attr: "rp"
+ k:
+ attr: "k"
+ rp: "rp01" # Representative period [rp]
+
+
+dPower_WeightsK:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_k"
+ mapping:
+ k:
+ attr: "k"
+ pWeight_k:
+ attr: "pWeight_k"
+ pWeight_k: 1 # Weight of representative hour within representative period [h]
+
+
+dPower_WeightsRP:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_rp"
+ mapping:
+ rp:
+ attr: "rp"
+ pWeight_rp:
+ attr: "pWeight_rp"
+ rp: "rp01" # Representative period [rp]
+ pWeight_rp: 1 # Weight of representative period within full timespan [h]
+
+
+dPower_BusInfo:
+ source:
+ type: "attribute"
+ name: "buses"
+ filter: "carrier in ['AC', 'DC']"
+ index: "i"
+ mapping:
+ z: "country"
+ pBusBaseV: "v_nom"
+ pBusMaxV: 1.1 # Fixed value used, since v_mag_pu_max is currently not used in PyPSA
+ pBusMinV: 0.9 # Fixed value used, since v_mag_pu_min is currently not used in PyPSA
+ lat: "y"
+ lon: "x"
+
+
+dPower_Network:
+ BasePower: 100
+ source:
+ type: "helper"
+ name: "prepare_ac_lines_and_dc_links"
+ index:
+ i: "bus0"
+ j: "bus1"
+ c: "name"
+ mapping:
+ c: "c1"
+ pRline: "r"
+ pXline: "x"
+ pBcline: "b"
+ pPmax: "s_nom"
+ pFOMCost:
+ attr: "fom_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pEnableInvest:
+ attr: "s_nom_extendable"
+ conversion: "bool_to_binary"
+ pInvestCost:
+ attr: "capital_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pTecRepr:
+ attr: "carrier"
+ conversion: "line_carrier_to_tec_repr" # if no carrier is specified, the technical representation is set to DC-OPF
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+ pAngle: 0
+ pRatio: 1
+
+
+dPower_ThermalGen:
+ category: "ThermalGen"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd:
+ attr: "p_min_pu"
+ conversion: "pu_to_absolute"
+ RampUp:
+ attr: "ramp_limit_up"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ RampDw:
+ attr: "ramp_limit_down"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ MinUpTime:
+ attr: "min_up_time"
+ conversion: "greater_or_equal_to_one"
+ MinDownTime:
+ attr: "min_down_time"
+ conversion: "greater_or_equal_to_one"
+ Efficiency: "efficiency"
+ FuelCost:
+ attr: "carrier"
+ conversion: "get_fuel_cost_from_metadata"
+ CommitConsumption:
+ attr: "stand_by_cost"
+ conversion: "EUR_per_hour_to_MWh_per_hour"
+ OMVarCost: "marginal_cost"
+ StartupConsumption:
+ attr: "start_up_consumption"
+ conversion: "EUR_to_MWh"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ InvestCost: "capital_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_VRES: # including ROR
+ category: "VRES"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCost: "capital_cost"
+ OMVarCost: "marginal_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Storage:
+ category: "Storage"
+ source:
+ type: "attribute"
+ name: "storage_units"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd: 0
+ MaxCons:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ DisEffic: "efficiency_dispatch"
+ ChEffic: "efficiency_store"
+ Self Discharge: "standing_loss"
+ MinReserve: 0
+ IniReserve: "state_of_charge_initial"
+ IsLDES:
+ attr: "max_hours"
+ conversion: "is_ldes"
+ OMVarCost: "marginal_cost"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCostPerMW: "capital_cost"
+ InvestCostPerMWh:
+ attr: "carrier"
+ conversion: "get_storage_capacity_cost_from_metadata"
+ Ene2PowRatio: "max_hours"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Inflows:
+ category: "Inflows"
+ source:
+ type: "helper"
+ name: "prepare_inflow_profiles"
+ index:
+ rp: "rp"
+ k: "k"
+ g: "g"
+ mapping:
+ value: "Inflow"
+
+
+dPower_Demand:
+ source:
+ type: "helper"
+ name: "prepare_demand_profiles"
+ filter: "carrier in ['AC', 'DC']" # Should be the same filter as for the buses in dPower_BusInfo
+ index:
+ rp: "rp"
+ k: "k"
+ i: "n"
+ mapping:
+ value: "Demand"
+
+
+dPower_VRESProfiles:
+ category: "VRESProfiles"
+ source:
+ type: "helper"
+ name: "prepare_renewable_profiles"
+ index:
+ rp: "rp"
+ g: "generator_id"
+ k: "k"
+ mapping:
+ value: "Capacity"
diff --git a/PyPSA/pypsa_lego_mapping_config_2040-cy2023.yaml b/PyPSA/pypsa_lego_mapping_config_2040-cy2023.yaml
new file mode 100644
index 0000000..a9aad56
--- /dev/null
+++ b/PyPSA/pypsa_lego_mapping_config_2040-cy2023.yaml
@@ -0,0 +1,378 @@
+Metadata:
+ source:
+ type: "metadata"
+ dataPackage: "PyPSA-Import" # Name of data package [text]
+ dataSource: "PyPSA-Eur_2040_CY2023" # Name of data source [text]
+ LDES_threshold: 2160 # defines the threshold in hours for storage units to be classified as LDES (Long Duration Energy Storage)
+
+ # Define names of carriers and their corresponding fuel costs (if thermal generator) for every technology
+ # [technology]:
+ # filter: list of strings to filter the carrier name in the PyPSA network
+ # cost: fuel cost in EUR/MWh (only needed for thermal generators)
+ naturalGas:
+ filter: [ 'Gas', 'gas' ]
+ cost: 100 # EUR/MWh
+ openCycleGasTurbine:
+ filter: [ 'OCGT' ]
+ cost: 100 # EUR/MWh
+ closedCycleGasTurbine:
+ filter: [ 'CCGT' ]
+ cost: 100 # EUR/MWh
+ coal:
+ filter: [ 'coal' ]
+ cost: 50 # EUR/MWh
+ hardCoal:
+ filter: [ 'Hard coal' ]
+ cost: 15 # EUR/MWh
+ brownCoal_lignite:
+ filter: [ 'Brown coal', 'lignite' ]
+ cost: 22 # EUR/MWh
+ oil:
+ filter: [ 'oil', 'Oil' ]
+ cost: 70 # EUR/MWh
+ nuclear:
+ filter: [ 'Nuclear', 'nuclear' ]
+ cost: 50 # EUR/MWh
+ biomass:
+ filter: [ 'biomass' ]
+ cost: 70 # EUR/MWh
+ geothermal:
+ filter: [ 'Geothermal', 'geothermal' ]
+ cost: 60 # EUR/MWh
+ waste:
+ filter: [ 'waste', 'Waste' ]
+ cost: 40 # EUR/MWh
+ solar:
+ filter: [ 'solar', 'Solar', 'solar-hsat' ]
+ onshoreWind:
+ filter: [ 'Wind Onshore', 'onwind', 'onshore-wind' ]
+ offshoreWind:
+ filter: [ 'Wind Offshore', 'offwind-float', 'offwind-dc', 'offwind-ac' ]
+ ror:
+ filter: [ 'ror', 'Run of River' ]
+ hydro:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'Storage Hydro', 'hydro' ]
+ phs:
+ default_storage_capacity_cost: 1 # EUR/MWh, PyPSA does not have a separation between power and energy capacity for investment costs
+ filter: [ 'PHS', 'Pumped Hydro' ]
+
+ # Define categories of technologies and their corresponding technologies.
+ # The technologies must have the same name as specified in the mapping above.
+ ThermalGen:
+ technologies: [ 'naturalGas', 'openCycleGasTurbine', 'closedCycleGasTurbine', 'coal', 'hardCoal', 'brownCoal_lignite', 'oil', 'nuclear', 'biomass', 'geothermal', 'waste' ]
+ VRES:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind', 'ror' ]
+ Storage:
+ technologies: [ 'phs', 'hydro' ]
+ VRESProfiles:
+ technologies: [ 'solar', 'onshoreWind', 'offshoreWind' ]
+ Inflows:
+ technologies: [ 'phs', 'hydro', 'ror' ]
+
+
+dData_Packages:
+ source:
+ type: "helper"
+ name: "prepare_data_packages"
+ metadata_arg: "dataPackage"
+ mapping:
+ dataPackage:
+ attr: "dataPackage"
+ comments:
+ attr: "comments"
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dData_Sources:
+ source:
+ type: "helper"
+ name: "prepare_data_sources"
+ metadata_arg: "dataSource"
+ mapping:
+ dataSource:
+ attr: "dataSource"
+ sources:
+ attr: "sources"
+ licenseInfo:
+ attr: "licenseInfo"
+ personResponsible:
+ attr: "personResponsible"
+ lastEdited:
+ attr: "lastEdited"
+ comments:
+ attr: "comments"
+ sources: "PyPSA-Eur" # List of used sources (detailing which values are from where) [text]
+ licenseInfo: "Public" # Information about re-usability ['Public', 'Non-Public']
+ personResponsible: "Orhidea Shatri" # Person responsible that added this source and the corresponding entries [text]
+ lastEdited: "21.07.2026" # Date of last edit (most likely when it was created) [DD.MM.YYYY]
+ comments: "PyPSA-Eur dataset for target year 2025 and weather year 2013. The (fuel) costs for thermal generators defined in the Metadata section are default assumptions." # Additional comments, manual adjustments, etc. [text]
+
+
+dGlobal_Scenarios:
+ source:
+ type: "helper"
+ name: "prepare_global_scenarios"
+ mapping:
+ scenarioID:
+ attr: "scenarioID"
+ relativeWeight:
+ attr: "relativeWeight"
+ comments:
+ attr: "comments"
+ scenarioID: "ScenarioA" # Scenario name [text]
+ relativeWeight: 1.0 # Relative weight of scenario [float]
+ comments: "PyPSA-Eur 2040 CY2023" # Additional comments, manual adjustments, etc. [text]
+
+
+dPower_Hindex:
+ source:
+ type: "helper"
+ name: "prepare_power_hindex"
+ mapping:
+ p:
+ attr: "p"
+ rp:
+ attr: "rp"
+ k:
+ attr: "k"
+ rp: "rp01" # Representative period [rp]
+
+
+dPower_WeightsK:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_k"
+ mapping:
+ k:
+ attr: "k"
+ pWeight_k:
+ attr: "pWeight_k"
+ pWeight_k: 1 # Weight of representative hour within representative period [h]
+
+
+dPower_WeightsRP:
+ source:
+ type: "helper"
+ name: "prepare_power_weights_rp"
+ mapping:
+ rp:
+ attr: "rp"
+ pWeight_rp:
+ attr: "pWeight_rp"
+ rp: "rp01" # Representative period [rp]
+ pWeight_rp: 1 # Weight of representative period within full timespan [h]
+
+
+dPower_BusInfo:
+ source:
+ type: "attribute"
+ name: "buses"
+ filter: "carrier in ['AC', 'DC']"
+ index: "i"
+ mapping:
+ z: "country"
+ pBusBaseV: "v_nom"
+ pBusMaxV: 1.1 # Fixed value used, since v_mag_pu_max is currently not used in PyPSA
+ pBusMinV: 0.9 # Fixed value used, since v_mag_pu_min is currently not used in PyPSA
+ lat: "y"
+ lon: "x"
+
+
+dPower_Network:
+ BasePower: 100
+ source:
+ type: "helper"
+ name: "prepare_ac_lines_and_dc_links"
+ index:
+ i: "bus0"
+ j: "bus1"
+ c: "name"
+ mapping:
+ c: "c1"
+ pRline: "r"
+ pXline: "x"
+ pBcline: "b"
+ pPmax: "s_nom"
+ pFOMCost:
+ attr: "fom_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pEnableInvest:
+ attr: "s_nom_extendable"
+ conversion: "bool_to_binary"
+ pInvestCost:
+ attr: "capital_cost"
+ conversion: "EUR_per_MVA_to_MEUR"
+ pTecRepr:
+ attr: "carrier"
+ conversion: "line_carrier_to_tec_repr" # if no carrier is specified, the technical representation is set to DC-OPF
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+ pAngle: 0
+ pRatio: 1
+
+
+dPower_ThermalGen:
+ category: "ThermalGen"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd:
+ attr: "p_min_pu"
+ conversion: "pu_to_absolute"
+ RampUp:
+ attr: "ramp_limit_up"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ RampDw:
+ attr: "ramp_limit_down"
+ conversion: "pu_to_absolute_and_p_nom_if_nan_or_zero"
+ MinUpTime:
+ attr: "min_up_time"
+ conversion: "greater_or_equal_to_one"
+ MinDownTime:
+ attr: "min_down_time"
+ conversion: "greater_or_equal_to_one"
+ Efficiency: "efficiency"
+ FuelCost:
+ attr: "carrier"
+ conversion: "get_fuel_cost_from_metadata"
+ CommitConsumption:
+ attr: "stand_by_cost"
+ conversion: "EUR_per_hour_to_MWh_per_hour"
+ OMVarCost: "marginal_cost"
+ StartupConsumption:
+ attr: "start_up_consumption"
+ conversion: "EUR_to_MWh"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ InvestCost: "capital_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_VRES: # including ROR
+ category: "VRES"
+ source:
+ type: "attribute"
+ name: "generators"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCost: "capital_cost"
+ OMVarCost: "marginal_cost"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Storage:
+ category: "Storage"
+ source:
+ type: "attribute"
+ name: "storage_units"
+ index: "g"
+ mapping:
+ tec: "carrier"
+ i: "bus"
+ ExisUnits:
+ attr: "active"
+ conversion: "bool_to_binary_zero_if_p_nom_is_zero"
+ MaxProd:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ MinProd: 0
+ MaxCons:
+ attr: "p_nom"
+ conversion: "one_if_nan_or_zero"
+ DisEffic: "efficiency_dispatch"
+ ChEffic: "efficiency_store"
+ Self Discharge: "standing_loss"
+ MinReserve: 0
+ IniReserve: "state_of_charge_initial"
+ IsLDES:
+ attr: "max_hours"
+ conversion: "is_ldes"
+ OMVarCost: "marginal_cost"
+ EnableInvest:
+ attr: "p_nom_extendable"
+ conversion: "bool_to_binary"
+ MaxInvest:
+ attr: "p_nom_max"
+ conversion: "total_capacity_to_number_of_units"
+ InvestCostPerMW: "capital_cost"
+ InvestCostPerMWh:
+ attr: "carrier"
+ conversion: "get_storage_capacity_cost_from_metadata"
+ Ene2PowRatio: "max_hours"
+ YearCom: "build_year"
+ YearDecom:
+ attr: "decom_year"
+ conversion: "year_and_lifetime_to_year_decom"
+
+
+dPower_Inflows:
+ category: "Inflows"
+ source:
+ type: "helper"
+ name: "prepare_inflow_profiles"
+ index:
+ rp: "rp"
+ k: "k"
+ g: "g"
+ mapping:
+ value: "Inflow"
+
+
+dPower_Demand:
+ source:
+ type: "helper"
+ name: "prepare_demand_profiles"
+ filter: "carrier in ['AC', 'DC']" # Should be the same filter as for the buses in dPower_BusInfo
+ index:
+ rp: "rp"
+ k: "k"
+ i: "n"
+ mapping:
+ value: "Demand"
+
+
+dPower_VRESProfiles:
+ category: "VRESProfiles"
+ source:
+ type: "helper"
+ name: "prepare_renewable_profiles"
+ index:
+ rp: "rp"
+ g: "generator_id"
+ k: "k"
+ mapping:
+ value: "Capacity"
diff --git a/PypsaReader.py b/PypsaReader.py
deleted file mode 100644
index ea282b7..0000000
--- a/PypsaReader.py
+++ /dev/null
@@ -1,223 +0,0 @@
-import pypsa
-import pandas as pd
-import pypsa_helper as h
-import numpy as np
-import os
-
-class NetworkDataExtractor:
- def __init__(self, network: pypsa.Network):
- self.network = network
- self.columns = {
-
- "dPower_BusInfo": ['excl', 'id', 'z', 'pBusBaseV', 'pBusMaxV', 'pBusMinV', 'pBusB',
- 'pBusG', 'pBus_pf', 'YearCom', 'YearDecom', 'lat', 'lon', 'zoi',
- 'dataPackage', 'dataSource'],
- "dPower_Network": ['excl', 'id', 'pRline', 'pXline', 'pBcline', 'pAngle', 'pRatio',
- 'pPmax', 'pEnableInvest', 'pFOMCost', 'pInvestCost', 'pTecRepr',
- 'YearCom', 'YearDecom', 'dataPackage', 'dataSource'],
- "dPower_ThermalGen": ['excl', 'id', 'tec', 'i', 'ExisUnits', 'MaxProd', 'MinProd', 'RampUp',
- 'RampDw', 'MinUpTime', 'MinDownTime', 'Qmax', 'Qmin', 'InertiaConst',
- 'FuelCost', 'Efficiency', 'CommitConsumption', 'OMVarCost',
- 'StartupConsumption', 'EFOR', 'EnableInvest', 'InvestCost',
- 'FirmCapCoef', 'CO2Emis', 'YearCom', 'YearDecom', 'lat', 'long',
- 'dataPackage', 'dataSource', 'pSlopeVarCostEUR', 'pInterVarCostEUR',
- 'pStartupCostEUR', 'MaxInvest', 'InvestCostEUR'],
- "dPower_VRESProfiles": ['Capacity'],
- "dPower_VRES": ['excl', 'id', 'tec', 'i', 'ExisUnits', 'MaxProd', 'EnableInvest',
- 'MaxInvest', 'InvestCost', 'OMVarCost', 'FirmCapCoef', 'Qmax', 'Qmin',
- 'InertiaConst', 'YearCom', 'YearDecom', 'lat', 'lon', 'dataPackage',
- 'dataSource', 'MinProd', 'InvestCostEUR'],
- "dPower_Storage": ['tec', 'i', 'ExisUnits', 'MaxProd', 'MinProd', 'MaxCons', 'DisEffic',
- 'ChEffic', 'Qmax', 'Qmin', 'InertiaConst', 'MinReserve', 'IniReserve',
- 'IsHydro', 'OMVarCost', 'EnableInvest', 'MaxInvest', 'InvestCostPerMW',
- 'InvestCostPerMWh', 'Ene2PowRatio', 'ReplaceCost', 'ShelfLife',
- 'FirmCapCoef', 'CDSF_alpha', 'CDSF_beta', 'PPName', 'YearCom',
- 'YearDecom', 'lat', 'long', 'pOMVarCostEUR', 'InvestCostEUR'],
- "dPower_RoR": ['tec', 'i', 'ExisUnits', 'MaxProd', 'MinProd', 'MaxCons', 'DisEffic',
- 'ChEffic', 'Qmax', 'Qmin', 'InertiaConst', 'MinReserve', 'IniReserve',
- 'IsHydro', 'OMVarCost', 'EnableInvest', 'MaxInvest', 'InvestCostPerMW',
- 'InvestCostPerMWh', 'Ene2PowRatio', 'ReplaceCost', 'ShelfLife',
- 'FirmCapCoef', 'CDSF_alpha', 'CDSF_beta', 'PPName', 'YearCom',
- 'YearDecom', 'lat', 'long', 'InvestCostEUR'],
- "dPower_Demand": ['Capacity'],
- "dPower_Inflows": ['Inflow'],
- }
-
-
- self.component_definitions = {
- "dPower_BusInfo": {
- "source": lambda net: net.buses[net.buses["carrier"] == "AC"],
- "index": lambda Buses: Buses.index.rename("i"),
- "z": lambda Buses: Buses["country"] ,
- "pBusBaseV": lambda Buses: Buses["v_nom"],
- "pBusMaxV": lambda Buses: 1.1,
- "pBusMinV": lambda Buses: 0.9,
- "lat": lambda Buses: Buses["y"],
- "lon": lambda Buses: Buses["x"],
- },
- "dPower_Network": {
- "source": lambda net: pd.concat([h.prepare_ac_lines(net),
- h.prepare_dc_links(net)
- ], ignore_index=True),
- "index": lambda df: pd.MultiIndex.from_frame(
- df[["bus0", "bus1", "name"]].rename(columns={"bus0": "i", "bus1": "j", "name": "c"})
- ).set_names(["i", "j", "c"]),
- "pRline": lambda df: df["r"],
- "pXline": lambda df: df["x"],
- "pBcline": lambda df: df["b"],
- "pMax": lambda df: df["pmax"]
-
- },
- "dPower_ThermalGen": {
- "source": lambda net: h.prepare_thermal_generators(net),
- "index": lambda df: df["id"].rename("g"),
- "tec": lambda df: df["carrier"],
- "i": lambda df: df["bus"],
- "MaxProd": lambda df: df["max_prod"],
- "MinProd": lambda df: df["min_prod"],
- "RampUp": lambda df: df["ramp_up"],
- "RampDown": lambda df: df["ramp_down"],
- "pStartupCostEUR": lambda df: df["start_up_cost"],
- "EnableInvest": lambda df: df["enable_invest"],
- "InvestCost": lambda df: df["capital_cost"],
- "OMVarCost": lambda df: df["marginal_cost"],
- },
- "dPower_VRESProfiles": {
- "source": lambda net: h.prepare_renewable_profiles(net),
- "Capacity": lambda df: df["Capacity"],
-
- "index": lambda df: pd.MultiIndex.from_frame(
- df[["generator_id", "snapshot"]].rename(columns={"generator_id": "g", "snapshot": "k"}))
- },
- "dPower_VRES": {
- "source": lambda net: h.prepare_renewable_generators(net),
- "index": lambda df: df["id"].rename("g"),
- "tec": lambda df: df["carrier"],
- "i": lambda df: df["bus"],
- "MaxProd": lambda df: df["max_prod"],
- "enableinvest": lambda df: df["enable_invest"],
- "MaxInvest": lambda df: df["p_nom_max"],
- "InvestCost": lambda df: df["capital_cost"],
- "OMVarCost": lambda df: df["marginal_cost"]
- },
- "dPower_RoR": {
- "source": lambda net: h.prepare_ror_generators(net),
- "tec": lambda df: df["carrier"],
- "index": lambda df: df["id"].rename("g"),
- "i": lambda df: df["bus"],
- "MaxProd": lambda df: df["max_prod"],
- "MinProd": lambda df: df["min_prod"],
- "DisEffic": lambda df: df["discharge"],
- "IsHydro": lambda df: df["is_hydro"],
- "OMVarCost": lambda df: df["marginal_cost"],
- "EnableInvest": lambda df: df["enable_invest"],
- "MaxInvest": lambda df: df["p_nom_max"],
- "InvestCostPerMW": lambda df: df["capital_cost"]
- },
- "dPower_Storage": {
- "source": lambda net: h.prepare_storage_units(net),
- "index": lambda df: df["id"].rename("g"),
- "tec": lambda df: df["carrier"],
- "i": lambda df: df["bus"],
- "MaxProd": lambda df: df["max_prod"],
- "MinProd": lambda df: df["min_prod"],
- "DisEffic": lambda df: df["discharge"],
- "ChEffic": lambda df: df["charge"],
- "IniReserve": lambda df: df["ini_reserve"],
- "IsHydro": lambda df: df["is_hydro"],
- "OMVarCost": lambda df: df["marginal_cost"],
- "EnableInvest": lambda df: df["enable_invest"],
- "MaxInvest": lambda df: df["p_nom_max"],
- "InvestCostPerMWh": lambda df: df["capital_cost"],
- "Ene2PowRatio": lambda df: df["max_hours"],
- "ShelfLife": lambda df: df["lifetime"]
- },
- "dPower_Inflows": {
- "source": lambda net: h.prepare_inflow_profiles(net),
- "rp": lambda df: df["rp"],
- "g": lambda df: df["g"],
- "k": lambda df: df["k"],
- "Inflow": lambda df: df["Inflow"],
- "index": lambda df: pd.MultiIndex.from_frame(df[["rp","k", "g"]])
- },
- "dPower_Demand": {
- "source": lambda net: h.prepare_demand_profiles(net),
- "rp": lambda df: df["rp"],
- "g": lambda df: df["g"],
- "k": lambda df: df["k"],
- "Demand": lambda df: df["Demand"],
- "index": lambda df: pd.MultiIndex.from_frame(df[["rp", "k", "g"]])
- }
- }
-
- self.dataframes = self._extract_dataframes()
- # add empty columns
- self.dataframes = self._add_empty_columns()
- # reorder columns
- self.dataframes = self._reorder_columns()
-
- def _extract_dataframes(self):
- df_dict = {}
-
- for name, config in self.component_definitions.items():
- source_df = config["source"](self.network)
-
- column_data = {
- column_name: transform(source_df)
- for column_name, transform in config.items()
- if column_name not in ("source", "index")
- }
-
- df = pd.DataFrame(column_data)
-
- # Set custom index if defined
- if "index" in config:
- index_values = config["index"](source_df)
-
- df.index = index_values
-
- # Drop the columns used in the index if they exist in the DataFrame
- if isinstance(index_values, pd.MultiIndex):
- df = df.drop(columns=[col for col in index_values.names if col in df.columns], errors="ignore")
- elif isinstance(index_values, pd.Index):
- if index_values.name in df.columns:
- df = df.drop(columns=[index_values.name], errors="ignore")
-
- # df.index.name = None # Optional: remove index name
-
-
- else:
- df = df.reset_index(drop=True)
-
- df_dict[name] = df
-
- return df_dict
-
- def _add_empty_columns(self):
- for name, df in self.dataframes.items():
- for col in self.columns[name]:
- if col not in df.columns:
- df[col] = np.nan
- return self.dataframes
-
- def _reorder_columns(self):
- for name, df in self.dataframes.items():
- if name in self.columns:
- cols = self.columns[name]
- # Reorder the DataFrame columns
- df = df.reindex(columns=cols)
- # Update the DataFrame in the dictionary
- self.dataframes[name] = df
- return self.dataframes
-
- def get_dataframes(self):
- return self.dataframes
-
-filepath = os.path.join(os.path.dirname(__file__), "..", "pypsa-eur/resources/test/networks/base_s_39_elec_1year.nc")
-net = pypsa.Network(filepath)
-extractor = NetworkDataExtractor(net)
-dfs = extractor.get_dataframes()
-for name, df in dfs.items():
- print(f"DataFrame: {name}")
- print(df.head())
- print("\n")
diff --git a/README.md b/README.md
index 6c553be..2fd18ec 100644
--- a/README.md
+++ b/README.md
@@ -1 +1,7 @@
-# InOutModule
\ No newline at end of file
+# InOutModule
+
+This repository contains the LEGO input/output utilities.
+
+PyPSA-related files are located in the folder [`PyPSA`](PyPSA). This folder contains the PyPSA reader, helper functions,
+the mapping configuration, and a [`README.md`](PyPSA/README.md) with documentation and instructions for translating a
+PyPSA-Eur dataset to LEGO Excel files.
diff --git a/environment.yml b/environment.yml
index c066c5b..b204fbc 100644
--- a/environment.yml
+++ b/environment.yml
@@ -15,5 +15,7 @@ dependencies:
- rich=13.9.4
- rich-argparse=1.7.1
- tsam=2.3.9
+ - pypsa=1.1.2
+ - pyyaml=6.0.3
- pip:
- pulp==2.9.0
diff --git a/pyproject.toml b/pyproject.toml
index 25294be..48a3406 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,7 +8,11 @@ version = "0.0.0-dev"
readme = "README.md"
[tool.setuptools]
-py-modules = ['CaseStudy', 'ExcelReader', 'ExcelWriter', 'printer', 'PypsaReader', 'pypsa_helper', 'SQLiteWriter', 'TableDefinition']
+py-modules = ['CaseStudy', 'ExcelReader', 'ExcelWriter', 'printer', 'SQLiteWriter', 'TableDefinition']
+packages = ['PyPSA']
+
+[tool.setuptools.package-data]
+PyPSA = ['pypsa_lego_mapping_config.yaml', 'README.md']
[tool.pytest.ini_options]
pythonpath = [".."]
diff --git a/pypsa_helper.py b/pypsa_helper.py
deleted file mode 100644
index 1fe31d9..0000000
--- a/pypsa_helper.py
+++ /dev/null
@@ -1,146 +0,0 @@
-import pandas as pd
-
-def prepare_ac_lines(net):
- lines = net.lines.copy()
- types = net.line_types
-
- lines["r"] = lines.apply(
- lambda row: row["r"] if row["r"] != 0 else types.loc[row["type"]].r_per_length * row["length"], axis=1)
- lines["x"] = lines.apply(
- lambda row: row["x"] if row["x"] != 0 else types.loc[row["type"]].x_per_length * row["length"], axis=1)
- lines["b"] = lines.apply(
- lambda row: row["b"] if row["b"] != 0 else types.loc[row["type"]].x_per_length * row["length"], axis=1)
-
- lines["pmax"] = lines["s_nom"] * lines["s_max_pu"]
- lines["id"] = lines.index
- lines["name"] = [f"Line_{i}" for i in range(len(lines))]
-
- return lines[["bus0", "bus1", "r", "x", "b", "pmax", "id", "name"]]
-
-def prepare_dc_links(net):
- links = net.links[net.links["carrier"] == "DC"].copy()
- links["r"] = 0.0
- links["x"] = 0.0
- links["b"] = 0.0
- links["pmax"] = links["p_nom"]
- links["id"] = links.index
- links["name"] = [f"DC_Link_{i}" for i in range(len(links))]
-
- return links[["bus0", "bus1", "r", "x", "b", "pmax", "id", "name"]]
-
-def prepare_thermal_generators(net):
- thermal_types = ['OCGT', 'biomass', 'CCGT', 'nuclear', 'oil', 'coal', 'lignite']
- gens = net.generators.copy()
- gens = gens[gens.carrier.isin(thermal_types)]
-
- gens["max_prod"] = gens["p_max_pu"] * gens["p_nom"]
- gens["min_prod"] = gens["p_min_pu"] * gens["p_nom"]
- gens["ramp_up"] = gens["ramp_limit_up"] * gens["p_nom"]
- gens["ramp_down"] = gens["ramp_limit_down"] * gens["p_nom"]
- gens["enable_invest"] = gens["p_nom_extendable"].astype(int)
-
- gens["id"] = gens.index
- return gens[[
- "id", "carrier", "bus", "max_prod", "min_prod",
- "ramp_up", "ramp_down", "start_up_cost",
- "enable_invest", "capital_cost", "marginal_cost"
- ]]
-
-def prepare_renewable_profiles(net):
- renewable_types = ['solar-hsat', 'onwind', 'solar']
- gens = net.generators.copy()
- vres_gens = gens[gens.carrier.isin(renewable_types)]
- vres_ids = vres_gens.index.to_list()
-
- profiles = net.generators_t.p_max_pu[vres_ids].copy()
- profiles = profiles.reset_index().melt(
- id_vars="snapshot", var_name="generator_id", value_name="Capacity"
- )
- profiles = profiles.rename(columns={"index": "k"})
- profiles["rp"] = "rp01" # add a dummy column for compatibility
-
- return profiles # flat, column-based, no index set yet
-
-def prepare_renewable_generators(net):
- renewable_types = ['solar-hsat', 'onwind', 'solar']
- gens = net.generators.copy()
- vres = gens[gens.carrier.isin(renewable_types)].copy()
-
- vres["max_prod"] = vres["p_max_pu"] * vres["p_nom"]
- vres["enable_invest"] = vres["p_nom_extendable"].astype(int)
-
- vres["id"] = vres.index.values
- return vres[[
- "id", "carrier", "bus", "max_prod",
- "enable_invest", "p_nom_max", "capital_cost", "marginal_cost"
- ]]
-
-def prepare_ror_generators(net):
- ror = net.generators[net.generators.carrier == "ror"].copy()
-
- ror["id"] = ror.index
- ror["max_prod"] = ror["p_max_pu"] * ror["p_nom"]
- ror["min_prod"] = ror["p_min_pu"] * ror["p_nom"]
- ror["discharge"] = ror["efficiency"]
- ror["is_hydro"] = 1
- ror["enable_invest"] = ror["p_nom_extendable"].astype(int)
-
- return ror[[
- "id", "carrier", "bus", "max_prod", "min_prod", "discharge", "is_hydro",
- "marginal_cost", "enable_invest", "p_nom_max", "capital_cost"
- ]]
-
-def prepare_storage_units(net):
- su = net.storage_units.copy()
-
- su["id"] = su.index
- su["max_prod"] = su["p_nom"] * su["p_max_pu"]
- su["min_prod"] = su["p_nom"] * su["p_min_pu"] # note: often negative
- su["discharge"] = su["efficiency_dispatch"]
- su["charge"] = su["efficiency_store"]
- su["ini_reserve"] = su["state_of_charge_initial"]
- su["is_hydro"] = su["carrier"].isin(["PHS", "hydro"]).astype(int)
- su["enable_invest"] = su["p_nom_extendable"].astype(int)
-
- # Note: "min_reserve" is not present in PyPSA by default — we'll skip it
- return su[[
- "id", "carrier", "bus", "max_prod", "min_prod", "discharge", "charge",
- "ini_reserve", "is_hydro", "marginal_cost", "enable_invest", "p_nom_max",
- "capital_cost", "max_hours", "lifetime"
- ]]
-
-def prepare_inflow_profiles(net):
- # Get hydro storage inflows
- hydro_ids = net.storage_units[net.storage_units["carrier"] == "hydro"].index.to_list()
- inflow_storage = net.storage_units_t.inflow[hydro_ids].copy()
-
- # Get RoR generator inflows
- ror = net.generators[net.generators.carrier == "ror"]
- ror_ids = ror.index.to_list()
- p_nom = ror["p_nom"]
-
- inflow_ror = net.generators_t.p_max_pu[ror_ids].copy()
- inflow_ror = inflow_ror.mul(p_nom, axis=1)
-
- # Concatenate both: hydro + RoR inflows → [time, generator]
- combined = pd.concat([inflow_storage, inflow_ror], axis=1)
- combined = combined.T # index: generator_id, columns: time
-
- # Convert to long format
- inflow_long = combined.reset_index().melt(
- id_vars="index", var_name="k", value_name="Inflow"
- )
- inflow_long = inflow_long.rename(columns={"index": "g"})
- inflow_long["rp"] = "rp01" # add a dummy column for compatibility
-
- return inflow_long[["rp", "g", "k", "Inflow"]]
-
-def prepare_demand_profiles(net):
- df = net.loads_t.p_set.copy() # shape: [time, load_id]
- df = df.rename_axis("k").reset_index() # 'k' = time
-
- demand_long = df.melt(id_vars="k", var_name="g", value_name="Demand")
- demand_long["rp"] = "rp01"
- return demand_long[["rp", "g", "k", "Demand"]]
-
-