Skip to content

scenario_loader

scenario_loader

Utilities for loading scenarios from various sources.

Classes

ScenarioLoader

Bases: Protocol

Protocol for loading, copying, and creating scenarios.

Different implementations interpret scenario IDs differently: - SessionLoader: IDs refer to ETEngine Sessions - SavedScenarioLoader: IDs refer to MyETM SavedScenarios

Functions
load 
load(scenario_id, area_code, end_year, row_label, metadata_updates)

Load an existing scenario by ID and apply metadata updates.

Source code in src/pyetm/models/scenario_loader.py 
19
20
21
22
23
24
25
26
27
28
def load(
    self,
    scenario_id: int,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Load an existing scenario by ID and apply metadata updates."""
    ...
copy 
copy(scenario_id, row_label, metadata_updates)

Create a deep copy of a scenario (no template link).

Source code in src/pyetm/models/scenario_loader.py 
30
31
32
33
34
35
36
37
def copy(
    self,
    scenario_id: int,
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Create a deep copy of a scenario (no template link)."""
    ...
create_new 
create_new(area_code, end_year, row_label, metadata_updates)

Create a brand new scenario.

Source code in src/pyetm/models/scenario_loader.py 
39
40
41
42
43
44
45
46
47
def create_new(
    self,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Create a brand new scenario."""
    ...

SessionLoader 

SessionLoader(packer_helper)

Loader for ETEngine Sessions.

Interprets IDs as ETEngine scenario/session IDs.

Parameters:

Name Type Description Default
packer_helper Any

Reference to ScenarioPacker instance for helper methods

 required
Source code in src/pyetm/models/scenario_loader.py 
57
58
59
60
61
62
def __init__(self, packer_helper: Any) -> None:
    """
    Args:
        packer_helper: Reference to ScenarioPacker instance for helper methods
    """
    self._helper = packer_helper
Functions
load 
load(scenario_id, area_code, end_year, row_label, metadata_updates)

Load an ETEngine Session by ID.

Source code in src/pyetm/models/scenario_loader.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def load(
    self,
    scenario_id: int,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Load an ETEngine Session by ID."""
    scenario = self._helper._load_or_create_scenario(
        scenario_id, area_code, end_year, row_label, **metadata_updates
    )
    if scenario is None:
        return None
    self._helper._apply_metadata_to_scenario(scenario, metadata_updates)
    return cast(Session, scenario)
copy 
copy(scenario_id, row_label, metadata_updates)

Deep copy an ETEngine Session.

Source code in src/pyetm/models/scenario_loader.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def copy(
    self,
    scenario_id: int,
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Deep copy an ETEngine Session."""
    try:
        source_scenario = Session.load(scenario_id)
        return source_scenario.copy(**metadata_updates)
    except Exception as e:
        logger.warning(
            "Failed to copy from Session '%s' for row '%s': %s",
            scenario_id,
            row_label,
            e,
        )
        return None
create_new 
create_new(area_code, end_year, row_label, metadata_updates)

Create a new ETEngine Session.

Source code in src/pyetm/models/scenario_loader.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def create_new(
    self,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Create a new ETEngine Session."""
    scenario = self._helper._load_or_create_scenario(
        None, area_code, end_year, row_label, **metadata_updates
    )
    if scenario is None:
        return None
    self._helper._apply_metadata_to_scenario(scenario, metadata_updates)
    return cast(Session, scenario)

SavedScenarioLoader 

SavedScenarioLoader(packer_helper)

Loader for MyETM SavedScenarios.

Interprets IDs as MyETM SavedScenario IDs and automatically saves new scenarios.

Parameters:

Name Type Description Default
packer_helper Any

Reference to ScenarioPacker instance for helper methods

 required
Source code in src/pyetm/models/scenario_loader.py 
124
125
126
127
128
129
def __init__(self, packer_helper: Any) -> None:
    """
    Args:
        packer_helper: Reference to ScenarioPacker instance for helper methods
    """
    self._helper = packer_helper
Functions
load 
load(scenario_id, area_code, end_year, row_label, metadata_updates)

Load a SavedScenario from MyETM.

Source code in src/pyetm/models/scenario_loader.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def load(
    self,
    scenario_id: int,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Load a SavedScenario from MyETM."""
    from pyetm.models.scenario import Scenario

    try:
        saved_scenario = Scenario.load(scenario_id)
        return cast(Session, saved_scenario)
    except Exception as e:
        error_msg = str(e)
        if "does not exist" in error_msg or "not found" in error_msg.lower():
            logger.warning(
                "Scenario %s does not exist on this ETM environment (row '%s')",
                scenario_id,
                row_label,
            )
        else:
            logger.warning(
                "Could not load Scenario %s for row '%s': %s",
                scenario_id,
                row_label,
                e,
            )
        return None
copy 
copy(scenario_id, row_label, metadata_updates)

Copy a SavedScenario and save the copy to MyETM.

Source code in src/pyetm/models/scenario_loader.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def copy(
    self,
    scenario_id: int,
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Copy a SavedScenario and save the copy to MyETM."""
    from pyetm.models.scenario import Scenario

    # Validate authentication before attempting to save
    self._require_authentication()

    try:
        saved_scenario = Scenario.load(scenario_id)
        copied_session = saved_scenario.session.copy(**metadata_updates)

        title = metadata_updates.get("title") or f"Copy of {saved_scenario.title}"
        try:
            saved_copy = copied_session.save(title=title)
            # Validate that we got a proper SavedScenario back with required attributes
            if hasattr(saved_copy, 'id') and hasattr(saved_copy, 'scenario_id'):
                logger.info(
                    "Automatically saved copy to MyETM with ID %s (session ID: %s)",
                    saved_copy.id,
                    saved_copy.scenario_id,
                )
                return cast(Session, saved_copy)
            else:
                logger.warning(
                    "Save operation for row '%s' returned unexpected object type. Returning session instead.",
                    row_label,
                )
                return copied_session
        except Exception as save_error:
            logger.warning(
                "Failed to save copy to MyETM for row '%s': %s. Returning session instead.",
                row_label,
                save_error,
            )
            return copied_session
    except Exception as e:
        error_msg = str(e)
        if "does not exist" in error_msg or "not found" in error_msg.lower():
            logger.warning(
                "Scenario %s does not exist on this ETM environment. Cannot copy for row '%s'",
                scenario_id,
                row_label,
            )
        else:
            logger.warning(
                "Failed to copy from Scenario %s for row '%s': %s",
                scenario_id,
                row_label,
                e,
            )
        return None
create_new 
create_new(area_code, end_year, row_label, metadata_updates)

Create a new scenario and save it to MyETM.

Source code in src/pyetm/models/scenario_loader.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def create_new(
    self,
    area_code: Any,
    end_year: Optional[int],
    row_label: str,
    metadata_updates: Dict[str, Any],
) -> Optional[Session]:
    """Create a new scenario and save it to MyETM."""
    from pyetm.models.scenario import Scenario

    # Validate authentication before attempting to save
    self._require_authentication()

    scenario = self._helper._load_or_create_scenario(
        None, area_code, end_year, row_label, **metadata_updates
    )
    if scenario is None:
        return None
    self._helper._apply_metadata_to_scenario(scenario, metadata_updates)

    title = metadata_updates.get("title") or f"Scenario {area_code} {end_year}"

    try:
        saved_scenario = scenario.save(title=title)
        # Validate that we got a proper SavedScenario back with required attributes
        if hasattr(saved_scenario, 'id') and hasattr(saved_scenario, 'scenario_id'):
            logger.info(
                "Saved new scenario to MyETM with ID %s (session ID: %s)",
                saved_scenario.id,
                saved_scenario.scenario_id,
            )
            return cast(Session, saved_scenario)
        else:
            logger.warning(
                "Save operation for row '%s' returned unexpected object type. Returning session instead.",
                row_label,
            )
            return cast(Session, scenario)
    except Exception as e:
        logger.warning(
            "Failed to save new scenario to MyETM for row '%s': %s. Returning session instead.",
            row_label,
            e,
        )
        return cast(Session, scenario)