Coverage for scripts / stage_3_scenarios / electricity / solar_build_curves.py: 52%

1""" 

2Aggregate NIWA solar hourly profiles to TIMES-NZ timeslices. 

3""" 

4 

5from __future__ import annotations 

6 

7import pandas as pd 

8from prepare_times_nz.utilities.data_in_out import _save_data 

9from prepare_times_nz.utilities.filepaths import ASSUMPTIONS, STAGE_3_DATA 

10 

11STATIC_RENEWABLE_CURVES_FILE = ( 

12 ASSUMPTIONS / "electricity_generation/renewable_curves/RenewableCurves.csv" 

13) 

14SOLAR_ZONE_WEIGHTS_FILE = ( 

15 ASSUMPTIONS / "electricity_generation/renewable_curves/SolarZoneWeights.csv" 

16) 

17 

18OUTPUT_ROOT = STAGE_3_DATA / "electricity/solar_af" 

19HOURLY_DIR = OUTPUT_ROOT / "hourly" 

20SOLAR_AF_DIR = OUTPUT_ROOT / "timeslices" 

21 

22SOLAR_AF_FILE = SOLAR_AF_DIR / "solar_availability_factors.csv" 

23SOLAR_AF_BY_ZONE_FILE = SOLAR_AF_DIR / "solar_availability_factors_by_zone.csv" 

24RENEWABLE_CURVES_FILE = STAGE_3_DATA / "electricity/renewable_curves.csv" 

25 

26SOLAR_TECHS = {"SolarDistSmall", "SolarDistBifacial", "SolarTrack", "SolarFixed"} 

27 

28 

29def load_zone_weights() -> pd.DataFrame: 

30 """ 

31 Load configured solar zone weights and normalize them within each island. 

32 """ 

33 zone_weights = pd.read_csv(SOLAR_ZONE_WEIGHTS_FILE) 

34 required_cols = ["ZoneCode", "Region", "Island", "Weight"] 

35 missing_cols = set(required_cols).difference(zone_weights.columns) 

36 if missing_cols: 

37 missing = ", ".join(sorted(missing_cols)) 

38 raise ValueError( 

39 f"Missing required columns in {SOLAR_ZONE_WEIGHTS_FILE}: {missing}" 

40 ) 

41 

42 zone_weights = zone_weights[required_cols].copy() 

43 if zone_weights["ZoneCode"].duplicated().any(): 

44 duplicates = zone_weights.loc[ 

45 zone_weights["ZoneCode"].duplicated(), "ZoneCode" 

46 ].tolist() 

47 raise ValueError( 

48 f"Duplicate ZoneCode entries in {SOLAR_ZONE_WEIGHTS_FILE}: {duplicates}" 

49 ) 

50 

51 zone_weights["Weight"] = pd.to_numeric(zone_weights["Weight"], errors="raise") 

52 if (zone_weights["Weight"] < 0).any(): 

53 raise ValueError( 

54 f"Negative weights are not allowed in {SOLAR_ZONE_WEIGHTS_FILE}" 

55 ) 

56 

57 island_weight_totals = zone_weights.groupby("Island", as_index=False)[ 

58 "Weight" 

59 ].sum() 

60 zero_weight_islands = island_weight_totals.loc[ 

61 island_weight_totals["Weight"] <= 0, "Island" 

62 ].tolist() 

63 if zero_weight_islands: 

64 raise ValueError( 

65 "Configured solar zone weights must sum to a positive value for each island. " 

66 f"Invalid islands: {zero_weight_islands}" 

67 ) 

68 

69 zone_weights = zone_weights.merge( 

70 island_weight_totals.rename(columns={"Weight": "IslandWeightTotal"}), 

71 on="Island", 

72 how="left", 

73 ) 

74 zone_weights["NormalizedWeight"] = ( 

75 zone_weights["Weight"] / zone_weights["IslandWeightTotal"] 

76 ) 

77 return zone_weights.sort_values(["Island", "ZoneCode"]).reset_index(drop=True) 

78 

79 

80def aggregate_zone_availability_factors(hourly: pd.DataFrame) -> pd.DataFrame: 

81 """ 

82 Aggregate hourly zone output into TIMES-NZ availability factors. 

83 """ 

84 grouped = hourly.groupby( 

85 ["Tech_TIMES", "Scenario", "ZoneCode", "Region", "Island", "TimeSlice"], 

86 as_index=False, 

87 ).agg( 

88 AvailabilityFactor=("generation_kw_per_kw", "mean"), 

89 HoursInTimeSlice=("generation_kw_per_kw", "size"), 

90 ) 

