paulin/EasyGA
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Cleaned up the database.

Browse Source 
Fixed sqlite3 Error import.
Changed attributes from annotations to dataclass fields.
Changed decorators into functions.
Cleaned up formatting (query strings/doc-strings).
This commit is contained in:
SimpleArt
2021-07-08 11:30:34 -04:00
parent b81d1614af
commit 3f2b2cb756
1 changed files with 97 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
EasyGA/database/sql_database.py
View File

@ -1,4 +1,5 @@
import sqlite3 import sqlite3
from sqlite3 import Error
from tabulate import tabulate from tabulate import tabulate
@ -11,11 +12,13 @@ class SQL_Database:
self.conn = None self.conn = None
self.config_id = None self.config_id = None
self._database_name = 'database.db' self._database_name = 'database.db'
self.config_structure = f""" self.config_structure = """
CREATE TABLE IF NOT EXISTS config ( CREATE TABLE IF NOT EXISTS config (
config_id INTEGER, config_id INTEGER,
attribute_name TEXT, attribute_name TEXT,
attribute_value TEXT)""" attribute_value TEXT
;)
"""
#=====================================# #=====================================#
@ -23,23 +26,18 @@ class SQL_Database:
#=====================================# #=====================================#
def create_all_tables(self, ga): def create_all_tables(self, ga):
"""Create the database if it doenst exist and then the data and config """Create the database if it doenst exist and then the data and config tables."""
tables.""" # Create the database connection.
# Create the database connection
self.create_connection() self.create_connection()
# No connection.
if self.conn is not None: if self.conn is None:
# Create data table
self.create_table(ga.sql_create_data_structure)
# Creare config table
self.create_table(self.config_structure)
# Set the config id
self.config_id = self.get_current_config()
else:
raise Exception("Error! Cannot create the database connection.") raise Exception("Error! Cannot create the database connection.")
# Create data table.
self.create_table(ga.sql_create_data_structure)
# Creare config table.
self.create_table(self.config_structure)
# Set the config id.
self.config_id = self.get_current_config()
def insert_config(self, ga): def insert_config(self, ga):
@ -48,7 +46,7 @@ class SQL_Database:
Notes: Notes:
"Attributes" here refers to ga.__annotations__.keys(), "Attributes" here refers to ga.__dataclass_fields__.keys(),
which allows the attributes to be customized. which allows the attributes to be customized.
Only attributes that are bool, float, int, or str will be used. Only attributes that are bool, float, int, or str will be used.
@ -58,61 +56,26 @@ class SQL_Database:
self.config_id = self.get_current_config() self.config_id = self.get_current_config()
# Setting the config_id index if there is no file # Setting the config_id index if there is no file
if self.config_id == None: if self.config_id is None:
self.config_id = 0 self.config_id = 0
else: else:
self.config_id = self.config_id + 1 self.config_id = self.config_id + 1
# Getting all the attributes from the attributes class # Getting all attribute fields from the attributes class
db_config = [ db_config = [
(self.config_id, attr_name, attr_value) (self.config_id, attr_name, attr_value)
for attr_name for attr_name
in ga.__annotations__ in ga.__dataclass_fields__
if isinstance((attr_value := getattr(ga, attr_name)), (bool, float, int, str)) if isinstance((attr_value := getattr(ga, attr_name)), (bool, float, int, str))
] ]
query = f""" query = """
INSERT INTO config(config_id, attribute_name, attribute_value) INSERT INTO config(config_id, attribute_name, attribute_value)
VALUES (?, ?, ?); VALUES (?, ?, ?);
""" """
self.conn.executemany(query, db_config) self.conn.executemany(query, db_config)
self.config_id = self.get_current_config() self.config_id = self.get_current_config()
#=====================================#
# Decorators: #
#=====================================#
def default_config_id(method):
"""Decorator used to set the default config_id inside other functions."""
def new_method(self, config_id = None):
input_id = self.config_id if config_id is None else config_id
return method(self, input_id)
return new_method
def format_query_data(method):
"""Decorator used to format query data"""
def new_method(self, config_id):
query = method(self, config_id)
# Unpack elements if they are lists with only 1 element
if type(query[0]) in (list, tuple) and len(query[0]) == 1:
query = [i[0] for i in query]
# Unpack list if it is a list with only 1 element
if type(query) in (list, tuple) and len(query) == 1:
query = query[0]
return query
return new_method
#=====================================# #=====================================#
# Request information Queries: # # Request information Queries: #
#=====================================# #=====================================#
@ -125,79 +88,91 @@ class SQL_Database:
def past_runs(self): def past_runs(self):
"""Show a summerization of the past runs that the user has done.""" """Show a summerization of the past runs that the user has done."""
query_data = self.query_all(f""" query_data = self.query_all("""
SELECT config_id,attribute_name,attribute_value SELECT config_id, attribute_name, attribute_value
FROM config;""") FROM config;
""")
print( table = tabulate(
tabulate( query_data,
query_data, headers = [
headers = [ 'config_id',
'config_id', 'attribute_name',
'attribute_name', 'attribute_value',
'attribute_value' ]
]
)
) )
print(table)
return table
@default_config_id
def get_generation_total_fitness(self, config_id): def get_generation_total_fitness(self, config_id):
"""Get each generations total fitness sum from the database """ """Get each generations total fitness sum from the database """
config_id = self.config_id if config_id is None else config_id
return self.query_all(f""" return self.query_all(f"""
SELECT SUM(fitness) SELECT SUM(fitness)
FROM data FROM data
WHERE config_id={config_id} WHERE config_id={config_id}
GROUP BY generation;""") GROUP BY generation;
""")
@default_config_id
def get_total_generations(self, config_id): def get_total_generations(self, config_id):
"""Get the total generations from the database""" """Get the total generations from the database"""
config_id = self.config_id if config_id is None else config_id
return self.query_one_item(f""" return self.query_one_item(f"""
SELECT COUNT(DISTINCT generation) SELECT COUNT(DISTINCT generation)
FROM data FROM data
WHERE config_id={config_id};""") WHERE config_id={config_id};
""")
@default_config_id
def get_highest_chromosome(self, config_id): def get_highest_chromosome(self, config_id):
"""Get the highest fitness of each generation""" """Get the highest fitness of each generation"""
config_id = self.config_id if config_id is None else config_id
return self.query_all(f""" return self.query_all(f"""
SELECT max(fitness) SELECT max(fitness)
FROM data FROM data
WHERE config_id={config_id} WHERE config_id={config_id}
GROUP by generation;""") GROUP by generation;
""")
@default_config_id
def get_lowest_chromosome(self, config_id): def get_lowest_chromosome(self, config_id):
"""Get the lowest fitness of each generation""" """Get the lowest fitness of each generation"""
config_id = self.config_id if config_id is None else config_id
return self.query_all(f""" return self.query_all(f"""
SELECT min(fitness) SELECT min(fitness)
FROM data FROM data
WHERE config_id={config_id} WHERE config_id={config_id}
GROUP by generation;""") GROUP by generation;
""")
def get_all_config_id(self): def get_all_config_id(self):
"""Get an array of all the DISTINCT config_id in the database""" """Get an array of all the DISTINCT config_id in the database"""
return self.query_all(f""" return self.query_all(f"""
SELECT DISTINCT config_id SELECT DISTINCT config_id
FROM config;""") FROM config;
""")
def get_each_generation_number(self, config_id): def get_each_generation_number(self, config_id):
"""Get an array of all the generation numbers""" """Get an array of all the generation numbers"""
return self.query_all(f""" return self.query_all(f"""
SELECT DISTINCT generation SELECT DISTINCT generation
FROM data FROM data
WHERE config_id={config_id};""") WHERE config_id={config_id};
""")
@ -214,12 +189,14 @@ class SQL_Database:
self.config_id, self.config_id,
generation, generation,
chromosome.fitness, chromosome.fitness,
repr(chromosome) repr(chromosome),
) )
# Create sql query structure # Create sql query structure
sql = """INSERT INTO data(config_id, generation, fitness, chromosome) sql = """
VALUES(?,?,?,?)""" INSERT INTO data(config_id, generation, fitness, chromosome)
VALUES(?, ?, ?, ?)
"""
cur = self.conn.cursor() cur = self.conn.cursor()
cur.execute(sql, db_chromosome) cur.execute(sql, db_chromosome)
@ -243,8 +220,10 @@ class SQL_Database:
] ]
# Create sql query structure # Create sql query structure
sql = """INSERT INTO data(config_id, generation, fitness, chromosome) sql = """
VALUES(?,?,?,?)""" INSERT INTO data(config_id, generation, fitness, chromosome)
VALUES(?,?,?,?)
"""
cur = self.conn.cursor() cur = self.conn.cursor()
cur.executemany(sql, db_chromosome_list) cur.executemany(sql, db_chromosome_list)
@ -257,8 +236,7 @@ class SQL_Database:
#=====================================# #=====================================#
def create_connection(self): def create_connection(self):
"""Create a database connection to the SQLite database """Create a database connection to the SQLite database specified by db_file."""
specified by db_file."""
try: try:
self.conn = sqlite3.connect(self.database_name) self.conn = sqlite3.connect(self.database_name)
@ -266,6 +244,7 @@ class SQL_Database:
self.conn = None self.conn = None
print(e) print(e)
def create_table(self, create_table_sql): def create_table(self, create_table_sql):
"""Create a table from the create_table_sql statement.""" """Create a table from the create_table_sql statement."""
@ -276,22 +255,31 @@ class SQL_Database:
print(e) print(e)
@format_query_data def format_query_data(self, data):
"""Used to format query data."""
# Unpack elements if they are lists with only 1 element
if isinstance(data[0], (list, tuple)) and len(data[0]) == 1:
data = [i[0] for i in data]
# Unpack list if it is a list with only 1 element
if isinstance(data, (list, tuple)) and len(data) == 1:
data = data[0]
return data
def query_all(self, query): def query_all(self, query):
"""Query for muliple rows of data""" """Query for muliple rows of data"""
cur = self.conn.cursor() cur = self.conn.cursor()
cur.execute(query) cur.execute(query)
return cur.fetchall() return self.format_query_data(cur.fetchall())
@format_query_data
def query_one_item(self, query): def query_one_item(self, query):
"""Query for single data point""" """Query for single data point"""
cur = self.conn.cursor() cur = self.conn.cursor()
cur.execute(query) cur.execute(query)
return cur.fetchone() return self.format_query_data(cur.fetchone())
def remove_database(self): def remove_database(self):
@ -330,7 +318,7 @@ class SQL_Database:
# If the connection doesnt exist then print error # If the connection doesnt exist then print error
except: except:
raise Exception("""You are required to run a ga before you can connect to the database. Run ga.evolve() or ga.active()""") raise Exception("You are required to run a ga before you can connect to the database. Run ga.evolve() or ga.active()")
@conn.setter @conn.setter
@ -357,7 +345,7 @@ class SQL_Database:
# If the config_id doesnt exist then print error # If the config_id doesnt exist then print error
except: except:
raise Exception("""You are required to run a ga before you can connect to the database. Run ga.evolve() or ga.active()""") raise Exception("You are required to run a ga before you can connect to the database. Run ga.evolve() or ga.active()")
@config_id.setter @config_id.setter