21263611097

Committed 22 Jan 2026 08:21PM UTC coverage: 89.055%. Remained the same

Build # 21263611097

Build Type

push

github

Committed by

anamileva

Commit Message

Lint with black==26.1.0

Coverage Stats

90 of 143 new or added lines in 74 files covered. (62.94%)

20 existing lines in 10 files now uncovered.

27550 of 30936 relevant lines covered (89.05%)

0.89 hits per line

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

72.53

/db/create_database.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.

"""
Create an empty GridPath database with the appropriate table structure.

The user may specify the name and location of the GridPath database path using the
*--database* flag.

>>> gridpath_create_database --database PATH/DO/DB

The default schema for the GridPath SQLite database is in db/db_schema.sql.

.. _database-structure-section-ref:

To create a database for GridPath raw data, point to the schema in
../data_toolkit/raw_data_db_schema.sql instead and also specify the
--omit_data flag.

"""

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

from db.common_functions import spin_on_database_lock, spin_on_database_lock_generic


def parse_arguments(arguments):
    """

    :return:
    """
    parser = ArgumentParser(add_help=True)

    # Scenario name and location options
    parser.add_argument(
        "--database",
        default="./io.db",
        help="The database file path relative to the current "
        "working directory. Defaults to ./io.db ",
    )
    parser.add_argument(
        "--db_schema",
        default="./db_schema.sql",
        help="Name of the SQL file containing the database "
        "schema. Assumed to be in same directory as"
        "create_database.py",
    )
    parser.add_argument(
        "--in_memory",
        default=False,
        action="store_true",
        help="Create in-memory database. The database " "argument will be inactive.",
    )
    parser.add_argument(
        "--data_directory",
        default="./data",
        help="Directory of model defaults data.",
    )
    parser.add_argument(
        "--omit_data",
        default=False,
        action="store_true",
        help="Don't load the model defaults data from the data directory.",
    )
    parser.add_argument(
        "--custom_units",
        default=False,
        action="store_true",
        help="Ask the user for custom units.",
    )

    # Parse arguments
    parsed_arguments = parser.parse_known_args(args=arguments)[0]

    return parsed_arguments


def create_database_schema(conn, parsed_arguments):
    """
    :param conn: database connection
    :param parsed_arguments:

    """
    schema_path = os.path.join(os.path.dirname(__file__), parsed_arguments.db_schema)

    with open(schema_path, "r") as db_schema_script:
        schema = db_schema_script.read()
        conn.executescript(schema)


def load_data(conn, data_directory, custom_units):
    """
    Load GridPath structural data (e.g. defaults, allowed modules, validation
    data, UI component data, etc.)
    :param conn: database connection
    :param data_directory:
    :param omit_data:
    :param custom_units: Boolean, True if user-specified units
    :param omit_data:
    :return:
    """
    expected_files = [
        "mod_availability_types",
        "mod_capacity_types",
        "mod_features",
        "mod_feature_subscenarios",
        "mod_horizon_boundary_types",
        "mod_months",
        "mod_operational_types",
        "mod_prm_types",
        "mod_reserve_types",
        "mod_run_status_types",
        "mod_tx_availability_types",
        "mod_tx_capacity_types",
        "mod_tx_operational_types",
        "mod_tx_capacity_and_tx_operational_type_invalid_combos",
        "mod_capacity_and_operational_type_invalid_combos",
        "mod_units",
        "mod_validation_status_types",
        "ui_scenario_detail_table_metadata",
        "ui_scenario_detail_table_row_metadata",
        "ui_scenario_results_plot_metadata",
        "ui_scenario_results_table_metadata",
        "viz_technologies",
    ]
    for f in expected_files:
        load_aux_data(conn=conn, data_directory=data_directory, filename=f)

    set_custom_units(conn=conn, custom_units=custom_units)


