Skip to content

fetch_curves_generic

fetch_curves_generic

Service for fetch curves generic operations.

Classes

GenericCurveDownloadRunner

Bases: BaseRunner[Any]

Generic runner for downloading any curve as CSV data. Supports both custom curves and output curves.

Functions
run staticmethod 
run(client, scenario, curve_name, curve_type='output', **kwargs)

Execute the curve download operation.

Returns:

Type Description
ServiceResult[Any]

ServiceResult[io.StringIO]: Success case contains StringIO with CSV data; failure case contains error messages.
Source code in src/pyetm/services/scenario_runners/fetch_curves_generic.py 
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
@staticmethod
def run(
    client: BaseClient,
    scenario: Any,
    curve_name: str,
    curve_type: Literal["custom", "output"] = "output",
    **kwargs: Any,
) -> ServiceResult[Any]:
    """Execute the curve download operation.

    Returns:
        ServiceResult[io.StringIO]: Success case contains StringIO with CSV data;
            failure case contains error messages.
    """
    # Normalize to Session to get ETEngine session ID
    from pyetm.models.scenario import Scenario
    session = scenario.session if isinstance(scenario, Scenario) else scenario

    path = (
        f"/scenarios/{session.id}/custom_curves/{curve_name}.csv"
        if curve_type == "custom"
        else f"/scenarios/{session.id}/curves/{curve_name}.csv"
    )
    req = [
        {
            "method": "get",
            "path": path,
            "payload": None,
            "kwargs": {"headers": {"Accept": "text/csv"}},
        }
    ]
    try:
        result = GenericCurveDownloadRunner._make_batch_requests(client, req)[0]
    except Exception as e:
        return ServiceResult.fail([str(e)])

    if not result.success:
        return ServiceResult.fail(result.errors)
    try:
        resp = result.data
        # TODO: is this ok to return a io object??
        # Is this what causes IO pressure?
        return ServiceResult.ok(data=io.StringIO(resp.content.decode("utf-8")))  # type: ignore[union-attr]
    except Exception as e:
        return ServiceResult.fail([f"Failed to parse curve data: {e}"])

GenericCurveBulkRunner

Bases: BaseRunner[Dict[str, StringIO]]

Large output curves can cause high memory + IO pressure if fetched all at once; a batch size limit prevents overwhelming the event loop while still benefiting from concurrency.