update_couplings

update_couplings

Service for update couplings operations.

Classes

UpdateCouplingsRunner

Bases: BaseRunner[Dict[str, Any]]

Runner for updating coupling groups in a scenario.

POST /api/v3/scenarios/{scenario_id}/couple POST /api/v3/scenarios/{scenario_id}/uncouple

Functions

run staticmethod

run(client, scenario, coupling_groups, action='couple', force=False, **kwargs)

Update coupling groups for a scenario.

Parameters:

Name	Type	Description	Default
client	BaseClient	The API client	required
scenario	Any	The scenario object with an id attribute	required
coupling_groups	List[str]	List of coupling group names to couple/uncouple	required
action	str	Either "couple" or "uncouple"	'couple'
force	bool	If True and action is "uncouple", force uncouple all groups	False

Source code in src/pyetm/services/scenario_runners/update_couplings.py

@staticmethod
def run(
    client: BaseClient,
    scenario: Any,
    coupling_groups: List[str],
    action: str = "couple",
    force: bool = False,
    **kwargs: Any,
) -> ServiceResult[Dict[str, Any]]:
    """
    Update coupling groups for a scenario.

    Args:
        client: The API client
        scenario: The scenario object with an id attribute
        coupling_groups: List of coupling group names to couple/uncouple
        action: Either "couple" or "uncouple"
        force: If True and action is "uncouple", force uncouple all groups
    """
    if action not in ["couple", "uncouple"]:
        return ServiceResult.fail(
            errors=[f"Invalid action: {action}. Must be 'couple' or 'uncouple'"]
        )

    # Prepare request data
    data: Dict[str, Union[List[str], bool]] = {"groups": coupling_groups}

    if action == "uncouple" and force:
        data["force"] = True

    result = UpdateCouplingsRunner._make_request(
        client=client,
        method="post",
        path=f"/scenarios/{scenario.id}/{action}",
        payload=data,
    )

    if not result.success:
        return result

    # The response should be the updated scenario data
    body = result.data
    coupling_data: Dict[str, Any] = {}
    warnings: list[str] = []

    # Extract relevant coupling information from the response
    coupling_keys = [
        "active_couplings",
        "inactive_couplings",
    ]

    for key in coupling_keys:
        if key in body:  # type: ignore[operator]
            coupling_data[key] = body[key]  # type: ignore[index]
        else:
            coupling_data[key] = None
            warnings.append(f"Missing coupling field in response: {key!r}")

    return ServiceResult.ok(data=coupling_data, errors=warnings)