grid_feedback_optimizer package

Subpackages

Submodules

grid_feedback_optimizer.main module

grid_feedback_optimizer.main.main(file_path: str, save_path: str | None = None, max_iter: int = 1000, tol: float = 0.001, delta_p: float = 1.0, delta_q: float = 1.0, algorithm: str = 'gp', alpha: float = 0.5, alpha_v: float = 10.0, alpha_l: float = 10.0, alpha_t: float = 10.0, no_record_iterates: bool = False, solver: str = 'CLARABEL', loading_meas_side: str = 'from', rel_tol=0.0001, rel_tol_line=0.01, **solver_kwargs)

Run grid feedback optimizer from a JSON/EXCEL file path provided as string.

Module contents

class grid_feedback_optimizer.Bus(*, index: int, u_rated: float, u_pu_max: float, u_pu_min: float)

Bases: BaseModel

index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

u_pu_max: float
u_pu_min: float
u_rated: float
class grid_feedback_optimizer.GradientProjectionOptimizer(opt_model_data: OptimizationModelData, sensitivities: dict, alpha: float = 0.5, solver: str = 'CLARABEL', **solver_kwargs)

Bases: object

Gradient projection optimizer. Caches the CVXPY problem to allow fast updates of parameters.

solve_problem(opt_input: OptimizationInputs)

Update CVXPY parameters using structured optimization inputs, solve the cached optimization problem, and return optimized setpoints.

Parameters:

opt_input (OptimizationInputs) –

Structured input model containing:
  • u_pu_measnp.ndarray

    Measured node voltages [p.u.]

  • P_line_meas, Q_line_measnp.ndarray

    Measured active/reactive line power flows

  • p_gen_last, q_gen_lastnp.ndarray

    Previous generator active/reactive power setpoints

  • P_transformer_meas, Q_transformer_measnp.ndarray, optional

    Measured transformer active/reactive power (if applicable)

Returns:

Optimized generator setpoints of shape (n_generators, 2), where each row contains [p_opt, q_opt].

Return type:

np.ndarray

class grid_feedback_optimizer.Line(*, index: int, from_bus: int, to_bus: int, r1: float, x1: float, c1: float, tan1: float, i_n: float)

Bases: BaseModel

c1: float
from_bus: int
i_n: float
index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

r1: float
tan1: float
to_bus: int
x1: float
class grid_feedback_optimizer.Load(*, index: int, bus: int, p_norm: float, q_norm: float)

Bases: BaseModel

Represents a non-controllable unit: either a load or a generator.

bus: int
index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_norm: float
q_norm: float
class grid_feedback_optimizer.Network(*, buses: List[Bus], lines: List[Line], transformers: List[Transformer] = [], sources: List[Source], renew_gens: List[RenewGen], loads: List[Load])

Bases: BaseModel

buses: List[Bus]
lines: List[Line]
loads: List[Load]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

renew_gens: List[RenewGen]
sources: List[Source]
transformers: List[Transformer]
class grid_feedback_optimizer.OptimizationInputs(*, u_pu_meas: ndarray, P_line_meas: ndarray, Q_line_meas: ndarray, p_gen_last: ndarray, q_gen_last: ndarray, P_transformer_meas: ndarray | None = None, Q_transformer_meas: ndarray | None = None)

Bases: BaseModel

Structured container for optimization inputs at each iteration.

P_line_meas: ndarray
P_transformer_meas: ndarray | None
Q_line_meas: ndarray
Q_transformer_meas: ndarray | None
check_lengths()

Ensure measurement and generator arrays have consistent lengths.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_gen_last: ndarray
q_gen_last: ndarray
to_dict()

Convert to a plain dict (for compatibility with existing code).

u_pu_meas: ndarray
class grid_feedback_optimizer.OptimizationModelData(*, p_min: ndarray, p_max: ndarray, q_min: ndarray | None = None, q_max: ndarray | None = None, s_inv: ndarray | None = None, pf_min: ndarray | None = None, c1_p: ndarray, c2_p: ndarray, c1_q: ndarray, c2_q: ndarray, p_norm: ndarray, q_norm: ndarray, u_pu_max: ndarray, u_pu_min: ndarray, s_line: ndarray, s_transformer: ndarray | None = None)

Bases: BaseModel

Minimal structure needed for the optimizer.

c1_p: ndarray
c1_q: ndarray
c2_p: ndarray
c2_q: ndarray
check_fill_optional_arrays()

Ensure optional arrays are initialized as arrays of None if missing.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_max: ndarray
p_min: ndarray
p_norm: ndarray
pf_min: ndarray | None
q_max: ndarray | None
q_min: ndarray | None
q_norm: ndarray
s_inv: ndarray | None
s_line: ndarray
s_transformer: ndarray | None
u_pu_max: ndarray
u_pu_min: ndarray
class grid_feedback_optimizer.PowerFlowSolver(network: Network)

