session

session

Async HTTP session management for ETM API.

Classes

ScenarioError

Bases: Exception

Base scenario error

Session

Session(**data)

Bases: Base

Pydantic model for an ETM Scenario, matching the DB schema, but with only id required so it can be used for API runners.

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

version property

version

Returns the version of the ETM the scenario was made in

couplings property

couplings

Get coupling groups for this scenario

preset_scenario_id property

preset_scenario_id

Backward-compatible property for template_id (API field name).

Functions

new classmethod

new(area_code=None, end_year=None, client=None, **kwargs)

Create a new scenario with the specified parameters.

Parameters:

Name	Type	Description	Default
area_code	str | None	Area code for the scenario (optional if template_id is provided)	None
end_year	int | None	End year for the scenario (optional if template_id is provided)	None
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None
**kwargs	Any	Additional parameters including: - template_id: Session ID to use as template - private, keep_compatible, source, title, metadata, start_year, scaling, url	{}

Returns:

Type	Description
'Session'	A new Scenario instance

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

@classmethod
def new(
    cls,
    area_code: str | None = None,
    end_year: int | None = None,
    client: Optional[BaseClient] = None,
    **kwargs: Any,
) -> "Session":
    """
    Create a new scenario with the specified parameters.

    Args:
        area_code: Area code for the scenario (optional if template_id is provided)
        end_year: End year for the scenario (optional if template_id is provided)
        client: Optional BaseClient instance for API communication
        **kwargs: Additional parameters including:
            - template_id: Session ID to use as template
            - private, keep_compatible, source, title, metadata, start_year, scaling, url

    Returns:
        A new Scenario instance
    """
    scenario_data = {**kwargs}
    if area_code is not None:
        scenario_data["area_code"] = area_code
    if end_year is not None:
        scenario_data["end_year"] = end_year

    result = CreateScenarioRunner.run(client or get_client(), scenario_data)

    if not result.success:
        raise ScenarioError(f"Could not create scenario: {result.errors}")

    scenario = cls.model_validate(result.data)
    for warning in result.errors:
        scenario.add_warning("base", warning)

    # Auto-display creation warnings
    area_code = getattr(scenario, "area_code", None)
    end_year = getattr(scenario, "end_year", None)
    context = f"Session #{scenario.id}"
    if area_code or end_year:
        context += f" (area_code={area_code}, end_year={end_year})"
    scenario.auto_show_warnings(context)

    for field, value in scenario_data.items():
        if (
            hasattr(scenario, field)
            and result.data is not None
            and field not in result.data
        ):
            setattr(scenario, field, value)

    return scenario

load classmethod

load(scenario_id, client=None)

Fetch metadata for scenario_id, return a Scenario (with warnings if any keys missing).

Parameters:

Name	Type	Description	Default
scenario_id	int	ETEngine session ID to load	required
client	Optional[BaseClient]	Optional BaseClient instance for API communication	None

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

@classmethod
def load(cls, scenario_id: int, client: Optional[BaseClient] = None) -> Session:
    """
    Fetch metadata for scenario_id, return a Scenario (with warnings if any keys missing).

    Args:
        scenario_id: ETEngine session ID to load
        client: Optional BaseClient instance for API communication
    """
    template = type("T", (), {"id": scenario_id})
    result = FetchMetadataRunner.run(client or get_client(), template)

    if not result.success:
        raise ScenarioError(
            f"Could not load session {scenario_id}: {result.errors}"
        )

    # parse into a Scenario
    scenario = cls.model_validate(result.data)

    # Optional metadata fields that shouldn't trigger warnings
    optional_fields = {
        "title",
        "preset_scenario_id",
        "short_name",
        "metadata",
        "start_year",
        "keep_compatible",
        "private",
        "source",
        "url",
        "scaling",
    }

    # Only add warnings for missing fields that are NOT optional
    for w in result.errors:
        # Filter out "Missing field" warnings for optional fields
        if "Missing field" in w:
            # Extract field name from warning message
            field_name = w.split("'")[-2] if "'" in w else None
            if field_name in optional_fields:
                continue  # Skip this warning
        scenario.add_warning("metadata", w)

    # Auto-display load warnings
    area_code = getattr(scenario, "area_code", None)
    end_year = getattr(scenario, "end_year", None)
    context = f"Session #{scenario.id}"
    if area_code or end_year:
        context += f" (area_code={area_code}, end_year={end_year})"
    scenario.auto_show_warnings(context)

    return scenario

