scenarios

scenarios

Collection and bulk operations for scenarios.

Classes

ScenarioCreationParams

Bases: TypedDict

Type definition for create_many parameter dicts.

Note: template_id must be a Session ID (ETEngine), not a SavedScenario ID (MyETM).

Scenarios

Scenarios(**data)

Bases: Base

A collection of SavedScenario and/or Session objects.

Can hold both Scenario and Session objects to support mixed collections loaded from Excel or other sources.

Warnings from bulk operations are collected in the inherited _warning_collector.

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

def __init__(self, **data: Any) -> None:
    """
    Initialize the model, converting validation errors to warnings.
    """
    super(BaseModel, self).__setattr__("__pydantic_private__", {})

    # Initialize all private attributes with their defaults
    private_dict: Dict[str, Any] = self.__pydantic_private__  # type: ignore[assignment]
    for attr_name, attr_info in self.__class__.__private_attributes__.items():
        if (
            hasattr(attr_info, "default_factory")
            and attr_info.default_factory is not None
        ):
            # Call factory - signature varies between pydantic versions
            private_dict[attr_name] = attr_info.default_factory()
        elif hasattr(attr_info, "default"):
            private_dict[attr_name] = attr_info.default
        else:
            private_dict[attr_name] = None

    try:
        super().__init__(**data)
    except ValidationError as e:
        # Check if data is None or empty - this indicates API error
        if not data or data is None:
            # Re-raise with clearer message
            raise ValueError(
                f"Cannot create {self.__class__.__name__} with empty data. "
                "This usually indicates an authentication or API error."
            ) from e

        # If validation fails, create model without validation and collect warnings
        # Use model_construct to bypass validation
        temp_instance = self.__class__.model_construct(**data)

        # Copy the constructed data to this instance
        for field_name, field_value in temp_instance.__dict__.items():
            if not field_name.startswith("_"):
                object.__setattr__(self, field_name, field_value)

        # Ensure required Pydantic slot attributes exist to prevent AttributeError
        for slot in ("__pydantic_fields_set__", "__pydantic_extra__"):
            try:
                value = object.__getattribute__(temp_instance, slot)
                object.__setattr__(self, slot, value)
            except AttributeError:
                # Initialize missing slot attributes with defaults
                if slot == "__pydantic_extra__":
                    object.__setattr__(self, slot, {})
                elif slot == "__pydantic_fields_set__":
                    object.__setattr__(self, slot, set())

        # Convert validation errors to warnings
        for error in e.errors():
            field_path = ".".join(str(part) for part in error.get("loc", []))
            message = error.get("msg", "Validation failed")
            self._warning_collector.add(field_path, message, "error")

Attributes

sessions property

sessions

Get the underlying ETEngine Session objects from all items.

combine property

combine

Helps users with quick access to a packer. The combine keyword makes sense when spelling out method calls to scenarios.

E.g. scenarios.combine.inputs.to_dataframe() or scenarios.combine.to_excel()

Functions

get_hourly_output_curves

get_hourly_output_curves(carrier_type)

Get hourly output curves for all scenarios by carrier type.

Parameters:

Name	Type	Description	Default
carrier_type	CarrierType	Carrier type alias (electricity, heat, hydrogen, methane)	required

Returns:

Type	Description
dict[str, dict[str, DataFrame]]	Dictionary mapping curve names to scenario data

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

def get_hourly_output_curves(
    self,
    carrier_type: CarrierType,
) -> dict[str, dict[str, pd.DataFrame]]:
    """
    Get hourly output curves for all scenarios by carrier type.

    Args:
        carrier_type: Carrier type alias (electricity, heat, hydrogen, methane)

    Returns:
        Dictionary mapping curve names to scenario data
    """
    # Pre-fetch curves for all scenarios using new API
    self._ensure_hourly_curves_fetched(carrier_type)

    # Now delegate to packer for organization
    return self.combine.hourly_output_curves(carrier_type)

get_annual_exports

get_annual_exports(exports=None)

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

Returns:

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

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

