3 Commits

Author SHA1 Message Date
6644d60fa2 Remove unused imports 2024-06-07 10:24:02 +02:00
bastien ollier
c190656165 fix file not load 2024-06-07 10:22:47 +02:00
bastien ollier
fd3d6e3b01 debut merge 2024-06-05 09:51:48 +02:00
6 changed files with 0 additions and 295 deletions

1
.gitignore vendored
View File

@@ -1 +0,0 @@
__pycache__

View File

@@ -13,7 +13,6 @@ 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!")

View File

@@ -1,179 +0,0 @@
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"

View File

@@ -1,35 +0,0 @@
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")

View File

@@ -1,44 +0,0 @@
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")

View File

@@ -1,35 +0,0 @@
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")