29131831494

Committed 11 Jul 2026 12:04AM UTC coverage: 88.581%. First build

Build # 29131831494

Build Type

push

github

Committed by

web-flow

Commit Message

GridPath v2026.6.0

Merge pull request #1396 from sylvan-energy/develop

Coverage Stats

1127 of 1215 new or added lines in 163 files covered. (92.76%)

29035 of 32778 relevant lines covered (88.58%)

0.89 hits per line

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

75.94

/gridpath/common_functions.py

# Copyright 2016-2025 Blue Marble Analytics LLC.
# Copyright 2026 Sylvan Energy 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.

import os.path
import sys
import warnings

from argparse import ArgumentParser

import pandas as pd

from version import __version__


def get_version_parser():
    """
    Create a parser for the --version argument, which prints the GridPath
    version and exits. Include this in the parents of every GridPath
    command-line entry point's parser.
    """
    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--version", action="version", version=f"GridPath {__version__}"
    )

    return parser


def determine_scenario_directory(scenario_location, scenario_name):
    """
    :param scenario_location: string, the base directory
    :param scenario_name: string, the scenario name
    :return: the scenario directory (string)

    Determine the scenario directory given a base directory and the scenario
    name. If no base directory is specified, use a directory named
    'scenarios' in the root directory (one level down from the current
    working directory).
    """
    if scenario_location is None:
        main_directory = os.path.join(os.getcwd(), "..", "scenarios")
    else:
        main_directory = scenario_location

    scenario_directory = os.path.join(main_directory, str(scenario_name))

    return scenario_directory


def create_directory_if_not_exists(directory):
    """
    :param directory: string; the directory path

    Check if a directory exists and create it if not.
    """
    if not os.path.exists(directory):
        os.makedirs(directory)


def get_required_e2e_arguments_parser():
    """
    :return: the common parser for all e2e arguments

    Create ArgumentParser object which has the common set of arguments all
    end-to-end scripts. This includes the information for accessing local
    scenario data and whether to print run output.

    We can then simply add 'parents=[get_required_e2e_arguments_parser()]'
    when we create a parser for a script to inherit these common arguments.

    Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
    (one for parent and one for each child), which will throw an error.
    """

    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--scenario_location",
        default="../scenarios",
        help="The path to the directory in which to create "
        "the scenario directory. Defaults to "
        "'../scenarios' if not specified.",
    )
    parser.add_argument(
        "--quiet", default=False, action="store_true", help="Don't print run output."
    )

    parser.add_argument(
        "--verbose",
        default=False,
        action="store_true",
        help="Print extra output, e.g. current module info.",
    )

    return parser


def get_scenario_name_parser():
    """
    Create ArgumentParser object which has the common set of arguments for
    getting the scenario name

    We can then simply add 'parents=[get_scenario_names_parser()]' when we
    create a parser for a script to inherit these common arguments.

    Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
    (one for parent and one for each child), which will throw an error.
    :return:
    """

    parser = ArgumentParser(add_help=False)
    required = parser.add_argument_group("required arguments")
    required.add_argument(
        "--scenario",
        required=True,
        type=str,
        help="Name of the scenario problem to solve.",
    )

    return parser


def get_db_parser():
    """
    Create ArgumentParser object which has the common set of arguments for
    accessing scenario data from the database.

    We can then simply add 'parents=[get_db_parser()]' when we create a
    parser for a script to inherit these common arguments.

    Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
    (one for parent and one for each child), which will throw an error.
    :return:
    """

    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--database",
        default="../db/io.db",
        help="The database file path relative to the current "
        "working directory. Defaults to ../db/io.db ",
    )
    parser.add_argument(
        "--scenario_id",
        type=int,
        help="The scenario_id from the database. Not needed "
        "if scenario is specified.",
    )
    parser.add_argument(
        "--scenario",
        type=str,
        help="The scenario_name from the database. Not "
        "needed if scenario_id is specified.",
    )

    return parser