def get_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.

    Returns:
        Dict mapping export names to dicts of {scenario_title: DataFrame}
    """
    return self.combine.annual_exports(exports)

load_all classmethod

load_all(client=None)

Load all saved scenarios belonging to the authenticated user.

Fetches all MyETM saved scenarios for the authenticated user in a single request. Returns all saved scenarios the user has access to (owned or shared).

Parameters:

Name	Type	Description	Default
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None

Returns:

Type	Description
'Scenarios'	Scenarios collection containing all user's saved scenarios

Raises:

Type	Description
ValueError	If authentication fails or API request fails

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

@classmethod
def load_all(
    cls,
    client: Optional[BaseClient] = None,
) -> "Scenarios":
    """Load all saved scenarios belonging to the authenticated user.

    Fetches all MyETM saved scenarios for the authenticated user in a single request.
    Returns all saved scenarios the user has access to (owned or shared).

    Args:
        client: Optional BaseClient instance for API communication

    Returns:
        Scenarios collection containing all user's saved scenarios

    Raises:
        ValueError: If authentication fails or API request fails
    """
    from pyetm.services.scenario_runners.fetch_user_saved_scenarios import (
        FetchUserSavedScenariosRunner,
    )

    if client is None:
        client = get_client()

    result = FetchUserSavedScenariosRunner.run(client=client)

    if not result.success:
        raise ValueError(
            f"Failed to fetch user saved scenarios: {'; '.join(result.errors)}"
        )

    if result.data is None:
        raise ValueError("No data returned from API")

    # Use model_validate to avoid N+1 API calls
    saved_scenarios = [Scenario.model_validate(data) for data in result.data]
    return cls(items=cast(List[Union[Scenario, Session]], saved_scenarios))

load_many classmethod

load_many(saved_scenario_ids, client=None)

Load multiple SavedScenario objects by their MyETM saved scenario IDs.

This is a bulk operation - individual failures are collected as warnings to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

Parameters:

Name	Type	Description	Default
saved_scenario_ids	Iterable[int]	Iterable of MyETM saved scenario IDs to load	required
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None

Returns:

Type	Description
'Scenarios'	SavedScenarios collection containing the loaded SavedScenario objects.
'Scenarios'	Warnings from failures are collected in the warnings property.

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

@classmethod
def load_many(cls, saved_scenario_ids: Iterable[int], client: Optional[BaseClient] = None) -> "Scenarios":
    """
    Load multiple SavedScenario objects by their MyETM saved scenario IDs.

    This is a bulk operation - individual failures are collected as warnings
    to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

    Args:
        saved_scenario_ids: Iterable of MyETM saved scenario IDs to load
        client: Optional BaseClient instance for API communication

    Returns:
        SavedScenarios collection containing the loaded SavedScenario objects.
        Warnings from failures are collected in the warnings property.
    """
    scenarios = cls(items=[])
    scenarios.set_bulk_context(True)

    for ssid in saved_scenario_ids:
        try:
            scenario = Scenario.load(ssid, client=client)
            scenario.set_bulk_context(True)
            scenarios.items.append(scenario)
        except SavedScenarioError as e:
            scenarios.add_warning(
                "load_many",
                f"Could not load saved scenario {ssid}: {e}",
                severity="error",
            )

    # Show warnings if any
    if len(scenarios.warnings) > 0:
        scenarios.show_warnings()

    return scenarios

create_many classmethod

create_many(scenario_params, area_code=None, end_year=None, client=None)

Create multiple SavedScenario objects from parameter dicts.

If scenario_id is not provided in params, creates a new Session first.

This is a bulk operation - individual failures are collected as warnings to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

Parameters:

Name	Type	Description	Default
scenario_params	Iterable[ScenarioCreationParams]	Iterable of ScenarioCreationParams dicts, each containing: - title: Title for the saved scenario (required) - scenario_id: (Optional) ETEngine scenario ID to save. If not provided, a new Session will be created using area_code and end_year - template_id: (Optional) Session ID to use as template. Inherits area_code and end_year from the template. - area_code: (Optional if using defaults or template_id) Area code for new session - end_year: (Optional if using defaults or template_id) End year for new session - user_values: (Optional) Dict of user input values to apply after creation - custom_curves: (Optional) Dict of custom curves to upload after creation - sortables: (Optional) Dict of sortables to apply after creation	required
area_code	str | None	Default area_code for all scenarios (if not in params)	None
end_year	int | None	Default end_year for all scenarios (if not in params)	None
client	BaseClient | None	Optional BaseClient instance (shared across all creates)	None

Returns:

Name	Type	Description
Scenarios	'Scenarios'	Collection of created scenarios. Warnings from individual failures are collected in the warnings property.

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

@classmethod
def create_many(
    cls,
    scenario_params: Iterable[ScenarioCreationParams],
    area_code: str | None = None,
    end_year: int | None = None,
    client: BaseClient | None = None,
) -> "Scenarios":
    """
    Create multiple SavedScenario objects from parameter dicts.

    If scenario_id is not provided in params, creates a new Session first.

    This is a bulk operation - individual failures are collected as warnings
    to allow partial success. Use PYETM_ERROR_MODE=safe to raise on first error.

    Args:
        scenario_params: Iterable of ScenarioCreationParams dicts, each containing:
            - title: Title for the saved scenario (required)
            - scenario_id: (Optional) ETEngine scenario ID to save. If not provided,
              a new Session will be created using area_code and end_year
            - template_id: (Optional) Session ID to use as template. Inherits area_code
              and end_year from the template.
            - area_code: (Optional if using defaults or template_id) Area code for new session
            - end_year: (Optional if using defaults or template_id) End year for new session
            - user_values: (Optional) Dict of user input values to apply after creation
            - custom_curves: (Optional) Dict of custom curves to upload after creation
            - sortables: (Optional) Dict of sortables to apply after creation
        area_code: Default area_code for all scenarios (if not in params)
        end_year: Default end_year for all scenarios (if not in params)
        client: Optional BaseClient instance (shared across all creates)

    Returns:
        Scenarios: Collection of created scenarios. Warnings from individual
                  failures are collected in the warnings property.
    """
    if client is None:
        client = get_client()

    # Create the collection with bulk context enabled
    scenarios_list: List[Union[Scenario, Session]] = []
    scenarios = cls(items=scenarios_list)
    scenarios.set_bulk_context(True)

    # Separate data parameters from creation parameters
    DATA_PARAMS = ["user_values", "custom_curves", "sortables"]
    creation_params_list: List[Dict[str, Any]] = []
    data_to_apply: List[tuple[int, Dict[str, Any]]] = (
        []
    )  # List of (scenario_index, data_dict)

    for idx, params in enumerate(scenario_params):
        # Make a copy to avoid modifying original
        params_copy: Dict[str, Any] = dict(params)

        # Extract all data params declaratively
        data = {key: params_copy.pop(key, None) for key in DATA_PARAMS}

        creation_params_list.append(params_copy)

        # Only add to data_to_apply if at least one data param is present
        if any(data.values()):
            data_to_apply.append((idx, data))

    # Create scenarios sequentially
    saved_scenarios = []
    for params_item in creation_params_list:
        params = params_item  # type: ignore[assignment]
        title = params.get("title")

        if title is None:
            scenarios.add_warning(
                "create_many",
                f"Could not create saved scenario with {params}: Missing required 'title'",
                severity="warning",
            )
            continue

        try:
            # If scenario_id is provided, use existing session
            if "scenario_id" in params:
                session_id = params.get("scenario_id")
                private = bool(params.get("private", False))
                saved_scenarios.append(
                    Scenario.new(
                        title=title,
                        session_id=session_id,
                        client=client,
                        private=private,
                    )
                )
            else:
                # Create new session and save it
                area = params.get("area_code") or area_code
                year = params.get("end_year") or end_year

                # Check if template_id is provided (allows inheriting area_code/end_year)
                has_template = "template_id" in params

                if not has_template and (area is None or year is None):
                    scenarios.add_warning(
                        "create_many",
                        f"Could not create saved scenario with {params}: "
                        "Missing area_code or end_year. Provide them in each dict or as defaults.",
                        severity="warning",
                    )
                    continue

                # Extract private flag and ensure it's a bool
                private = bool(params.get("private", False))

                # Extract additional session params (like template_id)
                extra_params = {
                    k: v
                    for k, v in params.items()
                    if k not in ("title", "private", "area_code", "end_year")
                }

                # For template_id, we need to create session first, then wrap it
                if "template_id" in extra_params:
                    session = Session.new(area, year, client=client, **extra_params)
                    saved_scenarios.append(
                        Scenario.new(
                            title=title,
                            session_id=session.id,
                            client=client,
                            private=private,
                        )
                    )
                else:
                    # Use unified create for simple case
                    saved_scenarios.append(
                        Scenario.new(
                            title=title,
                            area_code=area,
                            end_year=year,
                            client=client,
                            private=private,
                        )
                    )

        except (SavedScenarioError, Exception) as e:
            scenarios.add_warning(
                "create_many",
                f"Could not create saved scenario with {params}: {e}",
                severity="error",
            )

    # Set bulk context on all created scenarios and add to collection
    for scenario in saved_scenarios:
        scenario.set_bulk_context(True)
        scenarios.items.append(scenario)

    # Apply data parameters concurrently after all scenarios are created
    if data_to_apply and saved_scenarios:
        failure_warnings = cls._apply_data_concurrently(
            saved_scenarios, data_to_apply, client
        )

        # Add data application failures as warnings to the collection
        for warning in failure_warnings:
            scenarios.add_warning(
                "data_application",
                warning,
                severity="error",
            )

    # Merge warnings from individual scenarios into collection warnings
    for scenario in saved_scenarios:
        scenarios._merge_submodel_warnings(scenario)

    # Display summary if there were any warnings
    if len(scenarios.warnings) > 0 or any(
        len(s.warnings) > 0 for s in saved_scenarios
    ):
        print("\n=== Batch Creation Summary ===")
        scenarios.show_warnings()

    return scenarios

to_excel

to_excel(path, **export_options)

Export all scenarios to Excel.

Note: This exports the underlying session data from each SavedScenario. The scenario_id column will contain Scenario IDs (MyETM).

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

def to_excel(self, path: PathLike[str] | str, **export_options: Any) -> None:
    """
    Export all scenarios to Excel.

    Note: This exports the underlying session data from each SavedScenario.
    The scenario_id column will contain Scenario IDs (MyETM).
    """

    if not self.items:
        raise ValueError("No scenarios to export")

    from pyetm.models.scenario_packer import ScenarioPacker

    packer = ScenarioPacker()
    packer.add(*self.items)
    packer.to_excel(str(Path(path).expanduser().resolve()), **export_options)

from_excel classmethod

from_excel(xlsx_path, update=False)

Import all scenarios from Excel file.

Loads all scenarios from the Excel file, including both: - SavedScenarios (where 'session' column is False or missing) - Sessions (where 'session' column is True)

Returns:

Type	Description
'Scenarios'	Scenarios collection containing mixed Scenario and Session objects

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

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

    Loads all scenarios from the Excel file, including both:
    - SavedScenarios (where 'session' column is False or missing)
    - Sessions (where 'session' column is True)

    Returns:
        Scenarios collection containing mixed Scenario and Session objects
    """
    from pyetm.models.scenario_packer import ScenarioPacker

    resolved_path = Path(xlsx_path).expanduser().resolve()

    packer = ScenarioPacker.from_excel(str(resolved_path), update=update)
    all_scenarios = list(packer._scenarios())

    scenarios = cls(items=[])
    if not all_scenarios:
        scenarios.add_warning(
            "from_excel",
            f"No scenarios found in Excel file: {resolved_path}",
            severity="warning",
        )
    else:
        all_scenarios.sort(key=lambda s: s.id if hasattr(s, "id") else 0)
        scenarios_list: List[Union[Scenario, Session]] = all_scenarios  # type: ignore[assignment]
        scenarios.items = scenarios_list

    scenarios._packer = packer

    # Auto-display warnings if any
    if len(scenarios.warnings) > 0:
        scenarios.show_warnings()

    return scenarios