91 

92 return grouped.sort_values(["Tech_TIMES", "ZoneCode", "TimeSlice"]).reset_index( 

93 drop=True 

94 ) 

95 

96 

97def aggregate_island_curves( 

98 zone_factors: pd.DataFrame, zone_weights: pd.DataFrame 

99) -> pd.DataFrame: 

100 """ 

101 Collapse zone-level solar factors to the island granularity used by TIMES-NZ. 

102 """ 

103 weighted = zone_factors.merge( 

104 zone_weights, 

105 on=["ZoneCode", "Region", "Island"], 

106 how="left", 

107 validate="many_to_one", 

108 ) 

109 missing_weights = ( 

110 weighted.loc[weighted["NormalizedWeight"].isna(), "ZoneCode"] 

111 .drop_duplicates() 

112 .tolist() 

113 ) 

114 if missing_weights: 

115 raise ValueError( 

116 "Missing configured solar zone weights for zone codes: " 

117 f"{missing_weights}" 

118 ) 

119 

120 weighted["WeightedAvailability"] = ( 

121 weighted["AvailabilityFactor"] * weighted["NormalizedWeight"] 

122 ) 

123 island = weighted.groupby( 

124 ["Tech_TIMES", "TimeSlice", "Island"], as_index=False 

125 ).agg( 

126 WeightedAvailability=("WeightedAvailability", "sum"), 

127 WeightInTimeSlice=("NormalizedWeight", "sum"), 

128 ) 

129 island["AvailabilityFactor"] = ( 

130 island["WeightedAvailability"] / island["WeightInTimeSlice"] 

131 ) 

132 island = island.pivot( 

133 index=["TimeSlice", "Tech_TIMES"], 

134 columns="Island", 

135 values="AvailabilityFactor", 

136 ).reset_index() 

137 

138 island.columns.name = None 

139 island["NI"] = island["NI"].fillna(0.0) 

140 island["SI"] = island["SI"].fillna(0.0) 

141 return island[["TimeSlice", "Tech_TIMES", "NI", "SI"]].sort_values( 

142 ["Tech_TIMES", "TimeSlice"] 

143 ) 

144 

145 

146def merge_solar_and_static_curves( 

147 solar_curves: pd.DataFrame, static_curves: pd.DataFrame 

148) -> pd.DataFrame: 

149 """ 

150 Replace the static solar rows with generated NIWA-based solar rows. 

151 """ 

152 static = static_curves.copy() 

153 if "TechCode" in static.columns: 

154 static = static.rename(columns={"TechCode": "Tech_TIMES"}) 

155 

156 static = static[~static["Tech_TIMES"].isin(SOLAR_TECHS)] 

157 merged = pd.concat([static, solar_curves], ignore_index=True) 

158 return merged.sort_values(["Tech_TIMES", "TimeSlice"]).reset_index(drop=True) 

159 

160 

161def build_solar_curves(): 

162 """ 

163 Aggregate hourly solar output to TIMES-NZ timeslices and merge with the 

164 non-solar renewable curve assumptions. 

165 """ 

166 SOLAR_AF_DIR.mkdir(parents=True, exist_ok=True) 

167 RENEWABLE_CURVES_FILE.parent.mkdir(parents=True, exist_ok=True) 

168 

169 hourly = pd.read_csv( 

170 HOURLY_DIR / "all_scenarios_hourly_long.csv", parse_dates=["Trading_Date"] 

171 ) 

172 zone_factors = aggregate_zone_availability_factors(hourly) 

173 zone_weights = load_zone_weights() 

174 island_curves = aggregate_island_curves(zone_factors, zone_weights) 

175 static_curves = pd.read_csv(STATIC_RENEWABLE_CURVES_FILE) 

176 merged_curves = merge_solar_and_static_curves(island_curves, static_curves) 

177 

178 _save_data( 

179 zone_factors, 

180 SOLAR_AF_BY_ZONE_FILE.name, 

181 "Zone-level solar availability factors", 

182 SOLAR_AF_BY_ZONE_FILE.parent, 

183 ) 

184 _save_data( 

185 island_curves, 

186 SOLAR_AF_FILE.name, 

187 "Island-level solar availability factors", 

188 SOLAR_AF_FILE.parent, 

189 ) 

190 _save_data( 

191 merged_curves, 

192 RENEWABLE_CURVES_FILE.name, 

193 "Renewable curves with generated solar availability factors", 

194 RENEWABLE_CURVES_FILE.parent, 

195 ) 

196 

197 

198if __name__ == "__main__": 

199 build_solar_curves()