copy_with_preset

copy_with_preset(**overrides)

Create a copy of this scenario using ETEngine's copy utility. The copied scenario will have its template field set to this scenario's ID.

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

def copy_with_preset(self, **overrides: Any) -> "Session":
    """
    Create a copy of this scenario using ETEngine's copy utility.
    The copied scenario will have its template field set to this scenario's ID.
    """
    result = CopyScenarioRunner.run(get_client(), self.id, overrides=overrides)

    if not result.success:
        raise ScenarioError(f"Failed to copy scenario: {result.errors}")

    scenario = Session.model_validate(result.data)
    for warning in result.errors:
        scenario.add_warning("base", warning)

    return scenario

copy

copy(**overrides)

Create a copy with no template link to the original scenario.

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

def copy(self, **overrides: Any) -> "Session":
    """
    Create a copy with no template link to the original scenario.
    """
    new_scenario = self.copy_with_preset(**overrides)

    result = BreakPresetLinkRunner.run(get_client(), new_scenario)

    if not result.success:
        raise ScenarioError(
            f"Copied scenario but failed to break template link: {result.errors}"
        )

    if result.data and "scenario" in result.data:
        scenario_data = result.data["scenario"]
        for field, value in scenario_data.items():
            # Handle alias: preset_scenario_id → template_id
            attr_name = "template_id" if field == "preset_scenario_id" else field
            if hasattr(new_scenario, attr_name):
                setattr(new_scenario, attr_name, value)

    return new_scenario

interpolate classmethod

interpolate(sessions, *end_years, client=None)

Interpolate one or more sessions to create scenarios at target years.

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

@classmethod
def interpolate(
    cls,
    sessions: Union["Session", List["Session"]],
    *end_years: int,
    client: Optional[BaseClient] = None,
) -> List["Session"]:
    """
    Interpolate one or more sessions to create scenarios at target years.
    """
    client = client or get_client()

    session_list = sessions if isinstance(sessions, list) else [sessions]

    # Validate area codes are consistent
    area_codes = [s.area_code for s in session_list]
    if len(set(area_codes)) > 1:
        raise ValueError(
            f"All sessions must have the same area_code. Found: {set(area_codes)}"
        )

    # Validate no duplicate end years
    end_years_in_sessions = [s.end_year for s in session_list]
    duplicates = [
        year
        for year in set(end_years_in_sessions)
        if end_years_in_sessions.count(year) > 1
    ]
    if duplicates:
        raise ValueError(
            f"Sessions must have unique end_year values. Found duplicate(s): {duplicates}"
        )

    scenario_ids = [s.id for s in session_list]
    end_years_list = list(end_years)

    result = InterpolateScenariosRunner.run(client, scenario_ids, end_years_list)

    if not result.success:
        raise ScenarioError(f"Interpolation failed: {result.errors}")

    # Create Session instances from the response data
    interpolated_sessions: List[Session] = []
    if result.data is not None:
        for scenario_data in result.data:
            scenario = cls.model_validate(scenario_data)
            for warning in result.errors:
                scenario.add_warning("base", warning)
            interpolated_sessions.append(scenario)

    return interpolated_sessions

from_excel classmethod

from_excel(xlsx_path)

Load a single scenario from Excel.

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

@classmethod
def from_excel(cls, xlsx_path: PathLike[str] | str) -> "Session":
    """Load a single scenario from Excel."""
    from pyetm.models.scenario_packer import ScenarioPacker

    packer = ScenarioPacker.from_excel(xlsx_path)
    scenarios = list(packer._scenarios())
    if len(scenarios) != 1:
        raise ScenarioError(
            f"Expected one scenario, found {len(scenarios)}. "
            "Use Scenarios.from_excel() for multi-scenario workbooks."
        )
    return scenarios[0]

to_excel

to_excel(path, **export_options)

Export this scenario to Excel.

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

