Skip to content
Snippets Groups Projects
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