Coverage for tests/cim_converter/unit/test_processors.py: 93%

1# tests/unit/test_processors.py 

2import math 

3import pytest 

4 

5from cimgraph.data_profile import cimhub_2023 as cim 

6from distopf.cim_importer.processors.base_processor import BaseProcessor 

7from distopf.cim_importer.processors.line_processor import LineProcessor 

8from distopf.cim_importer.processors.switch_processor import SwitchProcessor 

9from distopf.cim_importer.processors.capacitor_processor import CapacitorProcessor 

10from distopf.cim_importer.processors.generator_processor import GeneratorProcessor 

11from distopf.cim_importer.utils.phase_utils import PhaseUtils 

12 

13 

14class ConcreteBase(BaseProcessor): 

15 """Concrete subclass so we can instantiate BaseProcessor for tests.""" 

16 

17 def __init__(self, s_base: float = 1e6): 

18 super().__init__(s_base) 

19 

20 def process(self, network): 

21 return [] 

22 

23 

24def make_connectivity_node_with_voltage(nominal_voltage: float): 

25 """Create a mock ConnectivityNode whose Terminals include one with a ConductingEquipment.BaseVoltage.nominalVoltage.""" 

26 base_voltage = cim.BaseVoltage(nominalVoltage=float(nominal_voltage)) 

27 conducting_equipment = cim.ConductingEquipment() 

28 conducting_equipment.BaseVoltage = base_voltage 

29 terminal = cim.Terminal(ConductingEquipment=conducting_equipment) 

30 node = cim.ConnectivityNode(mRID="mrid_mock", name="mockbus") 

31 terminal.ConnectivityNode = node 

32 node.Terminals = [terminal] 

33 return node 

34 

35 

36def make_line_with_phase_impedance(nominal_voltage=480.0, length=100.0): 

37 """Create an ACLineSegment with PhaseImpedanceData for LineProcessor tests.""" 

38 node = make_connectivity_node_with_voltage(nominal_voltage) 

39 terminal0 = cim.Terminal(ConnectivityNode=node) 

40 terminal1 = cim.Terminal(ConnectivityNode=node) 

41 # Create PhaseImpedanceData entries (dataclass) 

42 p1 = cim.PhaseImpedanceData(row=1, column=1, r=0.1, x=0.2) 

43 p2 = cim.PhaseImpedanceData(row=1, column=2, r=0.01, x=0.02) 

44 p3 = cim.PhaseImpedanceData(row=2, column=2, r=0.11, x=0.21) 

45 # Prefer using a real PerLengthPhaseImpedance if available; else attach a lightweight holder 

46 try: 

47 per_length = cim.PerLengthPhaseImpedance(PhaseImpedanceData=[p1, p2, p3]) 

48 except AttributeError: 

49 

50 class _PerLength: 

51 def __init__(self, phase_list): 

52 self.PhaseImpedanceData = phase_list 

53 

54 per_length = _PerLength([p1, p2, p3]) 

55 line = cim.ACLineSegment( 

56 name="testline", Terminals=[terminal0, terminal1], length=length 

57 ) 

58 line.PerLengthImpedance = per_length 

59 return line 

60 

61 

62def test_base_processor_create_and_terminals_and_voltage(): 

63 bp = ConcreteBase(1e6) 

64 base = bp._create_base_branch_dict() 

65 # Keys existence 

66 assert "name" in base and "raa" in base and "xaa" in base 

67 # Test terminals info raises with insufficient terminals; use a real object with a name attribute 

68 equip = cim.ConductingEquipment() 

69 equip.name = "equip1" 

70 equip.Terminals = [cim.Terminal(ConnectivityNode=cim.ConnectivityNode(name="n1"))] 

71 with pytest.raises(ValueError): 

72 bp._get_terminals_info(equip) 

73 # Test _get_bus_voltage_base picks nominalVoltage 

74 node = make_connectivity_node_with_voltage(480.0) 

75 v_ln = bp._get_bus_voltage_base(node) 

76 assert pytest.approx(v_ln * math.sqrt(3), rel=1e-6) == 480.0 

77 

78 

79def test_line_processor_impedance_mapping_and_scaling(): 

80 lp = LineProcessor(1e6) 

81 line = make_line_with_phase_impedance(nominal_voltage=480.0, length=100.0) 

82 data = lp._process_line(line) 

83 # v_ln_base should be nominal/sqrt(3) 

84 assert pytest.approx(data["v_ln_base"] * math.sqrt(3), rel=1e-6) == 480.0 