def to_excel(self, path: PathLike[str] | str, **export_options: Any) -> None:
    """Export this scenario to Excel."""
    from pyetm.models.scenario_packer import ScenarioPacker

    packer = ScenarioPacker()
    packer.add(self)
    packer.to_excel(str(path), **export_options)

collect_export_data

collect_export_data(**export_options)

Collect export data in format-agnostic structure.

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

def collect_export_data(self, **export_options: Any) -> ExportDataCollection:
    """Collect export data in format-agnostic structure."""
    from pyetm.models.scenario_packer import ScenarioPacker

    packer = ScenarioPacker()
    packer.add(self)
    return packer.collect_export_data(**export_options)

save

save(client=None, title=None, **kwargs)

Save this scenario to MyETM as a SavedScenario.

Note

Requires authentication via ETM_API_TOKEN environment variable. SavedScenarios are persisted in MyETM and associated with your user account.

Parameters:

Name	Type	Description	Default
client	Optional[BaseClient]	Optional BaseClient instance (defaults to global client)	None
title	Optional[str]	Title for the saved scenario (defaults to self.title)	None
**kwargs	Any	Additional parameters (e.g., private, description)	{}

Returns:

Type	Description
Any	SavedScenario instance

Raises:

Type	Description
PermissionError	If ETM_API_TOKEN is not configured
ScenarioError	If title is not provided

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

def save(
    self,
    client: Optional[BaseClient] = None,
    title: Optional[str] = None,
    **kwargs: Any,
) -> Any:
    """
    Save this scenario to MyETM as a SavedScenario.

    Note:
        Requires authentication via ETM_API_TOKEN environment variable.
        SavedScenarios are persisted in MyETM and associated with your user account.

    Args:
        client: Optional BaseClient instance (defaults to global client)
        title: Title for the saved scenario (defaults to self.title)
        **kwargs: Additional parameters (e.g., private, description)

    Returns:
        SavedScenario instance

    Raises:
        PermissionError: If ETM_API_TOKEN is not configured
        ScenarioError: If title is not provided
    """
    from pyetm.models.scenario import Scenario

    client = client or get_client()

    save_title = title or self.title
    if not save_title:
        raise ScenarioError(
            "Title is required to save scenario. Provide title parameter or set scenario.title"
        )

    private = kwargs.pop(
        "private", self.private if self.private is not None else False
    )

    return Scenario.new(
        title=save_title,
        session_id=self.id,
        client=client,
        private=private,
        **kwargs,
    )

update_metadata

update_metadata(**kwargs)

Update metadata for this scenario.

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

def update_metadata(self, **kwargs: Any) -> Optional[Dict[str, Any]]:
    """
    Update metadata for this scenario.
    """
    result = UpdateMetadataRunner.run(get_client(), self, kwargs)

    if not result.success:
        raise ScenarioError(f"Could not update metadata: {result.errors}")

    # Add any warnings from the update
    for w in result.errors:
        self.add_warning("metadata", w)

    # Update the current scenario object with the server response
    if result.data and "scenario" in result.data:
        scenario_data = result.data["scenario"]
        for field, value in scenario_data.items():
            if hasattr(self, field):
                setattr(self, field, value)

        for field, value in kwargs.items():
            if hasattr(self, field) and field not in scenario_data:
                setattr(self, field, value)

    return result.data

set_short_name

set_short_name(short_name)

Set the short_name attribute and persist in metadata.

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

def set_short_name(self, short_name: str) -> None:
    """Set the short_name attribute and persist in metadata."""
    self.short_name = str(short_name)
    if self.metadata is not None:
        self.metadata["short_name"] = str(short_name)
    else:
        self.metadata = {"short_name": str(short_name)}

user_values

user_values()

Returns the values set by the user for inputs

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

def user_values(self) -> Dict[str, Any]:
    """
    Returns the values set by the user for inputs
    """
    return {inp.key: inp.user for inp in self.inputs if inp.user is not None}

set_user_values_from_dataframe

set_user_values_from_dataframe(dataframe)

Extract df to dict, set None/NaN sliders to reset, and call update_inputs. This ensures the dataframe exactly represents the inputs.

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

