17809960349

Committed 17 Sep 2025 08:32PM UTC coverage: 88.945% (-0.01%) from 88.959%

Build # 17809960349

Build Type

Pull #1289

github

Committed by

web-flow

Commit Message

Merge 7a39b1128 into bdb0a2ac5

Pull Request Pull Request #1289: Maintenance dependency upgrades

Coverage Stats

8 of 13 new or added lines in 6 files covered. (61.54%)

143 existing lines in 44 files now uncovered.

27347 of 30746 relevant lines covered (88.94%)

0.89 hits per line

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

89.68

/gridpath/validate_inputs.py

# Copyright 2016-2023 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.

"""
This script iterates over all modules required for a GridPath scenario and
calls their *validate_inputs()* method, which performs various validations
of the input data and scenario setup.
"""

import sqlite3
import sys
from argparse import ArgumentParser

from db.common_functions import connect_to_database, spin_on_database_lock
from gridpath.auxiliary.db_interface import (
    get_required_capacity_types_from_database,
    get_scenario_id_and_name,
)
from gridpath.auxiliary.validations import write_validation_to_database
from gridpath.common_functions import get_db_parser
from gridpath.auxiliary.module_list import determine_modules, load_modules
from gridpath.auxiliary.scenario_chars import (
    OptionalFeatures,
    SubScenarios,
    get_scenario_structure_from_db,
)


def validate_inputs(
    subproblems,
    loaded_modules,
    scenario_id,
    weather_iteration,
    hydro_iteration,
    availability_iteration,
    subscenarios,
    conn,
):
    """ "
    For each module, load the inputs from the database and validate them

    :param subproblems: SubProblems object with info on the subproblem/stage
        structure
    :param loaded_modules: list of imported modules (Python <class 'module'>
        objects)
    :param subscenarios: SubScenarios object with all subscenario info
    :param conn: database connection
    :return:
    """

    # TODO: check if we even need database cursor (and subscenarios?)
    #   since presumably we already have our data in the inputs.
    #   need to go through each module's input validation to check this

    # TODO: see if we can do some sort of automatic dtype validation for
    #  each table in the database? Problem is that you don't necessarily want
    #  to check the full table but only the appropriate subscenario

    subproblems_list = subproblems.SUBPROBLEM_STAGES.keys()
    for subproblem in subproblems_list:
        stages = subproblems.SUBPROBLEM_STAGES[subproblem]
        for stage in stages:
            # 1. input validation within each module
            for m in loaded_modules:
                if hasattr(m, "validate_inputs"):
                    m.validate_inputs(
                        scenario_id=scenario_id,
                        subscenarios=subscenarios,
                        weather_iteration=weather_iteration,
                        hydro_iteration=hydro_iteration,
                        availability_iteration=availability_iteration,
                        subproblem=subproblem,
                        stage=stage,
                        conn=conn,
                    )

            # 2. input validation across modules
            #    make sure geography and projects are in line
            #    ... (see Evernote validation list)
            #    create separate function for each validation that you call here


def validate_subscenario_ids(scenario_id, subscenarios, optional_features, conn):
    """
    Check whether subscenarios_ids are consistent with:
     - core required subscenario_ids
     - data dependent subscenario_ids (e.g. new build)
     - optional features
    data.

    :param subscenarios:
    :param optional_features:
    :param conn:
    :return: Boolean, True is all subscenario IDs are valid.
    """

    valid_core = validate_required_subscenario_ids(scenario_id, subscenarios, conn)

    if valid_core:
        valid_data_dependent = validate_data_dependent_subscenario_ids(
            scenario_id, subscenarios, conn
        )
    else:
        valid_data_dependent = False

    valid_feature = validate_feature_subscenario_ids(
        scenario_id, subscenarios, optional_features, conn
    )

    return valid_core and valid_data_dependent and valid_feature


