29355760456

Committed 14 Jul 2026 03:31PM UTC coverage: 88.58% (-0.001%) from 88.581%

Build # 29355760456

Build Type

push

github

Committed by

anamileva

Commit Message

Use absolute path for DB schema file

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

21 existing lines in 1 file now uncovered.

29034 of 32777 relevant lines covered (88.58%)

0.89 hits per line

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

72.9

/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.
A custom schema may be specified with the *--db_schema* flag; relative paths
are resolved against the current working directory.

.. _database-structure-section-ref:

To create a database for GridPath raw data, point *--db_schema* to the
raw_data_db_schema.sql file in the data_toolkit package directory instead
and also specify the --omit_data flag.

"""

from argparse import ArgumentParser
from gridpath.common_functions import get_version_parser
import csv
import datetime
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
from version import __version__


def parse_arguments(arguments):
    """

    :return:
    """
    parser = ArgumentParser(add_help=True, parents=[get_version_parser()])

    # 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=os.path.join(os.path.dirname(__file__), "db_schema.sql"),
        help="Path to the SQL file containing the database "
        "schema. Relative paths are resolved against the "
        "current working directory. Defaults to the "
        "db_schema.sql file that ships with GridPath.",
    )
    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:

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


def write_db_metadata(conn):
    """
    :param conn: database connection

    Record the GridPath version used to create the database and, if the
    schema tracks it (the raw-data database schema does not), the database
    creation datetime.
    """
    columns = [
        row[1] for row in conn.execute("PRAGMA table_info(db_metadata);").fetchall()
    ]
    if "created_datetime" in columns:
        sql = """INSERT INTO db_metadata (gridpath_version, created_datetime)
            VALUES (?, ?);"""
        data = (
            __version__,
            datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        )
    else:
        sql = "INSERT INTO db_metadata (gridpath_version) VALUES (?);"
        data = (__version__,)

    spin_on_database_lock(
        conn=conn,
        cursor=conn.cursor(),
        sql=sql,
        data=data,
        many=False,
    )


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

sylvan-energy / gridpath / 29355760456

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