Coverage for src/prepare_times_nz/stage_4/baseyear/residential.py: 40%

1"""All baseyear residential veda files

2Mostly built off of one input table, with additional inputs

3including the variable selection/renaming

4And a few other basic inputs defined in the constants section."""

6import numpy as np

7import pandas as pd

8from prepare_times_nz.stage_4.common import (

9 add_extra_input_to_topology,

10 get_processes_with_input_commodity,

11)

12from prepare_times_nz.utilities.data_in_out import _save_data

13from prepare_times_nz.utilities.filepaths import (

14 ASSUMPTIONS,

15 DATA_RAW,

16 STAGE_2_DATA,

17 STAGE_4_DATA,

18)

19from prepare_times_nz.utilities.helpers import select_and_rename

21# FILEPATHS ---------------------------------------------------------------

23INPUT_FILE = STAGE_2_DATA / "residential/baseyear_residential_demand.csv"

24OUTPUT_DIR = STAGE_4_DATA / "base_year_res"

25DEMAND_FLEX_ENABLED_TECHS_FILE = (

26 ASSUMPTIONS / "residential/demand_flex_enabled_techs.csv"

27)

28MODEL_SWITCHES_FILE = DATA_RAW / "user_config/settings/model_switches.csv"

29DEMAND_FLEX_INTERMEDIATES_SWITCH = "ResidentialDemandFlexIntermediates"

31# should instead use save function pattern here!!

32OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

34# CONSTANTS ---------------------------------------------------------------

35ACTIVITY_UNIT = "PJ"

36CAPACITY_UNIT = "GW"

37CAP2ACT = 31.536

39# pylint: disable=duplicate-code

41RESIDENTIAL_DEMAND_VARIABLE_MAP = {

42 "Process": "TechName",

43 "CommodityIn": "Comm-IN",

44 "CommodityOut": "Comm-OUT",

45 "Island": "Region",

46 "Capacity": "PRC_RESID",

47 "AFA": "AFA",

48 "CAPEX": "INVCOST",

49 "OPEX": "FIXOM",

50 "Efficiency": "EFF",

51 "Life": "Life",

52 "CAP2ACT": "CAP2ACT",

53 "OutputEnergy": "ACT_BND",

54}

56DELIVERY_COST_ASSUMPTIONS = {

57 # put me in an assumptions file!!

58 # these are NZDm/PJ or NZD/GJ

59 # anything not listed is assumed 0 (incl LPG)

60 "RESDSL": 0.92,

61 "RESPET": 0.92,

62 "RESWOD": 10,

63}

64# Helpers -----------------------------------------------------------------------

67def parse_switch_value(value):

68 """Parse a user switch value from CSV."""

69 if isinstance(value, bool):

70 return value

72 normalized = str(value).strip().lower()

73 if normalized in {"true", "1", "yes", "y", "on"}:

74 return True

75 if normalized in {"false", "0", "no", "n", "off"}:

76 return False

78 raise ValueError(f"Invalid switch value: {value}")

81def get_model_switch(switch_name, default=True, filepath=MODEL_SWITCHES_FILE):

82 """Read a named user-defined model switch from a simple CSV file."""

83 if not filepath.exists():

84 return default

86 df = pd.read_csv(filepath, encoding="utf-8-sig")

87 if not {"Switch", "Enabled"}.issubset(df.columns):

88 raise ValueError(f"{filepath} must include 'Switch' and 'Enabled' columns")

90 matches = df[df["Switch"] == switch_name]

91 if matches.empty:

92 return default

93 if len(matches) > 1:

94 raise ValueError(f"Duplicate model switch entries found for {switch_name}")

96 return parse_switch_value(matches.iloc[0]["Enabled"])

99def use_demand_flex_intermediates():

100 """Return whether residential demand-flex intermediates are enabled."""

101 return get_model_switch(DEMAND_FLEX_INTERMEDIATES_SWITCH, default=True)

102

103

104def save_residential_veda_file(df, name, label, filepath=OUTPUT_DIR):

105 """Wraps _save_data to send a file to the veda output"""

106 label = f"Saving VEDA table for {label}"

107 _save_data(df=df, name=name, label=label, filepath=filepath)

108

109

110# Main input data =--------------------------------------------------------------

111

112

