Coverage for src/distopf/cim_importer/processors/capacitor_processor.py: 92%

1from distopf.cim_importer.processors.base_processor import BaseProcessor 

2from distopf.cim_importer.utils import PhaseUtils 

3import cimgraph.data_profile.cimhub_2023 as cim 

4 

5 

6class CapacitorProcessor(BaseProcessor): 

7 """Processor for LinearShuntCompensator (capacitor) objects.""" 

8 

9 def process(self, network) -> list[dict]: 

10 """Process all LinearShuntCompensator objects for cap_data.csv.""" 

11 results = [] 

12 for capacitor in network.list_by_class(cim.LinearShuntCompensator): 

13 cap_data = self._process_single_capacitor(capacitor) 

14 if cap_data: 

15 results.append(cap_data) 

16 return results 

17 

18 def _process_single_capacitor(self, capacitor) -> dict | None: 

19 """Process an individual capacitor.""" 

20 if not hasattr(capacitor, "Terminals") or not capacitor.Terminals: 

21 return None 

22 bus = capacitor.Terminals[0].ConnectivityNode 

23 cap_data = { 

24 "id": bus.name, 

25 "name": capacitor.name, 

26 "qa": 0.0, 

27 "qb": 0.0, 

28 "qc": 0.0, 

29 "phases": "", 

30 } 

31 

32 v_ln_base = self._get_bus_voltage_base(bus) 

33 

34 phase_data = {} 

35 active_phases = set() 

36 

37 capacitor_phases = ( 

38 capacitor.ShuntCompensatorPhase 

39 if hasattr(capacitor, "ShuntCompensatorPhase") 

40 else [] 

41 ) 

42 

43 if capacitor_phases: 

44 for phase_comp in capacitor_phases: 

45 phase_letter = self._get_phase_str(phase_comp.phase) 

46 if phase_letter: 

47 # bPerSection is Susceptance (B). Q = V^2 * B 

48 b_per_section = ( 

49 float(phase_comp.bPerSection) if phase_comp.bPerSection else 0.0 

50 ) 

51 q_var = (v_ln_base**2) * b_per_section 

52 phase_data[phase_letter] = q_var / self.s_base 

53 active_phases.add(phase_letter) 

54 else: 

55 b_per_section = ( 

56 float(capacitor.bPerSection) 

57 if hasattr(capacitor, "bPerSection") and capacitor.bPerSection 

58 else 0.0 

59 ) 

60 eq_phases_str = PhaseUtils.get_equipment_phases(capacitor) 

61 dist_phases = [p for p in eq_phases_str if p in ["a", "b", "c"]] or [ 

62 "a", 

63 "b", 

64 "c", 

65 ] 

66 

67 for phase_letter in dist_phases: 

68 q_var = (v_ln_base**2) * b_per_section 

69 phase_data[phase_letter] = q_var / self.s_base 

70 active_phases.add(phase_letter) 

71 

72 cap_data["qa"] = phase_data.get("a", 0.0) 

73 cap_data["qb"] = phase_data.get("b", 0.0) 

74 cap_data["qc"] = phase_data.get("c", 0.0) 

75 cap_data["phases"] = "".join(sorted(active_phases)) 

76 return cap_data 

77 

78 def _get_phase_str(self, phase_code) -> str | None: 

79 if not hasattr(phase_code, "value"): 

80 return None 

81 phase_str = str(phase_code.value).lower() 

82 if "a" in phase_str: 

83 return "a" 

84 elif "b" in phase_str: 

85 return "b" 

86 elif "c" in phase_str: 

87 return "c" 

88 return None