Coverage for src/distopf/cim_converter/validators/topology_validator.py: 71%

1# cim_converter/validators/topology_validator.py 

2import logging 

3import pandas as pd 

4from typing import Dict, List, Set 

5import networkx as nx 

6 

7_log = logging.getLogger(__name__) 

8 

9 

10class TopologyValidator: 

11 """Validates power system topology for electrical correctness.""" 

12 

13 def __init__(self): 

14 self.tolerance = 1e-6 

15 

16 def validate_tree_topology(self, branch_data: pd.DataFrame) -> Dict: 

17 """ 

18 Validate that the power system forms a proper tree topology. 

19 

20 Args: 

21 branch_data: DataFrame containing branch information 

22 

23 Returns: 

24 Dict with validation results 

25 """ 

26 validation_result = {"valid": True, "issues": [], "warnings": []} 

27 

28 if branch_data.empty: 

29 validation_result["valid"] = False 

30 validation_result["issues"].append("No branch data found") 

31 return validation_result 

32 

33 try: 

34 graph = self._build_graph(branch_data) 

35 self._check_connectivity(graph, validation_result) 

36 self._check_radial_topology(graph, validation_result) 

37 self._check_electrical_consistency(branch_data, validation_result) 

38 self._check_orphaned_buses(branch_data, validation_result) 

39 

40 except Exception as e: 

41 validation_result["valid"] = False 

42 validation_result["issues"].append(f"Validation error: {str(e)}") 

43 _log.exception("Error during topology validation") 

44 

45 return validation_result 

46 

47 def _check_connectivity(self, graph: nx.Graph, result: Dict): 

48 """Check that all buses are connected.""" 

49 if not graph.nodes(): 

50 result["issues"].append("No buses found in network") 

51 result["valid"] = False 

52 return 

53 

54 if not nx.is_connected(graph): 

55 num_components = nx.number_connected_components(graph) 

56 result["issues"].append( 

57 f"Network has {num_components} disconnected components" 

58 ) 

59 result["valid"] = False 

60 

61 def _check_radial_topology(self, graph: nx.Graph, result: Dict): 

62 """Check for radial (tree-like) topology - no loops allowed.""" 

63 if not graph.nodes(): 

64 return 

65 

66 num_nodes = graph.number_of_nodes() 

67 num_edges = graph.number_of_edges() 

68 expected_edges = num_nodes - 1 

69 

70 if num_edges > expected_edges: 

71 result["warnings"].append( 

72 f"Network may have loops: {num_edges} edges for {num_nodes} nodes" 

73 ) 

74 elif num_edges < expected_edges and nx.is_connected(graph): 

75 # This case is already covered by check_connectivity if it leads to disconnection 

76 pass 

77 

78 try: 

79 cycles = list(nx.simple_cycles(graph)) 

80 if cycles: 

81 result["warnings"].append(f"Found {len(cycles)} cycles in network.") 

82 except Exception: 

83 pass 

84 

85 def _check_electrical_consistency(self, branch_data: pd.DataFrame, result: Dict): 

86 """Check for electrical parameter consistency.""" 

87 issues = [] 

88 impedance_cols = ["raa", "rbb", "rcc", "xaa", "xbb", "xcc"] 

89 

90 for col in impedance_cols: 

91 if col in branch_data.columns: 

92 negative_values = branch_data[branch_data[col] < -self.tolerance] 

93 if not negative_values.empty: 

94 issues.append( 

95 f"Found negative {col} values in {len(negative_values)} branches" 

96 ) 

97 

98 if ( 

99 "type" in branch_data.columns 

100 and "raa" in branch_data.columns 

101 and "xaa" in branch_data.columns 

102 ): 

103 non_switch_types = ["ACLineSegment", "transformer"] 

104 for _, row in branch_data.iterrows(): 

105 if row.get("type") in non_switch_types: 

106 total_z = (row.get("raa", 0) ** 2 + row.get("xaa", 0) ** 2) ** 0.5 

107 if total_z < self.tolerance: 

108 issues.append( 

109 f"Branch {row.get('name', 'unknown')} has zero impedance" 

110 ) 

111 

112 if "v_ln_base" in branch_data.columns: 

113 missing_voltage = branch_data[branch_data["v_ln_base"].isna()] 

114 if not missing_voltage.empty: 

115 issues.append( 

116 f"Missing voltage base for {len(missing_voltage)} branches" 

117 ) 

118 

119 if issues: 

120 result["warnings"].extend(issues) 

121 

122 def _check_orphaned_buses(self, branch_data: pd.DataFrame, result: Dict): 

123 """Check for buses that aren't connected to any branches.""" 

124 from_buses = set(branch_data["from_name"].dropna()) 

125 to_buses = set(branch_data["to_name"].dropna()) 

126 all_buses = from_buses.union(to_buses) 

127 

128 if len(all_buses) < 2: 

129 result["issues"].append("Network has fewer than 2 buses") 

130 result["valid"] = False 

131 

132 def _build_graph(self, branch_data: pd.DataFrame) -> nx.Graph: 

133 """Build NetworkX graph from branch data.""" 

134 graph = nx.Graph() 

135 for _, row in branch_data.iterrows(): 

136 from_bus = row.get("from_name") 

137 to_bus = row.get("to_name") 

138 if pd.notna(from_bus) and pd.notna(to_bus): 

139 graph.add_edge(from_bus, to_bus, **row.to_dict()) 

140 return graph 

141 

142 def validate_power_flow_data( 

143 self, bus_data: pd.DataFrame, branch_data: pd.DataFrame 

144 ) -> Dict: 

145 """Validate data for power flow analysis.""" 

146 validation_result = {"valid": True, "issues": [], "warnings": []} 

147 if bus_data.empty: 

148 return validation_result 

149 

150 swing_buses = bus_data[bus_data["bus_type"] == "SWING"] 

151 if swing_buses.empty: 

152 validation_result["issues"].append("No swing bus found") 

153 validation_result["valid"] = False 

154 elif len(swing_buses) > 1: 

155 validation_result["warnings"].append( 

156 f"Multiple swing buses found: {len(swing_buses)}" 

157 ) 

158 

159 load_cols = ["pl_a", "pl_b", "pl_c", "ql_a", "ql_b", "ql_c"] 

160 for col in load_cols: 

161 if col in bus_data.columns and bus_data[col].isna().any(): 

162 validation_result["warnings"].append( 

163 f"Missing load data in column {col}" 

164 ) 

165 

166 return validation_result