113def get_residential_veda_table(df, input_map, enable_biogas=True):

114 """convert input table to veda format

115 Option to add biogas to input topology for specific processes

116 """

117 df = df.drop(columns="Unit")

118 # we work wide - pivot

119 index_vars = [col for col in df.columns if col not in ["Variable", "Value"]]

120 df = df.pivot(index=index_vars, columns="Variable", values="Value").reset_index()

121 # add some things

122 df["CAP2ACT"] = CAP2ACT

123 # shape output

124 res_df = select_and_rename(df, input_map)

125

126 if enable_biogas:

127 # if a tech could use nga, we say it can also use biogas

128 res_nga_processes = get_processes_with_input_commodity(res_df, "RESNGA")

129 res_df = add_extra_input_to_topology(res_df, res_nga_processes, "RESBIM")

130

131 return res_df

132

133

134def get_commodity_demand(df):

135 """Aggregate total service demand per commodity"""

136 agg_df = df.groupby(["Region", "Comm-OUT"], as_index=False)["ACT_BND"].sum()

137 # Note: have set label as "Demand" rather than "Demand~2023". Demand should default to base year

138 agg_df = agg_df.rename(columns={"Comm-OUT": "CommName", "ACT_BND": "Demand"})

139 return agg_df

140

141

142def get_demand_flex_enabled_techs(filepath=DEMAND_FLEX_ENABLED_TECHS_FILE):

143 """Load demand-flex-enabled residential technologies."""

144 if not filepath.exists():

145 return set()

146

147 df = pd.read_csv(filepath, encoding="utf-8-sig")

148 return set(df["TechName"].dropna())

149

150

151def get_intermediate_commodity_name(tech_name):

152 """Return the detailed service commodity for a residential demand process."""

153 parts = tech_name.split("-", maxsplit=3)

154 if len(parts) != 4 or parts[0] != "RES":

155 raise ValueError(f"Unexpected residential TechName format: {tech_name}")

156

157 _, region, _fuel, tech_detail = parts

158 return f"{region}-{tech_detail}"

159

160

161def get_demand_flex_topology(df, demand_flex_enabled_techs):

162 """Create intermediate commodity mapping for flex-enabled demand processes."""

163 flex_df = df[df["TechName"].isin(demand_flex_enabled_techs)].copy()

164 if flex_df.empty:

165 return pd.DataFrame(columns=["TechName", "Comm-OUT", "Comm-IN"])

166

167 flex_df["Comm-IN"] = flex_df["TechName"].map(get_intermediate_commodity_name)

168 flex_df = flex_df[["TechName", "Comm-OUT", "Comm-IN"]].drop_duplicates()

169

170 duplicates = flex_df.duplicated("TechName", keep=False)

171 if duplicates.any():

172 duplicate_techs = sorted(flex_df.loc[duplicates, "TechName"].unique())

173 raise ValueError(

174 "Demand-flex technologies map to multiple output commodities: "

175 + ", ".join(duplicate_techs)

176 )

177

178 return flex_df

179

180

181def add_demand_flex_intermediate_outputs(df, demand_flex_topology):

182 """Route demand-flex technologies through detailed intermediate commodities."""

183 if demand_flex_topology.empty:

184 return df

185

186 intermediate_commodities = demand_flex_topology.set_index("TechName")["Comm-IN"]

187 df = df.copy()

188 flex_mask = df["TechName"].isin(intermediate_commodities.index)

189 df.loc[flex_mask, "Comm-OUT"] = df.loc[flex_mask, "TechName"].map(

190 intermediate_commodities

191 )

192 return df

193

194

195# Define processes ----------------------------------------------------------

196

197

198def define_demand_processes(df, filename, label, demand_flex_enabled_techs=None):

199 """Distinct processes for the FI_PRocess table

200 Also add activity and capacity units just for clarity"""

201

202 processes = df["TechName"].unique()

203 demand_flex_enabled_techs = demand_flex_enabled_techs or set()

204

205 demand_df = pd.DataFrame()

206 demand_df["TechName"] = processes

207 demand_df["Sets"] = "DMD"

208 demand_df["Tact"] = ACTIVITY_UNIT

209 demand_df["Tcap"] = CAPACITY_UNIT

