Limit the number of neighbors based on the dataframe

Reviewed-on: https://codefirst.iut.uca.fr/git/clement.freville2/miner/pulls/6
Reviewed-by: Clément FRÉVILLE <clement.freville2@etu.uca.fr>

.gitignore

@@ -0,0 +1 @@
+__pycache__

frontend/exploration.py

@@ -0,0 +1,48 @@
+import pandas as pd
+import streamlit as st
+
+st.set_page_config(
+    page_title="Project Miner",
+    layout="wide"
+)
+
+st.title("Home")
+
+### Exploration
+uploaded_file = st.file_uploader("Upload your CSV file", type=["csv"])
+
+if uploaded_file is not None:
+    st.session_state.data = pd.read_csv(uploaded_file)
+    st.session_state.original_data = st.session_state.data
+    st.success("File loaded successfully!")
+
+
+if "data" in st.session_state:
+    data = st.session_state.data
+    st.write(data.head(10))
+    st.write(data.tail(10))
+
+    st.header("Data Preview")
+    
+    st.subheader("First 5 Rows")
+    st.write(data.head())
+    
+    st.subheader("Last 5 Rows")
+    st.write(data.tail())
+    
+    st.header("Data Summary")
+    
+    st.subheader("Basic Information")
+    col1, col2 = st.columns(2)
+    col1.metric("Number of Rows", data.shape[0])
+    col2.metric("Number of Columns", data.shape[1])
+    
+    st.write(f"Column Names: {list(data.columns)}")
+    
+    st.subheader("Missing Values by Column")
+    missing_values = data.isnull().sum()
+    st.write(missing_values)
+    
+    st.subheader("Statistical Summary")
+    st.write(data.describe())
+    

frontend/main.py

@@ -1,64 +0,0 @@
-import pandas as pd
-import streamlit as st
-import matplotlib.pyplot as plt
-import seaborn as sns
-from pandas.api.types import is_numeric_dtype
-
-st.set_page_config(
-    page_title="Project Miner",
-    layout="wide"
-)
-
-### Exploration
-uploaded_file = st.file_uploader("Upload your CSV file", type=["csv"])
-
-if uploaded_file:
-    data = pd.read_csv(uploaded_file)
-    st.success("File loaded successfully!")
-
-    st.header("Data Preview")
-    
-    st.subheader("First 5 Rows")
-    st.write(data.head())
-    
-    st.subheader("Last 5 Rows")
-    st.write(data.tail())
-    
-    st.header("Data Summary")
-    
-    st.subheader("Basic Information")
-    col1, col2 = st.columns(2)
-    col1.metric("Number of Rows", data.shape[0])
-    col2.metric("Number of Columns", data.shape[1])
-    
-    st.write(f"Column Names: {list(data.columns)}")
-    
-    st.subheader("Missing Values by Column")
-    missing_values = data.isnull().sum()
-    st.write(missing_values)
-    
-    st.subheader("Statistical Summary")
-    st.write(data.describe())
-    
-    ### Visualization
-
-    st.header("Data Visualization")
-
-    st.subheader("Histogram")
-    column_to_plot = st.selectbox("Select Column for Histogram", data.columns)
-    if column_to_plot:
-        fig, ax = plt.subplots()
-        ax.hist(data[column_to_plot].dropna(), bins=20, edgecolor='k')
-        ax.set_title(f'Histogram of {column_to_plot}')
-        ax.set_xlabel(column_to_plot)
-        ax.set_ylabel('Frequency')
-        st.pyplot(fig)
-    
-    st.subheader("Boxplot")
-    dataNumeric = data.select_dtypes(include='number')
-    column_to_plot = st.selectbox("Select Column for Boxplot", dataNumeric.columns)
-    if column_to_plot:
-        fig, ax = plt.subplots()
-        sns.boxplot(data=data, x=column_to_plot, ax=ax)
-        ax.set_title(f'Boxplot of {column_to_plot}')
-        st.pyplot(fig)

frontend/normstrategy.py

