Skip to content

custom_curves

custom_curves

Custom curve data models for scenario inputs.

Classes

CustomCurveError

Bases: Exception

Base custom curve error

CustomCurve 

CustomCurve(**data)

Bases: Base

Wrapper around a single custom curve. Curves are getting saved to the filesystem, as bulk processing of scenarios could end up with several 100 MBs of curves, which we don't want to keep in memory.

Source code in src/pyetm/models/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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")
Functions
retrieve 
retrieve(client, scenario)

Process curve from client, save to file, set file_path

Source code in src/pyetm/models/custom_curves.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def retrieve(self, client: Any, scenario: Any) -> Optional[pd.Series[Any]]:
    """Process curve from client, save to file, set file_path"""
    # Normalize to Session to get ETEngine session ID
    session = self._normalize_to_session(scenario)

    file_path = (
        get_settings().path_to_tmp(str(session.id)) / f"{self.key.replace('/', '-')}.csv"
    )

    # TODO: Examine the caching situation in the future if time permits: could be particularly
    # relevant for bulk processing
    # if file_path.is_file():
    #     self.file_path = file_path
    #     return self.contents()
    try:
        result = DownloadCustomCurveRunner.run(client, session, self.key)

        if result.success and result.data is not None:
            try:
                raw_df = pd.read_csv(result.data, header=None, index_col=False, dtype=float)
                df_or_series = raw_df.squeeze("columns")
                if isinstance(df_or_series, pd.DataFrame):
                    curve = df_or_series.dropna(how="all").iloc[:, 0]
                elif isinstance(df_or_series, pd.Series):
                    curve = df_or_series.dropna(how="all")
                else:
                    # If it's somehow something else, convert to Series
                    curve = pd.Series(df_or_series, dtype=float).dropna(how="all")
                if len(curve) != 8760:
                    self.add_warning(
                        self.key,
                        f"Curve length should be 8760, got {len(curve)}; proceeding with current data",
                    )

                self.file_path = file_path
                curve.to_csv(self.file_path, index=False, header=False)
                return pd.Series(curve.values, name=self.key)
            except Exception as e:
                # File processing error - add warning and return None
                self.add_warning(self.key, f"Failed to process curve data: {e}")
                return None
        else:
            # API call failed - add warning for each error
            for error in result.errors:
                self.add_warning(self.key, f"Failed to retrieve curve: {error}")
            return None

    except Exception as e:
        # Unexpected error - add warning
        self.add_warning(self.key, f"Unexpected error retrieving curve: {e}")
        return None
contents 
contents()

Open file from path and return contents

Source code in src/pyetm/models/custom_curves.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def contents(self) -> Optional[pd.Series[Any]]:
    """Open file from path and return contents"""
    if not self.available():
        self.add_warning(self.key, f"Curve not available - no file path set")
        return None

    try:
        if self.file_path is None:
            self.add_warning(self.key, f"Curve has no file path set")
            return None
        raw_df = pd.read_csv(self.file_path, header=None, index_col=False, dtype=float)
        df_or_series = raw_df.squeeze("columns")
        if isinstance(df_or_series, pd.DataFrame):
            series = df_or_series.dropna(how="all").iloc[:, 0]
        elif isinstance(df_or_series, pd.Series):
            series = df_or_series.dropna(how="all")
        else:
            # If it's somehow something else, convert to Series
            series = pd.Series(df_or_series, dtype=float).dropna(how="all")
        if len(series) != 8760:
            self.add_warning(
                self.key,
                f"Curve length should be 8760, got {len(series)}; using available data",
            )
        return pd.Series(series.values, name=self.key)
    except Exception as e:
        self.add_warning(self.key, f"Failed to read curve file: {e}")
        return None
remove 
remove()

Remove file and clear path

Source code in src/pyetm/models/custom_curves.py 
140
141
142
143
144
145
146
147
148
149
150
151
def remove(self) -> bool:
    """Remove file and clear path"""
    if not self.available() or self.file_path is None:
        return True

    try:
        self.file_path.unlink(missing_ok=True)
        self.file_path = None
        return True
    except Exception as e:
        self.add_warning(self.key, f"Failed to remove curve file: {e}")
        return False
from_json classmethod 
from_json(data)