def set_user_values_from_dataframe(self, dataframe: pd.DataFrame) -> None:
    """
    Extract df to dict, set None/NaN sliders to reset, and call update_inputs.
    This ensures the dataframe exactly represents the inputs.
    """
    series = dataframe["user"]
    # If MultiIndex with 'unit', drop it
    if isinstance(series.index, pd.MultiIndex) and "unit" in (
        series.index.names or []
    ):
        series = series.droplevel("unit")
    user_values_dict: Dict[str, Any] = {
        str(k): v for k, v in series.fillna("reset").to_dict().items()
    }
    self.update_user_values(user_values_dict)

update_user_values

update_user_values(update_inputs, skip_upload=False)

Parameters:

Name	Type	Description	Default
update_inputs	Dict[str, Any]	Dictionary of input key-value pairs to update	required
skip_upload	bool	If True, skip API call and validation, only update local cache	False

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

def update_user_values(
    self, update_inputs: Dict[str, Any], skip_upload: bool = False
) -> None:
    """
    Args:
        update_inputs: Dictionary of input key-value pairs to update
        skip_upload: If True, skip API call and validation, only update local cache
    """
    if not skip_upload:
        # Update them in the Inputs object, and check validation
        validity_errors = self.inputs.is_valid_update(update_inputs)
        self._handle_validity_errors(validity_errors, "user values")

        result = UpdateInputsRunner.run(get_client(), self, update_inputs)

        if not result.success:
            raise ScenarioError(f"Could not update user values: {result.errors}")

    self.inputs.update(update_inputs)

remove_user_values

remove_user_values(input_keys)

Remove user values for specified inputs, resetting them to default values.

Parameters:

Name	Type	Description	Default
input_keys	Union[List[str], Set[str]]	List or set of input keys to reset to default values	required

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

def remove_user_values(self, input_keys: Union[List[str], Set[str]]) -> None:
    """
    Remove user values for specified inputs, resetting them to default values.

    Args:
        input_keys: List or set of input keys to reset to default values
    """
    reset_inputs = {key: "reset" for key in input_keys}
    result = UpdateInputsRunner.run(get_client(), self, reset_inputs)

    if not result.success:
        raise ScenarioError(f"Could not remove inputs: {result.errors}")

    # Update them in the Inputs object
    self.inputs.update(reset_inputs)

set_sortables_from_dataframe

set_sortables_from_dataframe(dataframe, skip_upload=False)

Extract sortables from dataframe and update them. The dataframe should have sortable names as columns and orders as rows.

Parameters:

Name	Type	Description	Default
dataframe	DataFrame	DataFrame with sortable names as columns and order values as rows	required
skip_upload	bool	If True, skip API call and validation, only update local cache	False

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

def set_sortables_from_dataframe(
    self, dataframe: pd.DataFrame, skip_upload: bool = False
) -> None:
    """
    Extract sortables from dataframe and update them.
    The dataframe should have sortable names as columns and orders as rows.

    Args:
        dataframe: DataFrame with sortable names as columns and order values as rows
        skip_upload: If True, skip API call and validation, only update local cache
    """
    coll = Sortables._from_dataframe(dataframe)
    updates = coll.to_updates_dict()
    if updates:
        self.update_sortables(updates, skip_upload=skip_upload)

update_sortables

update_sortables(update_sortables, skip_upload=False)

Update the order of specified sortables.

Parameters:

Name	Type	Description	Default
update_sortables	Dict[str, List[Any]]	Dictionary mapping sortable names to their new orders	required
skip_upload	bool	If True, skip API calls and validation, only update local cache	False

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

def update_sortables(
    self, update_sortables: Dict[str, List[Any]], skip_upload: bool = False
) -> None:
    """
    Update the order of specified sortables.

    Args:
        update_sortables: Dictionary mapping sortable names to their new orders
        skip_upload: If True, skip API calls and validation, only update local cache
    """
    if not skip_upload:
        # Validate the updates first
        validity_errors = self.sortables.is_valid_update(update_sortables)
        self._handle_validity_errors(validity_errors, "sortables")

        # Make individual API calls for each sortable as there is no bulk endpoint
        for name, order in update_sortables.items():
            if name.startswith("heat_network_"):
                subtype = name.replace("heat_network_", "")
                result = UpdateSortablesRunner.run(
                    get_client(), self, "heat_network", order, subtype=subtype
                )
            else:
                result = UpdateSortablesRunner.run(get_client(), self, name, order)

            if not result.success:
                raise ScenarioError(
                    f"Could not update sortable '{name}': {result.errors}"
                )

    self.sortables.update(update_sortables)

