scenario_packer

scenario_packer

Scenario packing utilities for creating multiple scenario variations.

Classes

ScenarioPacker

Bases: BaseModel

Packs one or multiple scenarios for export to dataframes or excel

Functions

add

add(*scenarios)

Add scenarios to all packs. Supports Sessions and Scenarios

Source code in src/pyetm/models/scenario_packer.py

def add(self, *scenarios: Any) -> None:
    """Add scenarios to all packs. Supports Sessions and Scenarios"""
    self.add_custom_curves(*scenarios)
    self.add_inputs(*scenarios)
    self.add_sortables(*scenarios)
    self.add_hourly_output_curves(*scenarios)
    self.add_annual_exports(*scenarios)
    self._query_pack.add(*scenarios)
    self._users.add(*scenarios)

main_info

main_info()

Create main info DataFrame with scenarios as rows.

Returns DataFrame with scenarios as rows. The scenario_id column contains scenario identifiers, and the DataFrame index is set to match for proper Excel export. The session_id and saved_scenario_id columns from _to_dataframe() provide the export-friendly IDs.

Source code in src/pyetm/models/scenario_packer.py

def main_info(self) -> pd.DataFrame:
    """
    Create main info DataFrame with scenarios as rows.

    Returns DataFrame with scenarios as rows. The scenario_id column contains
    scenario identifiers, and the DataFrame index is set to match for proper
    Excel export. The session_id and saved_scenario_id columns from
    _to_dataframe() provide the export-friendly IDs.
    """
    scenarios = self._scenarios()
    if not scenarios:
        return pd.DataFrame()
    df = pd.concat([scenario._to_dataframe() for scenario in scenarios], axis=1)
    # Transpose and create scenario_id column from the index
    result = df.T.reset_index(names=["scenario_id"])
    # Set the index to scenario_id for proper Excel row labels, while keeping the column
    result.index = result["scenario_id"]
    result.index.name = None
    return result

custom_curves

custom_curves(as_dict=False, curves=None)

Get custom curves for all scenarios.

Parameters:

Name	Type	Description	Default
as_dict	bool	If True, returns dict[curve_name, dict[scenario_id, DataFrame]]. If False (default), returns concatenated DataFrame for backward compatibility.	False
curves	Optional[Sequence[str]]	Optional filter for specific curve names (only used when as_dict=True)	None

Returns:

Type	Description
DataFrame | dict[str, dict[str, DataFrame]]	DataFrame (default) or dict depending on as_dict parameter

Source code in src/pyetm/models/scenario_packer.py

def custom_curves(
    self, as_dict: bool = False, curves: Optional[Sequence[str]] = None
) -> pd.DataFrame | dict[str, dict[str, pd.DataFrame]]:
    """
    Get custom curves for all scenarios.

    Args:
        as_dict: If True, returns dict[curve_name, dict[scenario_id, DataFrame]].
                 If False (default), returns concatenated DataFrame for backward compatibility.
        curves: Optional filter for specific curve names (only used when as_dict=True)

    Returns:
        DataFrame (default) or dict depending on as_dict parameter
    """
    if as_dict:
        return self._custom_curves.to_dict_per_curve(curves=curves)
    return self._custom_curves.to_dataframe()

hourly_output_curves

hourly_output_curves(curves=None, carrier_type=None)

Get hourly output curves for all scenarios, organized by curve name.

Parameters:

Name	Type	Description	Default
curves	Optional[Sequence[str]]	Specific curve names to retrieve. If provided, carrier_type is ignored.	None
carrier_type	Optional[CarrierType]	Carrier type to get all associated curves for. One of: "electricity", "heat", "hydrogen", "methane"	None

Returns:

Type	Description
dict[str, dict[str, DataFrame]]	Dict mapping curve names to dicts of {scenario_title: DataFrame}

Note

For concatenated DataFrame format, use: packer._hourly_output_curves.to_dataframe(curves)

Source code in src/pyetm/models/scenario_packer.py

def hourly_output_curves(
    self,
    curves: Optional[Sequence[str]] = None,
    carrier_type: Optional[CarrierType] = None,
) -> dict[str, dict[str, pd.DataFrame]]:
    """
    Get hourly output curves for all scenarios, organized by curve name.

    Args:
        curves: Specific curve names to retrieve. If provided, carrier_type is ignored.
        carrier_type: Carrier type to get all associated curves for.
                     One of: "electricity", "heat", "hydrogen", "methane"

    Returns:
        Dict mapping curve names to dicts of {scenario_title: DataFrame}

    Note:
        For concatenated DataFrame format, use: packer._hourly_output_curves.to_dataframe(curves)
    """
    self._validate_curve_params(curves, carrier_type)
    curve_names = self._resolve_curve_names(curves, carrier_type)
    return self._hourly_output_curves.to_dict_per_curve(curves=curve_names)