85 # z_base = v_ln^2 / s_base => check raa = length * r / z_base 

86 z_base = data["z_base"] 

87 expected_raa = 100.0 * 0.1 / z_base 

88 assert pytest.approx(data["raa"], rel=1e-9) == expected_raa 

89 # check ab value present 

90 expected_rab = 100.0 * 0.01 / z_base 

91 assert pytest.approx(data["rab"], rel=1e-9) == expected_rab 

92 assert data["length"] == 100.0 

93 assert data["type"] == "ACLineSegment" 

94 

95 

96def test_switch_processor_impedance_and_status(monkeypatch): 

97 # Patch PhaseUtils.get_equipment_phases for deterministic behavior 

98 monkeypatch.setattr( 

99 PhaseUtils, "get_equipment_phases", staticmethod(lambda eq: "ac") 

100 ) 

101 sp = SwitchProcessor(1e6) 

102 node = make_connectivity_node_with_voltage(480.0) 

103 term0 = cim.Terminal(ConnectivityNode=node) 

104 term1 = cim.Terminal(ConnectivityNode=node) 

105 # Use a real switch class (LoadBreakSwitch) and set 'open' attribute as string "true" 

106 switch = cim.LoadBreakSwitch(name="mysw", Terminals=[term0, term1]) 

107 # The code expects switch.open to be the string 'true' for open, so set it to that 

108 setattr(switch, "open", "true") 

109 data = sp._process_switch(switch) 

110 # Check phases applied and only aa/cc filled (a and c), bb remains zero 

111 assert data["phases"] == "ac" 

112 assert data["raa"] > 0 

113 assert data["rcc"] > 0 

114 assert data["rbb"] == 0.0 

115 assert data["status"] == "open" 

116 

117 

118def test_capacitor_processor_shunt_phases(monkeypatch): 

119 cap_proc = CapacitorProcessor(1e6) 

120 node = make_connectivity_node_with_voltage(480.0) 

121 terminal = cim.Terminal(ConnectivityNode=node) 

122 # Create a Phase-like object with .value attribute (avoid SimpleNamespace) 

123 val_obj = type("Val", (), {})() 

124 val_obj.value = "A" 

125 phase_comp = cim.LinearShuntCompensatorPhase() 

126 phase_comp.phase = val_obj 

127 phase_comp.bPerSection = 1e-6 

128 cap = cim.LinearShuntCompensator( 

129 name="cap1", Terminals=[terminal], ShuntCompensatorPhase=[phase_comp] 

130 ) 

131 cap_data = cap_proc._process_single_capacitor(cap) 

132 assert cap_data["phases"] == "a" 

133 assert cap_data["qa"] >= 0.0 

134 assert cap_data["qb"] == 0.0 

135 assert cap_data["qc"] == 0.0 

136 

137 

138def test_generator_processor_power_electronics_phases(monkeypatch): 

139 monkeypatch.setattr( 

140 PhaseUtils, 

141 "get_phase_str", 

142 staticmethod(lambda val: "a" if "A" in str(val).upper() else None), 

143 ) 

144 gp = GeneratorProcessor(1e6) 

145 node = make_connectivity_node_with_voltage(480.0) 

146 terminal = cim.Terminal(ConnectivityNode=node) 

147 # Build a PowerElectronicsConnection and a PowerElectronicsConnectionPhase 

148 pec = cim.PowerElectronicsConnection() 

149 pec.Terminals = [terminal] 

150 pec.mRID = "mrid_pec" 

151 pec.name = "PV1" 

152 pec.ratedS = 10000.0 

153 pec.p = 1000.0 

154 pec.q = 200.0 

155 pec.maxQ = 50.0 

156 pec.minQ = -50.0 

157 # Create a phase object with .value attribute 

158 phase_val = type("Phase", (), {})() 

159 phase_val.value = "A" 

160 pec_phase = cim.PowerElectronicsConnectionPhase() 

161 pec_phase.phase = phase_val 

162 pec_phase.p = 1000.0 

163 pec_phase.q = 200.0 

164 pec.PowerElectronicsConnectionPhases = [pec_phase] 

165 gen = gp._process_power_electronics_connection(pec) 

166 assert isinstance(gen, dict) 

167 expected_total_pu = pec.p / gp.s_base 

168 if "p" in gen: 

169 assert pytest.approx(gen["p"], rel=1e-9) == expected_total_pu 

170 else: 

171 sum_abc = sum(gen.get(k, 0.0) for k in ("pa", "pb", "pc")) 

172 assert pytest.approx(sum_abc, rel=1e-6) == expected_total_pu