Bases: object

Run a power flow simulation and return bus voltages and line currents.

build_network(network: Network)
is_congested(output_data: Dict[ComponentType, ndarray] | None = None, tol_v: float = 0.0001, tol_l: float = 0.0001, tol_t: float = 0.0001)

Check if the network is congested. Returns True if any of the following occur: - Bus voltage exceeds limits (u_pu_max or u_pu_min) - Line or transformer loading exceeds 1.0 (100%)

obtain_sensitivity(delta_p: float = 1.0, delta_q: float = 1.0, loading_meas_side: str = 'from', rel_tol: float = 0.0001, rel_tol_line: float = 0.01)

Compute sensitivities of bus voltages and line/transformer power flows to small perturbations in generator power injections (p and q) around the default operating point.

Parameters:

  • delta_p (float) – Active power perturbation (W).

  • delta_q (float) – Reactive power perturbation (VAr).

  • loading_meas_side (str) – From which side branches are monitored: “from” or “to”.

Returns:

sensitivities – { “du_dp”: array (n_bus, n_gen), “du_dq”: array (n_bus, n_gen), “dP_line_dp”: array (n_line, n_gen), “dQ_line_dp”: array (n_line, n_gen), “dP_line_dq”: array (n_line, n_gen), “dQ_line_dq”: array (n_line, n_gen), “dP_transformer_dp”: array (n_transformer, n_gen), “dQ_transformer_dp”: array (n_transformer, n_gen), “dP_transformer_dq”: array (n_transformer, n_gen), “dQ_transformer_dq”: array (n_transformer, n_gen) }

Return type:

dict

static prune_relative(A: ndarray, rel_tol: float = 0.0001)

Zero very small elements relative to the matrix scale.

run(gen_update: ndarray = None, load_update: ndarray = None)

Re-run power flow in optimization iterations. The size of the arrays should match the total numbers of gens.

class grid_feedback_optimizer.PrimalDualOptimizer(opt_model_data: OptimizationModelData, sensitivities: dict, alpha: float = 0.5, alpha_v: float = 10.0, alpha_l: float = 10.0, alpha_t: float = 10.0, solver: str = 'CLARABEL', **solver_kwargs)

Bases: object

A primal-dual gradient projection feedback optimizer.

static calc_loading(p: ndarray | float, q: ndarray | float, s: ndarray | float) → ndarray | float
static calc_pf(p: ndarray | float, q: ndarray | float) → ndarray | float
static calc_rpf(p: ndarray | float, q: ndarray | float) → ndarray | float
solve_problem(opt_input: OptimizationInputs, grad_callback: Callable | None = None, **callback_kwargs)

Update parameters and implement primal-dual gradient projection.

Parameters:

  • opt_input (OptimizationInputs) –

    Structured input model containing:
    • u_pu_measnp.ndarray

      Measured node voltages [p.u.]

    • P_line_meas, Q_line_measnp.ndarray

      Measured active/reactive line power flows

    • p_gen_last, q_gen_lastnp.ndarray

      Previous generator active/reactive power setpoints

    • P_transformer_meas, Q_transformer_measnp.ndarray, optional

      Measured transformer active/reactive power (if applicable)

  • grad_callback (callable | None = None) – A function which takes grad_p and grad_q and outputs grad_p and grad_q

Returns:

Optimized generator setpoints of shape (n_generators, 2), where each row contains [p_opt, q_opt].

Return type:

np.ndarray

class grid_feedback_optimizer.RenewGen(*, index: int, bus: int, p_max: float, s_inv: float | None = None, q_min: float | None = None, q_max: float | None = None, pf_min: float | None = None, p_min: float | None = 0.0, q_norm: float | None = 0.0, p_norm: float | None = None, c1_p: float | None = 0.0, c2_p: Annotated[float, Ge(ge=0)] = 1.0, c1_q: float | None = 0.0, c2_q: Annotated[float, Ge(ge=0)] = 0.1)

Bases: BaseModel

Represents a renewable generator or power-consuming device.

Notes

  • If p_max > 0 and p_min >= 0 → behaves as a generator.

  • If p_max <= 0 and p_min <= 0 → behaves as a load/consuming device.

  • Mixed cases (p_min < 0 < p_max) → flexible device (can consume or generate).

bus: int
c1_p: float | None
c1_q: float | None
c2_p: float
c2_q: float
compute_p_norm()

Compute p_norm only if not provided by user.

index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p_max: float
p_min: float | None
p_norm: float | None
pf_min: float | None
q_max: float | None
q_min: float | None
q_norm: float | None
s_inv: float | None
class grid_feedback_optimizer.RenewGenProjection(solver: str = 'CLARABEL', **solver_kwargs)