annual_exports

annual_exports(exports=None)

Get annual exports for all scenarios, organized by export type.

Source code in src/pyetm/models/scenario_packer.py

def annual_exports(
    self,
    exports: Optional[AnnualExportType | Sequence[AnnualExportType]] = None,
) -> dict[str, dict[str, pd.DataFrame]]:
    """
    Get annual exports for all scenarios, organized by export type.
    """
    validated_exports: Optional[List[str]] = None
    if exports is not None:
        # If exports is a string, validate_export_names expects it directly
        # If exports is a Sequence (but not str), pass it directly too
        validated_exports = validate_export_names(cast("str | list[str]", exports))
    return self._annual_exports.to_dict_per_export(exports=validated_exports)

collect_export_data

collect_export_data(*, hourly_curves=None, include_inputs=None, include_sortables=None, include_custom_curves=None, include_gqueries=None, include_hourly_curves=None, include_input_defaults=None, include_input_min_max=None, include_users=None, include_annual_exports=None)

Collect export data in format-agnostic structure.

Parameters:

Name	Type	Description	Default
hourly_curves	Optional[Sequence[str]]	Carrier types for hourly output curves (e.g., ["electricity", "heat"]). Only used if include_hourly_curves is True.	None
include_inputs	Optional[bool]	Include input parameters. Defaults based on scenario export config.	None
include_sortables	Optional[bool]	Include sortable technologies. Defaults to False.	None
include_custom_curves	Optional[bool]	Include custom curves. Defaults to False.	None
include_gqueries	Optional[bool]	Include query results. Defaults to False.	None
include_hourly_curves	Optional[bool]	Include hourly output curves. Defaults to False.	None
include_input_defaults	Optional[bool]	Include default values for inputs. Defaults to False.	None
include_input_min_max	Optional[bool]	Include min/max bounds for inputs. Defaults to False.	None
include_users	Optional[bool]	Include user permissions. Defaults to False.	None
include_annual_exports	Optional[Sequence[str]]	List of annual export names to include. Examples: ["energy_flow", "sankey", "production_parameters"] Defaults to None (no annual exports).	None

Returns:

Type	Description
ExportDataCollection	ExportDataCollection containing all requested data in format-agnostic structures

Source code in src/pyetm/models/scenario_packer.py

def collect_export_data(
    self,
    *,
    hourly_curves: Optional[Sequence[str]] = None,
    include_inputs: Optional[bool] = None,
    include_sortables: Optional[bool] = None,
    include_custom_curves: Optional[bool] = None,
    include_gqueries: Optional[bool] = None,
    include_hourly_curves: Optional[bool] = None,
    include_input_defaults: Optional[bool] = None,
    include_input_min_max: Optional[bool] = None,
    include_users: Optional[bool] = None,
    include_annual_exports: Optional[Sequence[str]] = None,
) -> ExportDataCollection:
    """
    Collect export data in format-agnostic structure.

    Args:
        hourly_curves: Carrier types for hourly output curves (e.g., ["electricity", "heat"]).
                      Only used if include_hourly_curves is True.
        include_inputs: Include input parameters. Defaults based on scenario export config.
        include_sortables: Include sortable technologies. Defaults to False.
        include_custom_curves: Include custom curves. Defaults to False.
        include_gqueries: Include query results. Defaults to False.
        include_hourly_curves: Include hourly output curves. Defaults to False.
        include_input_defaults: Include default values for inputs. Defaults to False.
        include_input_min_max: Include min/max bounds for inputs. Defaults to False.
        include_users: Include user permissions. Defaults to False.
        include_annual_exports: List of annual export names to include.
                               Examples: ["energy_flow", "sankey", "production_parameters"]
                               Defaults to None (no annual exports).

    Returns:
        ExportDataCollection containing all requested data in format-agnostic structures
    """
    if not self._scenarios():
        raise ValueError("Packer was empty, nothing to export")

    global_config = self._get_global_export_config()
    resolved_flags = self._resolve_export_flags(
        global_config,
        include_inputs,
        include_sortables,
        include_custom_curves,
        include_gqueries,
        include_hourly_curves,
        include_input_defaults,
        include_input_min_max,
        include_users,
        include_annual_exports,
        hourly_curves,
    )

    hourly_curve_carriers = self._determine_hourly_curve_carriers(
        hourly_curves, resolved_flags, global_config
    )
    config = self._build_export_config(resolved_flags, hourly_curve_carriers)
    collected_data = self._collect_data_by_flags(
        resolved_flags, hourly_curve_carriers
    )

    return ExportDataCollection(
        main_info=self.main_info(),
        inputs=collected_data.get("inputs"),
        inputs_detailed=collected_data.get("inputs_detailed"),
        sortables=collected_data.get("sortables"),
        custom_curves=collected_data.get("custom_curves"),
        hourly_output_curves=collected_data.get("hourly_output_curves"),
        annual_exports=collected_data.get("annual_exports"),
        gquery_results=collected_data.get("gquery_results"),
        users=collected_data.get("users"),
        config=config,
    )