remove_sortables

remove_sortables(sortable_names)

Reset specified sortables to their default/empty orders.

Parameters:

Name	Type	Description	Default
sortable_names	Union[List[str], Set[str]]	List or set of sortable names to reset	required

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

def remove_sortables(self, sortable_names: Union[List[str], Set[str]]) -> None:
    """
    Reset specified sortables to their default/empty orders.

    Args:
        sortable_names: List or set of sortable names to reset
    """
    # Make individual API calls to reset each sortable
    for name in sortable_names:
        if name.startswith("heat_network_"):
            # Handle heat_network with subtype
            subtype = name.replace("heat_network_", "")
            result = UpdateSortablesRunner.run(
                get_client(), self, "heat_network", [], subtype=subtype
            )
        else:
            result = UpdateSortablesRunner.run(get_client(), self, name, [])

        if not result.success:
            raise ScenarioError(
                f"Could not remove sortable '{name}': {result.errors}"
            )

    reset_sortables: Dict[str, List[Any]] = {name: [] for name in sortable_names}
    self.sortables.update(reset_sortables)

custom_curves_series

custom_curves_series()

Yield all Series

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

def custom_curves_series(self) -> Any:
    """Yield all Series"""
    for key in self.custom_curves.attached_keys():
        yield self.custom_curve_series(key)

update_custom_curves

update_custom_curves(custom_curves, skip_upload=False)

Upload/update custom curves for this scenario.

Parameters:

Name	Type	Description	Default
custom_curves	Any	CustomCurves object containing curves to upload	required
skip_upload	bool	If True, skip API call and validation, only update local cache	False

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

def update_custom_curves(
    self, custom_curves: Any, skip_upload: bool = False
) -> None:
    """
    Upload/update custom curves for this scenario.

    Args:
        custom_curves: CustomCurves object containing curves to upload
        skip_upload: If True, skip API call and validation, only update local cache
    """
    if not skip_upload:
        # Validate curves before uploading
        validity_errors = custom_curves.validate_for_upload()
        self._handle_validity_errors(validity_errors, "custom curves")

        # Upload curves
        result = UpdateCustomCurvesRunner.run(get_client(), self, custom_curves)
        if not result.success:
            raise ScenarioError(f"Could not update custom curves: {result.errors}")

    # TODO: this should be done in custom curves
    # Update the scenario's custom curves object
    for new_curve in custom_curves.curves:
        existing_curve = self.custom_curves._find(new_curve.key)
        if existing_curve:
            existing_curve.file_path = new_curve.file_path
        else:
            self.custom_curves.curves.append(new_curve)
    try:
        self.custom_curves._scenario = self
    except Exception:
        pass

remove_custom_curves

remove_custom_curves(curve_keys)

Remove custom curves from this scenario.

Deletes curves from the ETM API and removes them from the local collection. Local cache files are automatically cleared.

Parameters:

Name	Type	Description	Default
curve_keys	Union[List[str], Set[str]]	List or set of curve keys to remove	required

Raises:

Type	Description
ScenarioError	If API deletion fails

Example

session.remove_custom_curves(["weather/solar_pv", "weather/wind"])

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

def remove_custom_curves(
    self, curve_keys: Union[List[str], Set[str]]
) -> None:
    """
    Remove custom curves from this scenario.

    Deletes curves from the ETM API and removes them from the local collection.
    Local cache files are automatically cleared.

    Args:
        curve_keys: List or set of curve keys to remove

    Raises:
        ScenarioError: If API deletion fails

    Example:
        >>> session.remove_custom_curves(["weather/solar_pv", "weather/wind"])
    """
    from pyetm.services.scenario_runners.delete_custom_curves import (
        DeleteCustomCurvesRunner,
    )

    curve_keys_list = list(curve_keys) if isinstance(curve_keys, set) else curve_keys

    if not curve_keys_list:
        return

    # Delete from API
    result = DeleteCustomCurvesRunner.run(get_client(), self, curve_keys_list)
    if not result.success:
        raise ScenarioError(f"Could not remove custom curves: {result.errors}")

    # Remove from local collection and clear cache files
    keys_set = set(curve_keys_list)
    for curve in list(self.custom_curves.curves):
        if curve.key in keys_set:
            curve.remove()  # Clear cache file

    # Remove from curves list
    self.custom_curves.curves = [
        c for c in self.custom_curves.curves if c.key not in keys_set
    ]