def get_temporal_structure_csv_overwrite_parser():
    """ """
    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--temporal_structure_csv_overwrite",
        default=False,
        action="store_true",
        help="Overwrite the temporal structure from the database with the "
        "provided CSV file.",
    )
    parser.add_argument(
        "--temporal_structure_csv_path",
        help="Path to the CSV where the temporal structure is defined.",
    )

    return parser


def get_get_inputs_parser():
    """ """

    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--n_parallel_get_inputs",
        default=1,
        help="Get inputs for n subproblems in parallel.",
    )

    return parser


def get_run_scenario_parser():
    """
    Create ArgumentParser object which has the common set of arguments for
    solving a scenario (see run_scenario.py and run_end_to_end.py).

    We can then simply add 'parents=[get_solve_parser()]' when we create a
    parser for a script to inherit these common arguments.

    Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
    (one for parent and one for each child), which will throw an error.
    :return:
    """

    parser = ArgumentParser(add_help=False)

    # Output options
    parser.add_argument(
        "--log",
        default=False,
        action="store_true",
        help="Log output to a file in the scenario's 'logs' "
        "directory as well as the terminal.",
    )
    # Problem files and solutions
    parser.add_argument(
        "--create_lp_problem_file_only",
        default=False,
        action="store_true",
        help="Create and save the problem file, but don't solve yet.",
    )
    parser.add_argument(
        "--load_cplex_solution",
        default=False,
        action="store_true",
        help="Skip solve and load results from a CPLEX solution file instead.",
    )
    parser.add_argument(
        "--load_gurobi_solution",
        default=False,
        action="store_true",
        help="Skip solve and load results from a Gurobi solution file instead.",
    )
    parser.add_argument(
        "--load_highs_solution",
        default=False,
        action="store_true",
        help="Skip solve and load results from a HiGHS solution file instead.",
    )
    # Duals
    parser.add_argument(
        "--skip_duals",
        default=False,
        action="store_true",
        help="Don't import or save constraint duals. Duals are imported "
        "for every constraint in the model, which adds significant memory "
        "and solution-load time; skip them if you don't need shadow prices "
        "(e.g. LMPs). Dual-based results files will not be written.",
    )
    # Solver options
    parser.add_argument(
        "--solver",
        help="Name of the solver to use. "
        "GridPath will use Cbc if solver is "
        "not specified here and a "
        "'solver_options.csv' file does not "
        "exist in the scenario directory.",
    )
    parser.add_argument(
        "--solver_executable",
        help="The path to the solver executable to use. This "
        "is optional; if you don't specify it, "
        "Pyomo will look for the solver executable in "
        "your PATH. The solver specified with the "
        "--solver option must be the same as the solver "
        "for which you are providing an executable.",
    )
    parser.add_argument(
        "--mute_solver_output",
        default=False,
        action="store_true",
        help="Don't print solver output.",
    )
    parser.add_argument(
        "--write_solver_files_to_logs_dir",
        default=False,
        action="store_true",
        help="Write the temporary " "solver files to the logs " "directory.",
    )
    parser.add_argument(
        "--keepfiles",
        default=False,
        action="store_true",
        help="Save temporary solver files.",
    )
    parser.add_argument(
        "--symbolic",
        default=False,
        action="store_true",
        help="Use symbolic labels in solver files.",
    )
    parser.add_argument(
        "--report_timing",
        default=False,
        action="store_true",
    )
    # Flag for test runs (various changes in behavior)
    parser.add_argument(
        "--testing",
        default=False,
        action="store_true",
        help="Flag for test suite runs.",
    )

    # Parallel solve
    parser.add_argument(
        "--n_parallel_solve",
        default=1,
        help="Solve n subproblems in parallel.",
    )

    # Solve only incomplete subproblems
    parser.add_argument(
        "--incomplete_only",
        default=False,
        action="store_true",
        help="Solve only incomplete subproblems, i.e. do no re-solve if "
        "results are found. The subproblem is assumed complete if the"
        "termination_condition.txt file is found.",
    )

    # Results export rule name
    parser.add_argument(
        "--results_export_rule",
        help="The name of the rule to use to decide whether to export results.",
    )

    parser.add_argument(
        "--results_export_summary_rule",
        help="The name of the rule to use to decide whether to export "
        "summary results.",
    )

    parser.add_argument(
        "--skip_quick_summary",
        default=False,
        action="store_true",
        help="Skip quick summary text file",
    )

    return parser


