initial commit

README.md

@@ -1,11 +1,11 @@
 ---
-title: Interactive Decision Boundary
-emoji: 😻
-colorFrom: indigo
-colorTo: green
+title: ELVIS Decision Boundary
+emoji: 📊
+colorFrom: green
+colorTo: red
 sdk: gradio
 sdk_version: 5.44.1
-app_file: app.py
+app_file: decision_boundary.py
 pinned: false
 ---
 

decision_boundary.py

@@ -0,0 +1,452 @@
+import gradio as gr
+import matplotlib.pyplot as plt
+import inspect
+import numpy as np
+import pandas as pd
+import io
+from PIL import Image
+import sklearn
+from sklearn.linear_model import LogisticRegression
+from sklearn.svm import LinearSVC
+from sklearn.base import ClassifierMixin
+from sklearn.datasets import load_iris
+from sklearn.decomposition import PCA
+from sklearn.metrics import classification_report
+
+from collections import deque
+
+from util import *
+
+import logging
+# Configure the logger once at the start of your program
+logging.basicConfig(
+    level=logging.INFO,  # set minimum level to capture (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+    format="%(asctime)s [%(levelname)s] %(message)s",  # log format
+)
+logger = logging.getLogger("ELVIS")
+
+# TODO:
+# - support for session: load a previous session and continue from there
+
+def label2color(labels):
+    color_deque = deque(["red", "green", "blue", "yellow", "orange", "purple", "pink", "brown", "gray", "black"])
+    
+    colors = []
+
+    for label in labels:
+        if label.lower() in color_deque:
+            colors.append(label.lower())
+            color_deque.remove(label.lower())
+        else:
+            colors.append(color_deque.popleft())
+
+    return colors
+
+def toydata():
+    points = [['Red', 0.12375, 0.8516666666666667], 
+              ['Red', 0.19, 0.8916666666666666], 
+              ['Red', 0.27375, 0.9233333333333333], 
+              ['Blue', 0.50625, 0.785], 
+              ['Blue', 0.38375, 0.6733333333333333], 
+              ['Blue', 0.28875, 0.595]]
+    df = pd.DataFrame(points, columns=['label', 'F1', 'F2'])
+    return df
+
+class CoordinateProjection:
+    '''
+    TODO: allow user to specify different coordinates
+    '''
+
+    def __init__(self, n_components=None, dims=[0, 1]):
+        self.dims = dims
+
+    def fit_transform(self, X):
+        print(X)
+        self.mean = X.mean(axis=0)
+        return X[:, self.dims]
+
+    def transform(self, X):
+        return X[:, self.dims]
+
+    def inverse_transform(self, Z):
+        X = np.ones((len(Z), 1)) * self.mean
+        X[:, self.dims] = Z
+        return X
+
+class InteractiveDecisionBoundary:
+    def __init__(self, width, height):
+        # initialized in draw_plot
+        #self.canvas_width = -1
+        #self.canvas_height = -1
+
+        self.canvas_width = width
+        self.canvas_height = height
+
+        self.classifiers = get_sklearn_classifiers() 
+        self.dataloaders = get_sklearn_dataloaders()
+
+        self.embedders = {
+                'CoordinateProjection': CoordinateProjection,
+                'GaussianRandomProjection': sklearn.random_projection.GaussianRandomProjection,
+                'SparseRandomProjection': sklearn.random_projection.SparseRandomProjection,
+                }
+        module = getattr(sklearn, 'decomposition')
+        for cls_name, cls in inspect.getmembers(module, inspect.isclass):
+            self.embedders[cls_name] = cls
+
+        self.Embedder = CoordinateProjection
+        #self.Embedder = PCA
+
+        # default classifier model
+        #self.Model = LogisticRegression
+        self.Model = LinearSVC
+        self.model_args = {}
+
+        # todo: support arbitrary number of classes and user-defined class labels 
+        #self.dataset = toydata()
+        #iris = load_iris(as_frame=True)
+        #self.dataset = pd.concat([iris.data, iris.target], axis=1)
+        #self.dataset = self.dataset.rename(columns={'target': 'label'})
+        self.dataset = pd.DataFrame(columns=['target', 'F1', 'F2'])
+        self.dataset_type = 'Draw2D'
+        self.custom_selected = True
+
+        self.data_image = None
+        self.boundary_image = None
+
+        self.css ="""
+#my-button {
+    height: 30px;
+    font-size: 16px;
+}
+
+#rowheight {
+    height: 90px;
+}
+
+.report-table {
+    border: 0 !important;
+}
+.report-table tr, .report-table th, .report-table td, .report-table tbody, .report-table thead {
+    border: 0 !important;
+    padding: 6px 12px;
+    text-align: center;
+}"""
+    def plot(self, decision_boundary=False):
+        '''
+        Plot data and decision boundary with matplotlib and return as PIL image.
+        '''
+
+        logger.info("Initializing figure")
+        fig = plt.figure(figsize=(self.canvas_width/100., self.canvas_height/100.0), dpi=100)
+        # set entire figure to be the canvas to allow simple conversion of mouse
+        # position to coordinates in the figure
+        ax = fig.add_axes([0., 0., 1., 1.]) # 
+        ax.margins(x=0, y=0) # no padding in both directions
+
+        if self.dataset_type == 'Draw2D':
+            # draw canvas boundary
+            #ax.scatter([0, 0, 1, 1], [0, 1, 0, 1], color='brown')
+            ax.plot([0, 0, 1, 1, 0], [0, 1, 1, 0, 0], color='brown')
+        ax.axis('off')
+
+        # TODO: allow showing x and y axes with ticks and labels
+        if (self.dataset is not None and len(self.dataset) > 0):
+            try: 
+                X = self.dataset.loc[:, self.dataset.columns != 'target'].values
+                y = self.dataset.target.values
+                logger.info("Data:\n" + str(X))
+                logger.info("Target:\n" + str(y)) 
+
+                # compute embedding
+                embedder = self.Embedder(n_components=2)
+                self.embedder = embedder
+                logger.info("Embedder = " + str(self.embedder))
+
+                Z = embedder.fit_transform(X)
+                logger.info("Projected data:\n" + str(Z))
+
+                #ax.set_title("Click to add points")
+                labels = np.unique(y)
+                colors = label2color(labels)
+                logger.info("Classes:\n" + str(labels)) 
+                logger.info("Colors:\n" + str(colors)) 
+                l2c = dict(zip(labels, colors))
+
+                # scatter plots for data
+                for l, label in enumerate(labels):
+                    #print('class', label)
+                    #ax.scatter(*zip(*self.dataset[self.dataset.label == label].features), color=label, label=label)
+                    subset = Z[y == label]
+                    ax.scatter(subset[:, 0], subset[:, 1], color=colors[l], label=label)
+                    ax.legend()
+
+                # plot the decision boundary
+                if decision_boundary:
+                    model = self.Model(**parse_param_string(self.model_args))
+                    model.fit(X, y)
+                    self.model = model
+
+                    # plot decision boundary in the projected space
+                    # xx, yy = np.meshgrid(np.linspace(Z[:, 0].min(), 1, 100), np.linspace(0, 1, 100))
+                    if self.dataset_type == 'Draw2D':
+                        xx, yy = np.meshgrid(np.linspace(0, 1, 100),
+                                             np.linspace(0, 1, 100))
+                    else:
+                        xx, yy = np.meshgrid(np.linspace(Z[:, 0].min(), Z[:, 0].max(), 100),
+                                             np.linspace(Z[:, 1].min(), Z[:, 1].max(), 100))
+
+                    grid = np.c_[xx.ravel(), yy.ravel()]
+                    #scores = clf.decision_function(grid)[:, 1].reshape(xx.shape)
+                    #scores = clf.decision_function(grid).reshape(xx.shape)
+                    #ax.contour(xx, yy, scores)#, levels=[0], colors="black", linestyles="--")
+                    print('grid', grid)
+                    print('inverse', embedder.inverse_transform(grid))
+                    preds = model.predict(embedder.inverse_transform(grid)).reshape(xx.shape)
+                    #print(preds.shape, xx.shape, yy.shape)
+                    ax.scatter(xx.ravel(), yy.ravel(), c=[l2c[l] for l in preds.ravel()], s=1, alpha=0.5)
+            except Exception as e:
+                raise gr.Error(f"⚠️   {e}")
+
+        buf = io.BytesIO()
+        ax.figure.savefig(buf, format="png", bbox_inches="tight", pad_inches=0)
+        plt.close(fig)
+        buf.seek(0)
+        img = Image.open(buf)
+        
+        # TODO: add a save function for saving screenshot
+        #img.save('image.png')
+
+        return img
+
+    def add_point(self, evt: gr.SelectData, label):
+        '''
+        Mouse click to add a point.
+        '''
+        if self.custom_selected:
+            if self.dataset_type != 'Draw2D':
+                self.dataset = pd.DataFrame(columns=['target', 'F1', 'F2'])
+                self.dataset_type = 'Draw2D'
+
+            # normalize clicked position to [0, 1]
+            x = evt.index[0] / self.canvas_width
+            y = 1 - evt.index[1] / self.canvas_height # flip y-axis to match matplotlib
+
+            self.dataset.loc[len(self.dataset)] = [label, x, y]
+
+            logger.info(f'clicked ({evt.index[0]}, {evt.index[1]}), mapped to ({x}, {y})')
+        
+        vis = self.plot()
+        data_table = gr.Dataframe(value = self.dataset[:100], visible=True,
+                                  headers=list(self.dataset.columns))
+
+        return vis, data_table
+
+    # train a model and show decision boundary
+    def train(self):
+        image = self.plot(decision_boundary=True)
+        X = self.dataset.loc[:, self.dataset.columns != 'target'].values
+        y = self.dataset.target.values
+        pred = self.model.predict(X)
+        df = pd.DataFrame(classification_report(y, pred, output_dict=True)).T
+        summary = df.to_html(classes="report-table", float_format="%.2f")
+
+        return image, gr.HTML(visible=True), "<b>Classification report</b><br>" + summary
+
+    # clear dataset and replot
+    def clear(self):
+        self.dataset = self.dataset[0:0]
+        return self.plot()
+
+    # save dataset 
+    def save(self):
+        # TODO: allow user-specified filename
+        self.dataset.to_csv('dataset.csv', index=False)
+
+    def update_model(self, classifier_name):
+        self.Model = self.classifiers[classifier_name]
+        self.args_textbox.value = ""
+        logger.info(f'Updated model to {self.model}')
+
+        return ""
+
+    def update_args(self, model_args):
+        self.model_args = model_args
+        print('updated model_args:', self.model_args)
+
+    def update_embedder(self, embedder):
+        self.Embedder = self.embedders[embedder]
+        print('updated embedder:', self.Embedder)
+        return self.plot()
+
+    def handle_dataset_radio(self, type):
+        if type == 'Draw2D':
+            self.custom_selected = True
+            return gr.File(visible=False), gr.Dropdown(visible=False), gr.Dropdown(visible=False), gr.Textbox(visible=True), gr.Button(visible=True), gr.Button(visible=True)
+        elif type == 'Upload':
+            self.custom_selected = False
+            return gr.File(visible=True), gr.Dropdown(visible=False), gr.Dropdown(visible=True), gr.Textbox(visible=False), gr.Button(visible=False), gr.Button(visible=False)
+        elif type == 'sklearn':
+            self.custom_selected = False
+            return gr.File(visible=False), gr.Dropdown(visible=True), gr.Dropdown(visible=True), gr.Textbox(visible=False), gr.Button(visible=False), gr.Button(visible=False)
+        else:
+            # TODO: better error handling
+            print('Error - unknown dataset type:', type)
+
+    def load_local_data_and_plot(self, filename):
+        if filename is not None:
+            self.dataset = read(filename)
+            self.dataset.target = self.dataset.target.astype(str)
+            self.dataset_type = 'Upload'
+            logger.info(f'Loaded dataset from {filename}') 
+
+        vis = self.plot()
+        #data_html = self.dataset.to_html(classes="report-table", float_format="%.2f")
+        data_table = gr.Dataframe(value = self.dataset[:100], visible=True,
+                                  headers=list(self.dataset.columns))
+
+        return vis, data_table
+
+    def load_sklearn_data_and_plot(self, datasetname):
+        if datasetname is not None:
+            dataset = self.dataloaders[datasetname]()
+            X = dataset.data
+            y = dataset.target
+            if hasattr(dataset, 'feature_names'):
+                feature_names = dataset.feature_names
+            else:
+                feature_names = ['F{%d}' % i for i in range(len(X[0]))]
+
+            if hasattr(dataset, 'target_names'):
+                labels = dataset.target_names
+            else:
+                labels = ['C{%d}' % i for i in range(len(np.unique(y)))]
+
+            self.dataset = pd.DataFrame(X, columns=feature_names)
+            self.dataset['target'] = y.astype(str)
+            self.dataset_type = 'sklearn'
+            logger.info(f'Loaded dataset {datasetname}') 
+
+        vis = self.plot()
+        #data_html = self.dataset.to_html(classes="report-table", float_format="%.2f")
+        data_table = gr.Dataframe(value = self.dataset[:100], visible=True,
+                                  headers=list(self.dataset.columns))
+        return vis, data_table
+ 
+    def launch(self):
+        # build the Gradio interface
+        with gr.Blocks(css=self.css) as demo:
+            # app title
+            gr.Markdown("<div style='text-align:left; font-size:40px; font-weight: bold;'>ELVIS Interactive Decision Boundary Visualizer</div>")
+
+            # GUI elements and layout 
+            with gr.Row():
+                with gr.Column(scale=2):
+                    self.data_image = gr.Image(value=self.plot(), container=True)
+
+                with gr.Column(scale=1): 
+                    with gr.Tab("Dataset"):
+                        dataset_radio = gr.Radio(["Draw2D", "Upload", "sklearn"], 
+                                                value="Draw2D", label="Dataset type", elem_id="rowheight")
+
+                        # upload data
+                        file_chooser = gr.File(label="Choose a file", visible=False, elem_id="rowheight")
+                        self.file_chooser = file_chooser
+
+                        # sklearn data dropdown menu 
+                        sklearn_data_selector = gr.Dropdown(choices=self.dataloaders, 
+                                                            label='Select dataset',
+                                                            value='None',
+                                                            visible=False,
+                                                            allow_custom_value=True)
+                        self.sklearn_data_selector = sklearn_data_selector
+ 
+                        # embedder
+                        embedder_selector = gr.Dropdown(choices=self.embedders, 
+                                                        label='Select embedder', 
+                                                        value='CoordinateProjection',
+                                                        visible=False,
+                                                        allow_custom_value=True)
+
+                        # custom data
+                        label = gr.Radio(["Red", "Green", "Blue"], value="Red", label="Choose point label", visible=True, elem_id="rowheight")
+                        self.label = label
+      
+                        with gr.Row():
+                            btn_clear = gr.Button("Clear", visible=True, elem_id="my-button")
+                            self.btn_clear = btn_clear
+
+                            btn_save = gr.Button("Save", visible=True, elem_id="my-button")
+                            self.btn_save = btn_save
+                        
+                        #data_html = gr.HTML(visible=True)
+                        data_table = gr.Dataframe(visible=False)
+
+                    # classifier selector
+                    with gr.Tab("Classifier"):
+                        # specify model
+                        model_selector = gr.Dropdown(choices=self.classifiers,
+                                               #label='', 
+                                               #value='Select classifier',
+                                               label='Select Classifier', 
+                                               value='LinearSVC',
+                                               allow_custom_value=True)
+                        self.model_selector = model_selector
+                   
+                        # specify arguments
+                        args_textbox = gr.Textbox(label="Classifier arguments")
+                        self.args_textbox = args_textbox
+
+                        model_selector.change(fn=self.update_model, inputs=model_selector, outputs=args_textbox)
+                        
+                        btn_train = gr.Button("Train Model")
+
+                        classification_summary = gr.HTML(visible=False)
+                       
+                    with gr.Tab("Export"):
+                        btn_export_data = gr.Button('Data')
+                        
+                        btn_export_model = gr.Button('Model')
+
+                        btn_export_code = gr.Button('Code')
+
+                    with gr.Tab("Options"):
+                        pass
+
+                    with gr.Tab("Usage"):
+                        pass
+
+
+            # event handlers for GUI elements
+            self.data_image.select(self.add_point, inputs=label,
+                                   outputs=(self.data_image, data_table))
+
+            dataset_radio.change(fn=self.handle_dataset_radio,
+                                 inputs=dataset_radio, 
+                                 outputs=(file_chooser, sklearn_data_selector, embedder_selector, label, btn_clear, btn_save))
+
+            # events for custom dataset
+            btn_clear.click(fn=self.clear, outputs=self.data_image)
+            btn_save.click(fn=self.save)
+
+            # events for local dataset
+            file_chooser.change(fn=self.load_local_data_and_plot,
+                                inputs=file_chooser, 
+                                outputs=(self.data_image, data_table))
+
+            # events for sklearn dataset
+            sklearn_data_selector.change(fn=self.load_sklearn_data_and_plot,
+                                         inputs=sklearn_data_selector,
+                                         outputs=(self.data_image, data_table))
+
+            embedder_selector.change(fn=self.update_embedder,
+                                     inputs=embedder_selector,
+                                     outputs=self.data_image)
+
+            btn_train.click(fn=self.update_args, inputs=args_textbox)
+            btn_train.click(fn=self.train, outputs=(self.data_image, classification_summary, classification_summary))
+
+        demo.launch()
+
+visualizer = InteractiveDecisionBoundary(width=1200, height=900)
+visualizer.launch()