Initialize a CustomCurve from JSON data

Source code in src/pyetm/models/custom_curves.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@classmethod
def from_json(cls, data: dict[str, Any]) -> "CustomCurve":
    """
    Initialize a CustomCurve from JSON data
    """
    try:
        curve = cls(**data)
        missing = [k for k in ("key", "type") if k not in data]
        if missing:
            curve.add_warning(
                "base",
                f"Failed to create curve from data: missing required fields: {', '.join(missing)}",
            )

        return curve
    except Exception as e:
        basic_data = {
            "key": data.get("key", "unknown"),
            "type": data.get("type", "unknown"),
        }
        curve = cls.model_construct(**basic_data)
        curve.add_warning("base", f"Failed to create curve from data: {e}")
        return curve

CustomCurves 

CustomCurves(**data)

Bases: Base

Wrapper around a collection of custom curves.

Source code in src/pyetm/models/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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")
Functions
is_attached 
is_attached(curve_name)

Returns true if that curve is attached

Source code in src/pyetm/models/custom_curves.py 
235
236
237
def is_attached(self, curve_name: str) -> bool:
    """Returns true if that curve is attached"""
    return any((curve_name == key for key in self.attached_keys()))
attached_keys 
attached_keys()

Returns the keys of attached curves

Source code in src/pyetm/models/custom_curves.py 
239
240
241
def attached_keys(self) -> Any:
    """Returns the keys of attached curves"""
    yield from (curve.key for curve in self.curves)
from_json classmethod 
from_json(data)

Initialize CustomCurves collection from JSON data

Source code in src/pyetm/models/custom_curves.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
@classmethod
def from_json(cls, data: list[dict[str, Any]]) -> "CustomCurves":
    """
    Initialize CustomCurves collection from JSON data
    """
    curves = []

    for curve_data in data:
        try:
            curve = CustomCurve.from_json(curve_data)
            curves.append(curve)
        except Exception as e:
            # Create a basic curve and continue processing
            key = curve_data.get("key", "unknown")
            basic_curve = CustomCurve.model_construct(key=key, type="unknown")
            basic_curve.add_warning(key, f"Skipped invalid curve data: {e}")
            curves.append(basic_curve)

    collection = cls.model_validate({"curves": curves})
    collection._merge_submodel_warnings(*curves, key_attr="key")
    return collection
validate_for_upload 
validate_for_upload()

Validate all curves for upload

Source code in src/pyetm/models/custom_curves.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
def validate_for_upload(self) -> dict[str, WarningCollector]:
    """
    Validate all curves for upload
    """
    validation_errors: dict[str, WarningCollector] = {}

    for curve in self.curves:
        curve_warnings = WarningCollector()

        if not curve.available():
            curve_warnings.add(curve.key, "Curve has no data available")
            validation_errors[curve.key] = curve_warnings
            continue

        try:
            try:
                # Read without dtype conversion to preserve non-numeric values
                if curve.file_path is None:
                    curve_warnings.add(curve.key, "Curve has no file path")
                    validation_errors[curve.key] = curve_warnings
                    continue
                raw_data = pd.read_csv(curve.file_path, header=None, index_col=False)
                if raw_data.empty:
                    curve_warnings.add(curve.key, "Curve contains no data")
                    validation_errors[curve.key] = curve_warnings
                    continue

                # Check length first
                if len(raw_data) != 8760:
                    curve_warnings.add(
                        curve.key,
                        f"Curve must contain exactly 8,760 values, found {len(raw_data)}",
                    )
                else:
                    try:
                        # Try to convert to numeric, this will raise if there are non-numeric values
                        pd.to_numeric(raw_data.iloc[:, 0], errors="raise")
                    except (ValueError, TypeError):
                        curve_warnings.add(curve.key, "Curve contains non-numeric values")

            except pd.errors.EmptyDataError:
                curve_warnings.add(curve.key, "Curve contains no data")
            except Exception as e:
                curve_warnings.add(curve.key, f"Error reading curve data: {str(e)}")

        except Exception as e:
            curve_warnings.add(curve.key, f"Error reading curve data: {str(e)}")

        if len(curve_warnings) > 0:
            validation_errors[curve.key] = curve_warnings

    return validation_errors
is_valid_update 
is_valid_update(updates)