get_hourly_curve

get_hourly_curve(identifier)

Get a single hourly output curve by name or carrier type alias.

Carrier types ('electricity', 'heat', 'hydrogen', 'methane') are treated as convenient aliases for their primary curves.

Parameters:

Name	Type	Description	Default
identifier	str	Curve name (e.g., 'merit_order') or carrier type alias	required

Returns:

Type	Description
Optional[DataFrame]	DataFrame with hourly data, or None if not found

Examples:

>>> session.get_hourly_curve("merit_order")      # by name
>>> session.get_hourly_curve("electricity")      # by carrier alias

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

def get_hourly_curve(self, identifier: str) -> Optional[pd.DataFrame]:
    """
    Get a single hourly output curve by name or carrier type alias.

    Carrier types ('electricity', 'heat', 'hydrogen', 'methane') are treated
    as convenient aliases for their primary curves.

    Args:
        identifier: Curve name (e.g., 'merit_order') or carrier type alias

    Returns:
        DataFrame with hourly data, or None if not found

    Examples:
        >>> session.get_hourly_curve("merit_order")      # by name
        >>> session.get_hourly_curve("electricity")      # by carrier alias
    """
    return self.hourly_output_curves.get_curve(identifier, self)

get_hourly_curves

get_hourly_curves(identifiers)

Get multiple hourly output curves by names or carrier type aliases.

Parameters:

Name	Type	Description	Default
identifiers	list[str]	List of curve names and/or carrier type aliases	required

Returns:

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

Examples:

>>> session.get_hourly_curves(["electricity", "heat"])
{'merit_order': DataFrame, 'heat_network': DataFrame}

>>> session.get_hourly_curves(["merit_order", "electricity_price"])
{'merit_order': DataFrame, 'electricity_price': DataFrame}

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

def get_hourly_curves(self, identifiers: list[str]) -> dict[str, pd.DataFrame]:
    """
    Get multiple hourly output curves by names or carrier type aliases.

    Args:
        identifiers: List of curve names and/or carrier type aliases

    Returns:
        Dictionary mapping curve names to DataFrames

    Examples:
        >>> session.get_hourly_curves(["electricity", "heat"])
        {'merit_order': DataFrame, 'heat_network': DataFrame}

        >>> session.get_hourly_curves(["merit_order", "electricity_price"])
        {'merit_order': DataFrame, 'electricity_price': DataFrame}
    """
    return self.hourly_output_curves.get_curves(identifiers, self)

all_hourly_output_curves

all_hourly_output_curves()

Generator yielding all hourly output curves.

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

def all_hourly_output_curves(self) -> Any:
    """Generator yielding all hourly output curves."""
    for key in self.hourly_output_curves.attached_keys():
        yield self.get_hourly_curve(key)

clear_hourly_curves_cache

clear_hourly_curves_cache()

Clear all hourly output curves cache files and LRU cache.

Returns:

Type	Description
int	Number of files successfully removed

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

def clear_hourly_curves_cache(self) -> int:
    """Clear all hourly output curves cache files and LRU cache.

    Returns:
        Number of files successfully removed
    """
    return self.hourly_output_curves.clear_cache()

clear_custom_curves_cache

clear_custom_curves_cache()

Clear all custom curves cache files.

Returns:

Type	Description
int	Number of files successfully removed

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

def clear_custom_curves_cache(self) -> int:
    """Clear all custom curves cache files.

    Returns:
        Number of files successfully removed
    """
    return self.custom_curves.clear_cache()

clear_all_curve_caches

clear_all_curve_caches()