def validate_feature_subscenario_ids(
    scenario_id, subscenarios, optional_features, conn
):
    """

    :param subscenarios:
    :param optional_features:
    :param conn:
    :return:
    """

    subscenario_ids_by_feature = determine_subscenarios_by_feature(conn)
    feature_list = optional_features.get_active_features()

    errors = {"High": [], "Low": []}  # errors by severity
    for feature, subscenario_ids in subscenario_ids_by_feature.items():
        if feature not in ["core", "optional", "data_dependent"]:
            for sc_id in subscenario_ids:
                # If the feature is requested, and the associated subscenarios
                # are not specified, raise a validation error
                if feature in feature_list and getattr(subscenarios, sc_id) == "NULL":
                    errors["High"].append(
                        "Requested feature '{}' requires an input for '{}'".format(
                            feature, sc_id
                        )
                    )

                # If the feature is not requested, and the associated
                # subscenarios are specified, raise a validation error
                # TODO: need to add handling of subscenarios shared among
                #  features; commenting out for now
                # elif feature not in feature_list and \
                #         getattr(subscenarios, sc_id) != "NULL":
                #     errors["Low"].append(
                #         "Detected inputs for '{}' while related feature '{}' "
                #          "is not requested".format(sc_id, feature)
                #     )

    for severity, error_list in errors.items():
        write_validation_to_database(
            conn=conn,
            scenario_id=scenario_id,
            weather_iteration="N/A",
            hydro_iteration="N/A",
            availability_iteration="N/A",
            subproblem_id="N/A",
            stage_id="N/A",
            gridpath_module="N/A",
            db_table="scenarios",
            severity=severity,
            errors=error_list,
        )

    # Return True if all required subscenario_ids are valid (list is empty)
    return not bool(sum(errors.values(), []))


def validate_required_subscenario_ids(scenario_id, subscenarios, conn):
    """
    Check whether the required subscenario_ids are specified in the db
    :param subscenarios:
    :param conn:
    :return: boolean, True if all required subscenario_ids are specified
    """

    required_subscenario_ids = determine_subscenarios_by_feature(conn)["core"]

    errors = []
    for sc_id in required_subscenario_ids:
        if getattr(subscenarios, sc_id) is None:
            errors.append(
                "'{}' is a required input in the 'scenarios' table".format(sc_id)
            )

    write_validation_to_database(
        conn=conn,
        scenario_id=scenario_id,
        weather_iteration="N/A",
        hydro_iteration="N/A",
        availability_iteration="N/A",
        subproblem_id="N/A",
        stage_id="N/A",
        gridpath_module="N/A",
        db_table="scenarios",
        severity="High",
        errors=errors,
    )

    # Return True if all required subscenario_ids are valid (list is empty)
    return not bool(errors)


def validate_data_dependent_subscenario_ids(scenario_id, subscenarios, conn):
    """

    :param subscenarios:
    :param conn:
    :return:
    """

    assert subscenarios.PROJECT_PORTFOLIO_SCENARIO_ID is not None

    req_cap_types = set(get_required_capacity_types_from_database(conn, scenario_id))

    new_build_types = {
        "gen_new_lin,",
        "gen_new_bin",
        "stor_new_lin",
        "stor_new_bin",
        "dr_new",
    }
    existing_build_types = {
        "gen_spec",
        "gen_ret_bin",
        "gen_ret_lin",
        "stor_spec",
    }
    dr_types = {"dr_new"}

    # Determine required subscenario_ids
    sc_id_type = []
    if bool(req_cap_types & new_build_types):
        sc_id_type.append(("PROJECT_NEW_COST_SCENARIO_ID", "New Build"))
    if bool(req_cap_types & existing_build_types):
        sc_id_type.append(("PROJECT_SPECIFIED_CAPACITY_SCENARIO_ID", "Existing"))
        sc_id_type.append(("PROJECT_SPECIFIED_FIXED_COST_SCENARIO_ID", "Existing"))
    if bool(req_cap_types & dr_types):
        sc_id_type.append(("PROJECT_NEW_POTENTIAL_SCENARIO_ID", "Demand Response (DR)"))

    # Check whether required subscenario_ids are present
    errors = []
    for sc_id, build_type in sc_id_type:
        if getattr(subscenarios, sc_id) is None:
            errors.append(
                "'{}' is a required input in the 'scenarios' table if there "
                "are '{}' resources in the portfolio".format(sc_id, build_type)
            )

    write_validation_to_database(
        conn=conn,
        scenario_id=scenario_id,
        weather_iteration="N/A",
        hydro_iteration="N/A",
        availability_iteration="N/A",
        subproblem_id="N/A",
        stage_id="N/A",
        gridpath_module="N/A",
        db_table="scenarios",
        severity="High",
        errors=errors,
    )

    # Return True if all required subscenario_ids are valid (list is empty)
    return not bool(errors)