def set_custom_units(conn, custom_units):
    """
    Load the units
    :param conn:
    :param custom_units: Boolean, True if user-specified units
    :return:
    """
    c = conn.cursor()
    if custom_units:
        # Retrieve settings from user
        power = input("""
            Specify the unit of power, e.g. kW, MW, GW, etc.
            Note: the unit of energy will be derived from the unit of power by 
            multiplying by 1 hour, e.g. MW -> MWh.
            Use `default` to keep the defaults (MW). 
            """)
        fuel_energy = input("""
            Specify the unit of fuel energy content, e.g. MMBtu, J, MJ, etc.
            Use 'default' to keep defaults (MMBtu). 
            """)
        cost = input("""
            Specify the unit of cost, e.g. USD, EUR, INR, etc.
            Use 'default' to keep defaults (USD).
            """)
        carbon_emissions = input("""
            Specify the unit of carbon emissions, e.g. tCO2, MtCO2, etc. 
            Use 'default' to keep defaults (tCO2; metric tonne)
            """)

        # Update table with user settings
        if power != "default":
            sql = """UPDATE mod_units
                SET unit = ?
                WHERE metric = 'power'"""
            spin_on_database_lock(
                conn=conn, cursor=c, sql=sql, many=False, data=(power,)
            )
            # add energy units based on user's power units
            energy = power + "h"
            sql = """UPDATE mod_units
                SET unit = ?
                WHERE metric = 'energy'"""
            spin_on_database_lock(
                conn=conn, cursor=c, sql=sql, many=False, data=(energy,)
            )
        if fuel_energy != "default":
            sql = """UPDATE mod_units
                SET unit = ?
                WHERE metric = 'fuel_energy'"""
            spin_on_database_lock(
                conn=conn, cursor=c, sql=sql, many=False, data=(fuel_energy,)
            )
        if cost != "default":
            sql = """UPDATE mod_units
                SET unit = ?
                WHERE metric = 'cost'"""
            spin_on_database_lock(
                conn=conn, cursor=c, sql=sql, many=False, data=(cost,)
            )
        if carbon_emissions != "default":
            sql = """UPDATE mod_units
                SET unit = ?
                WHERE metric = 'carbon_emissions'"""
            spin_on_database_lock(
                conn=conn, cursor=c, sql=sql, many=False, data=(carbon_emissions,)
            )

    # Derive secondary units
    df = pd.read_sql(sql="SELECT * FROM mod_units", con=conn, index_col="metric")
    for sec_metric in df[df["type"] == "secondary"].index:
        numerator = df.loc[sec_metric, "numerator_core_units"]
        if pd.isna(numerator) or numerator == "":
            num_str = "1"
        else:
            num_metrics = numerator.split("*")
            num_units = [df.loc[m, "unit"] for m in num_metrics]
            num_str = "-".join(num_units)

        denominator = df.loc[sec_metric, "denominator_core_units"]
        if pd.isna(denominator) or denominator == "":
            denom_str = ""
        else:
            denom_metrics = denominator.split("*")
            denom_units = [df.loc[m, "unit"] for m in denom_metrics]
            denom_str = "/" + "-".join(denom_units)

        sec_unit = num_str + denom_str

        sql = """UPDATE mod_units
            SET unit = ?
            WHERE metric = ?"""
        spin_on_database_lock(
            conn=conn, cursor=c, sql=sql, many=False, data=(sec_unit, sec_metric)
        )


def load_aux_data(conn, data_directory, filename):
    """
    :param conn:
    :param data_directory:
    :param filename:
    :param sql:
    :return:

    """
    data = []
    cursor = conn.cursor()

    file_path = os.path.join(
        os.path.dirname(__file__), data_directory, f"{filename}.csv"
    )
    df = pd.read_csv(file_path, delimiter=",")
    spin_on_database_lock_generic(
        command=df.to_sql(
            name=filename,
            con=conn,
            if_exists="append",
            index=False,
        )
    )


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

    if parsed_args.in_memory:
        db_path = ":memory:"
    else:
        db_path = parsed_args.database
        if os.path.isfile(db_path):
            print(
                """WARNING: The database file {} already exists. Please 
                delete it before re-creating the database""".format(
                    os.path.abspath(db_path)
                )
            )
            sys.exit()

    # Connect to the database
    conn = sqlite3.connect(database=db_path)
    # Allow concurrent reading and writing
    conn.execute("PRAGMA journal_mode=WAL")
    # Enforce foreign keys (default = not enforced)
    conn.execute("PRAGMA foreign_keys=ON;")
    # Create schema
    create_database_schema(conn=conn, parsed_arguments=parsed_args)
    # Load data
    if not parsed_args.omit_data:
        load_data(
            conn=conn,
            data_directory=parsed_args.data_directory,
            custom_units=parsed_args.custom_units,
        )
    # Close the database
    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	Create an empty GridPath database with the appropriate table structure.