requirements.txt

@@ -0,0 +1,6 @@
+matplotlib
+numpy
+pandas
+scikit-learn
+mpu
+pillow

util.py

@@ -0,0 +1,92 @@
+import inspect
+import sklearn
+from sklearn.linear_model import LogisticRegression
+from sklearn.base import ClassifierMixin
+from sklearn.datasets import *
+import pkgutil
+import importlib
+import warnings
+import ast
+import pandas as pd
+
+def safe_import_module(name):
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            return importlib.import_module(name)
+    except Exception:
+        return None  # or raise/log if desired
+
+def get_sklearn_classifiers():
+    classifiers = {}
+
+    #for modname in dir(sklearn):
+    for _, modname, _ in pkgutil.walk_packages(sklearn.__path__, prefix="sklearn."):
+        if '._' in modname: # exclude hidden modules
+            continue
+
+        if modname.count('.') > 1: # exclude modules more than two levels deep
+            continue
+
+        #print(modname)
+        try:
+            #with warnings.catch_warnings():
+                #warnings.simplefilter("ignore")
+ 
+            module = importlib.import_module(modname)
+            for cls_name, cls in inspect.getmembers(module, inspect.isclass):
+                if '._' not in cls_name and ('ClassifierMixin' not in cls_name):
+                    if issubclass(cls, ClassifierMixin) and cls.__module__.startswith("sklearn"):
+                        classifiers[cls_name] = cls
+                        #classifiers.append(f"{cls.__module__}.{cls_name}")
+        except:
+            continue
+
+    return classifiers
+
+def get_sklearn_dataloaders():
+    dataloaders = {}
+    dataloaders['20newsgroup'] = fetch_20newsgroups
+    dataloaders['20newsgroup_vectorized'] = fetch_20newsgroups_vectorized
+    dataloaders['covtype'] = fetch_covtype
+    dataloaders['kddcup99'] = fetch_kddcup99
+    dataloaders['iris'] = load_iris
+
+    return dataloaders
+
+def parse_param_string(param_str):
+    param_str = param_str.replace("*,", "")  # Remove '*' if present
+    params = {}
+    for item in param_str.split(','):
+        if not item.strip():
+            continue
+        if '=' not in item:
+            continue
+        key, value = item.split('=', 1)
+        key = key.strip()
+        try:
+            value = ast.literal_eval(value.strip())
+        except Exception:
+            value = value.strip()  # fallback: treat as string
+        params[key] = value
+    return params
+
+def read(filename):
+    if filename.endswith(".csv"):
+        return pd.read_csv(filename)
+    elif filename.endswith(".xlsx") or filename.endswith(".xls"):
+        return pd.read_excel(filename)
+    elif filename.endswith(".parquet"):
+        return pd.read_parquet(filename)
+    elif filename.endswith(".feather"):
+        return pd.read_feather(filename)
+    elif filename.endswith(".json"):
+        return pd.read_json(filename)
+    else:
+        raise ValueError("Unsupported file format.")
+
+if __name__ == '__main__':
+    #print(classifier_list)
+    s = "penalty='l2', *, dual=False, tol=0.0001, C=1.0, fit_intercept=True, intercept_scaling=1"
+    parsed = parse_param_string(s)
+    print(parsed)