def reset_input_validation(conn, scenario_id):
    """
    Reset input validation: delete old input validation outputs and reset the
    input validation status.
    :param conn: database connection
    :param scenario_id: scenario_id
    :return:
    """
    c = conn.cursor()

    sql = """
        DELETE FROM status_validation
        WHERE scenario_id = ?;
        """
    spin_on_database_lock(conn=conn, cursor=c, sql=sql, data=(scenario_id,), many=False)

    sql = """
        UPDATE scenarios
        SET validation_status_id = 0
        WHERE scenario_id = ?;
        """
    spin_on_database_lock(conn=conn, cursor=c, sql=sql, data=(scenario_id,), many=False)


def update_validation_status(conn, scenario_id):
    """

    :param conn:
    :param scenario_id:
    :return:
    """
    c = conn.cursor()
    validations = c.execute(
        """SELECT scenario_id 
        FROM status_validation
        WHERE scenario_id = {}""".format(
            str(scenario_id)
        )
    ).fetchall()

    if validations:
        status = 2
        # Print the errors
        for e in c.execute(
            f"""SELECT * FROM status_validation WHERE scenario_id = {scenario_id};"""
        ).fetchall():
            print(e)
    else:
        status = 1

    sql = """
        UPDATE scenarios
        SET validation_status_id = ?
        WHERE scenario_id = ?;
        """
    spin_on_database_lock(
        conn=conn, cursor=c, sql=sql, data=(status, scenario_id), many=False
    )


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, parents=[get_db_parser()])

    # Add quiet flag which can suppress run output
    parser.add_argument(
        "--quiet", default=False, action="store_true", help="Don't print run output."
    )

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

    return parsed_arguments


def determine_subscenarios_by_feature(conn):
    """

    :param conn:
    :return:
    """
    c = conn.cursor()

    feature_sc = c.execute(
        """SELECT feature, subscenario_id
        FROM mod_feature_subscenarios"""
    ).fetchall()
    feature_sc_dict = {}
    for f, sc in feature_sc:
        if f in feature_sc_dict:
            feature_sc_dict[f].append(sc.upper())
        else:
            feature_sc_dict[f] = [sc.upper()]
    return feature_sc_dict


def main(args=None):
    """

    :return:
    """

    # Retrieve scenario_id and/or name from args + "quiet" flag
    if args is None:
        args = sys.argv[1:]
    parsed_arguments = parse_arguments(args=args)

    if not parsed_arguments.quiet:
        print("Validating inputs...")

    db_path = parsed_arguments.database
    scenario_id_arg = parsed_arguments.scenario_id
    scenario_name_arg = parsed_arguments.scenario

    conn = connect_to_database(db_path=db_path, detect_types=sqlite3.PARSE_DECLTYPES)
    c = conn.cursor()

    scenario_id, scenario_name = get_scenario_id_and_name(
        scenario_id_arg=scenario_id_arg,
        scenario_name_arg=scenario_name_arg,
        c=c,
        script="validate_inputs",
    )

    # Reset input validation status and results
    reset_input_validation(conn, scenario_id)

    # TODO: this is very similar to what's in get_scenario_inputs.py,
    #  so we should consolidate
    # Get scenario characteristics (features, scenario_id, subscenarios, subproblems)
    optional_features = OptionalFeatures(conn=conn, scenario_id=scenario_id)
    subscenarios = SubScenarios(conn=conn, scenario_id=scenario_id)
    scenario_structure = get_scenario_structure_from_db(
        conn=conn, scenario_id=scenario_id
    )

    # Check whether subscenario_ids are valid
    is_valid = validate_subscenario_ids(
        scenario_id, subscenarios, optional_features, conn
    )

    # Only do the detailed input validation if all required subscenario_ids
    # are specified (otherwise will get errors when loading data)
    if is_valid:
        # Load modules for all requested features
        feature_list = optional_features.get_active_features()
        # If any subproblem's stage list is non-empty, we have stages, so set
        # the stages_flag to True to pass to determine_modules below
        # This tells the determine_modules function to include the
        # stages-related modules
        stages_flag = any(
            [
                len(scenario_structure.SUBPROBLEM_STAGES[subp]) > 1
                for subp in list(scenario_structure.SUBPROBLEM_STAGES.keys())
            ]
        )
        modules_to_use = determine_modules(
            features=feature_list, multi_stage=stages_flag
        )
        loaded_modules = load_modules(modules_to_use=modules_to_use)

        # Read in inputs from db and validate inputs for loaded modules
        for weather_iteration in scenario_structure.ITERATION_STRUCTURE.keys():
            for hydro_iteration in scenario_structure.ITERATION_STRUCTURE[
                weather_iteration
            ].keys():
                for availability_iteration in scenario_structure.ITERATION_STRUCTURE[
                    weather_iteration
                ][hydro_iteration]:
                    validate_inputs(
                        scenario_structure,
                        loaded_modules,
                        scenario_id,
                        weather_iteration,
                        hydro_iteration,
                        availability_iteration,
                        subscenarios,
                        conn,
                    )

    else:
        if not parsed_arguments.quiet:
            print("Invalid subscenario ID(s). Skipped detailed input validation.")

    # Update validation status:
    update_validation_status(conn, scenario_id)

    # Close the database connection explicitly
    conn.close()


