Coverage for src/distopf/cim_converter/utils/phase_utils.py: 97%

1""" 

2Utility functions for handling phase information in CIM objects. 

3Centralizes phase determination logic to ensure consistency across processors. 

4""" 

5 

6import cimgraph.data_profile.cimhub_2023 as cim 

7 

8 

9class PhaseUtils: 

10 """Utility class for phase-related operations.""" 

11 

12 @staticmethod 

13 def get_phase_str( 

14 phase_code: cim.OrderedPhaseCodeKind | cim.SinglePhaseKind | None, 

15 ) -> str: 

16 """ 

17 Convert CIM phase code to phase letter. 

18 

19 Args: 

20 phase_code: CIM phase code object or value 

21 

22 Returns: 

23 Phase letters 

24 """ 

25 if phase_code is None: 

26 return "" 

27 assert phase_code is not None 

28 return phase_code.value.lower().replace("n", "") 

29 

30 @staticmethod 

31 def get_equipment_phases(equipment) -> str: 

32 """ 

33 Get phase string from equipment by checking various phase-specific objects. 

34 

35 Args: 

36 equipment: CIM equipment object 

37 

38 Returns: 

39 Sorted phase string (e.g., 'abc', 'ac', 's1s2') 

40 """ 

41 phases = set() 

42 

43 # Check for phase-specific objects 

44 phase_attrs = [ 

45 "ACLineSegmentPhases", 

46 "EnergyConsumerPhase", 

47 "PowerElectronicsConnectionPhases", 

48 "LinearShuntCompensatorPhases", 

49 "ShuntCompensatorPhase", 

50 "PowerTransformerEnd", 

51 "SwitchPhases", 

52 "TransformerTankEnds", 

53 ] 

54 

55 for attr in phase_attrs: 

56 if not hasattr(equipment, attr): 

57 continue 

58 phase_objects = getattr(equipment, attr) 

59 if phase_objects is None: 

60 continue 

61 for phase_obj in phase_objects: 

62 if hasattr(phase_obj, "phase"): 

63 phase_letter = PhaseUtils.get_phase_str(phase_obj.phase) 

64 phases.add(phase_letter) 

65 if hasattr(phase_obj, "orderedPhases"): 

66 phase_letter = PhaseUtils.get_phase_str(phase_obj.orderedPhases) 

67 phases.add(phase_letter) 

68 

69 # If no phase-specific data found, assume three-phase 

70 if not phases: 

71 phases = {"a", "b", "c"} 

72 

73 return "".join(sorted(phases)) 

74 

75 @staticmethod 

76 def filter_standard_phases(phases: str) -> str: 

77 """ 

78 Filter to only include standard three-phase system phases (a, b, c). 

79 

80 Args: 

81 phases: Phase string that may include secondary phases 

82 

83 Returns: 

84 Filtered phase string with only a, b, c phases 

85 """ 

86 standard_phases = set() 

87 for phase in phases: 

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

89 standard_phases.add(phase) 

90 

91 # Default to three-phase if no standard phases found 

92 if not standard_phases: 

93 standard_phases = {"a", "b", "c"} 

94 

95 return "".join(sorted(standard_phases))