to_excel

to_excel(path, *, hourly_curves=None, include_inputs=None, include_sortables=None, include_custom_curves=None, include_gqueries=None, include_hourly_curves=None, include_input_defaults=None, include_input_min_max=None, include_users=None, include_annual_exports=None)

Export scenarios to Excel file.

Source code in src/pyetm/models/scenario_packer.py

def to_excel(
    self,
    path: str,
    *,
    hourly_curves: Optional[Sequence[str]] = None,
    include_inputs: Optional[bool] = None,
    include_sortables: Optional[bool] = None,
    include_custom_curves: Optional[bool] = None,
    include_gqueries: Optional[bool] = None,
    include_hourly_curves: Optional[bool] = None,
    include_input_defaults: Optional[bool] = None,
    include_input_min_max: Optional[bool] = None,
    include_users: Optional[bool] = None,
    include_annual_exports: Optional[Sequence[str]] = None,
) -> None:
    """Export scenarios to Excel file."""
    from pyetm.exporters.excel_exporter import ExcelExporter

    export_data = self.collect_export_data(
        hourly_curves=hourly_curves,
        include_inputs=include_inputs,
        include_sortables=include_sortables,
        include_custom_curves=include_custom_curves,
        include_gqueries=include_gqueries,
        include_hourly_curves=include_hourly_curves,
        include_input_defaults=include_input_defaults,
        include_input_min_max=include_input_min_max,
        include_users=include_users,
        include_annual_exports=include_annual_exports,
    )

    ExcelExporter.write(
        export_data=export_data, path=path, scenarios=list(self._scenarios())
    )

from_excel classmethod

from_excel(xlsx_path, update=False)

Import scenarios from Excel file.

Uses per-row 'session' column to determine loader type for each scenario.

Parameters:

Name	Type	Description	Default
xlsx_path	PathLike[str] | str	Path to Excel file	required
update	bool | List[str]	Whether to upload data to ETM. Can be: - False (default): Load data locally without uploading - True: Upload all data types (user_values, custom_curves, sortables, users) - List of types: Upload only specified types (e.g., ['user_values', 'users'])	False

Source code in src/pyetm/models/scenario_packer.py

@classmethod
def from_excel(
    cls, xlsx_path: PathLike[str] | str, update: bool | List[str] = False
) -> "ScenarioPacker":
    """
    Import scenarios from Excel file.

    Uses per-row 'session' column to determine loader type for each scenario.

    Args:
        xlsx_path: Path to Excel file
        update: Whether to upload data to ETM. Can be:
            - False (default): Load data locally without uploading
            - True: Upload all data types (user_values, custom_curves, sortables, users)
            - List of types: Upload only specified types (e.g., ['user_values', 'users'])
    """
    packer = cls()
    update_set = cls._normalize_update(update)
    path = cls._resolve_excel_path(xlsx_path)

    excel_file = cls._open_excel_file(path, xlsx_path)
    if excel_file is None:
        return packer

    main_df = packer._import_main_sheet(excel_file)
    if main_df is None:
        return packer

    scenarios_by_column = packer._create_scenarios_from_main(main_df, update_set)
    if not scenarios_by_column:
        return packer

    export_config_df = packer._load_export_config_sheet(excel_file)
    if export_config_df is None:
        return packer

    packer._apply_export_configuration(
        main_df, scenarios_by_column, export_config_df
    )
    packer._import_all_sheets(excel_file, main_df, scenarios_by_column, update_set)

    return packer

clear

clear()

Clear all scenarios from all packs.

Source code in src/pyetm/models/scenario_packer.py

def clear(self) -> None:
    """Clear all scenarios from all packs."""
    for pack in self._packs():
        try:
            pack.clear()
        except Exception:
            pass

remove_scenario

remove_scenario(scenario)

Remove a specific scenario from all collections.

Source code in src/pyetm/models/scenario_packer.py

def remove_scenario(self, scenario: Session) -> None:
    """Remove a specific scenario from all collections."""
    for pack in self._packs():
        try:
            pack.discard(scenario)
        except Exception:
            pass

Functions