@@ -0,0 +1,179 @@
+from abc import ABC, abstractmethod
+from pandas import DataFrame, Series
+from pandas.api.types import is_numeric_dtype
+from sklearn.neighbors import KNeighborsClassifier
+from typing import Any, Union
+
+class DataFrameFunction(ABC):
+    """A command that may be applied in-place to a dataframe."""
+
+    @abstractmethod
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        """Apply the current function to the given dataframe, in-place.
+
+        The series is described by its label and dataframe."""
+        return df
+
+
+class MVStrategy(DataFrameFunction):
+    """A way to handle missing values in a dataframe."""
+
+    @staticmethod
+    def list_available(df: DataFrame, label: str, series: Series) -> list['MVStrategy']:
+        """Get all the strategies that can be used."""
+        choices = [DropStrategy(), ModeStrategy()]
+        if is_numeric_dtype(series):
+            choices.extend((MeanStrategy(), MedianStrategy(), LinearRegressionStrategy()))
+            other_columns = df.select_dtypes(include="number").drop(label, axis=1).columns.to_list()
+            if len(other_columns):
+                choices.append(KNNStrategy(other_columns))
+        return choices
+
+
+class ScalingStrategy(DataFrameFunction):
+    """A way to handle missing values in a dataframe."""
+
+    @staticmethod
+    def list_available(df: DataFrame, series: Series) -> list['MVStrategy']:
+        """Get all the strategies that can be used."""
+        choices = [KeepStrategy()]
+        if is_numeric_dtype(series):
+            choices.extend((MinMaxStrategy(), ZScoreStrategy()))
+            if series.sum() != 0:
+                choices.append(UnitLengthStrategy())
+        return choices
+
+
+class DropStrategy(MVStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        df.dropna(subset=label, inplace=True)
+        return df
+
+    def __str__(self) -> str:
+        return "Drop"
+
+
+class PositionStrategy(MVStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        series.fillna(self.get_value(series), inplace=True)
+        return df
+
+    @abstractmethod
+    def get_value(self, series: Series) -> Any:
+        pass
+
+
+class MeanStrategy(PositionStrategy):
+    #@typing.override
+    def get_value(self, series: Series) -> Union[int, float]:
+        return series.mean()
+
+    def __str__(self) -> str:
+        return "Use mean"
+
+
+class MedianStrategy(PositionStrategy):
+    #@typing.override
+    def get_value(self, series: Series) -> Union[int, float]:
+        return series.median()
+
+    def __str__(self) -> str:
+        return "Use median"
+
+
+class ModeStrategy(PositionStrategy):
+    #@typing.override
+    def get_value(self, series: Series) -> Any:
+        return series.mode()[0]
+
+    def __str__(self) -> str:
+        return "Use mode"
+
+
+class LinearRegressionStrategy(MVStrategy):
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        series.interpolate(inplace=True)
+        return df
+
+    def __str__(self) -> str:
+        return "Use linear regression"
+
+
+class KNNStrategy(MVStrategy):
+    def __init__(self, training_features: list[str]):
+        self.available_features = training_features
+        self.training_features = training_features
+        self.n_neighbors = 3
+
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        # Remove any training column that have any missing values
+        usable_data = df.dropna(subset=self.training_features)
+        # Select columns to impute from
+        train_data = usable_data.dropna(subset=label)
+        # Create train dataframe
+        x_train = train_data.drop(label, axis=1)
+        y_train = train_data[label]
+
+        reg = KNeighborsClassifier(self.n_neighbors).fit(x_train, y_train)
+
+        # Create test dataframe
+        test_data = usable_data[usable_data[label].isnull()]
+        if test_data.empty:
+            return df
+        x_test = test_data.drop(label, axis=1)
+        predicted = reg.predict(x_test)
+
+        # Fill with predicated values and patch the original data
+        usable_data[label].fillna(Series(predicted), inplace=True)
+        df.fillna(usable_data, inplace=True)
+        return df
+
+    def count_max(self, df: DataFrame, label: str) -> int:
+        usable_data = df.dropna(subset=self.training_features)
+        return usable_data[label].count()
+
+    def __str__(self) -> str:
+        return "kNN"
+
+
+class KeepStrategy(ScalingStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        return df
+
+    def __str__(self) -> str:
+        return "No-op"
+
+
+class MinMaxStrategy(ScalingStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        minimum = series.min()
+        maximum = series.max()
+        df[label] = (series - minimum) / (maximum - minimum)
+        return df
+
+    def __str__(self) -> str:
+        return "Min-max"
+
+
+class ZScoreStrategy(ScalingStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        df[label] = (series - series.mean()) / series.std()
+        return df
+
+    def __str__(self) -> str:
+        return "Z-Score"
+
+
+class UnitLengthStrategy(ScalingStrategy):
+    #@typing.override
+    def apply(self, df: DataFrame, label: str, series: Series) -> DataFrame:
+        df[label] = series / series.sum()
+        return df
+
+    def __str__(self) -> str:
+        return "Unit length"

frontend/pages/clustering_dbscan.py

@@ -0,0 +1,35 @@
+import streamlit as st
+import matplotlib.pyplot as plt
+from sklearn.cluster import DBSCAN
+
+st.header("Clustering: dbscan")
+
+
+if "data" in st.session_state:
+    data = st.session_state.data
+
+    with st.form("my_form"):
+        data_name = st.multiselect("Data Name", data.select_dtypes(include="number").columns, max_selections=3)
+        eps = st.slider("eps", min_value=0.0, max_value=1.0, value=0.5, step=0.01)
+        min_samples = st.number_input("min_samples", step=1, min_value=1, value=5)
+        st.form_submit_button("launch")
+
+    if len(data_name) >= 2 and len(data_name) <=3:
+        x = data[data_name].to_numpy()
+
+        dbscan = DBSCAN(eps=eps, min_samples=min_samples)
+        y_dbscan = dbscan.fit_predict(x)
+
+        fig = plt.figure()
+        if len(data_name) == 2:
+            ax = fig.add_subplot(projection='rectilinear')
+            plt.scatter(x[:, 0], x[:, 1], c=y_dbscan, s=50, cmap="viridis")
+        else:
+            ax = fig.add_subplot(projection='3d')
+            ax.scatter(x[:, 0], x[:, 1],x[:, 2], c=y_dbscan, s=50, cmap="viridis")
+        st.pyplot(fig)
+
+    
+
+else:
+    st.error("file not loaded")

frontend/pages/clustering_kmeans.py

@@ -0,0 +1,44 @@
+import streamlit as st
+from sklearn.cluster import KMeans
+import matplotlib.pyplot as plt
+
+st.header("Clustering: kmeans")
+
+
+if "data" in st.session_state:
+    data = st.session_state.data
+
+    with st.form("my_form"):
+        row1 = st.columns([1,1,1])
+        n_clusters = row1[0].selectbox("Number of clusters", range(1,data.shape[0]))
+        data_name = row1[1].multiselect("Data Name",data.select_dtypes(include="number").columns, max_selections=3)
+        n_init = row1[2].number_input("n_init",step=1,min_value=1)
+
+        row2 = st.columns([1,1])
+        max_iter = row1[0].number_input("max_iter",step=1,min_value=1)
+
+
+        st.form_submit_button("launch")
+
+    if len(data_name) >= 2 and len(data_name) <=3:
+        x = data[data_name].to_numpy()
+
+        kmeans = KMeans(n_clusters=n_clusters, init="random", n_init=n_init, max_iter=max_iter, random_state=111)
+        y_kmeans = kmeans.fit_predict(x)
+
+        fig = plt.figure()
+        if len(data_name) == 2:
+            ax = fig.add_subplot(projection='rectilinear')
+            plt.scatter(x[:, 0], x[:, 1], c=y_kmeans, s=50, cmap="viridis")
+            centers = kmeans.cluster_centers_
+            plt.scatter(centers[:, 0], centers[:, 1], c="black", s=200, marker="X")
+        else:
+            ax = fig.add_subplot(projection='3d')
+
+            ax.scatter(x[:, 0], x[:, 1],x[:, 2], c=y_kmeans, s=50, cmap="viridis")
+            centers = kmeans.cluster_centers_
+            ax.scatter(centers[:, 0], centers[:, 1],centers[:, 2], c="black", s=200, marker="X")
+        st.pyplot(fig)
+
+else:
+    st.error("file not loaded")

frontend/pages/normalization.py

@@ -0,0 +1,35 @@
+import streamlit as st
+from normstrategy import MVStrategy, ScalingStrategy, KNNStrategy
+
+if "data" in st.session_state:
+    data = st.session_state.original_data
+    st.session_state.original_data = data.copy()
+
+    for column, series in data.items():
+        col1, col2 = st.columns(2)
+        missing_count = series.isna().sum()
+        choices = MVStrategy.list_available(data, column, series)
+        option = col1.selectbox(
+            f"Missing values of {column} ({missing_count})",
+            choices,
+            index=1,
+            key=f"mv-{column}",
+        )
+        if isinstance(option, KNNStrategy):
+            option.training_features = st.multiselect("Training columns", option.training_features, default=option.available_features, key=f"cols-{column}")
+            option.n_neighbors = st.number_input("Number of neighbors", min_value=1, max_value=option.count_max(data, column), value=option.n_neighbors, key=f"neighbors-{column}")
+        # Always re-get the series to avoid reusing an invalidated series pointer
+        data = option.apply(data, column, data[column])
+
+        choices = ScalingStrategy.list_available(data, series)
+        option = col2.selectbox(
+            "Scaling",
+            choices,
+            key=f"scaling-{column}",
+        )
+        data = option.apply(data, column, data[column])
+
+    st.write(data)
+    st.session_state.data = data
+else:
+    st.error("file not loaded")

frontend/pages/visualization.py

@@ -0,0 +1,30 @@
+import streamlit as st
+import matplotlib.pyplot as plt
+import seaborn as sns
+
+st.header("Data Visualization")
+
+
+if "data" in st.session_state:
+    data = st.session_state.data
+
+    st.subheader("Histogram")
+    column_to_plot = st.selectbox("Select Column for Histogram", data.columns)
+    if column_to_plot:
+        fig, ax = plt.subplots()
+        ax.hist(data[column_to_plot].dropna(), bins=20, edgecolor='k')
+        ax.set_title(f"Histogram of {column_to_plot}")
+        ax.set_xlabel(column_to_plot)
+        ax.set_ylabel("Frequency")
+        st.pyplot(fig)
+    
+    st.subheader("Boxplot")
+    dataNumeric = data.select_dtypes(include="number")
+    column_to_plot = st.selectbox("Select Column for Boxplot", dataNumeric.columns)
+    if column_to_plot:
+        fig, ax = plt.subplots()
+        sns.boxplot(data=data, x=column_to_plot, ax=ax)
+        ax.set_title(f"Boxplot of {column_to_plot}")
+        st.pyplot(fig)
+else:
+    st.error("file not loaded")