Clear all curve caches (hourly output curves and custom curves).

Returns:

Type	Description
tuple[int, int]	Tuple of (hourly_curves_removed, custom_curves_removed)

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

def clear_all_curve_caches(self) -> tuple[int, int]:
    """Clear all curve caches (hourly output curves and custom curves).

    Returns:
        Tuple of (hourly_curves_removed, custom_curves_removed)
    """
    hourly = self.clear_hourly_curves_cache()
    custom = self.clear_custom_curves_cache()
    return (hourly, custom)

clear_session_cache

clear_session_cache()

Clear entire session temp directory (all cached files).

This removes all cached files for this session and clears all associated in-memory caches.

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

def clear_session_cache(self) -> None:
    """Clear entire session temp directory (all cached files).

    This removes all cached files for this session and clears
    all associated in-memory caches.
    """
    import shutil

    session_dir = get_settings().path_to_tmp(str(self.id))
    if session_dir.exists():
        shutil.rmtree(session_dir)
        # Recreate empty directory
        session_dir.mkdir(parents=True, exist_ok=True)

    # Clear LRU cache
    from pyetm.models.hourly_output_curves import _read_csv_cached_impl

    _read_csv_cached_impl.cache_clear()

    # Clear collection warnings
    self.hourly_output_curves.warnings.clear()
    self.custom_curves.warnings.clear()

get_annual_export

get_annual_export(export_name)

Get a single annual export by name.

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

def get_annual_export(self, export_name: str) -> Optional[pd.DataFrame]:
    """Get a single annual export by name."""
    return self.annual_exports.retrieve(get_client(), self, export_name)

get_annual_exports

get_annual_exports(export_names)

Get multiple annual exports by name.

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

def get_annual_exports(
    self, export_names: AnnualExportType | list[AnnualExportType]
) -> dict[str, pd.DataFrame]:
    """Get multiple annual exports by name."""
    validated_names = validate_export_names(cast("str | list[str]", export_names))
    return self.annual_exports.retrieve_multiple(
        get_client(), self, validated_names
    )

remove_queries

remove_queries(*query_keys)

Remove specific queries from the session.

Parameters:

Name	Type	Description	Default
query_keys	str	Query keys to remove from the collection	()

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

def remove_queries(self, *query_keys: str) -> None:
    """
    Remove specific queries from the session.

    Args:
        query_keys: Query keys to remove from the collection
    """
    if self._queries is not None:
        self._queries.remove(*query_keys)

clear_queries

clear_queries()

Remove all queries from the session.

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

def clear_queries(self) -> None:
    """
    Remove all queries from the session.
    """
    if self._queries is not None:
        self._queries.clear()

execute_queries

execute_queries()

Queries are executed explicitly, as we need to know when the user is ready collecting all of them

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

def execute_queries(self) -> None:
    """
    Queries are executed explicitly, as we need to know when the user is
    ready collecting all of them
    """
    if self._queries is not None:
        self._queries.execute(get_client(), self)
        self._merge_submodel_warnings(self._queries)

results

results(columns=None)

Returns the results of the requested queries in a dataframe

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

def results(self, columns: Optional[Any] = None) -> pd.DataFrame:
    """
    Returns the results of the requested queries in a dataframe
    """
    if columns is None:
        columns = ["present", "future"]

    if not self.queries_requested():
        return pd.DataFrame()

    # Check if queries need to be executed (none have been executed yet)
    if self._queries is not None:
        # Only execute if NO queries have data yet (all are None)
        has_any_results = any(
            v is not None for v in self._queries.query_dict.values()
        )
        if not has_any_results:
            self.execute_queries()

        # Show warning for any failed queries
        failed = [k for k, v in self._queries.query_dict.items() if v is None]
        if failed:
            print(f"WARNING: {len(failed)} query(ies) failed:")
            # Get specific error messages from warnings
            query_warnings = self._queries.warnings.get_by_field("results")
            for w in query_warnings:
                print(f"  - {w.message}")

    if self._queries is not None:
        return self._queries.to_dataframe(columns=columns)

    return pd.DataFrame()

queries_requested

queries_requested()

Returns True if queries have been requested

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