17
18	The user may specify the name and location of the GridPath database path using the
19	--database flag.
20
21	>>> gridpath_create_database --database PATH/DO/DB
22
23	The default schema for the GridPath SQLite database is in db/db_schema.sql.
24
25	.. _database-structure-section-ref:
26
27	To create a database for GridPath raw data, point to the schema in
28	../data_toolkit/raw_data_db_schema.sql instead and also specify the
29	--omit_data flag.
30
31	"""
32
33	from argparse import ArgumentParser	1✔
34	import csv	1✔
35	import os.path	1✔
36	import pandas as pd	1✔
37	import sqlite3	1✔
38	import sys	1✔
39
40	from db.common_functions import spin_on_database_lock, spin_on_database_lock_generic	1✔
41
42
43	def parse_arguments(arguments):	1✔
44	"""
45
46	:return:
47	"""
48	parser = ArgumentParser(add_help=True)	1✔
49
50	# Scenario name and location options
51	parser.add_argument(	1✔
52	"--database",
53	default="./io.db",
54	help="The database file path relative to the current "
55	"working directory. Defaults to ./io.db ",
56	)
57	parser.add_argument(	1✔
58	"--db_schema",
59	default="./db_schema.sql",
60	help="Name of the SQL file containing the database "
61	"schema. Assumed to be in same directory as"
62	"create_database.py",
63	)
64	parser.add_argument(	1✔
65	"--in_memory",
66	default=False,
67	action="store_true",
68	help="Create in-memory database. The database " "argument will be inactive.",
69	)
70	parser.add_argument(	1✔
71	"--data_directory",
72	default="./data",
73	help="Directory of model defaults data.",
74	)
75	parser.add_argument(	1✔
76	"--omit_data",
77	default=False,
78	action="store_true",
79	help="Don't load the model defaults data from the data directory.",
80	)
81	parser.add_argument(	1✔
82	"--custom_units",
83	default=False,
84	action="store_true",
85	help="Ask the user for custom units.",
86	)
87
88	# Parse arguments
89	parsed_arguments = parser.parse_known_args(args=arguments)[0]	1✔
90
91	return parsed_arguments	1✔
92
93
94	def create_database_schema(conn, parsed_arguments):	1✔
95	"""
96	:param conn: database connection
97	:param parsed_arguments:
98
99	"""
100	schema_path = os.path.join(os.path.dirname(__file__), parsed_arguments.db_schema)	1✔
101
102	with open(schema_path, "r") as db_schema_script:	1✔
103	schema = db_schema_script.read()	1✔
104	conn.executescript(schema)	1✔
105
106
107	def load_data(conn, data_directory, custom_units):	1✔
108	"""
109	Load GridPath structural data (e.g. defaults, allowed modules, validation
110	data, UI component data, etc.)
111	:param conn: database connection
112	:param data_directory:
113	:param omit_data:
114	:param custom_units: Boolean, True if user-specified units
115	:param omit_data:
116	:return:
117	"""
118	expected_files = [	1✔
119	"mod_availability_types",
120	"mod_capacity_types",
121	"mod_features",
122	"mod_feature_subscenarios",
123	"mod_horizon_boundary_types",
124	"mod_months",
125	"mod_operational_types",
126	"mod_prm_types",
127	"mod_reserve_types",
128	"mod_run_status_types",
129	"mod_tx_availability_types",
130	"mod_tx_capacity_types",
131	"mod_tx_operational_types",
132	"mod_tx_capacity_and_tx_operational_type_invalid_combos",
133	"mod_capacity_and_operational_type_invalid_combos",
134	"mod_units",
135	"mod_validation_status_types",
136	"ui_scenario_detail_table_metadata",
137	"ui_scenario_detail_table_row_metadata",
138	"ui_scenario_results_plot_metadata",
139	"ui_scenario_results_table_metadata",
140	"viz_technologies",
141	]
142	for f in expected_files:	1✔
143	load_aux_data(conn=conn, data_directory=data_directory, filename=f)	1✔
144
145	set_custom_units(conn=conn, custom_units=custom_units)	1✔
146
147
148	def set_custom_units(conn, custom_units):	1✔
149	"""
150	Load the units
151	:param conn:
152	:param custom_units: Boolean, True if user-specified units
153	:return:
154	"""
155	c = conn.cursor()	1✔
156	if custom_units:	1✔
157	# Retrieve settings from user
NEW 158	power = input("""	×
159	Specify the unit of power, e.g. kW, MW, GW, etc.
160	Note: the unit of energy will be derived from the unit of power by
161	multiplying by 1 hour, e.g. MW -> MWh.
162	Use `default` to keep the defaults (MW).
163	""")
NEW 164	fuel_energy = input("""	×
165	Specify the unit of fuel energy content, e.g. MMBtu, J, MJ, etc.
166	Use 'default' to keep defaults (MMBtu).
167	""")
NEW 168	cost = input("""	×
169	Specify the unit of cost, e.g. USD, EUR, INR, etc.
170	Use 'default' to keep defaults (USD).
171	""")
NEW 172	carbon_emissions = input("""	×
173	Specify the unit of carbon emissions, e.g. tCO2, MtCO2, etc.
174	Use 'default' to keep defaults (tCO2; metric tonne)
175	""")
176
177	# Update table with user settings
178	if power != "default":	×
179	sql = """UPDATE mod_units	×
180	SET unit = ?
181	WHERE metric = 'power'"""
182	spin_on_database_lock(	×
183	conn=conn, cursor=c, sql=sql, many=False, data=(power,)
184	)
185	# add energy units based on user's power units
186	energy = power + "h"	×
187	sql = """UPDATE mod_units	×
188	SET unit = ?
189	WHERE metric = 'energy'"""
190	spin_on_database_lock(	×
191	conn=conn, cursor=c, sql=sql, many=False, data=(energy,)
192	)
193	if fuel_energy != "default":	×
194	sql = """UPDATE mod_units	×
195	SET unit = ?
196	WHERE metric = 'fuel_energy'"""
197	spin_on_database_lock(	×
198	conn=conn, cursor=c, sql=sql, many=False, data=(fuel_energy,)
199	)
200	if cost != "default":	×
201	sql = """UPDATE mod_units	×
202	SET unit = ?
203	WHERE metric = 'cost'"""
204	spin_on_database_lock(	×
205	conn=conn, cursor=c, sql=sql, many=False, data=(cost,)
206	)
207	if carbon_emissions != "default":	×
208	sql = """UPDATE mod_units	×
209	SET unit = ?
210	WHERE metric = 'carbon_emissions'"""
211	spin_on_database_lock(	×
212	conn=conn, cursor=c, sql=sql, many=False, data=(carbon_emissions,)
213	)
214
215	# Derive secondary units
216	df = pd.read_sql(sql="SELECT * FROM mod_units", con=conn, index_col="metric")	1✔
217	for sec_metric in df[df["type"] == "secondary"].index:	1✔
218	numerator = df.loc[sec_metric, "numerator_core_units"]	1✔
219	if pd.isna(numerator) or numerator == "":	1✔
220	num_str = "1"	×
221	else:
222	num_metrics = numerator.split("*")	1✔
223	num_units = [df.loc[m, "unit"] for m in num_metrics]	1✔
224	num_str = "-".join(num_units)	1✔
225
226	denominator = df.loc[sec_metric, "denominator_core_units"]	1✔
227	if pd.isna(denominator) or denominator == "":	1✔
228	denom_str = ""	×
229	else:
230	denom_metrics = denominator.split("*")	1✔
231	denom_units = [df.loc[m, "unit"] for m in denom_metrics]	1✔
232	denom_str = "/" + "-".join(denom_units)	1✔
233
234	sec_unit = num_str + denom_str	1✔
235
236	sql = """UPDATE mod_units	1✔
237	SET unit = ?
238	WHERE metric = ?"""
239	spin_on_database_lock(	1✔
240	conn=conn, cursor=c, sql=sql, many=False, data=(sec_unit, sec_metric)
241	)
242
243
244	def load_aux_data(conn, data_directory, filename):	1✔
245	"""
246	:param conn:
247	:param data_directory:
248	:param filename:
249	:param sql:
250	:return:
251
252	"""
253	data = []	1✔
254	cursor = conn.cursor()	1✔
255
256	file_path = os.path.join(	1✔
257	os.path.dirname(__file__), data_directory, f"{filename}.csv"
258	)
259	df = pd.read_csv(file_path, delimiter=",")	1✔
260	spin_on_database_lock_generic(	1✔
261	command=df.to_sql(
262	name=filename,
263	con=conn,
264	if_exists="append",
265	index=False,
266	)
267	)
268
269
270	def main(args=None):	1✔
271	if args is None:	1✔
272	args = sys.argv[1:]	×
273	parsed_args = parse_arguments(arguments=args)	1✔
274
275	if parsed_args.in_memory:	1✔
276	db_path = ":memory:"	1✔
277	else:
278	db_path = parsed_args.database	1✔
279	if os.path.isfile(db_path):	1✔
280	print(	×
281	"""WARNING: The database file {} already exists. Please
282	delete it before re-creating the database""".format(
283	os.path.abspath(db_path)
284	)
285	)
286	sys.exit()	×
287
288	# Connect to the database
289	conn = sqlite3.connect(database=db_path)	1✔
290	# Allow concurrent reading and writing
291	conn.execute("PRAGMA journal_mode=WAL")	1✔
292	# Enforce foreign keys (default = not enforced)
293	conn.execute("PRAGMA foreign_keys=ON;")	1✔
294	# Create schema
295	create_database_schema(conn=conn, parsed_arguments=parsed_args)	1✔
296	# Load data
297	if not parsed_args.omit_data:	1✔
298	load_data(	1✔
299	conn=conn,
300	data_directory=parsed_args.data_directory,
301	custom_units=parsed_args.custom_units,
302	)
303	# Close the database
304	conn.close()	1✔
305
306
307	if __name__ == "__main__":	1✔
308	main()	×

blue-marble / gridpath / 21263611097

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