main.py

import kivy
from kivy import platform
import os

# Logging related
import logging
import sys
from kivy.logger import Logger
import warnings


# Redirect print (stdout) and errors (stderr)
# Redirect print statements
class LoggerWriter:
    def __init__(self, level):
        self.level = level

    def write(self, message):
        if message.strip():
            self.level(message.strip())

    def flush(self):
        pass


sys.stdout = LoggerWriter(Logger.info)
sys.stderr = LoggerWriter(Logger.error)


# Redirect warnings
def custom_showwarning(message, category, filename, lineno, file=None, line=None):
    Logger.warning(f"{filename}:{lineno}: {category.__name__}: {message}")


warnings.showwarning = custom_showwarning
warnings.simplefilter("always")


# Redirect uncaught exceptions
def log_exception(exctype, value, tb):
    Logger.exception(f"Uncaught exception: {value}", exc_info=(exctype, value, tb))


sys.excepthook = log_exception

# import matplotlib
from interactive_converter2D import main as plot2D
from interactive_converter3D import app_window_3D as plot3D
import api.visualise as vis

kivy.require('2.3.0')
os.environ['KIVY_LOG_LEVEL'] = 'info'
vis.logging.getLogger('PIL').setLevel(vis.logging.WARNING)


def plot_wrapper(x, y):
    """ Based on matplotlib.pyplot.plot """
    figure = vis.plot(x, y)
    plot2D(figure)
    return figure


def scatter_plot_2d_wrapper(x, y):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_2d(x, y)
    plot2D(figure)
    return figure


def scatter_plot_3d_to_2d_wrapper(x, y, z):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_3d_to_2d(x, y, z)
    plot2D(figure)
    return figure


def scatter_plot_4d_to_2d_wrapper(x, y, z, sizes):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_4d_to_2d(x, y, z, sizes)
    plot2D(figure)
    return figure


def scatter_plot_3d_wrapper(x, y, z):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_3d(x, y, z)
    plot3D(figure)
    return figure


def scatter_plot_4d_to_3D_wrapper(x, y, z, c):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_4d_to_3d(x, y, z, c)
    plot3D(figure)
    return figure


def scatter_plot_5d_to_3d_wrapper(x, y, z, c, sizes):
    """ Based on matplotlib.pyplot.scatter """
    figure = vis.scatter_plot_5d_to_3d(x, y, z, c, sizes)
    plot3D(figure)
    return figure


def bar_wrapper(categories, values):
    """ Based on matplotlib.pyplot.bar """
    figure = vis.bar(categories, values)
    plot2D(figure)
    return figure


def histogram_wrapper(values):
    """ Based on matplotlib.pyplot.hist """
    figure = vis.histogram(values)
    plot2D(figure)
    return figure


def pie_chart_wrapper(sizes, labels):
    """ Based on matplotlib.pyplot.pie """
    figure = vis.pie_chart(sizes, labels)
    plot2D(figure)
    return figure


def heatmap_2d_wrapper(data, x_labels, y_labels):
    """ Based on matplotlib.pyplot.imshow """
    figure = vis.heatmap_2d(data, x_labels, y_labels, 'Predicted', 'Actual')
    plot2D(figure)
    return figure


def contour_plot_wrapper(x, y):
    """ Based on matplotlib.pyplot.contour """
    figure = vis.contour_plot(x, y)
    plot2D(figure)
    return figure


def box_plot_wrapper(data):
    """ Based on matplotlib.pyplot.boxplot """
    figure = vis.box_plot(data)
    plot2D(figure)
    return figure


def violin_plot_wrapper(data):
    """ Based on matplotlib.pyplot.violinplot """
    figure = vis.violin_plot(data)
    plot2D(figure)
    return figure


def step_plot_wrapper(x, y):
    """ Based on matplotlib.pyplot.step """
    figure = vis.step_plot(x, y)
    plot2D(figure)
    return figure


def error_bar_plot_wrapper(x, y, errors):
    """ Based on matplotlib.pyplot.errorbar """
    figure = vis.error_bar_plot(x, y, errors)
    plot2D(figure)
    return figure


def quiver_plot_wrapper(x, y):
    """ Based on matplotlib.pyplot.quiver """
    figure = vis.quiver_plot(x, y)
    plot2D(figure)
    return figure


def dendrogram_wrapper(x, method):
    """ Based on scipy.cluster.hierarchy.dendrogram & linkage """
    figure = vis.dendrogram(x, method)
    plot2D(figure)
    return figure