def get_import_results_parser():
    parser = ArgumentParser(add_help=False)
    parser.add_argument(
        "--results_import_rule",
        help="The name of the rule to use to decide whether to import results.",
    )
    parser.add_argument(
        "--ignore_incomplete",
        default=False,
        action="store_true",
        help="Ignore problems with no results. Can be used to import results "
        "for subproblems before all other subproblems have been solved. "
        "Proceed with caution.",
    )

    return parser


def ensure_empty_string(string):
    empty_string_ensured = "" if string == "empty_string" else string

    return empty_string_ensured


def create_logs_directory_if_not_exists(
    scenario_directory,
    weather_iteration,
    hydro_iteration,
    availability_iteration,
    subproblem,
    stage,
):
    """
    Create a logs directory if it doesn't exist already
    :param scenario_directory:
    :param subproblem:
    :param stage:
    :return:
    """
    logs_directory = os.path.join(
        scenario_directory,
        weather_iteration,
        hydro_iteration,
        availability_iteration,
        subproblem,
        stage,
        "logs",
    )
    if not os.path.exists(logs_directory):
        os.makedirs(logs_directory)
    return logs_directory


class Logging(object):
    """
    Log output to both standard output and a log file. This will be
    accomplished by assigning this class to sys.stdout.
    """

    def __init__(self, logs_dir, start_time, e2e, process_id):
        """
        Assign sys.stdout and a log file as output destinations

        :param logs_dir:
        """
        self.terminal = sys.stdout

        # If logging only run_scenario, print to a file starting with opt_
        # and the datetime
        # If logging run_e2e, print to a file starting with e2e_, with the
        # datetime, and the process ID
        if not e2e:
            self.log_file_path = os.path.join(
                logs_dir, "opt_{}.log".format(string_from_time(start_time))
            )
        else:
            self.log_file_path = os.path.join(
                logs_dir,
                "e2e_{}_pid_{}.log".format(
                    string_from_time(start_time), str(process_id)
                ),
            )

        self.log_file = open(self.log_file_path, "a", buffering=1)

    def __getattr__(self, attr):
        """
        Default to sys.stdout when calling attributes for this class

        :param attr:
        :return:
        """
        return getattr(self.terminal, attr)

    def write(self, message):
        """
        Output to both terminal and a log file. The print statement will
        call the write() method of any object you assign to sys.stdout
        (in this case the Logging object)

        :param message:
        :return:
        """
        self.terminal.write(message)
        self.log_file.write(message)

        # Find a print statement
        # import collections
        # import inspect

        # if message.strip():
        #     Record = collections.namedtuple(
        #         'Record',
        #         'frame filename line_number function_name lines index')
        #
        #     record = Record(*inspect.getouterframes(inspect.currentframe())[1])
        #     self.terminal.write(
        #         '{f} {n}: '.format(f=record.filename, n=record.line_number))
        # self.terminal.write(message)

    def flush(self):
        """
        Flush both the terminal and the log file

        :return:
        """
        self.terminal.flush()
        self.log_file.flush()

    def close(self):
        """
        Close the log file to release the file descriptor.
        Critical for preventing "too many open files" errors.
        """
        if hasattr(self, "log_file") and self.log_file and not self.log_file.closed:
            self.log_file.close()

    def __del__(self):
        """
        Ensure log file is closed when object is garbage collected
        """
        self.close()

    def __enter__(self):
        """
        Support context manager protocol
        """
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Close log file when exiting context manager
        """
        self.close()
        return False


def string_from_time(datetime_string):
    """
    :param datetime_string: datetime string
    :return: formatted time string
    """
    return datetime_string.strftime("%Y-%m-%d_%H-%M-%S")


def append_to_timing_summary_file(timing_summary_file_path, line):
    """
    :param timing_summary_file_path: the timing summary file path; None if
        no summary file is being kept (i.e. when not logging)
    :param line: the line to append
    :return:

    Append a line to the timing summary file. The summary is written as we
    go, so that it is available (with the steps completed so far) even if
    the run fails or is interrupted.
    """
    if timing_summary_file_path is not None:
        with open(timing_summary_file_path, "a") as summary_file:
            summary_file.write(line + "\n")


def create_results_df(index_columns, results_columns, data):
    df = pd.DataFrame(
        columns=index_columns + results_columns,
        data=data,
    ).set_index(index_columns)

    return df


def update_results_df(target_df, results_df):
    """
    Add :code:`results_df`'s columns to :code:`target_df` (if not already
    present) and fill in its values; rows of :code:`target_df` not covered
    by :code:`results_df` are left as NaN.

    New columns are created with the source column's dtype. Avoids creating
    them as object columns (e.g. by first assigning None), as it would make
    numeric results columns store boxed Python floats (much higher memory
    requirements).
    """
    for c in results_df.columns:
        if c not in target_df.columns:
            target_df[c] = pd.Series(dtype=results_df[c].dtype)
    target_df.update(results_df)


def duals_wrapper(m, component, verbose=False):
    # AttributeError: no dual Suffix on the instance (--skip_duals);
    # KeyError: suffix present but no dual for this constraint
    if not hasattr(m, "dual"):
        return None
    try:
        return m.dual[component]
    except KeyError:
        if verbose:
            warnings.warn(f"""
                KeyError caught when saving duals for {component}. Duals were 
                not exported. This is expected if solving a MIP with CPLEX (and 
                possibly other solvers), not otherwise.
                """)
        return None


def none_dual_type_error_wrapper(component, coefficient):
    try:
        return component / coefficient
    except TypeError:
        return None

1	# Copyright 2016-2025 Blue Marble Analytics LLC.
2	# Copyright 2026 Sylvan Energy Analytics LLC.
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	import os.path	1✔
17	import sys	1✔
18	import warnings	1✔
19
20	from argparse import ArgumentParser	1✔
21
22	import pandas as pd	1✔
23
24	from version import __version__	1✔
25
26
27	def get_version_parser():	1✔
28	"""
29	Create a parser for the --version argument, which prints the GridPath
30	version and exits. Include this in the parents of every GridPath
31	command-line entry point's parser.
32	"""
33	parser = ArgumentParser(add_help=False)	1✔
34	parser.add_argument(	1✔
35	"--version", action="version", version=f"GridPath {__version__}"
36	)
37
38	return parser	1✔
39
40
41	def determine_scenario_directory(scenario_location, scenario_name):	1✔
42	"""
43	:param scenario_location: string, the base directory
44	:param scenario_name: string, the scenario name
45	:return: the scenario directory (string)
46
47	Determine the scenario directory given a base directory and the scenario
48	name. If no base directory is specified, use a directory named
49	'scenarios' in the root directory (one level down from the current
50	working directory).
51	"""
52	if scenario_location is None:	1✔
53	main_directory = os.path.join(os.getcwd(), "..", "scenarios")	×
54	else:
55	main_directory = scenario_location	1✔
56
57	scenario_directory = os.path.join(main_directory, str(scenario_name))	1✔
58
59	return scenario_directory	1✔
60
61
62	def create_directory_if_not_exists(directory):	1✔
63	"""
64	:param directory: string; the directory path
65
66	Check if a directory exists and create it if not.
67	"""
68	if not os.path.exists(directory):	1✔
69	os.makedirs(directory)	×
70
71
72	def get_required_e2e_arguments_parser():	1✔
73	"""
74	:return: the common parser for all e2e arguments
75
76	Create ArgumentParser object which has the common set of arguments all
77	end-to-end scripts. This includes the information for accessing local
78	scenario data and whether to print run output.
79
80	We can then simply add 'parents=[get_required_e2e_arguments_parser()]'
81	when we create a parser for a script to inherit these common arguments.
82
83	Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
84	(one for parent and one for each child), which will throw an error.
85	"""
86
87	parser = ArgumentParser(add_help=False)	1✔
88	parser.add_argument(	1✔
89	"--scenario_location",
90	default="../scenarios",
91	help="The path to the directory in which to create "
92	"the scenario directory. Defaults to "
93	"'../scenarios' if not specified.",
94	)
95	parser.add_argument(	1✔
96	"--quiet", default=False, action="store_true", help="Don't print run output."
97	)
98
99	parser.add_argument(	1✔
100	"--verbose",
101	default=False,
102	action="store_true",
103	help="Print extra output, e.g. current module info.",
104	)
105
106	return parser	1✔
107
108
109	def get_scenario_name_parser():	1✔
110	"""
111	Create ArgumentParser object which has the common set of arguments for
112	getting the scenario name
113
114	We can then simply add 'parents=[get_scenario_names_parser()]' when we
115	create a parser for a script to inherit these common arguments.
116
117	Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
118	(one for parent and one for each child), which will throw an error.
119	:return:
120	"""
121
122	parser = ArgumentParser(add_help=False)	1✔
123	required = parser.add_argument_group("required arguments")	1✔
124	required.add_argument(	1✔
125	"--scenario",
126	required=True,
127	type=str,
128	help="Name of the scenario problem to solve.",
129	)
130
131	return parser	1✔
132
133
134	def get_db_parser():	1✔
135	"""
136	Create ArgumentParser object which has the common set of arguments for
137	accessing scenario data from the database.
138
139	We can then simply add 'parents=[get_db_parser()]' when we create a
140	parser for a script to inherit these common arguments.
141
142	Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
143	(one for parent and one for each child), which will throw an error.
144	:return:
145	"""
146
147	parser = ArgumentParser(add_help=False)	1✔
148	parser.add_argument(	1✔
149	"--database",
150	default="../db/io.db",
151	help="The database file path relative to the current "
152	"working directory. Defaults to ../db/io.db ",
153	)
154	parser.add_argument(	1✔
155	"--scenario_id",
156	type=int,
157	help="The scenario_id from the database. Not needed "
158	"if scenario is specified.",
159	)
160	parser.add_argument(	1✔
161	"--scenario",
162	type=str,
163	help="The scenario_name from the database. Not "
164	"needed if scenario_id is specified.",
165	)
166
167	return parser	1✔
168
169
170	def get_temporal_structure_csv_overwrite_parser():	1✔
171	""" """
172	parser = ArgumentParser(add_help=False)	1✔
173	parser.add_argument(	1✔
174	"--temporal_structure_csv_overwrite",
175	default=False,
176	action="store_true",
177	help="Overwrite the temporal structure from the database with the "
178	"provided CSV file.",
179	)
180	parser.add_argument(	1✔
181	"--temporal_structure_csv_path",
182	help="Path to the CSV where the temporal structure is defined.",
183	)
184
185	return parser	1✔
186
187
188	def get_get_inputs_parser():	1✔
189	""" """
190
191	parser = ArgumentParser(add_help=False)	1✔
192	parser.add_argument(	1✔
193	"--n_parallel_get_inputs",
194	default=1,
195	help="Get inputs for n subproblems in parallel.",
196	)
197
198	return parser	1✔
199
200
201	def get_run_scenario_parser():	1✔
202	"""
203	Create ArgumentParser object which has the common set of arguments for
204	solving a scenario (see run_scenario.py and run_end_to_end.py).
205
206	We can then simply add 'parents=[get_solve_parser()]' when we create a
207	parser for a script to inherit these common arguments.
208
209	Note that 'add_help' is set to 'False' to avoid multiple `-h/--help` options
210	(one for parent and one for each child), which will throw an error.
211	:return:
212	"""
213
214	parser = ArgumentParser(add_help=False)	1✔
215
216	# Output options
217	parser.add_argument(	1✔
218	"--log",
219	default=False,
220	action="store_true",
221	help="Log output to a file in the scenario's 'logs' "
222	"directory as well as the terminal.",
223	)
224	# Problem files and solutions
225	parser.add_argument(	1✔
226	"--create_lp_problem_file_only",
227	default=False,
228	action="store_true",
229	help="Create and save the problem file, but don't solve yet.",
230	)
231	parser.add_argument(	1✔
232	"--load_cplex_solution",
233	default=False,
234	action="store_true",
235	help="Skip solve and load results from a CPLEX solution file instead.",
236	)
237	parser.add_argument(	1✔
238	"--load_gurobi_solution",
239	default=False,
240	action="store_true",
241	help="Skip solve and load results from a Gurobi solution file instead.",
242	)
243	parser.add_argument(	1✔
244	"--load_highs_solution",
245	default=False,
246	action="store_true",
247	help="Skip solve and load results from a HiGHS solution file instead.",
248	)
249	# Duals
250	parser.add_argument(	1✔
251	"--skip_duals",
252	default=False,
253	action="store_true",
254	help="Don't import or save constraint duals. Duals are imported "
255	"for every constraint in the model, which adds significant memory "
256	"and solution-load time; skip them if you don't need shadow prices "
257	"(e.g. LMPs). Dual-based results files will not be written.",
258	)
259	# Solver options
260	parser.add_argument(	1✔
261	"--solver",
262	help="Name of the solver to use. "
263	"GridPath will use Cbc if solver is "
264	"not specified here and a "
265	"'solver_options.csv' file does not "
266	"exist in the scenario directory.",
267	)
268	parser.add_argument(	1✔
269	"--solver_executable",
270	help="The path to the solver executable to use. This "
271	"is optional; if you don't specify it, "
272	"Pyomo will look for the solver executable in "
273	"your PATH. The solver specified with the "
274	"--solver option must be the same as the solver "
275	"for which you are providing an executable.",
276	)
277	parser.add_argument(	1✔
278	"--mute_solver_output",
279	default=False,
280	action="store_true",
281	help="Don't print solver output.",
282	)
283	parser.add_argument(	1✔
284	"--write_solver_files_to_logs_dir",
285	default=False,
286	action="store_true",
287	help="Write the temporary " "solver files to the logs " "directory.",
288	)
289	parser.add_argument(	1✔
290	"--keepfiles",
291	default=False,
292	action="store_true",
293	help="Save temporary solver files.",
294	)
295	parser.add_argument(	1✔
296	"--symbolic",
297	default=False,
298	action="store_true",
299	help="Use symbolic labels in solver files.",
300	)
301	parser.add_argument(	1✔
302	"--report_timing",
303	default=False,
304	action="store_true",
305	)
306	# Flag for test runs (various changes in behavior)
307	parser.add_argument(	1✔
308	"--testing",
309	default=False,
310	action="store_true",
311	help="Flag for test suite runs.",
312	)
313
314	# Parallel solve
315	parser.add_argument(	1✔
316	"--n_parallel_solve",
317	default=1,
318	help="Solve n subproblems in parallel.",
319	)
320
321	# Solve only incomplete subproblems
322	parser.add_argument(	1✔
323	"--incomplete_only",
324	default=False,
325	action="store_true",
326	help="Solve only incomplete subproblems, i.e. do no re-solve if "
327	"results are found. The subproblem is assumed complete if the"
328	"termination_condition.txt file is found.",
329	)
330
331	# Results export rule name
332	parser.add_argument(	1✔
333	"--results_export_rule",
334	help="The name of the rule to use to decide whether to export results.",
335	)
336
337	parser.add_argument(	1✔
338	"--results_export_summary_rule",
339	help="The name of the rule to use to decide whether to export "
340	"summary results.",
341	)
342
343	parser.add_argument(	1✔
344	"--skip_quick_summary",
345	default=False,
346	action="store_true",
347	help="Skip quick summary text file",
348	)
349
350	return parser	1✔
351
352
353	def get_import_results_parser():	1✔
354	parser = ArgumentParser(add_help=False)	1✔
355	parser.add_argument(	1✔
356	"--results_import_rule",
357	help="The name of the rule to use to decide whether to import results.",
358	)
359	parser.add_argument(	1✔
360	"--ignore_incomplete",
361	default=False,
362	action="store_true",
363	help="Ignore problems with no results. Can be used to import results "
364	"for subproblems before all other subproblems have been solved. "
365	"Proceed with caution.",
366	)
367
368	return parser	1✔
369
370
371	def ensure_empty_string(string):	1✔
372	empty_string_ensured = "" if string == "empty_string" else string	1✔
373
374	return empty_string_ensured	1✔
375
376
377	def create_logs_directory_if_not_exists(	1✔
378	scenario_directory,
379	weather_iteration,
380	hydro_iteration,
381	availability_iteration,
382	subproblem,
383	stage,
384	):
385	"""
386	Create a logs directory if it doesn't exist already
387	:param scenario_directory:
388	:param subproblem:
389	:param stage:
390	:return:
391	"""
392	logs_directory = os.path.join(	×
393	scenario_directory,
394	weather_iteration,
395	hydro_iteration,
396	availability_iteration,
397	subproblem,
398	stage,
399	"logs",
400	)
401	if not os.path.exists(logs_directory):	×
402	os.makedirs(logs_directory)	×
403	return logs_directory	×
404
405
406	class Logging(object):	1✔
407	"""
408	Log output to both standard output and a log file. This will be
409	accomplished by assigning this class to sys.stdout.
410	"""
411
412	def __init__(self, logs_dir, start_time, e2e, process_id):	1✔
413	"""
414	Assign sys.stdout and a log file as output destinations
415
416	:param logs_dir:
417	"""
418	self.terminal = sys.stdout	×
419
420	# If logging only run_scenario, print to a file starting with opt_
421	# and the datetime
422	# If logging run_e2e, print to a file starting with e2e_, with the
423	# datetime, and the process ID
424	if not e2e:	×
425	self.log_file_path = os.path.join(	×
426	logs_dir, "opt_{}.log".format(string_from_time(start_time))
427	)
428	else:
429	self.log_file_path = os.path.join(	×
430	logs_dir,
431	"e2e_{}_pid_{}.log".format(
432	string_from_time(start_time), str(process_id)
433	),
434	)
435
436	self.log_file = open(self.log_file_path, "a", buffering=1)	×
437
438	def __getattr__(self, attr):	1✔
439	"""
440	Default to sys.stdout when calling attributes for this class
441
442	:param attr:
443	:return:
444	"""
445	return getattr(self.terminal, attr)	×
446
447	def write(self, message):	1✔
448	"""
449	Output to both terminal and a log file. The print statement will
450	call the write() method of any object you assign to sys.stdout
451	(in this case the Logging object)
452
453	:param message:
454	:return:
455	"""
456	self.terminal.write(message)	×
457	self.log_file.write(message)	×
458
459	# Find a print statement
460	# import collections
461	# import inspect
462
463	# if message.strip():
464	# Record = collections.namedtuple(
465	# 'Record',
466	# 'frame filename line_number function_name lines index')
467	#
468	# record = Record(*inspect.getouterframes(inspect.currentframe())[1])
469	# self.terminal.write(
470	# '{f} {n}: '.format(f=record.filename, n=record.line_number))
471	# self.terminal.write(message)
472
473	def flush(self):	1✔
474	"""
475	Flush both the terminal and the log file
476
477	:return:
478	"""
479	self.terminal.flush()	×
480	self.log_file.flush()	×
481
482	def close(self):	1✔
483	"""
484	Close the log file to release the file descriptor.
485	Critical for preventing "too many open files" errors.
486	"""
487	if hasattr(self, "log_file") and self.log_file and not self.log_file.closed:	×
488	self.log_file.close()	×
489
490	def __del__(self):	1✔
491	"""
492	Ensure log file is closed when object is garbage collected
493	"""
494	self.close()	×
495
496	def __enter__(self):	1✔
497	"""
498	Support context manager protocol
499	"""
500	return self	×
501
502	def __exit__(self, exc_type, exc_val, exc_tb):	1✔
503	"""
504	Close log file when exiting context manager
505	"""
506	self.close()	×
507	return False	×
508
509
510	def string_from_time(datetime_string):	1✔
511	"""
512	:param datetime_string: datetime string
513	:return: formatted time string
514	"""
515	return datetime_string.strftime("%Y-%m-%d_%H-%M-%S")	×
516
517
518	def append_to_timing_summary_file(timing_summary_file_path, line):	1✔
519	"""
520	:param timing_summary_file_path: the timing summary file path; None if
521	no summary file is being kept (i.e. when not logging)
522	:param line: the line to append
523	:return:
524
525	Append a line to the timing summary file. The summary is written as we
526	go, so that it is available (with the steps completed so far) even if
527	the run fails or is interrupted.
528	"""
529	if timing_summary_file_path is not None:	1✔
NEW 530	with open(timing_summary_file_path, "a") as summary_file:	×
NEW 531	summary_file.write(line + "\n")	×
532
533
534	def create_results_df(index_columns, results_columns, data):	1✔
535	df = pd.DataFrame(	1✔
536	columns=index_columns + results_columns,
537	data=data,
538	).set_index(index_columns)
539
540	return df	1✔
541
542
543	def update_results_df(target_df, results_df):	1✔
544	"""
545	Add :code:`results_df`'s columns to :code:`target_df` (if not already
546	present) and fill in its values; rows of :code:`target_df` not covered
547	by :code:`results_df` are left as NaN.
548
549	New columns are created with the source column's dtype. Avoids creating
550	them as object columns (e.g. by first assigning None), as it would make
551	numeric results columns store boxed Python floats (much higher memory
552	requirements).
553	"""
554	for c in results_df.columns:	1✔
555	if c not in target_df.columns:	1✔
556	target_df[c] = pd.Series(dtype=results_df[c].dtype)	1✔
557	target_df.update(results_df)	1✔
558
559
560	def duals_wrapper(m, component, verbose=False):	1✔
561	# AttributeError: no dual Suffix on the instance (--skip_duals);
562	# KeyError: suffix present but no dual for this constraint
563	if not hasattr(m, "dual"):	1✔
NEW 564	return None	×
565	try:	1✔
566	return m.dual[component]	1✔
567	except KeyError:	×
568	if verbose:	×
569	warnings.warn(f"""	×
570	KeyError caught when saving duals for {component}. Duals were
571	not exported. This is expected if solving a MIP with CPLEX (and
572	possibly other solvers), not otherwise.
573	""")
574	return None	×
575
576
577	def none_dual_type_error_wrapper(component, coefficient):	1✔
578	try:	1✔
579	return component / coefficient	1✔
580	except TypeError:	×
581	return None	×

sylvan-energy / gridpath / 29131831494

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