Validate curve updates without applying them.

Parameters:

Name Type Description Default
updates dict[str, Series]

Dictionary mapping curve keys to pandas Series with 8760 values

 required

Returns:

Type Description
dict[str, WarningCollector]

Dictionary mapping curve keys to WarningCollector objects for errors
Source code in src/pyetm/models/custom_curves.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def is_valid_update(self, updates: dict[str, pd.Series]) -> dict[str, WarningCollector]:
    """
    Validate curve updates without applying them.

    Args:
        updates: Dictionary mapping curve keys to pandas Series with 8760 values

    Returns:
        Dictionary mapping curve keys to WarningCollector objects for errors
    """
    warnings: dict[str, WarningCollector] = {}

    for key, curve_data in updates.items():
        curve_warnings = WarningCollector()

        # Check if curve exists in collection
        if not self.is_attached(key):
            curve_warnings.add(key, f"Curve '{key}' is not attached to scenario")

        # Validate curve data type
        if not isinstance(curve_data, pd.Series):
            curve_warnings.add(key, "Curve data must be a pandas Series")
        elif len(curve_data) != 8760:
            curve_warnings.add(
                key, f"Curve must have 8760 values, found {len(curve_data)}"
            )
        else:
            # Validate all values are numeric
            try:
                pd.to_numeric(curve_data, errors="raise")
            except (ValueError, TypeError):
                curve_warnings.add(key, "Curve contains non-numeric values")

        if len(curve_warnings) > 0:
            warnings[key] = curve_warnings

    return warnings
update 
update(updates)

Update custom curve data with validation and warning display.

Invalid curves are rejected (not updated) to maintain data integrity. Warnings are automatically displayed for invalid curves and non-existent keys. Warnings from previous updates are cleared to show only current operation issues.

Parameters:

Name Type Description Default
updates dict[str, Series]

Dictionary mapping curve keys to pandas Series with 8760 values

 required
Source code in src/pyetm/models/custom_curves.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
def update(self, updates: dict[str, pd.Series]) -> None:
    """
    Update custom curve data with validation and warning display.

    Invalid curves are rejected (not updated) to maintain data integrity.
    Warnings are automatically displayed for invalid curves and non-existent keys.
    Warnings from previous updates are cleared to show only current operation issues.

    Args:
        updates: Dictionary mapping curve keys to pandas Series with 8760 values
    """
    # Auto-clear stale warnings from previous updates
    self.warnings.clear()

    # Pre-validate all updates
    validation_warnings = self.is_valid_update(updates)

    # Add validation warnings to collection
    for key, warnings in validation_warnings.items():
        for warning in warnings:
            self.add_warning(key, warning.message)

    # Apply valid updates only
    for key, curve_data in updates.items():
        if key not in validation_warnings and self.is_attached(key):
            curve = self._find(key)
            if curve and curve.file_path:
                try:
                    curve_data.to_csv(curve.file_path, index=False, header=False)
                except Exception as e:
                    self.add_warning(key, f"Failed to save curve: {e}")

    # Auto-display warnings if any exist
    if len(self.warnings) > 0:
        self.auto_show_warnings()
remove_curve 
remove_curve(curve_name)

Remove a specific custom curve's cache file.

Parameters:

Name Type Description Default
curve_name str

Name of the curve to remove

 required

Returns:

Type Description
bool

True if successfully removed, False otherwise
Source code in src/pyetm/models/custom_curves.py 
475
476
477
478
479
480
481
482
483
484
485
486
487
def remove_curve(self, curve_name: str) -> bool:
    """Remove a specific custom curve's cache file.

    Args:
        curve_name: Name of the curve to remove

    Returns:
        True if successfully removed, False otherwise
    """
    curve = self._find(curve_name)
    if curve is None:
        return False
    return curve.remove()
clear_cache 
clear_cache()

Clear all custom curve cache files.

Returns:

Type Description
int

Number of files successfully removed
Source code in src/pyetm/models/custom_curves.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
def clear_cache(self) -> int:
    """Clear all custom curve cache files.

    Returns:
        Number of files successfully removed
    """
    removed_count = 0
    for curve in self.curves:
        if curve.remove():
            removed_count += 1

    self.warnings.clear()
    return removed_count

Functions