26601056421

Committed 28 May 2026 08:42PM UTC coverage: 88.671% (-0.4%) from 89.062%

Build # 26601056421

Build Type

push

github

Committed by

web-flow

Commit Message

GridPath v2026.3.0

Merge pull request #1376 from blue-marble/develop

Coverage Stats

1314 of 1580 new or added lines in 77 files covered. (83.16%)

44 existing lines in 11 files now uncovered.

28318 of 31936 relevant lines covered (88.67%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.04

/data_toolkit/temporal/create_temporal_iteration_csv.py

# Copyright 2016-2024 Blue Marble Analytics LLC.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Create temporal iterations CSV from user-defined params. Does not cover all
possible cases yet.
"""

import sys
from argparse import ArgumentParser
import csv
import os.path
import pandas as pd
import random

N_PASSES_DEFAULT = 1


def parse_arguments(args):
    """
    :param args: the script arguments specified by the user
    :return: the parsed known argument values (<class 'argparse.Namespace'>
    Python object)

    Parse the known arguments.
    """

    parser = ArgumentParser(add_help=True)

    parser.add_argument(
        "-n",
        "--n_passes",
        default=N_PASSES_DEFAULT,
        help=f"Defaults to {N_PASSES_DEFAULT}.",
    )
    parser.add_argument("-csv", "--iterations_csv_path")

    parser.add_argument("-o", "--output_directory")

    parser.add_argument("-q", "--quiet", default=False, action="store_true")

    parsed_arguments = parser.parse_known_args(args=args)[0]

    return parsed_arguments


def create_temporal_scenario_iterations_csv(n_passes, filepath, output_directory):
    with open(os.path.join(output_directory, "iterations.csv"), "w") as f:
        writer = csv.writer(f, delimiter=",")
        writer.writerow(
            ["weather_iteration", "hydro_iteration", "availability_iteration"]
        )

    # print(os.path.abspath(filepath))
    df = pd.read_csv(filepath)

    weather_df = df["weather"]
    weather_list = [i[1] for i in weather_df.items()]
    weather_mode = weather_list[0]
    weather_iterations_pass = [i for i in weather_list[1:] if not pd.isna(i)]

    hydro_df = df["hydro"]
    hydro_list = [i[1] for i in hydro_df.items()]
    hydro_mode = hydro_list[0]
    hydro_iterations_pass = [i for i in hydro_list[1:] if not pd.isna(i)]

    availability_df = df["availability"]
    availability_list = [i[1] for i in availability_df.items()]
    availability_mode = availability_list[0]
    availability_iterations_pass = [i for i in availability_list[1:] if not pd.isna(i)]

    # TODO: possibly remove
    weather_iteration, hydro_iteration, availability_iteration = None, None, None

    for n in range(n_passes):
        weather_iterations = weather_iterations_pass.copy()
        hydro_iterations = hydro_iterations_pass.copy()
        availability_iterations = availability_iterations_pass.copy()

        av_current_index = 0
        hy_current_index = 0
        if weather_mode == "loop":
            for weather_iteration in weather_iterations:
                if hydro_mode == "loop":
                    for hydro_iteration in hydro_iterations:
                        (
                            availability_iteration,
                            av_current_index,
                        ) = get_availability_iteration(
                            availability_mode=availability_mode,
                            availability_iterations=availability_iterations,
                            av_current_index=av_current_index,
                        )

                        with open(
                            os.path.join(output_directory, "iterations.csv"), "a"
                        ) as f_out:
                            writer = csv.writer(f_out, delimiter=",")
                            writer.writerow(
                                [
                                    weather_iteration,
                                    hydro_iteration,
                                    availability_iteration,
                                ]
                            )
                else:
                    if hydro_mode == "ordered":
                        hydro_iteration = hydro_iterations[hy_current_index]
                        hy_current_index += 1
                    elif hydro_mode == "random_remove":
                        hydro_iteration = random_remove(hydro_iterations)
                    elif hydro_mode == "all":
                        hydro_iteration = hydro_iterations[0]
                    elif hydro_mode == "random_keep":
                        hydro_iteration = random_keep(hydro_iterations)
                    else:
                        print("Unknown hydro mode")
                    (
                        availability_iteration,
                        av_current_index,
                    ) = get_availability_iteration(
                        availability_mode=availability_mode,
                        availability_iterations=availability_iterations,
                        av_current_index=av_current_index,
                    )

                    with open(
                        os.path.join(output_directory, "iterations.csv"), "a"
                    ) as f_out:
                        writer = csv.writer(f_out, delimiter=",")
                        writer.writerow(
                            [weather_iteration, hydro_iteration, availability_iteration]
                        )


def random_remove(starting_list):
    # random.seed(0)
    i = random.randrange(len(starting_list))
    starting_list[i], starting_list[-1] = starting_list[-1], starting_list[i]
    iteration = starting_list.pop()

    return iteration


def random_keep(starting_list):
    # random.seed(0)
    i = random.randrange(len(starting_list))
    iteration = starting_list[i]

    return iteration


def sort_final_file(filepath):
    df = pd.read_csv(filepath, delimiter=",")

    df = df.sort_values(
        ["weather_iteration", "hydro_iteration", "availability_iteration"],
        ascending=[True, True, True],
    )

    df.to_csv(filepath, index=False)


def get_availability_iteration(
    availability_mode, availability_iterations, av_current_index
):
    availability_iteration = None
    if availability_mode == "ordered":
        availability_iteration = availability_iterations[av_current_index]
        av_current_index += 1
    elif availability_mode == "random_remove":
        availability_iteration = random_remove(availability_iterations)
    elif availability_mode == "random_keep":
        availability_iteration = random_keep(availability_iterations)
    elif availability_mode == "all":
        availability_iteration = availability_iterations[0]
    else:
        print("Unknown availability mode.")

    return availability_iteration, av_current_index


def main(args=None):
    if args is None:
        args = sys.argv[1:]

    parsed_args = parse_arguments(args=args)

    os.makedirs(parsed_args.output_directory, exist_ok=True)

    create_temporal_scenario_iterations_csv(
        n_passes=int(parsed_args.n_passes),
        filepath=parsed_args.iterations_csv_path,
        output_directory=parsed_args.output_directory,
    )
    sort_final_file(
        filepath=os.path.join(parsed_args.output_directory, "iterations.csv")
    )


if __name__ == "__main__":
    main()

1	# Copyright 2016-2024 Blue Marble Analytics LLC.
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# http://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	"""
16	Create temporal iterations CSV from user-defined params. Does not cover all
17	possible cases yet.
18	"""
19
20	import sys	1✔
21	from argparse import ArgumentParser	1✔
22	import csv	1✔
23	import os.path	1✔
24	import pandas as pd	1✔
25	import random	1✔
26
27	N_PASSES_DEFAULT = 1	1✔
28
29
30	def parse_arguments(args):	1✔
31	"""
32	:param args: the script arguments specified by the user
33	:return: the parsed known argument values (<class 'argparse.Namespace'>
34	Python object)
35
36	Parse the known arguments.
37	"""
38
39	parser = ArgumentParser(add_help=True)	1✔
40
41	parser.add_argument(	1✔
42	"-n",
43	"--n_passes",
44	default=N_PASSES_DEFAULT,
45	help=f"Defaults to {N_PASSES_DEFAULT}.",
46	)
47	parser.add_argument("-csv", "--iterations_csv_path")	1✔
48
49	parser.add_argument("-o", "--output_directory")	1✔
50
51	parser.add_argument("-q", "--quiet", default=False, action="store_true")	1✔
52
53	parsed_arguments = parser.parse_known_args(args=args)[0]	1✔
54
55	return parsed_arguments	1✔
56
57
58	def create_temporal_scenario_iterations_csv(n_passes, filepath, output_directory):	1✔
59	with open(os.path.join(output_directory, "iterations.csv"), "w") as f:	1✔
60	writer = csv.writer(f, delimiter=",")	1✔
61	writer.writerow(	1✔
62	["weather_iteration", "hydro_iteration", "availability_iteration"]
63	)
64
65	# print(os.path.abspath(filepath))
66	df = pd.read_csv(filepath)	1✔
67
68	weather_df = df["weather"]	1✔
69	weather_list = [i[1] for i in weather_df.items()]	1✔
70	weather_mode = weather_list[0]	1✔
71	weather_iterations_pass = [i for i in weather_list[1:] if not pd.isna(i)]	1✔
72
73	hydro_df = df["hydro"]	1✔
74	hydro_list = [i[1] for i in hydro_df.items()]	1✔
75	hydro_mode = hydro_list[0]	1✔
76	hydro_iterations_pass = [i for i in hydro_list[1:] if not pd.isna(i)]	1✔
77
78	availability_df = df["availability"]	1✔
79	availability_list = [i[1] for i in availability_df.items()]	1✔
80	availability_mode = availability_list[0]	1✔
81	availability_iterations_pass = [i for i in availability_list[1:] if not pd.isna(i)]	1✔
82
83	# TODO: possibly remove
84	weather_iteration, hydro_iteration, availability_iteration = None, None, None	1✔
85
86	for n in range(n_passes):	1✔
87	weather_iterations = weather_iterations_pass.copy()	1✔
88	hydro_iterations = hydro_iterations_pass.copy()	1✔
89	availability_iterations = availability_iterations_pass.copy()	1✔
90
91	av_current_index = 0	1✔
92	hy_current_index = 0	1✔
93	if weather_mode == "loop":	1✔
94	for weather_iteration in weather_iterations:	1✔
95	if hydro_mode == "loop":	1✔
96	for hydro_iteration in hydro_iterations:	1✔
97	(	1✔
98	availability_iteration,
99	av_current_index,
100	) = get_availability_iteration(
101	availability_mode=availability_mode,
102	availability_iterations=availability_iterations,
103	av_current_index=av_current_index,
104	)
105
106	with open(	1✔
107	os.path.join(output_directory, "iterations.csv"), "a"
108	) as f_out:
109	writer = csv.writer(f_out, delimiter=",")	1✔
110	writer.writerow(	1✔
111	[
112	weather_iteration,
113	hydro_iteration,
114	availability_iteration,
115	]
116	)
117	else:
118	if hydro_mode == "ordered":	1✔
119	hydro_iteration = hydro_iterations[hy_current_index]	1✔
120	hy_current_index += 1	1✔
121	elif hydro_mode == "random_remove":	×
122	hydro_iteration = random_remove(hydro_iterations)	×
123	elif hydro_mode == "all":	×
124	hydro_iteration = hydro_iterations[0]	×
125	elif hydro_mode == "random_keep":	×
126	hydro_iteration = random_keep(hydro_iterations)	×
127	else:
128	print("Unknown hydro mode")	×
129	(	1✔
130	availability_iteration,
131	av_current_index,
132	) = get_availability_iteration(
133	availability_mode=availability_mode,
134	availability_iterations=availability_iterations,
135	av_current_index=av_current_index,
136	)
137
138	with open(	1✔
139	os.path.join(output_directory, "iterations.csv"), "a"
140	) as f_out:
141	writer = csv.writer(f_out, delimiter=",")	1✔
142	writer.writerow(	1✔
143	[weather_iteration, hydro_iteration, availability_iteration]
144	)
145
146
147	def random_remove(starting_list):	1✔
148	# random.seed(0)
UNCOV 149	i = random.randrange(len(starting_list))	×
UNCOV 150	starting_list[i], starting_list[-1] = starting_list[-1], starting_list[i]	×
UNCOV 151	iteration = starting_list.pop()	×
152
UNCOV 153	return iteration	×
154
155
156	def random_keep(starting_list):	1✔
157	# random.seed(0)
158	i = random.randrange(len(starting_list))	×
159	iteration = starting_list[i]	×
160
161	return iteration	×
162
163
164	def sort_final_file(filepath):	1✔
165	df = pd.read_csv(filepath, delimiter=",")	1✔
166
167	df = df.sort_values(	1✔
168	["weather_iteration", "hydro_iteration", "availability_iteration"],
169	ascending=[True, True, True],
170	)
171
172	df.to_csv(filepath, index=False)	1✔
173
174
175	def get_availability_iteration(	1✔
176	availability_mode, availability_iterations, av_current_index
177	):
178	availability_iteration = None	1✔
179	if availability_mode == "ordered":	1✔
180	availability_iteration = availability_iterations[av_current_index]	1✔
181	av_current_index += 1	1✔
UNCOV 182	elif availability_mode == "random_remove":	×
UNCOV 183	availability_iteration = random_remove(availability_iterations)	×
184	elif availability_mode == "random_keep":	×
185	availability_iteration = random_keep(availability_iterations)	×
186	elif availability_mode == "all":	×
187	availability_iteration = availability_iterations[0]	×
188	else:
189	print("Unknown availability mode.")	×
190
191	return availability_iteration, av_current_index	1✔
192
193
194	def main(args=None):	1✔
195	if args is None:	1✔
196	args = sys.argv[1:]	×
197
198	parsed_args = parse_arguments(args=args)	1✔
199
200	os.makedirs(parsed_args.output_directory, exist_ok=True)	1✔
201
202	create_temporal_scenario_iterations_csv(	1✔
203	n_passes=int(parsed_args.n_passes),
204	filepath=parsed_args.iterations_csv_path,
205	output_directory=parsed_args.output_directory,
206	)
207	sort_final_file(	1✔
208	filepath=os.path.join(parsed_args.output_directory, "iterations.csv")
209	)
210
211
212	if __name__ == "__main__":	1✔
213	main()	×

blue-marble / gridpath / 26601056421

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous