Coverage for src/distopf/matrix_models/lindist_capacitor_mi.py: 22%

1from typing import Optional 

2import numpy as np 

3import pandas as pd 

4from numpy import zeros 

5from scipy.sparse import csr_array, lil_array, vstack 

6from distopf.matrix_models.base import LinDistBase 

7from distopf.utils import get 

8 

9 

10class LinDistModelCapMI(LinDistBase): 

11 """ 

12 LinDistFlow Model with support for capacitor bank control. 

13 

14 Parameters 

15 ---------- 

16 branch_data : pd.DataFrame 

17 DataFrame containing branch data (r and x values, limits) 

18 bus_data : pd.DataFrame 

19 DataFrame containing bus data (loads, voltages, limits) 

20 gen_data : pd.DataFrame 

21 DataFrame containing generator/DER data 

22 cap_data : pd.DataFrame 

23 DataFrame containing capacitor data 

24 reg_data : pd.DataFrame 

25 DataFrame containing regulator data 

26 

27 """ 

28 

29 def __init__( 

30 self, 

31 branch_data: Optional[pd.DataFrame] = None, 

32 bus_data: Optional[pd.DataFrame] = None, 

33 gen_data: Optional[pd.DataFrame] = None, 

34 cap_data: Optional[pd.DataFrame] = None, 

35 reg_data: Optional[pd.DataFrame] = None, 

36 ): 

37 super().__init__( 

38 branch_data, bus_data, gen_data, cap_data=cap_data, reg_data=reg_data 

39 ) 

40 self.build() 

41 

42 def initialize_variable_index_pointers(self): 

43 self.x_maps, self.n_x = self._variable_tables(self.branch) 

44 self.v_map, self.n_x = self._add_device_variables(self.n_x, self.all_buses) 

45 self.pg_map, self.n_x = self._add_device_variables(self.n_x, self.gen_buses) 

46 self.qg_map, self.n_x = self._add_device_variables(self.n_x, self.gen_buses) 

47 self.qc_map, self.n_x = self._add_device_variables(self.n_x, self.cap_buses) 

48 self.vx_map, self.n_x = self._add_device_variables(self.n_x, self.reg_buses) 

49 self.zc_map, self.n_x = self._add_device_variables(self.n_x, self.cap_buses) 

50 self.uc_map, self.n_x = self._add_device_variables(self.n_x, self.cap_buses) 

51 

52 def additional_variable_idx(self, var, node_j, phase): 

53 """ 

54 User added index function. Override this function to add custom variables. Return None if `var` is not found. 

55 Parameters 

56 ---------- 

57 var : name of variable 

58 node_j : node index (0 based; bus.id - 1) 

59 phase : "a", "b", or "c" 

60 

61 Returns 

62 ------- 

63 ix : index or list of indices of variable within x-vector or None if `var` is not found. 

64 """ 

65 if var in ["zc"]: 

66 return self.zc_map[phase].get(node_j, []) 

67 if var in ["uc"]: 

68 return self.uc_map[phase].get(node_j, []) 

69 return None 

70 

71 def add_capacitor_model( 

72 self, a_eq: lil_array, b_eq, j, a 

73 ) -> tuple[lil_array, np.ndarray]: 

74 qij = self.idx("qij", j, a) 

75 q_cap_nom = 0 

76 if self.cap is not None: 

77 q_cap_nom = get(self.cap[f"q{a}"], j, 0) 

78 # equation indexes 

79 zc = self.idx("zc", j, a) 

80 qc = self.idx("q_cap", j, a) 

81 a_eq[qij, qc] = 1 # add capacitor q variable to power flow equation 

82 a_eq[qc, qc] = 1 

83 a_eq[qc, zc] = -q_cap_nom 

84 return a_eq, b_eq 

85 

86 def create_capacitor_constraints(self) -> tuple[csr_array, np.ndarray]: 

87 """ 

88 Create inequality constraints for the optimization problem. 

89 """ 

90 

91 # ########## Aineq and Bineq Formation ########### 

92 n_rows_ineq = 4 * ( 

93 len(self.cap_buses["a"]) 

94 + len(self.cap_buses["b"]) 

95 + len(self.cap_buses["c"]) 

96 ) 

97 a_ineq = lil_array((n_rows_ineq, self.n_x)) 

98 b_ineq = zeros(n_rows_ineq) 

99 ineq1 = 0 

100 ineq2 = 1 

101 ineq3 = 2 

102 ineq4 = 3 

103 for j in self.cap.index: 

104 for a in "abc": 

105 if not self.phase_exists(a, j): 

106 continue 

107 # equation indexes 

108 v_max = get(self.bus["v_max"], j) ** 2 

109 a_ineq[ineq1, self.idx("zc", j, a)] = 1 

110 a_ineq[ineq1, self.idx("uc", j, a)] = -v_max 

111 a_ineq[ineq2, self.idx("zc", j, a)] = 1 

112 a_ineq[ineq2, self.idx("v", j, a)] = -1 

113 a_ineq[ineq3, self.idx("zc", j, a)] = -1 

114 a_ineq[ineq3, self.idx("v", j, a)] = +1 

115 a_ineq[ineq3, self.idx("uc", j, a)] = v_max 

116 b_ineq[ineq3] = v_max 

117 a_ineq[ineq4, self.idx("zc", j, a)] = -1 

118 ineq1 += 4 

119 ineq2 += 4 

120 ineq3 += 4 

121 ineq4 += 4 

122 

123 return csr_array(a_ineq), b_ineq 

124 

125 def create_inequality_constraints(self) -> tuple[csr_array, np.ndarray]: 

126 a_cap, b_cap = self.create_capacitor_constraints() 

127 a_inv, b_inv = self.create_octagon_constraints() 

128 a_ub = vstack([a_cap, a_inv]) 

129 b_ub = np.r_[b_cap, b_inv] 

130 return csr_array(a_ub), b_ub 

131 

132 def get_zc(self, x): 

133 return self.get_device_variables(x, self.zc_map) 

134 

135 def get_uc(self, x): 

136 return self.get_device_variables(x, self.uc_map)