210 demand_df["Tslvl"] = np.where(

211 demand_df["TechName"].isin(demand_flex_enabled_techs), "DAYNITE", ""

212 )

213

214 save_residential_veda_file(demand_df, name=filename, label=label)

215

216

217# Define commodities ---------------------------------------------------------

218

219

220def define_enduse_commodities(df, filename, label):

221 """Distinct enduse commodities for the FI_Comm table

222 Also add activity and capacity units just for clarity"""

223

224 commodities = df["Comm-OUT"].unique()

225

226 commodity_df = pd.DataFrame()

227 commodity_df["CommName"] = commodities

228 commodity_df["Csets"] = "DEM"

229 commodity_df["Unit"] = ACTIVITY_UNIT

230 commodity_df["TsLvl"] = "DAYNITE"

231

232 save_residential_veda_file(commodity_df, name=filename, label=label)

233

234

235def define_fuel_commodities(df, filename, label):

236 """Distinct fuel commodities for the FI_Comm table

237 Also add activity and capacity units just for clarity"""

238

239 fuels = df["Comm-IN"].unique()

240

241 fuel_df = pd.DataFrame()

242 fuel_df["CommName"] = fuels

243 fuel_df["Csets"] = "NRG"

244 fuel_df["Unit"] = ACTIVITY_UNIT

245 fuel_df["LimType"] = "FX"

246 fuel_df["TsLvl"] = np.where(fuel_df["CommName"] == "RESELC", "DAYNITE", "")

247

248 save_residential_veda_file(fuel_df, name=filename, label=label)

249

250

251def define_demand_flex_intermediates(demand_flex_topology):

252 """Generate intermediate commodity and pass-through process tables."""

253 if demand_flex_topology.empty:

254 intermediate_commodities = pd.DataFrame(

255 columns=["CommName", "Csets", "Unit", "LimType", "TsLvl"]

256 )

257 intermediate_definitions = pd.DataFrame(

258 columns=["TechName", "Sets", "Tact", "Tcap", "TsLvl"]

259 )

260 intermediate_parameters = pd.DataFrame(

261 columns=["Comm-OUT", "Comm-IN", "TechName", "LIFE", "EFF"]

262 )

263

264 save_residential_veda_file(

265 intermediate_commodities,

266 "intermediate_commodity_definitions.csv",

267 "demand flex intermediate commodity definitions",

268 )

269 save_residential_veda_file(

270 intermediate_definitions,

271 "intermediate_process_definitions.csv",

272 "demand flex intermediate process definitions",

273 )

274 save_residential_veda_file(

275 intermediate_parameters,

276 "intermediate_process_parameters.csv",

277 "demand flex intermediate process parameters",

278 )

279 return

280

281 intermediate_commodities = pd.DataFrame()

282 intermediate_commodities["CommName"] = demand_flex_topology["Comm-IN"].unique()

283 intermediate_commodities["Csets"] = "NRG"

284 intermediate_commodities["Unit"] = ACTIVITY_UNIT

285 intermediate_commodities["LimType"] = "FX"

286 intermediate_commodities["TsLvl"] = "DAYNITE"

287

288 intermediate_parameters = demand_flex_topology[

289 ["Comm-OUT", "Comm-IN"]

290 ].drop_duplicates()

291 intermediate_parameters["TechName"] = "FTE_" + intermediate_parameters["Comm-IN"]

292 intermediate_parameters["LIFE"] = 100

293 intermediate_parameters["EFF"] = 1

294

295 intermediate_definitions = pd.DataFrame(

296 {

297 "TechName": intermediate_parameters["TechName"].unique(),

298 "Sets": "PRE",

299 "Tact": ACTIVITY_UNIT,

300 "Tcap": "PJa",

301 "TsLvl": "DAYNITE",

302 }

303 )

304

305 save_residential_veda_file(

306 intermediate_commodities,

307 "intermediate_commodity_definitions.csv",

308 "demand flex intermediate commodity definitions",

309 )

310 save_residential_veda_file(

311 intermediate_definitions,

312 "intermediate_process_definitions.csv",

313 "demand flex intermediate process definitions",

314 )

315 save_residential_veda_file(

316 intermediate_parameters,

317 "intermediate_process_parameters.csv",

318 "demand flex intermediate process parameters",

319 )

320

321

