Coverage for tests/cim_converter/unit/test_cim_to_csv_helpser.py: 100%

1# tests/unit/test_cim_to_csv_helpers.py 

2import pandas as pd 

3import pytest 

4from distopf.cim_importer.cim_to_csv_converter import CIMToCSVConverter 

5 

6 

7def test_fix_bus_phases_from_branches_simple(): 

8 conv = CIMToCSVConverter(cim_file="unused") 

9 bus_df = pd.DataFrame( 

10 [ 

11 {"id": 1, "name": "sourcebus", "phases": "abc"}, 

12 {"id": 2, "name": "650", "phases": None}, 

13 {"id": 3, "name": "rg60", "phases": None}, 

14 ] 

15 ) 

16 branch_df = pd.DataFrame( 

17 [ 

18 {"fb": 1, "tb": 2, "phases": "ab"}, 

19 {"fb": 2, "tb": 3, "phases": "a"}, 

20 ] 

21 ) 

22 out = conv._fix_bus_phases_from_branches(bus_df.copy(), branch_df) 

23 # The bus with id 2 should get phases from tb mapping (first branch) 

24 assert out.loc[out["id"] == 2, "phases"].iloc[0] in ("ab", "a") 

25 

26 

27def test_fix_downstream_phase_consistency_intersection(): 

28 conv = CIMToCSVConverter(cim_file="unused") 

29 # branch 1: fb 1 -> tb 2 phases 'ab'; branch 2: fb 2 -> tb 3 phases 'abc' should be reduced to 'ab' intersect 'abc' => 'ab' 

30 branch_df = pd.DataFrame( 

31 [ 

32 {"fb": 1, "tb": 2, "phases": "ab"}, 

33 {"fb": 2, "tb": 3, "phases": "abc"}, 

34 ] 

35 ) 

36 out = conv._fix_downstream_phase_consistency(branch_df.copy()) 

37 assert out.loc[1, "phases"] in ("ab", "ba") or set( 

38 list(out.loc[1, "phases"]) 

39 ) == set(list("ab")) 

40 

41 

42def test_aggregate_generators_basic(): 

43 conv = CIMToCSVConverter(cim_file="unused") 

44 gen_df = pd.DataFrame( 

45 [ 

46 { 

47 "mrid": "m1", 

48 "id": 2, 

49 "name": "G1", 

50 "pa": 0.1, 

51 "pb": 0.0, 

52 "pc": 0.0, 

53 "qa": 0.01, 

54 "qb": 0.0, 

55 "qc": 0.0, 

56 "s_base": 100.0, 

57 "sa_max": 0.1, 

58 "sb_max": 0.0, 

59 "sc_max": 0.0, 

60 "phases": "a", 

61 "qa_max": 0.01, 

62 "qb_max": 0.0, 

63 "qc_max": 0.0, 

64 "qa_min": -0.01, 

65 "qb_min": 0.0, 

66 "qc_min": 0.0, 

67 "control_variable": "PQ", 

68 }, 

69 { 

70 "mrid": "m2", 

71 "id": 2, 

72 "name": "G1", 

73 "pa": 0.05, 

74 "pb": 0.0, 

75 "pc": 0.0, 

76 "qa": 0.005, 

77 "qb": 0.0, 

78 "qc": 0.0, 

79 "s_base": 50.0, 

80 "sa_max": 0.05, 

81 "sb_max": 0.0, 

82 "sc_max": 0.0, 

83 "phases": "a", 

84 "qa_max": 0.005, 

85 "qb_max": 0.0, 

86 "qc_max": 0.0, 

87 "qa_min": -0.005, 

88 "qb_min": 0.0, 

89 "qc_min": 0.0, 

90 "control_variable": "PQ", 

91 }, 

92 ] 

93 ) 

94 aggregated = conv._aggregate_generators(gen_df) 

95 # Should be single row for id 2 

96 assert len(aggregated) == 1 

97 row = aggregated.iloc[0] 

98 # MRIDs joined with '|' 

99 assert "m1" in row["mrid"] and "m2" in row["mrid"] 

100 # Name should indicate aggregation since >1 units 

101 assert "AggGen" in row["name"] 

102 # pa sums 

103 assert pytest.approx(row["pa"], rel=1e-12) == 0.15 

104 

105 

106def test_correct_generators_without_phases_and_distribution(): 

107 conv = CIMToCSVConverter(cim_file="unused") 

108 bus_df = pd.DataFrame( 

109 [ 

110 {"id": 1, "phases": "abc"}, 

111 {"id": 2, "phases": "a"}, 

112 ] 

113 ) 

114 # generator without phases but with p and q columns 

115 gen_df = pd.DataFrame( 

116 [ 

117 { 

118 "id": 2, 

119 "phases": "", 

120 "p": 0.06, 

121 "q": 0.006, 

122 "pa": 0.0, 

123 "pb": 0.0, 

124 "pc": 0.0, 

125 "qa": 0.0, 

126 "qb": 0.0, 

127 "qc": 0.0, 

128 } 

129 ] 

130 ) 

131 corrected = conv._correct_generators_without_phases(bus_df, gen_df.copy()) 

132 # Since bus 2 has phases 'a' only, pa should equal p (0.06) 

133 assert pytest.approx(corrected.loc[0, "pa"], rel=1e-12) == pytest.approx(0.06) 

134 assert corrected.loc[0, "phases"] == "a" or corrected.loc[0, "phases"] == "" 

135 

136 

137def test_convert_secondary_loads_moves_s1_s2_into_primary_phase(): 

138 conv = CIMToCSVConverter(cim_file="unused") 

139 # Create bus df with s1/s2 loads 

140 bus_df = pd.DataFrame( 

141 [ 

142 { 

143 "id": 10, 

144 "phases": "a", 

145 "pl_s1": 0.02, 

146 "pl_s2": 0.03, 

147 "ql_s1": 0.005, 

148 "ql_s2": 0.006, 

149 "pl_a": 0.0, 

150 "ql_a": 0.0, 

151 "pl_b": 0.0, 

152 "ql_b": 0.0, 

153 "pl_c": 0.0, 

154 "ql_c": 0.0, 

155 } 

156 ] 

157 ) 

158 out = conv._convert_secondary_loads(bus_df.copy()) 

159 # Should have moved s1+s2 into pl_a and ql_a 

160 assert pytest.approx( 

161 out.loc[out["id"] == 10, "pl_a"].iloc[0], rel=1e-12 

162 ) == pytest.approx(0.05) 

163 assert pytest.approx( 

164 out.loc[out["id"] == 10, "ql_a"].iloc[0], rel=1e-12 

165 ) == pytest.approx(0.011)