if __name__ == "__main__":
    main()

1	# Copyright 2016-2023 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	This script iterates over all modules required for a GridPath scenario and
17	calls their validate_inputs() method, which performs various validations
18	of the input data and scenario setup.
19	"""
20
21	import sqlite3	1✔
22	import sys	1✔
23	from argparse import ArgumentParser	1✔
24
25	from db.common_functions import connect_to_database, spin_on_database_lock	1✔
26	from gridpath.auxiliary.db_interface import (	1✔
27	get_required_capacity_types_from_database,
28	get_scenario_id_and_name,
29	)
30	from gridpath.auxiliary.validations import write_validation_to_database	1✔
31	from gridpath.common_functions import get_db_parser	1✔
32	from gridpath.auxiliary.module_list import determine_modules, load_modules	1✔
33	from gridpath.auxiliary.scenario_chars import (	1✔
34	OptionalFeatures,
35	SubScenarios,
36	get_scenario_structure_from_db,
37	)
38
39
40	def validate_inputs(	1✔
41	subproblems,
42	loaded_modules,
43	scenario_id,
44	weather_iteration,
45	hydro_iteration,
46	availability_iteration,
47	subscenarios,
48	conn,
49	):
50	""" "
51	For each module, load the inputs from the database and validate them
52
53	:param subproblems: SubProblems object with info on the subproblem/stage
54	structure
55	:param loaded_modules: list of imported modules (Python <class 'module'>
56	objects)
57	:param subscenarios: SubScenarios object with all subscenario info
58	:param conn: database connection
59	:return:
60	"""
61
62	# TODO: check if we even need database cursor (and subscenarios?)
63	# since presumably we already have our data in the inputs.
64	# need to go through each module's input validation to check this
65
66	# TODO: see if we can do some sort of automatic dtype validation for
67	# each table in the database? Problem is that you don't necessarily want
68	# to check the full table but only the appropriate subscenario
69
70	subproblems_list = subproblems.SUBPROBLEM_STAGES.keys()	1✔
71	for subproblem in subproblems_list:	1✔
72	stages = subproblems.SUBPROBLEM_STAGES[subproblem]	1✔
73	for stage in stages:	1✔
74	# 1. input validation within each module
75	for m in loaded_modules:	1✔
76	if hasattr(m, "validate_inputs"):	1✔
77	m.validate_inputs(	1✔
78	scenario_id=scenario_id,
79	subscenarios=subscenarios,
80	weather_iteration=weather_iteration,
81	hydro_iteration=hydro_iteration,
82	availability_iteration=availability_iteration,
83	subproblem=subproblem,
84	stage=stage,
85	conn=conn,
86	)
87
88	# 2. input validation across modules
89	# make sure geography and projects are in line
90	# ... (see Evernote validation list)
91	# create separate function for each validation that you call here
92
93
94	def validate_subscenario_ids(scenario_id, subscenarios, optional_features, conn):	1✔
95	"""
96	Check whether subscenarios_ids are consistent with:
97	- core required subscenario_ids
98	- data dependent subscenario_ids (e.g. new build)
99	- optional features
100	data.
101
102	:param subscenarios:
103	:param optional_features:
104	:param conn:
105	:return: Boolean, True is all subscenario IDs are valid.
106	"""
107
108	valid_core = validate_required_subscenario_ids(scenario_id, subscenarios, conn)	1✔
109
110	if valid_core:	1✔
111	valid_data_dependent = validate_data_dependent_subscenario_ids(	1✔
112	scenario_id, subscenarios, conn
113	)
114	else:
115	valid_data_dependent = False	×
116
117	valid_feature = validate_feature_subscenario_ids(	1✔
118	scenario_id, subscenarios, optional_features, conn
119	)
120
121	return valid_core and valid_data_dependent and valid_feature	1✔
122
123
124	def validate_feature_subscenario_ids(	1✔
125	scenario_id, subscenarios, optional_features, conn
126	):
127	"""
128
129	:param subscenarios:
130	:param optional_features:
131	:param conn:
132	:return:
133	"""
134
135	subscenario_ids_by_feature = determine_subscenarios_by_feature(conn)	1✔
136	feature_list = optional_features.get_active_features()	1✔
137
138	errors = {"High": [], "Low": []} # errors by severity	1✔
139	for feature, subscenario_ids in subscenario_ids_by_feature.items():	1✔
140	if feature not in ["core", "optional", "data_dependent"]:	1✔
141	for sc_id in subscenario_ids:	1✔
142	# If the feature is requested, and the associated subscenarios
143	# are not specified, raise a validation error
144	if feature in feature_list and getattr(subscenarios, sc_id) == "NULL":	1✔
145	errors["High"].append(	×
146	"Requested feature '{}' requires an input for '{}'".format(
147	feature, sc_id
148	)
149	)
150
151	# If the feature is not requested, and the associated
152	# subscenarios are specified, raise a validation error
153	# TODO: need to add handling of subscenarios shared among
154	# features; commenting out for now
155	# elif feature not in feature_list and \
156	# getattr(subscenarios, sc_id) != "NULL":
157	# errors["Low"].append(
158	# "Detected inputs for '{}' while related feature '{}' "
159	# "is not requested".format(sc_id, feature)
160	# )
161
162	for severity, error_list in errors.items():	1✔
163	write_validation_to_database(	1✔
164	conn=conn,
165	scenario_id=scenario_id,
166	weather_iteration="N/A",
167	hydro_iteration="N/A",
168	availability_iteration="N/A",
169	subproblem_id="N/A",
170	stage_id="N/A",
171	gridpath_module="N/A",
172	db_table="scenarios",
173	severity=severity,
174	errors=error_list,
175	)
176
177	# Return True if all required subscenario_ids are valid (list is empty)
178	return not bool(sum(errors.values(), []))	1✔
179
180
181	def validate_required_subscenario_ids(scenario_id, subscenarios, conn):	1✔
182	"""
183	Check whether the required subscenario_ids are specified in the db
184	:param subscenarios:
185	:param conn:
186	:return: boolean, True if all required subscenario_ids are specified
187	"""
188
189	required_subscenario_ids = determine_subscenarios_by_feature(conn)["core"]	1✔
190
191	errors = []	1✔
192	for sc_id in required_subscenario_ids:	1✔
193	if getattr(subscenarios, sc_id) is None:	1✔
194	errors.append(	×
195	"'{}' is a required input in the 'scenarios' table".format(sc_id)
196	)
197
198	write_validation_to_database(	1✔
199	conn=conn,
200	scenario_id=scenario_id,
201	weather_iteration="N/A",
202	hydro_iteration="N/A",
203	availability_iteration="N/A",
204	subproblem_id="N/A",
205	stage_id="N/A",
206	gridpath_module="N/A",
207	db_table="scenarios",
208	severity="High",
209	errors=errors,
210	)
211
212	# Return True if all required subscenario_ids are valid (list is empty)
213	return not bool(errors)	1✔
214
215
216	def validate_data_dependent_subscenario_ids(scenario_id, subscenarios, conn):	1✔
217	"""
218
219	:param subscenarios:
220	:param conn:
221	:return:
222	"""
223
224	assert subscenarios.PROJECT_PORTFOLIO_SCENARIO_ID is not None	1✔
225
226	req_cap_types = set(get_required_capacity_types_from_database(conn, scenario_id))	1✔
227
228	new_build_types = {	1✔
229	"gen_new_lin,",
230	"gen_new_bin",
231	"stor_new_lin",
232	"stor_new_bin",
233	"dr_new",
234	}
235	existing_build_types = {	1✔
236	"gen_spec",
237	"gen_ret_bin",
238	"gen_ret_lin",
239	"stor_spec",
240	}
241	dr_types = {"dr_new"}	1✔
242
243	# Determine required subscenario_ids
244	sc_id_type = []	1✔
245	if bool(req_cap_types & new_build_types):	1✔
246	sc_id_type.append(("PROJECT_NEW_COST_SCENARIO_ID", "New Build"))	1✔
247	if bool(req_cap_types & existing_build_types):	1✔
248	sc_id_type.append(("PROJECT_SPECIFIED_CAPACITY_SCENARIO_ID", "Existing"))	1✔
249	sc_id_type.append(("PROJECT_SPECIFIED_FIXED_COST_SCENARIO_ID", "Existing"))	1✔
250	if bool(req_cap_types & dr_types):	1✔
251	sc_id_type.append(("PROJECT_NEW_POTENTIAL_SCENARIO_ID", "Demand Response (DR)"))	×
252
253	# Check whether required subscenario_ids are present
254	errors = []	1✔
255	for sc_id, build_type in sc_id_type:	1✔
256	if getattr(subscenarios, sc_id) is None:	1✔
257	errors.append(	×
258	"'{}' is a required input in the 'scenarios' table if there "
259	"are '{}' resources in the portfolio".format(sc_id, build_type)
260	)
261
262	write_validation_to_database(	1✔
263	conn=conn,
264	scenario_id=scenario_id,
265	weather_iteration="N/A",
266	hydro_iteration="N/A",
267	availability_iteration="N/A",
268	subproblem_id="N/A",
269	stage_id="N/A",
270	gridpath_module="N/A",
271	db_table="scenarios",
272	severity="High",
273	errors=errors,
274	)
275
276	# Return True if all required subscenario_ids are valid (list is empty)
277	return not bool(errors)	1✔
278
279
280	def reset_input_validation(conn, scenario_id):	1✔
281	"""
282	Reset input validation: delete old input validation outputs and reset the
283	input validation status.
284	:param conn: database connection
285	:param scenario_id: scenario_id
286	:return:
287	"""
288	c = conn.cursor()	1✔
289
290	sql = """	1✔
291	DELETE FROM status_validation
292	WHERE scenario_id = ?;
293	"""
294	spin_on_database_lock(conn=conn, cursor=c, sql=sql, data=(scenario_id,), many=False)	1✔
295
296	sql = """	1✔
297	UPDATE scenarios
298	SET validation_status_id = 0
299	WHERE scenario_id = ?;
300	"""
301	spin_on_database_lock(conn=conn, cursor=c, sql=sql, data=(scenario_id,), many=False)	1✔
302
303
304	def update_validation_status(conn, scenario_id):	1✔
305	"""
306
307	:param conn:
308	:param scenario_id:
309	:return:
310	"""
311	c = conn.cursor()	1✔
312	validations = c.execute(	1✔
313	"""SELECT scenario_id
314	FROM status_validation
315	WHERE scenario_id = {}""".format(
316	str(scenario_id)
317	)
318	).fetchall()
319
320	if validations:	1✔
321	status = 2	×
322	# Print the errors
323	for e in c.execute(	×
324	f"""SELECT * FROM status_validation WHERE scenario_id = {scenario_id};"""
325	).fetchall():
326	print(e)	×
327	else:
328	status = 1	1✔
329
330	sql = """	1✔
331	UPDATE scenarios
332	SET validation_status_id = ?
333	WHERE scenario_id = ?;
334	"""
335	spin_on_database_lock(	1✔
336	conn=conn, cursor=c, sql=sql, data=(status, scenario_id), many=False
337	)
338
339
340	def parse_arguments(args):	1✔
341	"""
342	:param args: the script arguments specified by the user
343	:return: the parsed known argument values (<class 'argparse.Namespace'>
344	Python object)
345
346	Parse the known arguments.
347	"""
348	parser = ArgumentParser(add_help=True, parents=[get_db_parser()])	1✔
349
350	# Add quiet flag which can suppress run output
351	parser.add_argument(	1✔
352	"--quiet", default=False, action="store_true", help="Don't print run output."
353	)
354
355	parsed_arguments = parser.parse_known_args(args=args)[0]	1✔
356
357	return parsed_arguments	1✔
358
359
360	def determine_subscenarios_by_feature(conn):	1✔
361	"""
362
363	:param conn:
364	:return:
365	"""
366	c = conn.cursor()	1✔
367
368	feature_sc = c.execute(	1✔
369	"""SELECT feature, subscenario_id
370	FROM mod_feature_subscenarios"""
371	).fetchall()
372	feature_sc_dict = {}	1✔
373	for f, sc in feature_sc:	1✔
374	if f in feature_sc_dict:	1✔
375	feature_sc_dict[f].append(sc.upper())	1✔
376	else:
377	feature_sc_dict[f] = [sc.upper()]	1✔
378	return feature_sc_dict	1✔
379
380
381	def main(args=None):	1✔
382	"""
383
384	:return:
385	"""
386
387	# Retrieve scenario_id and/or name from args + "quiet" flag
388	if args is None:	1✔
UNCOV 389	args = sys.argv[1:]	×
390	parsed_arguments = parse_arguments(args=args)	1✔
391
392	if not parsed_arguments.quiet:	1✔
UNCOV 393	print("Validating inputs...")	×
394
395	db_path = parsed_arguments.database	1✔
396	scenario_id_arg = parsed_arguments.scenario_id	1✔
397	scenario_name_arg = parsed_arguments.scenario	1✔
398
399	conn = connect_to_database(db_path=db_path, detect_types=sqlite3.PARSE_DECLTYPES)	1✔
400	c = conn.cursor()	1✔
401
402	scenario_id, scenario_name = get_scenario_id_and_name(	1✔
403	scenario_id_arg=scenario_id_arg,
404	scenario_name_arg=scenario_name_arg,
405	c=c,
406	script="validate_inputs",
407	)
408
409	# Reset input validation status and results
410	reset_input_validation(conn, scenario_id)	1✔
411
412	# TODO: this is very similar to what's in get_scenario_inputs.py,
413	# so we should consolidate
414	# Get scenario characteristics (features, scenario_id, subscenarios, subproblems)
415	optional_features = OptionalFeatures(conn=conn, scenario_id=scenario_id)	1✔
416	subscenarios = SubScenarios(conn=conn, scenario_id=scenario_id)	1✔
417	scenario_structure = get_scenario_structure_from_db(	1✔
418	conn=conn, scenario_id=scenario_id
419	)
420
421	# Check whether subscenario_ids are valid
422	is_valid = validate_subscenario_ids(	1✔
423	scenario_id, subscenarios, optional_features, conn
424	)
425
426	# Only do the detailed input validation if all required subscenario_ids
427	# are specified (otherwise will get errors when loading data)
428	if is_valid:	1✔
429	# Load modules for all requested features
430	feature_list = optional_features.get_active_features()	1✔
431	# If any subproblem's stage list is non-empty, we have stages, so set
432	# the stages_flag to True to pass to determine_modules below
433	# This tells the determine_modules function to include the
434	# stages-related modules
435	stages_flag = any(	1✔
436	[
437	len(scenario_structure.SUBPROBLEM_STAGES[subp]) > 1
438	for subp in list(scenario_structure.SUBPROBLEM_STAGES.keys())
439	]
440	)
441	modules_to_use = determine_modules(	1✔
442	features=feature_list, multi_stage=stages_flag
443	)
444	loaded_modules = load_modules(modules_to_use=modules_to_use)	1✔
445
446	# Read in inputs from db and validate inputs for loaded modules
447	for weather_iteration in scenario_structure.ITERATION_STRUCTURE.keys():	1✔
448	for hydro_iteration in scenario_structure.ITERATION_STRUCTURE[	1✔
449	weather_iteration
450	].keys():
451	for availability_iteration in scenario_structure.ITERATION_STRUCTURE[	1✔
452	weather_iteration
453	][hydro_iteration]:
454	validate_inputs(	1✔
455	scenario_structure,
456	loaded_modules,
457	scenario_id,
458	weather_iteration,
459	hydro_iteration,
460	availability_iteration,
461	subscenarios,
462	conn,
463	)
464
465	else:
UNCOV 466	if not parsed_arguments.quiet:	×
UNCOV 467	print("Invalid subscenario ID(s). Skipped detailed input validation.")	×
468
469	# Update validation status:
470	update_validation_status(conn, scenario_id)	1✔
471
472	# Close the database connection explicitly
473	conn.close()	1✔
474
475
476	if __name__ == "__main__":	1✔
UNCOV 477	main()	×

blue-marble / gridpath / 17809960349

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