Skip to content

update_custom_curves

update_custom_curves

Service for update custom curves operations.

Classes

UpdateCustomCurvesRunner

Bases: BaseRunner[Dict[str, Any]]

Runner for uploading custom curves to a scenario.

Functions
build_requests staticmethod 
build_requests(scenario, custom_curves)

Build custom curve upload requests for concurrent batching.

Parameters:

Name Type Description Default
scenario Any

The scenario object (must have an 'id' attribute)

 required
custom_curves Dict[str, Any]

Dict mapping curve keys to curve data (pandas Series/DataFrame, list, or string)

 required

Returns:

Type Description
List[Dict[str, Any]]

List of request dicts ready for AsyncBatchRunner
Source code in src/pyetm/services/scenario_runners/update_custom_curves.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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
@staticmethod
def build_requests(scenario: Any, custom_curves: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    Build custom curve upload requests for concurrent batching.

    Args:
        scenario: The scenario object (must have an 'id' attribute)
        custom_curves: Dict mapping curve keys to curve data (pandas Series/DataFrame, list, or string)

    Returns:
        List of request dicts ready for AsyncBatchRunner
    """
    import pandas as pd
    from pathlib import Path

    requests = []

    for curve_key, curve_data in custom_curves.items():
        # Format curve data based on type
        if isinstance(curve_data, (str, Path)):
            path = Path(curve_data)
            try:
                if path.exists():
                    with open(path, "r") as f:
                        content = f.read()
                else:
                    content = str(curve_data)
            except (OSError, ValueError):
                # Path is too long, invalid, or not a valid path
                content = str(curve_data)
        elif isinstance(curve_data, (pd.Series, pd.DataFrame)):
            content = curve_data.to_csv(header=False, index=False)
        elif isinstance(curve_data, (list, tuple)):
            content = "\n".join(str(value) for value in curve_data)
        else:
            content = str(curve_data)

        requests.append(
            {
                "method": "put",
                "path": f"/scenarios/{scenario.id}/custom_curves/{curve_key}",
                "payload": None,
                "kwargs": {
                    "files": {
                        "file": (
                            f"{curve_key}.csv",
                            content,
                            "application/octet-stream",
                        )
                    },
                    "headers": {"Content-Type": None},
                },
            }
        )

    return requests