-
VulcanixFR authored3de03613
Database.py 8.39 KiB
"""File which contains the definition of the Database selector window
"""
import PySimpleGUI as sg
from pathlib import Path
from threading import Thread, Semaphore
from tools.database import Database, init_database
from tools.predictive_indicators import compute_indicators_all_robots
from .DB_Metadata import DBFileManagerWindow
class DBSelectWin (sg.Window):
def __init__ (self, db: Database | None = None):
self.db = db
self.edit_window = None
self.indicator_popup: sg.Window = None
self.indicator_popup_lock = Semaphore()
self.indicator_progress: int = 0
self.indicator_max_progress: int = 0
self.indicator_run_name: str = "None"
self.db_use_done = False
super().__init__("Database Selector", self._layout(), finalize=True)
self.update_display()
def _layout (self):
"""Creates the window layout
"""
self.buttons = {
"-open-close-": sg.Button("Open", key="-open-close-"),
"-edit-": sg.Button("Edit Data", key="-edit-"),
"-indicators-": sg.Button("Compute indicators", key="-indicators-"),
"-new-": sg.Button("New Database", key="-new-"),
"-browse-": sg.FileBrowse(
key="-browse-",
initial_folder=str(Path(".").absolute()),
file_types=(("SQLite Database", "*.sqlite"),)
)
}
self.inputs = {
"-db_path-": sg.Input(key="-db_path-")
}
layout = [
[
sg.Text("Database : "),
self.inputs["-db_path-"],
self.buttons["-browse-"]
],
[
self.buttons["-open-close-"],
self.buttons["-edit-"],
self.buttons["-indicators-"],
sg.Push(),
self.buttons["-new-"],
]
]
# Saves and returns the Window Layout
self._win_layout = layout
return layout
def update_display (self):
"""Updates the displayed elements
"""
has_db = self.db is not None
using_data = (self.edit_window is not None) or (self.indicator_popup is not None)
self.buttons["-edit-"].update(
disabled=not has_db or using_data,
button_color=sg.theme_button_color()
)
self.buttons["-indicators-"].update(
disabled=not has_db or using_data,
button_color=sg.theme_button_color()
)
self.buttons["-new-"].update(
disabled=has_db or using_data,
button_color=sg.theme_button_color()
)
self.buttons["-open-close-"].update(
text="Close" if has_db else "Open",
disabled=using_data,
button_color=sg.theme_button_color()
)
self.inputs["-db_path-"].update(
disabled=has_db or using_data,
text_color="#000"
)
self.buttons["-browse-"].update(
disabled=has_db or using_data,
button_color=sg.theme_button_color()
)
self.indicator_popup_lock.acquire()
if self.indicator_popup is not None:
self.indicator_popup["-run-name-"].update(value=self.indicator_run_name)
self.indicator_popup["-progress-"].update(current_count=self.indicator_progress)
self.indicator_popup.read(1)
self.indicator_popup_lock.release()
def open_database (self, path: str):
"""Tries to open a Database
Args:
path (str): The path to the database
"""
file_path = Path(path)
if not file_path.exists():
sg.popup(f'Not found : {file_path.stem}', title="Error")
return
if not file_path.is_file():
sg.popup(f'Not a file : {file_path.stem}', title="Error")
return
if file_path.suffix != ".sqlite":
sg.popup(f'Not a database : {file_path.stem}', title="Error")
return
self.db = Database(file_path)
def close_database (self):
"""Closes the currently opened database
"""
if self.db is not None:
self.db.close()
self.db = None
def create_db (self):
"""Create a new Database
"""
# Get the database name
text: str = sg.popup_get_text("Please enter the new Database name :", "New Database", "./db/new_db")
# Check cancelled
if text is not None:
# Add extension
if not text.endswith(".sqlite"):
text += ".sqlite"
path = Path(text)
# Create the containing subfolders if they do not exist
if not path.parent.exists():
path.parent.mkdir(parents=True)
# Create the file
init_database(path)
# Open the database
self.inputs["-db_path-"].update(value=text)
self.open_database(text)
def compute_indicators (self):
self.indicator_max_progress = len(self.db.list_files()) + 1
self.indicator_progress = 0
popup = sg.Window("Computing indicators", [
[ sg.Push(), sg.Text("Computing indicators."), sg.Push() ],
[
sg.Push(),
sg.Text(
"This might take a while",
justification="center"
),
sg.Push()
],
[
sg.Push(),
sg.Text("Waiting ...", key="-run-name-"),
sg.Push()
],
[
sg.ProgressBar(self.indicator_max_progress, expand_x=True, key="-progress-")
]
], finalize=True, disable_close=True, disable_minimize=True)
popup.read(1)
self.indicator_popup = popup
self.db.close()
def compute_thread (self: DBSelectWin):
def next (name: str):
print("Next", name)
self.indicator_popup_lock.acquire()
self.indicator_run_name = name
self.indicator_progress += 1
self.indicator_popup_lock.release()
print("Released lock")
self.db.open()
compute_indicators_all_robots(self.db, next)
self.db.close()
self.indicator_popup_lock.acquire()
self.db_use_done = True
self.indicator_popup_lock.release()
Thread(target=compute_thread, daemon=True, args=[self]).start()
def poll (self):
"""Runs the window
Returns:
bool: The window is still opened
"""
event, values = self.read(10)
# Window closed
if event == sg.WIN_CLOSED:
return False
# No interaction
if event == sg.TIMEOUT_EVENT:
# Check if we have the edit window opened
if self.edit_window is not None:
# Update the Edit Window
if not self.edit_window.poll():
# The window was closed
self.edit_window.close()
self.edit_window = None
self.update_display()
# DB management
self.indicator_popup_lock.acquire()
if self.db_use_done == True:
self.db_use_done = False
self.indicator_popup.close()
self.indicator_popup = None
self.db.open()
self.indicator_popup_lock.release()
# Update display
self.update_display()
return True
# User opens the Database
if event == "-open-close-":
if self.db is None:
self.open_database(values["-db_path-"])
else:
self.close_database()
# User creates a new Database
if event == "-new-":
self.create_db()
# User edits a Database
if event == "-edit-":
if self.db is None:
sg.popup(f'Please open a Database before trying to edit.', title="Error")
else:
self.edit_window = DBFileManagerWindow(self.db)
# User wants to plot some data
if event == "-indicators-":
self.compute_indicators()
# Update this window display
self.update_display()
return True