Coverage for src/prepare_times_nz/stage_4/common.py: 16%

1""" 

2This module stores some replicable patterns that are often useful in 

3building Veda tables 

4 

5""" 

6 

7import pandas as pd 

8 

9 

10def apply_share_constraints(df, share_constraints=None): 

11 """Apply optional share constraints to copied topology rows.""" 

12 if share_constraints is None: 

13 return df 

14 

15 base_year_share_up = share_constraints.get("base_year_share_up") 

16 future_share_year = share_constraints.get("future_share_year") 

17 future_share_up = share_constraints.get("future_share_up") 

18 

19 if base_year_share_up is not None: 

20 df["Share-I~UP"] = base_year_share_up 

21 if future_share_year is not None and future_share_up is not None: 

22 df[f"Share-I~UP~{future_share_year}"] = future_share_up 

23 

24 return df 

25 

26 

27def add_extra_input_to_topology( 

28 df, 

29 processes_to_expand, 

30 new_input, 

31 share_constraints=None, 

32): 

33 """ 

34 A method for adding an extra input option for base year parameters 

35 

36 Works by duplicating existing process information but changing the Comm-IN 

37 

38 input df must have process codes under 'TechName' 

39 and standard 'Comm-IN'/'Comm-OUT' variables 

40 as expected in baseyear FI_T tables 

41 

42 Optional share constraints can be added to keep the new input unavailable 

43 in the base year while allowing it in later years. 

44 """ 

45 

46 # if a tech could use these fuels, we say it can also use biogas 

47 # all other parameters remain the same 

48 new_input_df = df[df["TechName"].isin(processes_to_expand)].copy() 

49 # tech can use biogas 

50 new_input_df["Comm-IN"] = new_input 

51 # ACT_BND is a process-level base-year activity. Leave it blank on copied 

52 # input rows so we don't double-count demand or accidentally fix the 

53 # multi-input process activity through the duplicate row. 

54 if "ACT_BND" in new_input_df.columns: 

55 new_input_df["ACT_BND"] = pd.NA 

56 new_input_df = apply_share_constraints(new_input_df, share_constraints) 

57 # add to main table 

58 df = pd.concat([df, new_input_df]) 

59 # sort for clearer reads 

60 df = df.sort_values(["TechName", "Comm-IN"]) 

61 

62 return df 

63 

64 

65def get_processes_with_input_commodity(df, input_commodity): 

66 """ 

67 With a df containing 'TechName' and 'Comm-IN', 

68 returns the list of processes that use the specified input_commodity in "Comm-IN" 

69 

70 """ 

71 

72 processes = df[df["Comm-IN"].isin([input_commodity])].copy() 

73 process_list = processes["TechName"].unique().tolist() 

74 

75 return process_list