322# Fuel delivery tables ------------------------------------------------------

323

324

325def define_fuel_delivery(df):

326 """

327 Generates fuel delivery processes for each fuel used in residential sector

328 Adds fuel delivery costs by assumption

329 """

330

331 fuels = df["Comm-IN"].unique()

332

333 fuel_deliv_parameters = pd.DataFrame()

334 fuel_deliv_parameters["Comm-OUT"] = fuels

335 fuel_deliv_parameters["Comm-IN"] = fuel_deliv_parameters[

336 "Comm-OUT"

337 ].str.removeprefix("RES")

338 fuel_deliv_parameters["TechName"] = "FTE_" + fuel_deliv_parameters["Comm-OUT"]

339

340 fuel_deliv_parameters["LIFE"] = 100 # note default to ten years otherwise

341 fuel_deliv_parameters["EFF"] = 1 # pretty sure we don't need this

342

343 fuel_deliv_parameters["VAROM"] = fuel_deliv_parameters["Comm-OUT"].map(

344 DELIVERY_COST_ASSUMPTIONS

345 )

346

347 # Ensure this uses only distributed electricity, gas, or biomethanol

348 dist_fuels = ["ELC", "NGA", "BIM"]

349 fuel_deliv_parameters["Comm-IN"] = np.where(

350 fuel_deliv_parameters["Comm-IN"].isin(dist_fuels),

351 fuel_deliv_parameters["Comm-IN"] + "DD",

352 fuel_deliv_parameters["Comm-IN"],

353 )

354

355 # with the structure defined, we also define the new processes in a separate file (FI_Process)

356 fuel_deliv_definitions = pd.DataFrame(

357 {

358 "TechName": fuel_deliv_parameters["TechName"].unique(),

359 "Sets": "PRE",

360 "Tact": ACTIVITY_UNIT,

361 "Tcap": CAPACITY_UNIT,

362 }

363 )

364 fuel_deliv_definitions["TsLvl"] = np.where(

365 fuel_deliv_definitions["TechName"] == "FTE_RESELC", "DAYNITE", ""

366 )

367

368 save_residential_veda_file(

369 fuel_deliv_parameters,

370 "fuel_delivery_parameters.csv",

371 "fuel delivery parameters",

372 )

373 save_residential_veda_file(

374 fuel_deliv_definitions,

375 "fuel_delivery_definitions.csv",

376 "fuel delivery definitions",

377 )

378

379

380# Main ----------------------------------------------------------------------

381

382

383def main():

384 """script entry point"""

385 # get and transform data

386 raw_df = pd.read_csv(INPUT_FILE)

387 base_res_veda = get_residential_veda_table(raw_df, RESIDENTIAL_DEMAND_VARIABLE_MAP)

388 agg_df = get_commodity_demand(base_res_veda)

389 demand_flex_enabled_techs = (

390 get_demand_flex_enabled_techs() if use_demand_flex_intermediates() else set()

391 )

392 demand_flex_topology = get_demand_flex_topology(

393 base_res_veda, demand_flex_enabled_techs

394 )

395 res_veda = add_demand_flex_intermediate_outputs(base_res_veda, demand_flex_topology)

396

397 # main table

398 save_residential_veda_file(

399 res_veda,

400 name="residential_baseyear_details.csv",

401 label="residential baseyear details",

402 )

403

404 save_residential_veda_file(

405 agg_df,

406 name="residential_commodity_demand.csv",

407 label="residential commodity demand",

408 )

409 # commodity definitions for fi_comm

410 # (Note emissions commodity declared directly in user config file)

411 define_enduse_commodities(

412 base_res_veda,

413 filename="enduse_commodity_definitions.csv",

414 label="enduse commodity definitions",

415 )

416 define_fuel_commodities(

417 res_veda,

418 filename="fuel_commodity_definitions.csv",

419 label="fuel commodity definitions",

420 )

421 define_demand_flex_intermediates(demand_flex_topology)

422

423 # process definitions for fi_process

424 define_demand_processes(

425 res_veda,

426 filename="demand_process_definitions.csv",

427 label="demand process definitions",

428 demand_flex_enabled_techs=demand_flex_enabled_techs,

429 )

430

431 define_fuel_delivery(res_veda)

432

433

434if __name__ == "__main__":

435 main()