def queries_requested(self) -> bool:
    """
    Returns True if queries have been requested
    """
    if self._queries is None:
        return False

    return len(self._queries.query_keys()) > 0

show_all_warnings

show_all_warnings()

Display all warnings from the scenario and its submodels in a organized way.

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

def show_all_warnings(self) -> None:
    """
    Display all warnings from the scenario and its submodels in a organized way.
    """
    print(f"=== Warnings for Scenario {self.id} ===")

    # Show scenario-level warnings
    if len(self.warnings) > 0:
        print("\nScenario warnings:")
        self.show_warnings()

    # Show submodel warnings if they exist and are loaded
    submodels = [
        ("Inputs", self._inputs),
        ("Sortables", self._sortables),
        ("Custom Curves", self._custom_curves),
        ("Hourly Output Curves", self._hourly_output_curves),
        ("Annual Exports", self._annual_exports),
        ("Queries", self._queries),
        ("Couplings", self._couplings),
    ]

    for name, submodel in submodels:
        if submodel is not None and len(submodel.warnings) > 0:
            print(f"\n{name} warnings:")
            submodel.show_warnings()

list_users

list_users()

Fetch all users with access to this scenario.

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

def list_users(self) -> List[Dict[str, Any]]:
    """
    Fetch all users with access to this scenario.
    """
    result = ScenarioUsersIndexRunner.run(get_client(), self.id)

    if not result.success:
        raise ScenarioError(f"Could not fetch users: {result.errors}")

    if result.data is None:
        return []

    for user in result.data:
        user["role"] = user["role"].replace("scenario_", "", 1)

    return result.data

update_users

update_users(email, role, skip_upload=False)

Add, update, or remove a user's access to this scenario. - skip_upload: If True, store data locally without uploading (can be applied later)

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

def update_users(self, email: str, role: str, skip_upload: bool = False) -> None:
    """
    Add, update, or remove a user's access to this scenario.
    - skip_upload: If True, store data locally without uploading (can be applied later)
    """
    role = self._normalize_role(role)

    if skip_upload:
        self._pending_users[email] = role
        return

    if role == "remove":
        self._remove_user(email)
        return

    if self._user_exists(email):
        self._update_user_role(email, role)
    else:
        self._add_user(email, role)

apply_pending_users

apply_pending_users()

Apply all pending user updates that were loaded with skip_upload=True.

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

def apply_pending_users(self) -> int:
    """
    Apply all pending user updates that were loaded with skip_upload=True.
    """
    if not self._pending_users:
        return 0

    count = 0
    for email, role in list(self._pending_users.items()):
        try:
            self.update_users(email, role, skip_upload=False)
            count += 1
        except Exception as e:
            logger.warning(
                "Failed to apply pending user '%s' with role '%s': %s",
                email,
                role,
                e,
            )

    # Clear pending users after applying
    self._pending_users.clear()
    return count

delete

delete()

Permanently delete this ETEngine scenario/session (hard delete).

WARNING: This is a permanent deletion and cannot be undone. The scenario and all its data will be permanently removed from ETEngine.

This method should be used when you want to delete a temporary/ephemeral session that is not saved in MyETM. If this session is associated with a SavedScenario, use Scenario.delete() instead to ensure both are deleted.

Raises:

Type	Description
ScenarioError	If the deletion fails

Example

session = Session.new(area_code="nl2023", end_year=2050) session.delete()

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

def delete(self) -> None:
    """
    Permanently delete this ETEngine scenario/session (hard delete).

    WARNING: This is a permanent deletion and cannot be undone. The scenario
    and all its data will be permanently removed from ETEngine.

    This method should be used when you want to delete a temporary/ephemeral
    session that is not saved in MyETM. If this session is associated with a
    SavedScenario, use Scenario.delete() instead to ensure both are deleted.

    Raises:
        ScenarioError: If the deletion fails

    Example:
        >>> session = Session.new(area_code="nl2023", end_year=2050)
        >>> session.delete()
    """
    from pyetm.services.scenario_runners.destroy_session import DestroySessionRunner

    result = DestroySessionRunner.run(get_client(), self.id)

    if not result.success:
        raise ScenarioError(f"Could not delete session: {result.errors}")

Functions