Skip to content

custom_curves_pack

custom_curves_pack

Custom curves packing utilities.

Classes

CustomCurvesPack

Bases: Packable

A packable for managing custom curves data.

Functions
excel_read_kwargs staticmethod 
excel_read_kwargs()

Returns a dict representing the excel read kwargs like the header Availabale to overload for users own implementation

Source code in src/pyetm/models/packables/custom_curves_pack.py 
22
23
24
25
26
27
28
@staticmethod
def excel_read_kwargs() -> Dict[str, Any]:
    """
    Returns a dict representing the excel read kwargs like the header
    Availabale to overload for users own implementation
    """
    return {"header": None}
to_dict_per_curve 
to_dict_per_curve(curves=None)

Build dict organized by curve name, then by scenario.

Source code in src/pyetm/models/packables/custom_curves_pack.py 
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 to_dict_per_curve(
    self, curves: Optional[Sequence[str]] = None
) -> Dict[str, Dict[str, pd.DataFrame]]:
    """
    Build dict organized by curve name, then by scenario.
    """
    result: Dict[str, Dict[str, pd.DataFrame]] = {}

    for scenario in self.scenarios:
        try:
            scenario_key = self._key_for(scenario)

            # Determine which curves to fetch
            if curves is not None:
                # Filter to requested curves that exist
                curves_to_fetch = [
                    c for c in curves if c in list(scenario.custom_curves.attached_keys())
                ]
            else:
                # Fetch all attached curves
                curves_to_fetch = list(scenario.custom_curves.attached_keys())

            for curve_name in curves_to_fetch:
                curve_data = scenario.custom_curve_series(curve_name)
                if curve_data is None or curve_data.empty:
                    continue

                if curve_name not in result:
                    result[curve_name] = {}
                # Ensure curve_data is a DataFrame
                if isinstance(curve_data, pd.Series):
                    df_data = curve_data.to_frame()
                else:
                    df_data = curve_data
                result[curve_name][scenario_key] = df_data

            self.log_scenario_warnings(scenario, self.key, self.sheet_name)

        except Exception as e:
            logger.warning(
                "Failed building curves dict for scenario %s: %s",
                scenario.identifier(),
                e,
            )
            continue

    return result
load_from_dataframe 
load_from_dataframe(df, scenario, update_set=None)

Loads from a dataframe for a single scenario

Source code in src/pyetm/models/packables/custom_curves_pack.py 
87
88
89
90
91
92
93
94
95
96
def load_from_dataframe(self, df: pd.DataFrame, scenario: Any, update_set: Optional[Set[str]] = None) -> None:
    """
    Loads from a dataframe for a single scenario
    """
    normalized_data = excel_utils.normalize_sheet(df, helper_names=self._sheet_index)

    if normalized_data.empty:
        return

    self.apply_custom_curves_to_scenario(scenario, normalized_data, update_set)
apply_custom_curves_to_scenario 
apply_custom_curves_to_scenario(scenario, data, update_set=None)

Apply custom curves to scenario with validation and error handling.

Source code in src/pyetm/models/packables/custom_curves_pack.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def apply_custom_curves_to_scenario(
    self, scenario: Any, data: pd.DataFrame, update_set: Optional[Set[str]] = None
) -> None:
    """Apply custom curves to scenario with validation and error handling."""
    skip_upload = not self._should_include_upload(update_set)

    try:
        curves = CustomCurves._from_dataframe(data, scenario_id=scenario.id)

        # Log processing warnings
        curves.log_warnings(
            logger,
            prefix=f"Custom curves warning for '{scenario.identifier()}'",
        )

        # Validate curves and log validation issues (skip if read-only)
        if not skip_upload:
            self.validate_and_log_curves(curves, scenario)

        # Apply curves to scenario
        scenario.update_custom_curves(curves, skip_upload=skip_upload)

    except Exception as e:
        logger.warning("Failed processing custom curves for '%s': %s", scenario.identifier(), e)
validate_and_log_curves 
validate_and_log_curves(curves, scenario)

Validate curves and log any validation issues.

Source code in src/pyetm/models/packables/custom_curves_pack.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def validate_and_log_curves(self, curves: CustomCurves, scenario: Any) -> None:
    """Validate curves and log any validation issues."""
    try:
        validation_results = curves.validate_for_upload()
        for key, issues in (validation_results or {}).items():
            for issue in issues:
                logger.warning(
                    "Custom curve validation for '%s' in '%s' [%s]: %s",
                    key,
                    scenario.identifier(),
                    getattr(issue, "field", key),
                    getattr(issue, "message", str(issue)),
                )
    except Exception:
        # Validation errors are not critical, continue processing
        pass