29039031053

Committed 09 Jul 2026 05:59PM UTC coverage: 88.443% (-0.01%) from 88.454%

Build # 29039031053

Build Type

Pull #1390

github

Committed by

web-flow

Commit Message

Merge 38faf41ec into d74022cec

Pull Request Pull Request #1390: Memory efficiency enhancements

Coverage Stats

213 of 232 new or added lines in 72 files covered. (91.81%)

28630 of 32371 relevant lines covered (88.44%)

0.88 hits per line

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

75.81

/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


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

sylvan-energy / gridpath / 29039031053

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