if __name__ == '__main__':
    # VIP imports
    import multiprocessing

    multiprocessing.set_start_method('spawn')

    # Builtins
    import json
    from functools import partial

    # Kivy imports
    from kivy.config import Config
    from kivy.core.window import Window
    from kivy.properties import OptionProperty

    # KivyMD imports
    from kivymd.app import MDApp
    from kivymd.uix.screen import MDScreen
    from kivymd.uix.transition import MDSharedAxisTransition
    from kivymd.uix.boxlayout import MDBoxLayout
    from kivymd.uix.label import MDLabel

    # Toolkit imports
    from widgets.checkbox import ZCheckbox  # noqa
    from widgets.helptip import Helptip  # noqa
    from widgets.navdrawer import ZNavigationDrawer, ZNavigationDrawerMenu  # noqa
    from widgets.topappbar import ZTopAppBar  # noqa
    from widgets.card import ZCard  # noqa
    from widgets.switch import ZSwitch  # noqa
    from widgets.button import *  # noqa
    from widgets.scrollview import ZScrollViewHorizontal, ZScrollView  # noqa
    from widgets.table import ZTable  # noqa
    from widgets.settings import ZSettingsEntry, ZSettingsEntryMed, ZEntryMedItem  # noqa
    from widgets.listitem import ZListItem  # noqa
    from widgets.textfield import ZTextField  # noqa
    from widgets.divider import ZDividerDotted  # noqa
    from widgets.dialog import ZDialog

    from views.welcome import WelcomeLayout  # noqa
    from views.tutorial import TutorialLayout  # noqa
    from views.load import LoadLayout  # noqa
    from views.preprocessing import PreprocessingLayout  # noqa
    from views.split import SplitLayout  # noqa
    from views.technique import TechniqueLayout  # noqa
    from views.visualise import VisualiseLayout  # noqa
    from views.settings import SettingsLayout  # noqa

    import load_fonts  # noqa
    from utils import create_snackbar, set_and_dismiss
    from api.preprocessing import get_hash

    # Other libraries
    import numpy as np

    # Applying new theme
    from kivymd.theming import ThemeManager
    from kivy.utils import hex_colormap

    hex_colormap['uniwa'] = '#75AAFF'


    class CustomThemeMngr(ThemeManager):
        primary_palette = OptionProperty(
            None,
            options=[name_color.capitalize() for name_color in hex_colormap.keys()],
        )

        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.primary_palette = 'Uniwa'


    class RootScreen(MDScreen):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.md_bg_color = self.theme_cls.backgroundColor
            self.name = 'Main Screen'


    class DMToolkitApp(MDApp):
        def __init__(self, **kwargs):
            super().__init__(**kwargs)
            self.theme_cls = CustomThemeMngr()  # Overwrite the theme

            self.screen_manager = None

            self.load_path = None
            self.load_settings()  # Theme & other settings

            # Screen transition
            self.transition = MDSharedAxisTransition()
            self.transition.transition_axis = 'z'
            self.transition.duration = 0.2  # app.root.transition.opposite = False

            # Declarations and stubs
            self.pd_data = None
            self.ar_data = None
            self.ar_data_result = None

            self.X_train = None
            self.y_train = None
            self.X_test = None
            self.y_test = None
            self.test_set_name = ''
            self.split_state = 0  # 1 for splits, used by pre_train_check

            from sklearn.metrics import accuracy_score, recall_score, f1_score, roc_auc_score, mean_absolute_error, \
                mean_squared_error

            self.scorers = {
                'Accuracy': accuracy_score,
                'Recall': recall_score,
                'F1-score': f1_score,
                'ROC AUC': roc_auc_score,
                'Mean Absolute Error (MAE)': mean_absolute_error,
                'Mean Squared Error (MSE)': mean_squared_error
            }

            self.scorers_cv = {
                'Accuracy': 'accuracy',
                'Recall': 'recall',
                'F1-score': 'f1',
                'ROC AUC': 'roc_auc',
                'Mean Absolute Error (MAE)': 'neg_mean_absolute_error',
                'Mean Squared Error (MSE)': 'neg_Fmean_squared_error'
            }

            self.workflow_state = 0
            self.current_fit_btn = None  # Its on_release is called by other widgets

            self.snackbar_queue = []
            self.current_snackbar = None

            self.LoadLayout = None
            self.PreprocessingLayout = None
            self.TechniqueLayout = None
            self.VisualiseLayout = None

            self.table_container_ref = None
            self.ztable_ref = None
            self.ztable_header_ref = None
            self.ztable_index_ref = None

            self.dummy_bl = MDBoxLayout(size_hint_x=None, width=545)

            self.plot_processes = []  # Plots run in separate windows/processes

            self.d = ZDialog()
            self.dialog_widgets = dict()

        def build(self):
            # Window
            Window.always_on_top = False
            Window.size = 1190, 720
            Window.minimum_width = 800
            Window.minimum_height = 580
            Window.set_icon("assets/img/dmtoolkit_small.ico")

            # Build the app
            root = RootScreen()
            self.screen_manager = root.ids.screen_manager

            # Debug
            self.screen_manager.current = 'Welcome'

            return root

        def load_settings(self):
            try:
                with open('settings.json', 'r') as f:
                    settings = json.load(f)
                    self.theme_cls.theme_style = settings.get('theme_style', 'Light')
                    self.theme_cls.primary_palette = settings.get('primary_palette', 'Uniwa')
                    run_first_time = settings.get('run_first_time', '1')
                    self.load_path = settings.get('last_load_dataset_dir', None)
            except FileNotFoundError:
                # Set default settings if file does not exist
                self.theme_cls.theme_style = 'Light'
                self.theme_cls.primary_palette = 'Uniwa'
                run_first_time = '1'

            # Regardless of settings file
            self.theme_cls.theme_style_switch_animation = True
            self.theme_cls.theme_style_switch_animation_duration = 0.1
            # self.theme_cls.dynamic_scheme_name = 'TONAL_SPOT'

            if int(run_first_time):
                Config.set('input', 'mouse', 'mouse,disable_multitouch')
                Config.set('graphics', 'maxfps', '60')
                Config.set('kivy', 'exit_on_escape', '0')
                Config.write()

        def save_settings(self):
            if self.load_path is None:
                last_path = ' '
            else:
                if platform == 'win':
                    path_sep = '\\'
                else:
                    path_sep = '/'

                path_split = self.load_path.split(path_sep)

                if '.csv' in path_split[-1]:
                    last_path = path_sep.join(path_split[:-1])
                else:
                    last_path = self.load_path

            settings = {
                'theme_style': self.theme_cls.theme_style,
                'primary_palette': self.theme_cls.primary_palette,
                'run_first_time': str(0),
                'last_load_dataset_dir': last_path
            }
            with open('settings.json', 'w') as f:
                json.dump(settings, f)

        def on_stop(self):
            self.save_settings()

            for process in self.plot_processes:
                if process.is_alive():
                    process.terminate()

            super().on_stop()

        def enqueue_snackbar(self, message):
            self.snackbar_queue.append(message)
            self.show_next_snackbar()

        def show_next_snackbar(self):
            if self.current_snackbar or not self.snackbar_queue:
                return  # If there's already a snackbar showing or queue is empty

            message = self.snackbar_queue.pop(0)
            self.current_snackbar = create_snackbar(message)

            # Bind on_dismiss event to show the next snackbar
            self.current_snackbar.bind(on_dismiss=self.snackbar_dismiss)
            self.current_snackbar.open()

        def snackbar_dismiss(self, *args):
            """ Clear current snackbar reference & show next snackbar """
            self.current_snackbar = None
            self.show_next_snackbar()

        def reload_table(self, hash_before=None, body=True, header=False, silent=False):
            if hash_before and get_hash(self.pd_data) == hash_before:
                # Reload not needed
                self.enqueue_snackbar('Dataset has not been modified!')
            else:
                # Reload
                if body:
                    self.ztable_ref.load_dataframe(self.pd_data)
                if header:
                    self.ztable_header_ref.load_dataframe(self.pd_data)

                # Always update index
                self.ztable_index_ref.load_dataframe(self.pd_data)

                if not silent:
                    self.enqueue_snackbar('Success!')

        def to_screen(self, target_name):
            """ Dynamically switches screens, moving the table instance to keep it visible.
             A dummy widget of the same size is used for smooth animation """
            layouts_can_have_table = {
                'Preprocessing': self.PreprocessingLayout,
                'Split & Tune': self.SplitLayout,
                'Technique Selection': self.TechniqueLayout,
                'Visualise & Export': self.VisualiseLayout,
            }

            if target_name in layouts_can_have_table:
                target_layout = layouts_can_have_table[target_name]

                # Prepare the dummy for adding it, regardless.
                if self.dummy_bl.parent:
                    self.dummy_bl.parent.remove_widget(self.dummy_bl)

                # Remove the table from its current parent (it always has one)
                self.table_container_ref.parent.remove_widget(self.table_container_ref)

                # If origin screen is not a regular one, add the dummy!
                current_name = self.screen_manager.current
                if current_name in layouts_can_have_table and \
                        current_name != target_name:
                    origin_layout = layouts_can_have_table[current_name]
                    to_insert_dummy = len(origin_layout.ids.bl.children)
                    origin_layout.ids.bl.add_widget(self.dummy_bl, index=to_insert_dummy)

                # We attach the table after adding the dummy for a smooth animation...
                to_insert_table = len(target_layout.ids.bl.children)
                target_layout.ids.bl.add_widget(self.table_container_ref, index=to_insert_table)

            self.screen_manager.current = target_name

        def open_dialog(self, dialog_content_id, payload=None, custom_size=None):
            """ Handles the content of the dialog instance dynamically and displays it """
            # Remove previous child
            for widget in self.d.ids.content_container.children:
                if widget.visible is not False:
                    self.d.ids.content_container.remove_widget(widget)

            # Load new child
            new_content = self.dialog_widgets[dialog_content_id]

            # Update new child if necessary (dynamic)
            if payload:
                sv = new_content.children[0]
                lbl = sv.children[0]
                lbl.text = payload
                if custom_size:
                    sv.size = custom_size

            # Add new child & open dialogue
            self.d.ids.content_container.add_widget(new_content)
            self.d.open()

        def generate_buttons_cont(self, title, button_data, caller_id, set_extras=None):
            """ Dialog content generating helper method.
            Also sets certain properties depending on the button clicked """
            title_label = MDLabel(
                size_hint=(None, None), text=title,
                halign='center', valign='center',
                font_style='Settings Item Headline',
                bold=True, text_size=(None, None),
                pos_hint={'center_x': .5, 'center_y': .5}
            )

            title_label.text_size = [None, None]
            title_label.texture_update()
            title_label.size = title_label.texture_size
            title_label.height = title_label.height + 10

            buttons = []
            for i in range(len(button_data)):
                text = button_data[i]
                b = ZButton(text=text, style='outlined')
                widgets = [caller_id]
                attributes = ['text']
                values = [text]
                if set_extras:  # len(set_extras) == len(button_data)
                    for t in set_extras[i]:  # Specific data for each button
                        w, a, v = t
                        widgets.append(w)
                        attributes.append(a)
                        values.append(v)

                # Create a specific callback for each button
                b.on_release = partial(set_and_dismiss, widgets, attributes, values, self.d)
                buttons.append(b)

            container = MDBoxLayout(title_label, *buttons,
                                    orientation='vertical',
                                    adaptive_height=True,
                                    adaptive_width=True,
                                    spacing=15
                                    )
            return container

        def validate_col_selection(self, selection, empty_allowed=True, numeric_only=False, silent=False):
            """ Returns one or all names in a tuple if valid, and the empty tuple if invalid """
            empty_tuple = tuple()
            if selection.isnumeric():
                n_features = self.pd_data.columns.shape[0]
                if int(selection) < 0 or int(selection) >= n_features:
                    if not silent:
                        self.enqueue_snackbar('ERROR: Invalid column!')
                    return empty_tuple
                selection = self.pd_data.columns[int(selection)]  # Get the string

            if selection == '':
                if not empty_allowed:
                    if not silent:
                        self.enqueue_snackbar('Empty selection not allowed!')
                    return empty_tuple
                final = tuple(self.pd_data.columns)
            else:
                if selection not in self.pd_data.columns:
                    if not silent:
                        self.enqueue_snackbar('ERROR: Invalid column!')
                    return empty_tuple
                final = (selection,)

            if numeric_only:
                numeric_features = self.pd_data.select_dtypes(include=np.number).columns
                final_numeric_only = tuple([ft for ft in final if ft in numeric_features])
                if len(final_numeric_only) == 0:
                    if not silent:
                        self.enqueue_snackbar('ERROR: At least one numeric feature required!')
                    return empty_tuple
                elif len(final_numeric_only) != len(final):
                    if not silent:
                        self.enqueue_snackbar('WARNING: Not entire selection is numeric!')
                    final = final_numeric_only

            return final

        def validate_col_selection_multiple(self, selection, numeric_only=True):
            """ Returns one or more names in a list if valid, and the empty list if invalid """
            features = selection.split(',')

            # validate_col_selection always returns tuples
            valid_features_tups = [
                self.validate_col_selection(f.strip(), empty_allowed=False, numeric_only=numeric_only, silent=True) for
                f in
                features]

            if not all(valid_features_tups):
                bad_names = [f for f, v in zip(features, valid_features_tups) if not v]
                self.enqueue_snackbar(f'Aborting due to invalid columns: {bad_names}!')
                return []

            valid_features = [tups[0] for tups in valid_features_tups if tups]
            return valid_features

        def validate_rows(self, selection):
            rows = selection.split(',')
            rows = [r for r in rows if r]

            if len(rows) == 1 or len(rows) == 2:
                try:
                    rows = [int(r) for r in rows]
                except ValueError:
                    self.enqueue_snackbar('ERROR: Row indices must be integers!')
                    return []

                if len(rows) == 2:
                    start, end = rows
                    if start > end:
                        start, end = end, start
                    rows = list(range(start, end + 1))

                missing_indices = [idx for idx in rows if idx not in self.pd_data.index]
                rows = [idx for idx in rows if idx not in missing_indices]

                if rows:
                    if missing_indices:
                        self.enqueue_snackbar('WARNING: Missing index/indices in selected range!')
                        if len(missing_indices) > 15:
                            self.enqueue_snackbar(f'Skipped the missing indices!')
                        else:
                            self.enqueue_snackbar(f'Skipped: {missing_indices}')

                    return rows

                self.enqueue_snackbar('ERROR: No valid index found!')
                return []

            self.enqueue_snackbar("ERROR: Correct format is 'row' or 'row_1,row_2'!")
            return []

        def spawn_plot(self, plot_type, data, **kwargs):
            """ Creates and starts a new process for the plot. Does some additional checks """
            # Clean up dead plot processes
            for process in self.plot_processes:
                if not process.is_alive():
                    process.terminate()
                    self.plot_processes.remove(process)

            self.enqueue_snackbar('Spawning plot...')
            data_len = len(data)

            # Create a new independent process based on plot type
            if plot_type == 'plot':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=plot_wrapper, args=data)

            elif plot_type.startswith('scatter'):
                if '2D' in plot_type:
                    if data_len == 2:
                        plot_process = multiprocessing.Process(target=scatter_plot_2d_wrapper, args=data)
                    elif data_len == 3:
                        plot_process = multiprocessing.Process(target=scatter_plot_3d_to_2d_wrapper, args=data)
                    elif data_len == 4:
                        plot_process = multiprocessing.Process(target=scatter_plot_4d_to_2d_wrapper, args=data)
                    else:
                        return -1

                elif '3D' in plot_type:
                    if data_len == 3:
                        plot_process = multiprocessing.Process(target=scatter_plot_3d_wrapper, args=data)
                    elif data_len == 4:
                        plot_process = multiprocessing.Process(target=scatter_plot_4d_to_3D_wrapper, args=data)
                    elif data_len == 5:
                        plot_process = multiprocessing.Process(target=scatter_plot_5d_to_3d_wrapper, args=data)
                    else:
                        return -1

                else:
                    self.enqueue_snackbar('DEV ERROR: Unknown plot type.')  # Should never fire
                    return None
            elif plot_type == 'bar':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=bar_wrapper, args=data)
            elif plot_type == 'hist':
                if data_len != 1:
                    return -1
                # TODO Add bins control
                plot_process = multiprocessing.Process(target=histogram_wrapper, args=data)
            elif plot_type == 'pie':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=pie_chart_wrapper, args=data)
            elif plot_type == 'imshow':
                x_labels = kwargs.get('x_labels', None)
                y_labels = kwargs.get('y_labels', None)
                plot_process = multiprocessing.Process(target=heatmap_2d_wrapper, args=(data, x_labels, y_labels))
            elif plot_type == 'contour':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=contour_plot_wrapper, args=data)
            elif plot_type == 'boxplot':
                if data_len != 1:
                    return -1
                plot_process = multiprocessing.Process(target=box_plot_wrapper, args=data)
            elif plot_type == 'violinplot':
                if data_len != 1:
                    return -1
                plot_process = multiprocessing.Process(target=violin_plot_wrapper, args=data)
            elif plot_type == 'step':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=step_plot_wrapper, args=data)
            elif plot_type == 'errorbar':
                if data_len != 3:
                    return -1
                plot_process = multiprocessing.Process(target=error_bar_plot_wrapper, args=data)
            elif plot_type == 'quiver':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=quiver_plot_wrapper, args=data)
            elif plot_type == 'dendrogram':
                if data_len != 2:
                    return -1
                plot_process = multiprocessing.Process(target=dendrogram_wrapper, args=data)
            else:
                self.enqueue_snackbar('DEV ERROR: Unknown plot type.')  # Should never fire
                return None

            # Start & keep track of it for cleanup
            plot_process.start()
            self.plot_processes.append(plot_process)


    DMToolkitApp().run()