Bases: object

Project points onto the feasible inverter operating region. Supports analytical and CVXPY-based projections.

static analytic_projection(p_max: float, s_inv: float, p: float, q: float)

Project a point (p, q) onto the feasible operating region of a PV inverter defined by active power limit p_max and apparent power limit s_inv.

Based on: Optimal Power Flow Pursuit (Appendix B) — setpoint update rule.

Parameters:

  • p_max (float) – Active power limit (p_max >= 0)

  • s_inv (float) – Inverter apparent power capacity (s_inv >= 0)

  • p (float) – Active and reactive power values to project

  • q (float) – Active and reactive power values to project

Returns:

The projected point [p_proj, q_proj] lying within the feasible PV region.

Return type:

np.ndarray

Raises:

ValueError – If projection fails or inputs are inconsistent.

projection(p_max: float, p_min: float, p: float, q: float, s_inv: float | None = None, pf_min: float | None = None, q_min: float | None = None, q_max: float | None = None)

Hybrid projection: analytic if simple, CVXPY otherwise.

class grid_feedback_optimizer.SolveResults(*, final_output: Dict[Any, Any], final_gen_update: ndarray, iterations: List[Dict[str, Any]])

Bases: BaseModel

Container for optimization and power flow results. Provides structured access and convenient save/load utilities.

final_gen_update: ndarray
final_output: Dict[Any, Any]
iterations: List[Dict[str, Any]]
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

plot_iterations() → None

Plot evolution of voltages, line loadings, transformer loadings, and generator active/reactive powers over optimization iterations.

Layout: 5 rows * 1 column (shared x-axis: iteration number)

print_summary() → None

Print a concise summary of the optimization results.

save(output_file: str = 'optimization_results.json') → None

Save the results of the optimization and power flow to a JSON file.

class grid_feedback_optimizer.Source(*, index: int, bus: int, u_ref_pu: float)

Bases: BaseModel

bus: int
index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

u_ref_pu: float
class grid_feedback_optimizer.Transformer(*, index: int, from_bus: int, to_bus: int, u1: float, u2: float, sn: float, uk: float, pk: float, i0: float, p0: float, winding_from: WindingType, winding_to: WindingType, clock: int, tap_side: BranchSide, tap_min: int, tap_max: int, tap_size: float, tap_pos: int | None = 0)

Bases: BaseModel

clock: int
from_bus: int
i0: float
index: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

p0: float
pk: float
sn: float
tap_max: int
tap_min: int
tap_pos: int | None
tap_side: BranchSide
tap_size: float
to_bus: int
u1: float
u2: float
uk: float
winding_from: WindingType
winding_to: WindingType
class grid_feedback_optimizer.TransformerActivePowerTrackingCallback(sensitivity: ndarray, alpha: float = 1.0, n_transformer: int = 1)

Bases: object

grid_feedback_optimizer.load_network(file_path: str | Path) → Network

Load a network from a JSON file and validate it using Pydantic.

Parameters:

file_path – Path to the JSON network file.

Returns:

Validated Network object.

Return type:

Network

grid_feedback_optimizer.load_network_from_excel(file_path: str | Path) → Network

Load a Network object from an Excel file where each sheet corresponds to a component type (e.g., ‘buses’, ‘lines’, etc.).

Missing sheets are treated as empty lists.

grid_feedback_optimizer.main(file_path: str, save_path: str | None = None, max_iter: int = 1000, tol: float = 0.001, delta_p: float = 1.0, delta_q: float = 1.0, algorithm: str = 'gp', alpha: float = 0.5, alpha_v: float = 10.0, alpha_l: float = 10.0, alpha_t: float = 10.0, no_record_iterates: bool = False, solver: str = 'CLARABEL', loading_meas_side: str = 'from', rel_tol=0.0001, rel_tol_line=0.01, **solver_kwargs)

Run grid feedback optimizer from a JSON/EXCEL file path provided as string.

grid_feedback_optimizer.network_to_model_data(network: Network) → OptimizationModelData

Convert a Network object into OptimizationModelData.

grid_feedback_optimizer.solve(network: Network, max_iter: int = 1000, tol: float = 0.001, delta_p: float = 1.0, delta_q: float = 1.0, algorithm: str = 'gp', alpha: float = 0.5, alpha_v: float = 10.0, alpha_l: float = 10.0, alpha_t: float = 10.0, record_iterates: bool = True, solver: str = 'CLARABEL', loading_meas_side: str = 'from', rel_tol: float = 0.0001, rel_tol_line: float = 0.01, **solver_kwargs)

Solve the grid optimization problem by iterating between power flow and optimization.