Solvers

This module provides functions for running power flow calculations.

run_opf

Run Optimal Power Flow (OPF) calculation using Julia interface.

Parameters:

Name	Type	Description	Default
net	Network	A Network object containing the power system model.	required
jl	Any	Julia interface object for running OPF.	required

Returns:

Type	Description
Dict[str, Any]	OPF result containing termination status and solution data.

Raises:

Type	Description
RuntimeError	If OPF fails to converge or encounters an error.

Source code in gridfm_datakit/process/solvers.py

def run_opf(net: Network, jl: Any) -> Dict[str, Any]:
    """Run Optimal Power Flow (OPF) calculation using Julia interface.

    Args:
        net: A Network object containing the power system model.
        jl: Julia interface object for running OPF.

    Returns:
        OPF result containing termination status and solution data.

    Raises:
        RuntimeError: If OPF fails to converge or encounters an error.
    """
    # Create a temporary file for the MATPOWER case
    with tempfile.NamedTemporaryFile(mode="w", suffix=".m", delete=False) as temp_file:
        temp_filename = temp_file.name

    try:
        # Save network to temporary file
        net.to_mpc(temp_filename)

        result = jl.run_opf(temp_filename)

        if str(result["termination_status"]) != "LOCALLY_SOLVED":
            raise RuntimeError(f"OPF did not converge: {result['termination_status']}")

        return result

    except Exception as e:
        raise RuntimeError(f"Error running OPF: {e}")
    finally:
        # Clean up temporary file
        if os.path.exists(temp_filename):
            os.unlink(temp_filename)

run_pf

Run Power Flow (PF) calculation using Julia interface.

This function runs the power flow calculation using the Julia interface and returns the result with termination status.

Parameters:

Name	Type	Description	Default
net	Network	A network object containing the power system model.	required
jl	Any	Julia interface object for running power flow.	required
fast	Union[bool, None]	If True, use the direct (non-optimizer) computation. If None, defaults to False (uses optimizer-based solver).	None

Returns:

Type	Description
Dict[str, Any]	Power flow result containing termination status and solution data.

Raises:

Type	Description
RuntimeError	If power flow fails to converge or encounters an error.

Source code in gridfm_datakit/process/solvers.py

def run_pf(net: Network, jl: Any, fast: Union[bool, None] = None) -> Dict[str, Any]:
    """Run Power Flow (PF) calculation using Julia interface.

    This function runs the power flow calculation using the Julia interface
    and returns the result with termination status.

    Args:
        net: A network object containing the power system model.
        jl: Julia interface object for running power flow.
        fast: If True, use the direct (non-optimizer) computation. If None, defaults to False (uses optimizer-based solver).

    Returns:
        Power flow result containing termination status and solution data.

    Raises:
        RuntimeError: If power flow fails to converge or encounters an error.
    """

    # Create a temporary file for the MATPOWER case
    with tempfile.NamedTemporaryFile(mode="w", suffix=".m", delete=False) as temp_file:
        temp_filename = temp_file.name

    try:
        # Save network to temporary file
        net.to_mpc(temp_filename)

        # Run PF
        result = jl.run_pf_fast(temp_filename) if fast else jl.run_pf(temp_filename)
        if (
            fast
            and str(result["termination_status"]) != "True"
            or (not fast and str(result["termination_status"]) != "LOCALLY_SOLVED")
        ):
            raise RuntimeError(
                f"PF did not converge: {result['termination_status']}, fast={fast}",
            )

        return result

    except Exception as e:
        raise RuntimeError(f"Error running PF: {e}")
    finally:
        # Clean up temporary file
        if os.path.exists(temp_filename):
            os.unlink(temp_filename)

run_dcpf

Run DC Power Flow (DCPF) calculation using Julia interface.

This function runs the DC power flow calculation using the Julia interface and returns the result with termination status.

Parameters:

Name	Type	Description	Default
net	Network	A network object containing the power system model.	required
jl	Any	Julia interface object for running DC power flow.	required
fast	Union[bool, None]	If True, use the direct (non-optimizer) computation. If None, defaults to False (uses optimizer-based solver).	None

Returns:

Type	Description
Dict[str, Any]	DC power flow result containing termination status and solution data.

Raises:

Type	Description
RuntimeError	If DC power flow fails to converge or encounters an error.

Source code in gridfm_datakit/process/solvers.py

def run_dcpf(net: Network, jl: Any, fast: Union[bool, None] = None) -> Dict[str, Any]:
    """Run DC Power Flow (DCPF) calculation using Julia interface.

    This function runs the DC power flow calculation using the Julia interface
    and returns the result with termination status.

    Args:
        net: A network object containing the power system model.
        jl: Julia interface object for running DC power flow.
        fast: If True, use the direct (non-optimizer) computation. If None, defaults to False (uses optimizer-based solver).

    Returns:
        DC power flow result containing termination status and solution data.

    Raises:
        RuntimeError: If DC power flow fails to converge or encounters an error.
    """

    # Create a temporary file for the MATPOWER case
    with tempfile.NamedTemporaryFile(mode="w", suffix=".m", delete=False) as temp_file:
        temp_filename = temp_file.name

    try:
        # Save network to temporary file
        net.to_mpc(temp_filename)

        # Run DCPF (fast or standard)
        result = jl.run_dcpf_fast(temp_filename) if fast else jl.run_dcpf(temp_filename)

        if (
            fast
            and str(result["termination_status"]) != "True"
            or (not fast and str(result["termination_status"]) != "LOCALLY_SOLVED")
        ):
            raise RuntimeError(
                f"DC PF did not converge: {result['termination_status']}, fast={fast}",
            )

        return result

    except Exception as e:
        raise RuntimeError(f"Error running DC PF: {e}")
    finally:
        # Clean up temporary file
        if os.path.exists(temp_filename):
            os.unlink(temp_filename)