discard_many classmethod

discard_many(saved_scenario_ids, client=None)

Discard multiple saved scenarios in bulk (soft-delete).

The scenarios are marked as discarded and hidden from listings, but can be recovered through the MyETM web interface within 60 days. After 60 days, MyETM automatically removes discarded scenarios permanently.

Parameters:

Name	Type	Description	Default
saved_scenario_ids	List[int]	List of SavedScenario IDs to discard	required
client	Optional[BaseClient]	Optional BaseClient instance	None

Returns:

Type	Description
Dict[str, Any]	Dict with 'successful' and 'failed' keys containing lists of IDs

Example

result = Scenarios.discard_many([1, 2, 3]) print(f"Discarded: {result['successful']}") print(f"Failed: {result['failed']}")

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

@classmethod
def discard_many(
    cls,
    saved_scenario_ids: List[int],
    client: Optional[BaseClient] = None,
) -> Dict[str, Any]:
    """
    Discard multiple saved scenarios in bulk (soft-delete).

    The scenarios are marked as discarded and hidden from listings, but can be
    recovered through the MyETM web interface within 60 days. After 60 days,
    MyETM automatically removes discarded scenarios permanently.

    Args:
        saved_scenario_ids: List of SavedScenario IDs to discard
        client: Optional BaseClient instance

    Returns:
        Dict with 'successful' and 'failed' keys containing lists of IDs

    Example:
        result = Scenarios.discard_many([1, 2, 3])
        print(f"Discarded: {result['successful']}")
        print(f"Failed: {result['failed']}")
    """
    from pyetm.services.scenario_runners.discard_saved_scenario import (
        DiscardSavedScenarioRunner,
    )

    if client is None:
        client = get_client()

    if not saved_scenario_ids:
        return {"successful": [], "failed": []}

    # Build discard requests for all scenarios
    requests = []
    for scenario_id in saved_scenario_ids:
        request = DiscardSavedScenarioRunner.build_request(
            saved_scenario_id=scenario_id
        )
        requests.append(request)

    # Format requests for AsyncBatchRunner
    formatted_requests = []
    for req in requests:
        formatted = {
            "method": req["method"],
            "url": req["path"],
            "kwargs": {**req.get("kwargs", {})},
        }
        if req.get("payload"):
            formatted["kwargs"]["json"] = req["payload"]
        formatted_requests.append(formatted)

    # Execute batch discard
    results = AsyncBatchRunner.batch_requests_sync(
        client.session, formatted_requests, MAX_CONCURRENT
    )

    # Collect successes and failures
    successful = []
    failed = []
    for scenario_id, result in zip(saved_scenario_ids, results):
        if result.success:
            successful.append(scenario_id)
        else:
            failed.append(scenario_id)
            error_msg = "; ".join(result.errors) if result.errors else "Unknown error"
            logger.warning(
                f"Failed to discard saved scenario {scenario_id}: {error_msg}"
            )

    return {"successful": successful, "failed": failed}

Functions