Coverage for src/distopf/cim_converter/processors/switch_processor.py: 98%

1from distopf.cim_converter.processors.base_processor import BaseProcessor 

2from distopf.cim_converter.utils import PhaseUtils 

3import cimgraph.data_profile.cimhub_2023 as cim 

4 

5 

6class SwitchProcessor(BaseProcessor): 

7 """Processor for all switch types.""" 

8 

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

10 """Process all switch subclasses.""" 

11 results = [] 

12 for cim_class in network.graph.keys(): 

13 if ( 

14 hasattr(cim_class, "__mro__") 

15 and cim.Switch in cim_class.__mro__ 

16 and cim_class != cim.Switch 

17 ): 

18 for switch in network.list_by_class(cim_class): 

19 results.append(self._process_switch(switch)) 

20 return results 

21 

22 def _process_switch(self, switch) -> dict: 

23 """Process individual switch.""" 

24 data = self._create_base_branch_dict() 

25 

26 # Basic switch info 

27 data["name"] = switch.name 

28 data["type"] = switch.__class__.__name__.lower() 

29 

30 # Get terminal connections 

31 

32 terminals = switch.Terminals 

33 from_bus = terminals[0].ConnectivityNode 

34 to_bus = terminals[1].ConnectivityNode 

35 data["from_name"] = from_bus.name 

36 data["to_name"] = to_bus.name 

37 

38 # Get voltage base 

39 v_ln_base = self._get_bus_voltage_base(from_bus) 

40 z_base = v_ln_base**2 / self.s_base 

41 data["v_ln_base"] = v_ln_base 

42 data["z_base"] = z_base 

43 

44 # Determine actual phases using utility function 

45 phases = PhaseUtils.get_equipment_phases(switch) 

46 data["phases"] = phases 

47 

48 # Very small impedance for switches - apply only to active phases 

49 r, x = self._get_switch_impedance_per_phase(z_base) 

50 self._apply_switch_impedance(data, phases, r, x) 

51 

52 data["status"] = self._get_switch_status(switch) 

53 

54 return data 

55 

56 def _get_switch_impedance_per_phase(self, z_base: float) -> tuple[float, float]: 

57 """Get per-phase impedance values for switches.""" 

58 # Very small impedance for switches 

59 r = 1e-4 / z_base if z_base > 0 else 1e-4 

60 x = 1e-4 / z_base if z_base > 0 else 1e-4 

61 return r, x 

62 

63 def _apply_switch_impedance(self, data: dict, phases: str, r: float, x: float): 

64 """Apply impedance only to active phases.""" 

65 # Initialize all impedances to zero 

66 for phase_combo in ["aa", "bb", "cc", "ab", "ac", "bc"]: 

67 data[f"r{phase_combo}"] = 0.0 

68 data[f"x{phase_combo}"] = 0.0 

69 

70 # Apply impedance only to active phases 

71 for phase in phases: 

72 if phase in ["a", "b", "c"]: 

73 data[f"r{phase}{phase}"] = r 

74 data[f"x{phase}{phase}"] = x 

75 

76 def _get_switch_status(self, switch) -> str: 

77 """Get switch status (open/closed).""" 

78 if hasattr(switch, "open"): 

79 return "open" if switch.open == "true" else "closed" 

80 return "closed"