From c6623368834880b3b96270a3de1506b5244e3c09 Mon Sep 17 00:00:00 2001 From: GrgMdmn Date: Mon, 10 Nov 2025 16:40:12 +0100 Subject: [PATCH] Initial commit of scripts --- data_analysis_functions.py | 2046 ++++++++++++++++++------------------ pandas_df_management.py | 84 +- strings_management.py | 436 ++++---- 3 files changed, 1283 insertions(+), 1283 deletions(-) diff --git a/data_analysis_functions.py b/data_analysis_functions.py index da1f537..af8a399 100644 --- a/data_analysis_functions.py +++ b/data_analysis_functions.py @@ -1,1024 +1,1024 @@ -# Import built-in Python modules -from math import e - -# Import libraries for data manipulation -import pandas as pd -import numpy as np - -# Import libraries for data visualization -import matplotlib.pyplot as plt -import seaborn as sns - -# Import libraries for geospatial data and maps -import folium -from folium.plugins import MarkerCluster -from geopy.distance import geodesic -import branca - -# Import libraries for display (e.g., in Jupyter Notebook) -from IPython.display import display - -# Import libraries for statistics and modeling -from scipy.spatial import cKDTree -from scipy.stats import chi2_contingency -import pingouin as pg -from scipy.stats import pearsonr -from sklearn.preprocessing import StandardScaler -from sklearn.decomposition import PCA - - - - -def plot_comparison_between_original_and_corrected_dataset(df, df_refined): - # Calculate the number of non-missing values for each column - non_missing_original = df.notnull().sum() - non_missing_refined = df_refined.notnull().sum() - - # Prepare data for the bar plot - comparison_data = pd.DataFrame({ - 'Columns': non_missing_original.index, - 'Original Dataset': non_missing_original.values, - 'Corrected Dataset': non_missing_refined.values - }).melt(id_vars='Columns', var_name='Dataset', value_name='Non-Missing Count') - - - # Create the bar plot - plt.figure(figsize=(12, 6)) - ax = sns.barplot( - data=comparison_data, - x='Columns', - y='Non-Missing Count', - hue='Dataset', - palette='pastel' - ) - # Rotate the x-axis labels - plt.xticks(rotation=45, ha='right') - - # Add count annotations to each bar (inside the bars) - for p in ax.patches: - height = p.get_height() # Bar height (value of the bar) - if height > 0: # Only annotate bars with positive height - ax.annotate( - f'{int(height)}', # Display the count as an integer - (p.get_x() + p.get_width() / 2., height / 2), # Position inside the bar, vertically centered - ha='center', va='center', fontsize=10, color='black', rotation=90 - ) - - # Add plot titles and labels - plt.title('Comparison of Non-Missing Values by Column', fontsize=14) - plt.ylabel('Number of Non-Missing Values', fontsize=12) - plt.xlabel('Columns', fontsize=12) - plt.legend(title='Dataset', loc='upper right') - plt.grid(axis='y', linestyle='--', alpha=0.7) - plt.tight_layout() # Adjust layout to prevent overlap - plt.show() - -def plot_differences_between_original_and_corrected_dataset(df, df_refined): - # Calculate the number of non-missing values for each column - non_missing_original = df.notnull().sum() - non_missing_refined = df_refined.notnull().sum() - - # Create a DataFrame for plotting - diff_df = data_diff = (non_missing_refined - non_missing_original).reset_index() - diff_df.columns = ['Columns', 'Difference'] - - # Plot the data using Seaborn - plt.figure(figsize=(10, 6)) - ax = sns.barplot( - data=diff_df, - x=diff_df.columns[0], - y=diff_df.columns[1], - hue = diff_df.columns[0], - legend = False, - palette=["blue" if x > 0 else "red" for x in diff_df['Difference']] # Blue for positive, Red for negative - ) - - # Set the y-axis to be centered at 0 - ax.axhline(0, color='black',linewidth=1) - - # Add plot titles and labels - plt.title('Differences Between Original and Corrected Datasets counts', fontsize=14) - plt.ylabel('Difference (Original - Corrected)', fontsize=12) - plt.xlabel('Columns', fontsize=12) - - # Rotate the x-axis labels - plt.xticks(rotation=45, ha='right') - - # Display the values inside the bars - for p in ax.patches: - height = p.get_height() - ax.text( - p.get_x() + p.get_width() / 2, # Position at the center of the bar - height/2, # Position at the top of the bar - f'{int(height)}', # Annotate with the number - ha='center', va='center', fontsize=10, color='k', rotation=0 - ) - - # Show the plot - plt.tight_layout() # Adjust the layout to avoid overlap - plt.show() - - - -def plot_univariate_variable_analysis( - df: pd.DataFrame, - variable_name: str, - graph_type: str, - variable_title_type: str = None, - rotation: float = 90, - is_splitted: bool = False, -) -> None: - """ - Plots the distribution of a specified variable in a DataFrame based on the graph type. - - Args: - df (pd.DataFrame): The DataFrame containing the data. - variable_name (str): The name of the variable to plot. - graph_type (str): The type of graph to plot (e.g., 'pie', 'heat_map', 'bars', 'hist', 'bars_log'). - variable_title_type (str, optional): An optional title type for the variable. Defaults to None. - - Raises: - ValueError: If the graph_type is not recognized or if latitude/longitude columns are missing. - """ - if variable_title_type is None: - variable_title_type = graph_type - - graph_types = ['pie', 'heat_map', 'bars', 'hist', 'bars_log', 'boxplot'] - graph_type = graph_type.lower() - - graph_type_error = True - for g_type in graph_types: - if g_type in graph_type: - graph_type_error = False - - if graph_type_error: - raise ValueError('Error: wrong graph_type') - - if 'pie' in graph_type: - if '%' in graph_type: - threshold = float(graph_type.split('_')[-1][:-1]) - else: - threshold = 2 - value_counts = df[variable_name].value_counts() # Get counts without normalization - total = value_counts.sum() # Total count for percentage calculation - - # Group classes with less than 2% into "Others" (using raw counts) - others_count = value_counts[value_counts / total * 100 < threshold] - if not others_count.empty: - # others_label = f"Others ({', '.join(others_count.index)})" - others_label = "Others" - value_counts = value_counts[value_counts / total * 100 >= threshold] - value_counts[others_label] = others_count.sum() # Add the "Others" category - - plt.pie( - value_counts, - labels=[f"{label} ({count/total*100:.1f}%)" for label, count in zip(value_counts.index, value_counts.values)], # Percentage in labels - autopct=lambda p: f'{int(p * total / 100)}' # Count on wedges - ) - plt.title(f"{variable_name} Distribution") - plt.show() - - elif 'bars' in graph_type or 'bars_log' in graph_type: - value_counts = df[variable_name].value_counts() - - tail = graph_type.split('_')[-1] - if tail.isdigit(): - threshold = int(tail) - else: - threshold = 26 - - # Group smaller classes into "others" if there are more than 26 - if len(value_counts) > threshold: - top_categories = value_counts[:threshold-1] - others_count = value_counts[threshold-1:].sum() - value_counts = pd.concat([top_categories, pd.Series({f'others ({len(value_counts[threshold-1:])})': others_count})]) - - if is_splitted: - total = df.index.nunique() - else: - total = value_counts.sum() - plt.figure(figsize=(10, 6)) - ax = sns.barplot(x=value_counts.index, y=value_counts.values) - plt.title(f"{variable_name} Distribution") - - # Add percentages inside the bars - for p, percentage in zip(ax.patches, value_counts / total * 100): - rotation = rotation if len(value_counts) > 10 else 0 - ax.annotate( - f"{percentage:.1f}%", - (p.get_x() + p.get_width() * 0.50, p.get_height() * 0.90), - ha='center', va='center', fontsize=10, rotation=90, color='black' - ) - - plt.xlabel(variable_name) # Set the X-axis label - plt.ylabel('Count') # Set the Y-axis label - - # Rotate X-axis labels if there are too many categories - plt.xticks(rotation=rotation if len(value_counts) > 10 else 0) - - # Check if the graph should have a logarithmic scale - if graph_type == 'bars_log': - plt.yscale('log') # Apply a logarithmic scale to the Y-axis - - # Get the current Y-axis ticks - current_ticks = plt.yticks()[0] # Retrieve the existing tick positions - - - if graph_type =='bars_log': - new_ticks = list(current_ticks) + [total] # Add the custom tick - new_ticks = sorted(new_ticks) # Sort ticks to maintain proper tick order - else: # Handle the case for a cartesian (linear) scale - new_ticks = sorted(list(current_ticks) + [total]) # Add the custom tick - - plt.gca().set_yticks(new_ticks) # Update ticks on the Y-axis - # Display the final plot - plt.show() - - - elif graph_type == 'hist': - plt.figure(figsize=(10, 6)) - # sns.histplot(df[variable_name][df[variable_name] <= df[variable].quantile(0.99)], kde=True) - sns.histplot(df[variable_name], kde=True, color='blue', bins=15) - plt.title(f"{variable_name} Distribution") - plt.xlabel(variable_name) - plt.ylabel('Count') - plt.show() - - elif graph_type == "boxplot": - # sns.boxplot(y=df[variable_name][df[variable_name] <= df[variable_name].quantile(0.99)]) - sns.boxplot(y=df[variable_name]) - plt.title(f"{variable_name} Distribution (Boxplot)") - plt.ylabel(variable_name) - plt.show() - - elif graph_type == 'heat_map': - # Check if the DataFrame contains the necessary geographic columns - if 'geo_point_2d_a' in df.columns and 'geo_point_2d_b' in df.columns and variable_name in df.columns: - # Group by the variable of interest (e.g., 'city', 'region') and count the occurrences in each sector - sector_data = df.groupby([variable_name]).agg({ - 'geo_point_2d_a': 'mean', - 'geo_point_2d_b': 'mean', - }).reset_index() - - # Add a "population density" column based on the number of occurrences of the variable in each sector - sector_data['counts'] = df.groupby(variable_name).size().values - - # Calculate the occurrence percentages - total_counts = sector_data['counts'].sum() - sector_data['percentage'] = (sector_data['counts'] / total_counts) * 100 - - # Common zoom level used for city-level maps in Folium (and Leaflet) - zoom_level = 12 - # Create a map centered around the average geographic points - m = folium.Map(location=[sector_data['geo_point_2d_a'].mean(), sector_data['geo_point_2d_b'].mean()], zoom_start=zoom_level) - - # Normalize population densities for the heat map - min_density = sector_data['counts'].min() - max_density = sector_data['counts'].max() - - # Create the colormap with branca - colormap = branca.colormap.LinearColormap( - colors=['yellow', 'orange', 'red'], # Choose a classic heatmap palette (yellow-orange-red) - vmin=min_density, vmax=max_density - ).to_step(10) # Divide the colormap into 10 steps for the legend - - # Extract coordinates of the sectors - coords = sector_data[['geo_point_2d_a', 'geo_point_2d_b']].values - tree = cKDTree(coords) # Create the k-d tree with the coordinates - - # Initialize the minimum distance to a very large number - min_distance = float('inf') - closest_pair = None - - # Find the closest pairs - for i, point in enumerate(coords): - # Find the closest neighbor from point i (excluding the point itself) - distances, indices = tree.query(point, k=2) # k=2 to get the closest neighbor and itself - - # The second closest will be the nearest (the first is the point itself) - closest_distance = distances[1] - closest_index = indices[1] - - # If the found distance is smaller, update the minimum distance and associated points - if closest_distance < min_distance: - min_distance = closest_distance - closest_pair = (point, coords[closest_index]) - - # Convert the minimum Euclidean distance to geodesic distance for more accuracy - min_distance = geodesic(closest_pair[0], closest_pair[1]).meters - - # Limit the circle size to half the smallest distance between two points - max_circle_radius = min_distance / 2 - - # Average Earth radius in meters - EARTH_RADIUS = 6371000 - - # Calculate the size of a pixel in meters (at the given average latitude) - lat_rad = np.radians(np.mean(sector_data['geo_point_2d_b'])) - meters_per_pixel = (2 * np.pi * EARTH_RADIUS) / (256 * (2 ** zoom_level)) * np.cos(lat_rad) - - max_circle_radius_in_pixels = max_circle_radius / meters_per_pixel - - # Add each sector as a colored circle based on population density - for _, row in sector_data.iterrows(): - color = colormap(row['counts']) # Use the colormap to get the color based on density - - # Calculate the size of the marker based on density - # size = np.log(row['population_density'] + 1) * 5 # Logarithmic for a reasonable scale - # size = max(10, min(size, max_circle_radius)) # Limit the size to max_circle_radius - size = max_circle_radius_in_pixels - - # Add a circle to the map - folium.CircleMarker( - location=[row['geo_point_2d_a'], row['geo_point_2d_b']], - radius=size, # Circle size based on density and minimum distance - color=color, - fill=True, - fill_color=color, # Fill with the same color - fill_opacity=0.7, - popup=f"{variable_name}: {row[variable_name]}
Trees: {row['counts']}
Percentage: {row['percentage']:.3}%", - # Allow the size to adjust based on zoom level - zoom_on_click=True - ).add_to(m) - - # Add the colormap legend to the map - colormap.caption = f"Trees per {variable_name}" - colormap.add_to(m) - - # Force the legend to the bottom left corner using CSS style - m.get_root().html.add_child(folium.Element(""" - - """)) - - # Display the map - display(m) - m.save(f'./map_{variable_name}.html') - return # End the function - - else: - raise ValueError("The DataFrame must contain 'geo_point_2d_a', 'geo_point_2d_b', and the sector variable to display a map.") - - - -def plot_bivariate_variable_analysis( - x_series: pd.Series, - y_series: pd.Series, - graph_type: str, - title: str = None, -) -> None: - """ - Plots the distribution of a specified variable using two Pandas Series and a specified graph type. - - Args: - x_series (pd.Series): The first variable to plot (x-axis). - y_series (pd.Series): The second variable to plot (y-axis). - graph_type (str): The type of graph to plot (e.g., 'boxplot', 'scatter'). - variable_title_type (str, optional): An optional title for the graph. Defaults to None. - - Raises: - ValueError: If the graph_type is not recognized. - """ - - # Define the supported graph types - graph_types = ['boxplot', 'scatter'] - graph_type = graph_type.lower() # Convert the graph type to lowercase for case insensitivity - - # Raise an error if the specified graph type is not supported - if graph_type not in graph_types: - raise ValueError( - f"The graph type '{graph_type}' is not supported. " - f"Accepted types: {', '.join(graph_types)}" - ) - - # Plot a boxplot if the specified graph type is "boxplot" - if graph_type == "boxplot": - boxplot_bivariate_analysis(x_series, y_series, title) - - # Plot a scatter plot if the specified graph type is "scatter" - elif graph_type == "scatter": - scatter_plot(x_series, y_series, title) - -def boxplot_bivariate_analysis( - x_series: pd.Series, - y_series: pd.Series, - title: str = None, -) -> None: - """ - Plots a boxplot for the specified x_data and y_data. - - Args: - x_series (pd.Series): The data for the x-axis. - y_series (pd.Series): The data for the y-axis. - """ - # Filter out NaN or infinite values from both series - idx = np.isfinite(x_series) & np.isfinite(y_series) - x_clean = x_series[idx] - y_clean = y_series[idx] - - # Create the boxplot using Seaborn - plt.figure(figsize=(8, 6)) # Set the figure size - sns.boxplot(x=x_clean, y=y_clean) # Generate the boxplot - if title == None: - plt.title(f'Boxplot {x_series.name} vs. {y_series.name}') # Add the title - else : - plt.title(title) - plt.xlabel(x_series.name) # Set x-axis label - plt.ylabel(y_series.name) # Set y-axis label - plt.xticks(rotation=90) # Rotate the x-axis labels for better readability - plt.show() # Display the boxplot - -def scatter_plot( - x_series: pd.Series, - y_series: pd.Series, - title: str = None, -) -> None: - """ - Plots a scatter plot for the specified x_data and y_data and overlays a linear regression line. - - Args: - x_series (pd.Series): The data for the x-axis. - y_series (pd.Series): The data for the y-axis. - """ - # Filter out NaN or infinite values from both series - idx = np.isfinite(x_series) & np.isfinite(y_series) - x_clean = x_series[idx] - y_clean = y_series[idx] - - # Calculate the linear regression line coefficients (slope and intercept) - a, b = np.polyfit(x_clean, y_clean, 1) # Fit a polynomial of degree 1 (linear regression) - - # Create the scatter plot with Seaborn - plt.figure(figsize=(10, 6)) # Set the figure size - sns.scatterplot(x=x_clean, y=y_clean, color='blue', alpha=0.6) # Generate scatterplot - - # Plot the linear regression line - x_fit = np.linspace(min(x_clean), max(x_clean), 100) # Generate values for the x-axis of the line - y_fit = a * x_fit + b # Calculate corresponding y values for the line - plt.plot(x_fit, y_fit, color='red', label=f'y = {format_coefficient(a)}x + {format_coefficient(b)}') # Add the line - - # Customize the plot - if title == None: - plt.title("Scatter Plot with Linear Regression Line") # Add the plot title - else: - plt.title(title) - plt.xlabel(f'{x_series.name}') # Set x-axis label - plt.ylabel(f'{y_series.name}') # Set y-axis label - plt.legend() # Display the legend - plt.grid() # Add a grid for better readability - - # Display the plot - plt.show() - - -### Main bivariate indicators function -def compute_bivariate_stats_indicators(x_series: pd.Series, y_series: pd.Series) -> None: - """ - Detects the types of series (categorical or quantitative) and determines the - appropriate statistical analysis (ANOVA, Pearson Correlation, Chi-Squared Test). - - Args: - x_series (pd.Series): The first series. - y_series (pd.Series): The second series. - - Returns: - None: Results are printed to the console. - """ - # Step 1: Determine the type of each series - if isinstance(x_series.dropna().iloc[0], str): - x_type = 'categorical' - else: - x_type = 'quantitative' - - if isinstance(y_series.dropna().iloc[0], str): - y_type = 'categorical' - else: - y_type = 'quantitative' - - # Step 2: Call the appropriate function based on the types - if x_type == 'quantitative' and y_type == 'quantitative': - # Both series are quantitative -> Pearson Correlation - corr, p_value = perform_pearson_correlation(x_series, y_series) - result = corr, p_value - - elif x_type == 'categorical' and y_type == 'categorical': - # Both series are categorical -> Chi-Squared Test - contingency_table, v_cramer, p, chi2, dof, expected = perform_chi2_test(x_series, y_series) - result = contingency_table, v_cramer, p, chi2, dof, expected - - elif (x_type == 'categorical' and y_type == 'quantitative') or ( - x_type == 'quantitative' and y_type == 'categorical' - ): - # One series is categorical and the other is quantitative -> ANOVA - anova_table, eta_squared, p_value = perform_anova(x_series, y_series) - result = anova_table, eta_squared, p_value - - else: - raise ValueError("Invalid combination of variable types.") - - return result - -### Pearson Correlation Function -def perform_pearson_correlation(x_series: pd.Series, y_series: pd.Series) -> tuple: - """ - Computes the Pearson correlation coefficient for two quantitative variables. - - Args: - x_series (pd.Series): The first quantitative series. - y_series (pd.Series): The second quantitative series. - - Returns: - tuple: A tuple containing: - - the correlation coefficient (corr), - - the p-value (p_value). - """ - # Remove NaN values from both series - valid_idx = x_series.notna() & y_series.notna() - x_clean = x_series[valid_idx] - y_clean = y_series[valid_idx] - - # Compute Pearson correlation - corr, p_value = pearsonr(x_clean, y_clean) - - return corr, p_value - -def perform_anova(x_series: pd.Series, y_series: pd.Series) -> tuple: - """ - Performs an ANOVA (analysis of variance) test using Pingouin - and returns results. - - Args: - x_series (pd.Series): The categorical series. - y_series (pd.Series): The quantitative series. - - Returns: - tuple: A tuple containing: - - a DataFrame of the ANOVA results (anova_table), - - Eta-squared (eta_squared), - - the p-value (p_value). - """ - # Determine if x_series is categorical - if x_series.dtype == 'O' or isinstance(x_series.iloc[0], str): - category_column = 'x_series' - dependent_column = 'y_series' - else: - category_column = 'y_series' - dependent_column = 'x_series' - - # Create a temporary DataFrame for the ANOVA analysis - df_temp = pd.DataFrame({category_column: x_series, dependent_column: y_series}) - - # Exclure missing or empty values - df_temp = df_temp[df_temp[category_column] != ""].dropna() - - # unidirectional ANOVA with Pingouin - anova_table = pg.anova(data=df_temp, dv=dependent_column, between=category_column, detailed=True) - - # Compute Eta-squared - eta_squared = anova_table["np2"].iloc[0] # Pingouin fournit directement l'eta-carre partiel (np2) - - # Extraire la p-value - p_value = anova_table["p-unc"].iloc[0] - - return anova_table, eta_squared, p_value - -### Chi-Squared Test Function -def perform_chi2_test(x_series: pd.Series, y_series: pd.Series, adjusted=True) -> tuple: - """ - Performs Chi-Squared test and computes Cramer's V (optionally adjusted). - - Args: - x_series (pd.Series): First categorical variable. - y_series (pd.Series): Second categorical variable. - adjusted (bool): If True (default), returns adjusted Cramer's V. - False returns standard V. - - Returns: - tuple: contingency_table, v_retour, p, chi2, dof, expected - """ - # Create contingency table - contingency_table = pd.crosstab(x_series, y_series) - - # Compute Chi-Squared stats - chi2, p, dof, expected = chi2_contingency(contingency_table) - - # Prepare parameters - n = contingency_table.sum().sum() - min_dof = min(contingency_table.shape[0]-1, contingency_table.shape[1]-1) - k = min(contingency_table.shape) # min(r, c) - - # Base V (standard) - v_standard = np.sqrt(chi2 / (n * min_dof)) if min_dof > 0 else 0 - - # Adjusted V - if adjusted: - if k <= 1: - v_adj = 0.0 - else: - v_max = np.sqrt(min_dof / (k - 1)) - v_adj = v_standard / v_max if v_max != 0 else 0.0 - v_cramer_adj = v_adj - else: - v_cramer_adj = v_standard - - return contingency_table, v_cramer_adj, p, chi2, dof, expected - - -def correlation_matrix_with_p_values(df: pd.DataFrame, rotation=90): - """ - Computes a correlation matrix with statistical significance annotations. - - - ✔ : p < 0.05 (significant correlation) - - ✖ : p ≥ 0.05 (non-significant correlation) - - Based on the discussion here: https://stackoverflow.com/questions/25571882/pandas-columns-correlation-with-statistical-significance - """ - - # Compute the correlation matrix (rho) - rho = df.corr() - - # Compute the p-value matrix (pval) using pearsonr and adjust for diagonal values - pval = df.corr(method=lambda x, y: pearsonr(x, y)[1]) - np.eye(*rho.shape) - - # Add annotations (✔ for p < 0.05, ✖ otherwise) to each cell - p = pval.map(lambda x: '\n✔' if x < 0.05 else '\n✖') - - # Combine correlation values with significance annotations - annotated_matrix = rho.round(2).astype(str) + p - - # Plot the heatmap with annotations - plt.figure(figsize=(12, 10)) - sns.heatmap(rho, annot=annotated_matrix, fmt="", cmap='coolwarm', cbar=True, square=True) - plt.xticks(rotation=rotation) - plt.title('Correlation Matrix with Significance (✔ = p < 0.05, ✖ = p ≥ 0.05)', fontsize=16) - plt.show() - -def eta_squared_with_p_heatmap(df, dependent_var, rotation=90): - """ - Plots a heatmap of Eta-squared coefficients with p-value annotations. - - Handles both cases: - - dependent_var is quantitative, and other variables are categorical. - - dependent_var is categorical, and other variables are quantitative. - - Explodes list-like categorical columns automatically for proper ANOVA computation. - - Args: - - df : DataFrame containing the dataset. - - dependent_var : Name of the dependent variable. - - rotation : Rotation angle for variable labels on the X-axis. - """ - # Helper function to detect if a column is categorical - def is_categorical(col): - """A column is categorical if it contains only strings or lists of strings.""" - return ( - df[col].apply(lambda x: isinstance(x, (str, list))).all() - ) - - # Helper to explode list-like columns - def explode_if_needed(df, categorical_col): - """Explodes a column if it contains list-like elements.""" - if df[categorical_col].apply(lambda x: isinstance(x, list)).any(): - df = df.explode(categorical_col) - return df - - # Determine if `dependent_var` is categorical or quantitative - if is_categorical(dependent_var): - dependent_type = "categorical" - independent_vars = df.select_dtypes(include=["number"]).columns.tolist() - else: - dependent_type = "quantitative" - independent_vars = df.select_dtypes(include=["category", "object"]).columns.tolist() - - # Ensure the independent variables exclude the dependent variable - independent_vars = [col for col in independent_vars if col != dependent_var] - - # Containers for results - eta_squareds = [] - p_values = [] - - # Iterate over independent variables - for col in independent_vars: - # Copy the dataframe for processing - temp_df = df.copy() - - # If dependent_var is categorical, check and explode if needed - if dependent_type == "categorical": - temp_df = explode_if_needed(temp_df, dependent_var) - - # If the current independent variable is categorical, check and explode if needed - elif is_categorical(col): - temp_df = explode_if_needed(temp_df, col) - # Convert to 'category' type for ANOVA compatibility - if temp_df[col].dtype != "category": - temp_df[col] = temp_df[col].astype("category") - - # Perform ANOVA - try: - if dependent_type == "quantitative": - # Quantitative dependent variable -> Categorical independent - anova = pg.anova(dv=dependent_var, between=col, data=temp_df, detailed=True) - else: - # Categorical dependent variable -> Quantitative independent - anova = pg.anova(dv=col, between=dependent_var, data=temp_df, detailed=True) - - # Extract eta-squared and p-value - eta_squared = anova.loc[0, "np2"] - p_value = anova.loc[0, "p-unc"] - - except Exception as e: - # Skip problematic columns if ANOVA cannot be performed - print(f"Skipping column {col} due to error: {e}") - eta_squared, p_value = None, None - - eta_squareds.append(eta_squared) - p_values.append(p_value) - - # Create DataFrame for the heatmap - eta_sq_df = pd.DataFrame([eta_squareds], columns=independent_vars) - p_value_df = pd.DataFrame([p_values], columns=independent_vars) - - # Add significance annotations - annotations = p_value_df.map(lambda x: "\n✔" if x < 0.05 else "\n✖" if x is not None else "\n-") - annotated_matrix = eta_sq_df.round(2).astype(str).replace("nan", "") + annotations - - # Determine labels based on the nature of `dependent_var` - if dependent_type == "quantitative": - ylabel = "Quantitative Variable" - xlabel = "Categorical Variables" - else: - ylabel = "Categorical Variable" - xlabel = "Quantitative Variables" - - # Plot the heatmap - plt.figure(figsize=(12, 6)) - sns.heatmap(eta_sq_df, annot=annotated_matrix, fmt="", cmap="coolwarm", cbar=True, square=False) - plt.xticks(rotation=rotation, fontsize=10) - plt.yticks([0], [dependent_var], rotation=0, fontsize=12) - plt.title("η² Heatmap with Significance (✔ = p < 0.05, ✖ = p ≥ 0.05, - = Error)", fontsize=16) - plt.xlabel(xlabel) - plt.ylabel(ylabel) - plt.show() - - - -def cramers_v_heatmap(df: pd.DataFrame, dependent_var: str, rotation: int = 90) -> None: - """ - Displays a heatmap of adjusted Cramer's V between a dependent column and other categorical features, - with sequential list explosion handling. - - Args: - df: Input DataFrame containing categorical data - dependent_var: Target column name for association analysis - rotation: X-axis label rotation (degrees) - """ - import pandas as pd - import numpy as np - import seaborn as sns - import matplotlib.pyplot as plt - from scipy.stats import chi2_contingency - - # Duplicate columns check - if df.columns.duplicated().any(): - dup_cols = df.columns[df.columns.duplicated()].tolist() - raise ValueError(f"Duplicate columns detected: {dup_cols}") - - # Data preparation - df_clean = df[[dependent_var] + [c for c in df.columns if c != dependent_var]].copy() - numeric_cols = df_clean.select_dtypes(include=np.number).columns - df_clean[numeric_cols] = df_clean[numeric_cols].astype(str) # Convert numeric columns to categorical - - # Results storage - results = { - 'adj_cramers_v': [], - 'p_value': [], - 'k_value': [], - 'features': [] - } - - # Feature analysis loop - for feature in [c for c in df_clean.columns if c != dependent_var]: - try: - # Sequential explosion with index alignment - target_exploded = df_clean[[dependent_var]].explode(dependent_var).dropna() - feature_exploded = df_clean[[feature]].explode(feature).dropna() - - merged_df = target_exploded.merge( - feature_exploded, - left_index=True, - right_index=True, - how='inner' - ).dropna() - - if merged_df.empty: - print(f"Skipped {feature}: Not enough valid data after explosion") - continue - - # Contingency table creation - contingency_table = pd.crosstab(merged_df[dependent_var], merged_df[feature]) - - # Statistical parameters - n_obs = contingency_table.sum().sum() - n_rows, n_cols = contingency_table.shape - min_degrees = min(n_rows - 1, n_cols - 1) - min_dim = min(n_rows, n_cols) - - # Chi2 test results - chi2_stat, p_val, _, _ = chi2_contingency(contingency_table) - - # Adjusted Cramer's V calculation - if min_degrees == 0 or min_dim <= 1: - adj_v = 0.0 - else: - std_v = np.sqrt(chi2_stat / (n_obs * min_degrees)) - max_v = np.sqrt(min_degrees / (min_dim - 1)) - adj_v = std_v / max_v - - # Store results - results['adj_cramers_v'].append(adj_v) - results['p_value'].append(p_val) - results['k_value'].append(min_dim) - results['features'].append(feature) - - except Exception as e: - print(f"Error processing {feature}: {str(e)}") - continue - - # Heatmap data preparation - v_matrix = pd.DataFrame([results['adj_cramers_v']], columns=results['features']) - k_matrix = pd.DataFrame([results['k_value']], columns=results['features']) - p_matrix = pd.DataFrame([results['p_value']], columns=results['features']) - - # Annotation formatting - annotations = ( - v_matrix.round(2).astype(str) + - "\n(k=" + k_matrix.astype(str) + ")" + - p_matrix.map(lambda p: " ✔" if p < 0.05 else " ✖") - ) - - # Visualization settings - plt.figure(figsize=(12, 6)) - ax = sns.heatmap( - v_matrix, - annot=annotations, - fmt="", - cmap="coolwarm", - cbar_kws={'label': 'Adjusted Cramer\'s V'}, - linewidths=0.5, - annot_kws={"size": 10, "color": "white"} - ) - - plt.title(f"Cramer's V Heatmap Association Strength: {dependent_var} vs Features\n(✔ = p < 0.05, ✖ = p ≥ 0.05)", pad=15) - plt.xticks(rotation=rotation) - plt.yticks([]) - plt.xlabel("Features") - plt.show() - -def pca_analysis( - df: pd.DataFrame, - n_components: int = 2, - normalize: bool = True, - rotation: int = 90, - figsize: tuple = (14, 8) -) -> None: - """ - Performs PCA analysis with variance explanation and annotated component loadings. - - Args: - df: Quantitative variables DataFrame - n_components: Number of principal components to keep - normalize: Whether to standardize variables (default True) - rotation: X-axis label rotation - figsize: Output figure dimensions - """ - # Check for duplicates - if df.columns.duplicated().any(): - dup_cols = df.columns[df.columns.duplicated()].tolist() - raise ValueError(f"Duplicate columns detected: {dup_cols}") - - df_clean = df.dropna() - - # Standardization - if normalize: - scaler = StandardScaler() - X = scaler.fit_transform(df_clean) - else: - X = df_clean.values - - # PCA computation - pca = PCA(n_components=n_components) - principal_components = pca.fit_transform(X) - - # Build components dataframe - loadings = pca.components_.T * np.sqrt(pca.explained_variance_) - components_df = pd.DataFrame( - loadings, - index=df_clean.columns, - columns=[f"PC{i+1}" for i in range(n_components)] - ) - - # Create visualization grid - fig = plt.figure(figsize=figsize) - gs = fig.add_gridspec(2, 1, height_ratios=[1, 2.5]) - - # Variance explanation plot - ax1 = fig.add_subplot(gs[0]) - explained_var = pca.explained_variance_ratio_ - cum_var = np.cumsum(explained_var) - ax1.plot(range(1, n_components+1), cum_var, 'o-', color='#2e5e87') - ax1.set_ylabel('Cumulative Explained Variance', fontsize=10) - ax1.set_ylim(0, 1.1) - ax1.set_xticks(range(1, n_components+1)) - - # Annotate variance percentage - for i, (ev, cv) in enumerate(zip(explained_var, cum_var)): - ax1.text(i+1, cv + 0.03, - f"{cv*100:.1f}%\n(cumul.)\n——\n{ev*100:.1f}%", - ha='center', - fontsize=8, - color='#174a6e') - - # Loadings heatmap - ax2 = fig.add_subplot(gs[1]) - heatmap = sns.heatmap( - components_df.T, - annot=components_df.T.round(2), - fmt="", - cmap='coolwarm', - cbar_kws={'label': 'Loading Value'}, - linewidths=0.5, - ax=ax2 - ) - - # Formatting - heatmap.set_yticklabels( - [f"PC{i+1}\n({ev*100:.1f}%)" for i, ev in enumerate(explained_var)], - rotation=0 - ) - plt.xticks(rotation=rotation) - plt.title(f"PCA Component Loadings (Normalized={normalize})", pad=20, fontsize=14) - plt.tight_layout() - plt.show() - - - -def compute_distances(df, tree_index): - """ - Computes distances between a specific tree and all other trees in the DataFrame. - - Args: - df: DataFrame containing tree data with 'geo_point_2d_a' (longitude) and 'geo_point_2d_b' (latitude) columns. - tree_index: Index of the specific tree in the DataFrame. - - Returns: - A pandas Series containing distances to all other trees from the specified tree. - Returns None if the tree_index is out of range or if necessary columns are missing. - """ - - # Safety checks - if not all(col in df.columns for col in ['geo_point_2d_a', 'geo_point_2d_b']): - print("Error: 'geo_point_2d_a' or 'geo_point_2d_b' columns not found in DataFrame.") - return None - - if tree_index not in df.index: - print(f"Error: tree_index {tree_index} is out of range.") - return None - - # Get coordinates of the chosen tree - tree_coords = (df.loc[tree_index, 'geo_point_2d_a'], df.loc[tree_index, 'geo_point_2d_b']) - - # Calculate distances to all other trees - distances = [] - for index, row in df.iterrows(): - other_tree_coords = (row['geo_point_2d_a'], row['geo_point_2d_b']) # Assuming latitude, longitude order - distance = geodesic(tree_coords, other_tree_coords).meters - distances.append(distance) - - return pd.Series(distances, index=df.index) # Use original index - - -def format_coefficient(value, limit=5): - threshold = 0.01 - digits_after_coma = 2 - while abs(value) < threshold: - threshold /= 10 - digits_after_coma += 1 - if digits_after_coma > limit: - digits_after_coma = 0 - # return f'{round(value,digits_after_coma):.{digits_after_coma}f}' - return f'{round(value,digits_after_coma)}' - - -# to drop categories inside of my categorical variable each time the count < 10 -def drop_infrequent_categories(df, column, threshold=10): - """Drops categories from a categorical variable if their count is below a threshold.""" - value_counts = df[column].value_counts() - categories_to_keep = value_counts[value_counts >= threshold].index - df[column] = df[column].astype('string').apply(lambda x: x if x in categories_to_keep else 'Other') +# Import built-in Python modules +from math import e + +# Import libraries for data manipulation +import pandas as pd +import numpy as np + +# Import libraries for data visualization +import matplotlib.pyplot as plt +import seaborn as sns + +# Import libraries for geospatial data and maps +import folium +from folium.plugins import MarkerCluster +from geopy.distance import geodesic +import branca + +# Import libraries for display (e.g., in Jupyter Notebook) +from IPython.display import display + +# Import libraries for statistics and modeling +from scipy.spatial import cKDTree +from scipy.stats import chi2_contingency +import pingouin as pg +from scipy.stats import pearsonr +from sklearn.preprocessing import StandardScaler +from sklearn.decomposition import PCA + + + + +def plot_comparison_between_original_and_corrected_dataset(df, df_refined): + # Calculate the number of non-missing values for each column + non_missing_original = df.notnull().sum() + non_missing_refined = df_refined.notnull().sum() + + # Prepare data for the bar plot + comparison_data = pd.DataFrame({ + 'Columns': non_missing_original.index, + 'Original Dataset': non_missing_original.values, + 'Corrected Dataset': non_missing_refined.values + }).melt(id_vars='Columns', var_name='Dataset', value_name='Non-Missing Count') + + + # Create the bar plot + plt.figure(figsize=(12, 6)) + ax = sns.barplot( + data=comparison_data, + x='Columns', + y='Non-Missing Count', + hue='Dataset', + palette='pastel' + ) + # Rotate the x-axis labels + plt.xticks(rotation=45, ha='right') + + # Add count annotations to each bar (inside the bars) + for p in ax.patches: + height = p.get_height() # Bar height (value of the bar) + if height > 0: # Only annotate bars with positive height + ax.annotate( + f'{int(height)}', # Display the count as an integer + (p.get_x() + p.get_width() / 2., height / 2), # Position inside the bar, vertically centered + ha='center', va='center', fontsize=10, color='black', rotation=90 + ) + + # Add plot titles and labels + plt.title('Comparison of Non-Missing Values by Column', fontsize=14) + plt.ylabel('Number of Non-Missing Values', fontsize=12) + plt.xlabel('Columns', fontsize=12) + plt.legend(title='Dataset', loc='upper right') + plt.grid(axis='y', linestyle='--', alpha=0.7) + plt.tight_layout() # Adjust layout to prevent overlap + plt.show() + +def plot_differences_between_original_and_corrected_dataset(df, df_refined): + # Calculate the number of non-missing values for each column + non_missing_original = df.notnull().sum() + non_missing_refined = df_refined.notnull().sum() + + # Create a DataFrame for plotting + diff_df = data_diff = (non_missing_refined - non_missing_original).reset_index() + diff_df.columns = ['Columns', 'Difference'] + + # Plot the data using Seaborn + plt.figure(figsize=(10, 6)) + ax = sns.barplot( + data=diff_df, + x=diff_df.columns[0], + y=diff_df.columns[1], + hue = diff_df.columns[0], + legend = False, + palette=["blue" if x > 0 else "red" for x in diff_df['Difference']] # Blue for positive, Red for negative + ) + + # Set the y-axis to be centered at 0 + ax.axhline(0, color='black',linewidth=1) + + # Add plot titles and labels + plt.title('Differences Between Original and Corrected Datasets counts', fontsize=14) + plt.ylabel('Difference (Original - Corrected)', fontsize=12) + plt.xlabel('Columns', fontsize=12) + + # Rotate the x-axis labels + plt.xticks(rotation=45, ha='right') + + # Display the values inside the bars + for p in ax.patches: + height = p.get_height() + ax.text( + p.get_x() + p.get_width() / 2, # Position at the center of the bar + height/2, # Position at the top of the bar + f'{int(height)}', # Annotate with the number + ha='center', va='center', fontsize=10, color='k', rotation=0 + ) + + # Show the plot + plt.tight_layout() # Adjust the layout to avoid overlap + plt.show() + + + +def plot_univariate_variable_analysis( + df: pd.DataFrame, + variable_name: str, + graph_type: str, + variable_title_type: str = None, + rotation: float = 90, + is_splitted: bool = False, +) -> None: + """ + Plots the distribution of a specified variable in a DataFrame based on the graph type. + + Args: + df (pd.DataFrame): The DataFrame containing the data. + variable_name (str): The name of the variable to plot. + graph_type (str): The type of graph to plot (e.g., 'pie', 'heat_map', 'bars', 'hist', 'bars_log'). + variable_title_type (str, optional): An optional title type for the variable. Defaults to None. + + Raises: + ValueError: If the graph_type is not recognized or if latitude/longitude columns are missing. + """ + if variable_title_type is None: + variable_title_type = graph_type + + graph_types = ['pie', 'heat_map', 'bars', 'hist', 'bars_log', 'boxplot'] + graph_type = graph_type.lower() + + graph_type_error = True + for g_type in graph_types: + if g_type in graph_type: + graph_type_error = False + + if graph_type_error: + raise ValueError('Error: wrong graph_type') + + if 'pie' in graph_type: + if '%' in graph_type: + threshold = float(graph_type.split('_')[-1][:-1]) + else: + threshold = 2 + value_counts = df[variable_name].value_counts() # Get counts without normalization + total = value_counts.sum() # Total count for percentage calculation + + # Group classes with less than 2% into "Others" (using raw counts) + others_count = value_counts[value_counts / total * 100 < threshold] + if not others_count.empty: + # others_label = f"Others ({', '.join(others_count.index)})" + others_label = "Others" + value_counts = value_counts[value_counts / total * 100 >= threshold] + value_counts[others_label] = others_count.sum() # Add the "Others" category + + plt.pie( + value_counts, + labels=[f"{label} ({count/total*100:.1f}%)" for label, count in zip(value_counts.index, value_counts.values)], # Percentage in labels + autopct=lambda p: f'{int(p * total / 100)}' # Count on wedges + ) + plt.title(f"{variable_name} Distribution") + plt.show() + + elif 'bars' in graph_type or 'bars_log' in graph_type: + value_counts = df[variable_name].value_counts() + + tail = graph_type.split('_')[-1] + if tail.isdigit(): + threshold = int(tail) + else: + threshold = 26 + + # Group smaller classes into "others" if there are more than 26 + if len(value_counts) > threshold: + top_categories = value_counts[:threshold-1] + others_count = value_counts[threshold-1:].sum() + value_counts = pd.concat([top_categories, pd.Series({f'others ({len(value_counts[threshold-1:])})': others_count})]) + + if is_splitted: + total = df.index.nunique() + else: + total = value_counts.sum() + plt.figure(figsize=(10, 6)) + ax = sns.barplot(x=value_counts.index, y=value_counts.values) + plt.title(f"{variable_name} Distribution") + + # Add percentages inside the bars + for p, percentage in zip(ax.patches, value_counts / total * 100): + rotation = rotation if len(value_counts) > 10 else 0 + ax.annotate( + f"{percentage:.1f}%", + (p.get_x() + p.get_width() * 0.50, p.get_height() * 0.90), + ha='center', va='center', fontsize=10, rotation=90, color='black' + ) + + plt.xlabel(variable_name) # Set the X-axis label + plt.ylabel('Count') # Set the Y-axis label + + # Rotate X-axis labels if there are too many categories + plt.xticks(rotation=rotation if len(value_counts) > 10 else 0) + + # Check if the graph should have a logarithmic scale + if graph_type == 'bars_log': + plt.yscale('log') # Apply a logarithmic scale to the Y-axis + + # Get the current Y-axis ticks + current_ticks = plt.yticks()[0] # Retrieve the existing tick positions + + + if graph_type =='bars_log': + new_ticks = list(current_ticks) + [total] # Add the custom tick + new_ticks = sorted(new_ticks) # Sort ticks to maintain proper tick order + else: # Handle the case for a cartesian (linear) scale + new_ticks = sorted(list(current_ticks) + [total]) # Add the custom tick + + plt.gca().set_yticks(new_ticks) # Update ticks on the Y-axis + # Display the final plot + plt.show() + + + elif graph_type == 'hist': + plt.figure(figsize=(10, 6)) + # sns.histplot(df[variable_name][df[variable_name] <= df[variable].quantile(0.99)], kde=True) + sns.histplot(df[variable_name], kde=True, color='blue', bins=15) + plt.title(f"{variable_name} Distribution") + plt.xlabel(variable_name) + plt.ylabel('Count') + plt.show() + + elif graph_type == "boxplot": + # sns.boxplot(y=df[variable_name][df[variable_name] <= df[variable_name].quantile(0.99)]) + sns.boxplot(y=df[variable_name]) + plt.title(f"{variable_name} Distribution (Boxplot)") + plt.ylabel(variable_name) + plt.show() + + elif graph_type == 'heat_map': + # Check if the DataFrame contains the necessary geographic columns + if 'geo_point_2d_a' in df.columns and 'geo_point_2d_b' in df.columns and variable_name in df.columns: + # Group by the variable of interest (e.g., 'city', 'region') and count the occurrences in each sector + sector_data = df.groupby([variable_name]).agg({ + 'geo_point_2d_a': 'mean', + 'geo_point_2d_b': 'mean', + }).reset_index() + + # Add a "population density" column based on the number of occurrences of the variable in each sector + sector_data['counts'] = df.groupby(variable_name).size().values + + # Calculate the occurrence percentages + total_counts = sector_data['counts'].sum() + sector_data['percentage'] = (sector_data['counts'] / total_counts) * 100 + + # Common zoom level used for city-level maps in Folium (and Leaflet) + zoom_level = 12 + # Create a map centered around the average geographic points + m = folium.Map(location=[sector_data['geo_point_2d_a'].mean(), sector_data['geo_point_2d_b'].mean()], zoom_start=zoom_level) + + # Normalize population densities for the heat map + min_density = sector_data['counts'].min() + max_density = sector_data['counts'].max() + + # Create the colormap with branca + colormap = branca.colormap.LinearColormap( + colors=['yellow', 'orange', 'red'], # Choose a classic heatmap palette (yellow-orange-red) + vmin=min_density, vmax=max_density + ).to_step(10) # Divide the colormap into 10 steps for the legend + + # Extract coordinates of the sectors + coords = sector_data[['geo_point_2d_a', 'geo_point_2d_b']].values + tree = cKDTree(coords) # Create the k-d tree with the coordinates + + # Initialize the minimum distance to a very large number + min_distance = float('inf') + closest_pair = None + + # Find the closest pairs + for i, point in enumerate(coords): + # Find the closest neighbor from point i (excluding the point itself) + distances, indices = tree.query(point, k=2) # k=2 to get the closest neighbor and itself + + # The second closest will be the nearest (the first is the point itself) + closest_distance = distances[1] + closest_index = indices[1] + + # If the found distance is smaller, update the minimum distance and associated points + if closest_distance < min_distance: + min_distance = closest_distance + closest_pair = (point, coords[closest_index]) + + # Convert the minimum Euclidean distance to geodesic distance for more accuracy + min_distance = geodesic(closest_pair[0], closest_pair[1]).meters + + # Limit the circle size to half the smallest distance between two points + max_circle_radius = min_distance / 2 + + # Average Earth radius in meters + EARTH_RADIUS = 6371000 + + # Calculate the size of a pixel in meters (at the given average latitude) + lat_rad = np.radians(np.mean(sector_data['geo_point_2d_b'])) + meters_per_pixel = (2 * np.pi * EARTH_RADIUS) / (256 * (2 ** zoom_level)) * np.cos(lat_rad) + + max_circle_radius_in_pixels = max_circle_radius / meters_per_pixel + + # Add each sector as a colored circle based on population density + for _, row in sector_data.iterrows(): + color = colormap(row['counts']) # Use the colormap to get the color based on density + + # Calculate the size of the marker based on density + # size = np.log(row['population_density'] + 1) * 5 # Logarithmic for a reasonable scale + # size = max(10, min(size, max_circle_radius)) # Limit the size to max_circle_radius + size = max_circle_radius_in_pixels + + # Add a circle to the map + folium.CircleMarker( + location=[row['geo_point_2d_a'], row['geo_point_2d_b']], + radius=size, # Circle size based on density and minimum distance + color=color, + fill=True, + fill_color=color, # Fill with the same color + fill_opacity=0.7, + popup=f"{variable_name}: {row[variable_name]}
Trees: {row['counts']}
Percentage: {row['percentage']:.3}%", + # Allow the size to adjust based on zoom level + zoom_on_click=True + ).add_to(m) + + # Add the colormap legend to the map + colormap.caption = f"Trees per {variable_name}" + colormap.add_to(m) + + # Force the legend to the bottom left corner using CSS style + m.get_root().html.add_child(folium.Element(""" + + """)) + + # Display the map + display(m) + m.save(f'./map_{variable_name}.html') + return # End the function + + else: + raise ValueError("The DataFrame must contain 'geo_point_2d_a', 'geo_point_2d_b', and the sector variable to display a map.") + + + +def plot_bivariate_variable_analysis( + x_series: pd.Series, + y_series: pd.Series, + graph_type: str, + title: str = None, +) -> None: + """ + Plots the distribution of a specified variable using two Pandas Series and a specified graph type. + + Args: + x_series (pd.Series): The first variable to plot (x-axis). + y_series (pd.Series): The second variable to plot (y-axis). + graph_type (str): The type of graph to plot (e.g., 'boxplot', 'scatter'). + variable_title_type (str, optional): An optional title for the graph. Defaults to None. + + Raises: + ValueError: If the graph_type is not recognized. + """ + + # Define the supported graph types + graph_types = ['boxplot', 'scatter'] + graph_type = graph_type.lower() # Convert the graph type to lowercase for case insensitivity + + # Raise an error if the specified graph type is not supported + if graph_type not in graph_types: + raise ValueError( + f"The graph type '{graph_type}' is not supported. " + f"Accepted types: {', '.join(graph_types)}" + ) + + # Plot a boxplot if the specified graph type is "boxplot" + if graph_type == "boxplot": + boxplot_bivariate_analysis(x_series, y_series, title) + + # Plot a scatter plot if the specified graph type is "scatter" + elif graph_type == "scatter": + scatter_plot(x_series, y_series, title) + +def boxplot_bivariate_analysis( + x_series: pd.Series, + y_series: pd.Series, + title: str = None, +) -> None: + """ + Plots a boxplot for the specified x_data and y_data. + + Args: + x_series (pd.Series): The data for the x-axis. + y_series (pd.Series): The data for the y-axis. + """ + # Filter out NaN or infinite values from both series + idx = np.isfinite(x_series) & np.isfinite(y_series) + x_clean = x_series[idx] + y_clean = y_series[idx] + + # Create the boxplot using Seaborn + plt.figure(figsize=(8, 6)) # Set the figure size + sns.boxplot(x=x_clean, y=y_clean) # Generate the boxplot + if title == None: + plt.title(f'Boxplot {x_series.name} vs. {y_series.name}') # Add the title + else : + plt.title(title) + plt.xlabel(x_series.name) # Set x-axis label + plt.ylabel(y_series.name) # Set y-axis label + plt.xticks(rotation=90) # Rotate the x-axis labels for better readability + plt.show() # Display the boxplot + +def scatter_plot( + x_series: pd.Series, + y_series: pd.Series, + title: str = None, +) -> None: + """ + Plots a scatter plot for the specified x_data and y_data and overlays a linear regression line. + + Args: + x_series (pd.Series): The data for the x-axis. + y_series (pd.Series): The data for the y-axis. + """ + # Filter out NaN or infinite values from both series + idx = np.isfinite(x_series) & np.isfinite(y_series) + x_clean = x_series[idx] + y_clean = y_series[idx] + + # Calculate the linear regression line coefficients (slope and intercept) + a, b = np.polyfit(x_clean, y_clean, 1) # Fit a polynomial of degree 1 (linear regression) + + # Create the scatter plot with Seaborn + plt.figure(figsize=(10, 6)) # Set the figure size + sns.scatterplot(x=x_clean, y=y_clean, color='blue', alpha=0.6) # Generate scatterplot + + # Plot the linear regression line + x_fit = np.linspace(min(x_clean), max(x_clean), 100) # Generate values for the x-axis of the line + y_fit = a * x_fit + b # Calculate corresponding y values for the line + plt.plot(x_fit, y_fit, color='red', label=f'y = {format_coefficient(a)}x + {format_coefficient(b)}') # Add the line + + # Customize the plot + if title == None: + plt.title("Scatter Plot with Linear Regression Line") # Add the plot title + else: + plt.title(title) + plt.xlabel(f'{x_series.name}') # Set x-axis label + plt.ylabel(f'{y_series.name}') # Set y-axis label + plt.legend() # Display the legend + plt.grid() # Add a grid for better readability + + # Display the plot + plt.show() + + +### Main bivariate indicators function +def compute_bivariate_stats_indicators(x_series: pd.Series, y_series: pd.Series) -> None: + """ + Detects the types of series (categorical or quantitative) and determines the + appropriate statistical analysis (ANOVA, Pearson Correlation, Chi-Squared Test). + + Args: + x_series (pd.Series): The first series. + y_series (pd.Series): The second series. + + Returns: + None: Results are printed to the console. + """ + # Step 1: Determine the type of each series + if isinstance(x_series.dropna().iloc[0], str): + x_type = 'categorical' + else: + x_type = 'quantitative' + + if isinstance(y_series.dropna().iloc[0], str): + y_type = 'categorical' + else: + y_type = 'quantitative' + + # Step 2: Call the appropriate function based on the types + if x_type == 'quantitative' and y_type == 'quantitative': + # Both series are quantitative -> Pearson Correlation + corr, p_value = perform_pearson_correlation(x_series, y_series) + result = corr, p_value + + elif x_type == 'categorical' and y_type == 'categorical': + # Both series are categorical -> Chi-Squared Test + contingency_table, v_cramer, p, chi2, dof, expected = perform_chi2_test(x_series, y_series) + result = contingency_table, v_cramer, p, chi2, dof, expected + + elif (x_type == 'categorical' and y_type == 'quantitative') or ( + x_type == 'quantitative' and y_type == 'categorical' + ): + # One series is categorical and the other is quantitative -> ANOVA + anova_table, eta_squared, p_value = perform_anova(x_series, y_series) + result = anova_table, eta_squared, p_value + + else: + raise ValueError("Invalid combination of variable types.") + + return result + +### Pearson Correlation Function +def perform_pearson_correlation(x_series: pd.Series, y_series: pd.Series) -> tuple: + """ + Computes the Pearson correlation coefficient for two quantitative variables. + + Args: + x_series (pd.Series): The first quantitative series. + y_series (pd.Series): The second quantitative series. + + Returns: + tuple: A tuple containing: + - the correlation coefficient (corr), + - the p-value (p_value). + """ + # Remove NaN values from both series + valid_idx = x_series.notna() & y_series.notna() + x_clean = x_series[valid_idx] + y_clean = y_series[valid_idx] + + # Compute Pearson correlation + corr, p_value = pearsonr(x_clean, y_clean) + + return corr, p_value + +def perform_anova(x_series: pd.Series, y_series: pd.Series) -> tuple: + """ + Performs an ANOVA (analysis of variance) test using Pingouin + and returns results. + + Args: + x_series (pd.Series): The categorical series. + y_series (pd.Series): The quantitative series. + + Returns: + tuple: A tuple containing: + - a DataFrame of the ANOVA results (anova_table), + - Eta-squared (eta_squared), + - the p-value (p_value). + """ + # Determine if x_series is categorical + if x_series.dtype == 'O' or isinstance(x_series.iloc[0], str): + category_column = 'x_series' + dependent_column = 'y_series' + else: + category_column = 'y_series' + dependent_column = 'x_series' + + # Create a temporary DataFrame for the ANOVA analysis + df_temp = pd.DataFrame({category_column: x_series, dependent_column: y_series}) + + # Exclure missing or empty values + df_temp = df_temp[df_temp[category_column] != ""].dropna() + + # unidirectional ANOVA with Pingouin + anova_table = pg.anova(data=df_temp, dv=dependent_column, between=category_column, detailed=True) + + # Compute Eta-squared + eta_squared = anova_table["np2"].iloc[0] # Pingouin fournit directement l'eta-carre partiel (np2) + + # Extraire la p-value + p_value = anova_table["p-unc"].iloc[0] + + return anova_table, eta_squared, p_value + +### Chi-Squared Test Function +def perform_chi2_test(x_series: pd.Series, y_series: pd.Series, adjusted=True) -> tuple: + """ + Performs Chi-Squared test and computes Cramer's V (optionally adjusted). + + Args: + x_series (pd.Series): First categorical variable. + y_series (pd.Series): Second categorical variable. + adjusted (bool): If True (default), returns adjusted Cramer's V. + False returns standard V. + + Returns: + tuple: contingency_table, v_retour, p, chi2, dof, expected + """ + # Create contingency table + contingency_table = pd.crosstab(x_series, y_series) + + # Compute Chi-Squared stats + chi2, p, dof, expected = chi2_contingency(contingency_table) + + # Prepare parameters + n = contingency_table.sum().sum() + min_dof = min(contingency_table.shape[0]-1, contingency_table.shape[1]-1) + k = min(contingency_table.shape) # min(r, c) + + # Base V (standard) + v_standard = np.sqrt(chi2 / (n * min_dof)) if min_dof > 0 else 0 + + # Adjusted V + if adjusted: + if k <= 1: + v_adj = 0.0 + else: + v_max = np.sqrt(min_dof / (k - 1)) + v_adj = v_standard / v_max if v_max != 0 else 0.0 + v_cramer_adj = v_adj + else: + v_cramer_adj = v_standard + + return contingency_table, v_cramer_adj, p, chi2, dof, expected + + +def correlation_matrix_with_p_values(df: pd.DataFrame, rotation=90): + """ + Computes a correlation matrix with statistical significance annotations. + + - ✔ : p < 0.05 (significant correlation) + - ✖ : p ≥ 0.05 (non-significant correlation) + + Based on the discussion here: https://stackoverflow.com/questions/25571882/pandas-columns-correlation-with-statistical-significance + """ + + # Compute the correlation matrix (rho) + rho = df.corr() + + # Compute the p-value matrix (pval) using pearsonr and adjust for diagonal values + pval = df.corr(method=lambda x, y: pearsonr(x, y)[1]) - np.eye(*rho.shape) + + # Add annotations (✔ for p < 0.05, ✖ otherwise) to each cell + p = pval.map(lambda x: '\n✔' if x < 0.05 else '\n✖') + + # Combine correlation values with significance annotations + annotated_matrix = rho.round(2).astype(str) + p + + # Plot the heatmap with annotations + plt.figure(figsize=(12, 10)) + sns.heatmap(rho, annot=annotated_matrix, fmt="", cmap='coolwarm', cbar=True, square=True) + plt.xticks(rotation=rotation) + plt.title('Correlation Matrix with Significance (✔ = p < 0.05, ✖ = p ≥ 0.05)', fontsize=16) + plt.show() + +def eta_squared_with_p_heatmap(df, dependent_var, rotation=90): + """ + Plots a heatmap of Eta-squared coefficients with p-value annotations. + + Handles both cases: + - dependent_var is quantitative, and other variables are categorical. + - dependent_var is categorical, and other variables are quantitative. + + Explodes list-like categorical columns automatically for proper ANOVA computation. + + Args: + - df : DataFrame containing the dataset. + - dependent_var : Name of the dependent variable. + - rotation : Rotation angle for variable labels on the X-axis. + """ + # Helper function to detect if a column is categorical + def is_categorical(col): + """A column is categorical if it contains only strings or lists of strings.""" + return ( + df[col].apply(lambda x: isinstance(x, (str, list))).all() + ) + + # Helper to explode list-like columns + def explode_if_needed(df, categorical_col): + """Explodes a column if it contains list-like elements.""" + if df[categorical_col].apply(lambda x: isinstance(x, list)).any(): + df = df.explode(categorical_col) + return df + + # Determine if `dependent_var` is categorical or quantitative + if is_categorical(dependent_var): + dependent_type = "categorical" + independent_vars = df.select_dtypes(include=["number"]).columns.tolist() + else: + dependent_type = "quantitative" + independent_vars = df.select_dtypes(include=["category", "object"]).columns.tolist() + + # Ensure the independent variables exclude the dependent variable + independent_vars = [col for col in independent_vars if col != dependent_var] + + # Containers for results + eta_squareds = [] + p_values = [] + + # Iterate over independent variables + for col in independent_vars: + # Copy the dataframe for processing + temp_df = df.copy() + + # If dependent_var is categorical, check and explode if needed + if dependent_type == "categorical": + temp_df = explode_if_needed(temp_df, dependent_var) + + # If the current independent variable is categorical, check and explode if needed + elif is_categorical(col): + temp_df = explode_if_needed(temp_df, col) + # Convert to 'category' type for ANOVA compatibility + if temp_df[col].dtype != "category": + temp_df[col] = temp_df[col].astype("category") + + # Perform ANOVA + try: + if dependent_type == "quantitative": + # Quantitative dependent variable -> Categorical independent + anova = pg.anova(dv=dependent_var, between=col, data=temp_df, detailed=True) + else: + # Categorical dependent variable -> Quantitative independent + anova = pg.anova(dv=col, between=dependent_var, data=temp_df, detailed=True) + + # Extract eta-squared and p-value + eta_squared = anova.loc[0, "np2"] + p_value = anova.loc[0, "p-unc"] + + except Exception as e: + # Skip problematic columns if ANOVA cannot be performed + print(f"Skipping column {col} due to error: {e}") + eta_squared, p_value = None, None + + eta_squareds.append(eta_squared) + p_values.append(p_value) + + # Create DataFrame for the heatmap + eta_sq_df = pd.DataFrame([eta_squareds], columns=independent_vars) + p_value_df = pd.DataFrame([p_values], columns=independent_vars) + + # Add significance annotations + annotations = p_value_df.map(lambda x: "\n✔" if x < 0.05 else "\n✖" if x is not None else "\n-") + annotated_matrix = eta_sq_df.round(2).astype(str).replace("nan", "") + annotations + + # Determine labels based on the nature of `dependent_var` + if dependent_type == "quantitative": + ylabel = "Quantitative Variable" + xlabel = "Categorical Variables" + else: + ylabel = "Categorical Variable" + xlabel = "Quantitative Variables" + + # Plot the heatmap + plt.figure(figsize=(12, 6)) + sns.heatmap(eta_sq_df, annot=annotated_matrix, fmt="", cmap="coolwarm", cbar=True, square=False) + plt.xticks(rotation=rotation, fontsize=10) + plt.yticks([0], [dependent_var], rotation=0, fontsize=12) + plt.title("η² Heatmap with Significance (✔ = p < 0.05, ✖ = p ≥ 0.05, - = Error)", fontsize=16) + plt.xlabel(xlabel) + plt.ylabel(ylabel) + plt.show() + + + +def cramers_v_heatmap(df: pd.DataFrame, dependent_var: str, rotation: int = 90) -> None: + """ + Displays a heatmap of adjusted Cramer's V between a dependent column and other categorical features, + with sequential list explosion handling. + + Args: + df: Input DataFrame containing categorical data + dependent_var: Target column name for association analysis + rotation: X-axis label rotation (degrees) + """ + import pandas as pd + import numpy as np + import seaborn as sns + import matplotlib.pyplot as plt + from scipy.stats import chi2_contingency + + # Duplicate columns check + if df.columns.duplicated().any(): + dup_cols = df.columns[df.columns.duplicated()].tolist() + raise ValueError(f"Duplicate columns detected: {dup_cols}") + + # Data preparation + df_clean = df[[dependent_var] + [c for c in df.columns if c != dependent_var]].copy() + numeric_cols = df_clean.select_dtypes(include=np.number).columns + df_clean[numeric_cols] = df_clean[numeric_cols].astype(str) # Convert numeric columns to categorical + + # Results storage + results = { + 'adj_cramers_v': [], + 'p_value': [], + 'k_value': [], + 'features': [] + } + + # Feature analysis loop + for feature in [c for c in df_clean.columns if c != dependent_var]: + try: + # Sequential explosion with index alignment + target_exploded = df_clean[[dependent_var]].explode(dependent_var).dropna() + feature_exploded = df_clean[[feature]].explode(feature).dropna() + + merged_df = target_exploded.merge( + feature_exploded, + left_index=True, + right_index=True, + how='inner' + ).dropna() + + if merged_df.empty: + print(f"Skipped {feature}: Not enough valid data after explosion") + continue + + # Contingency table creation + contingency_table = pd.crosstab(merged_df[dependent_var], merged_df[feature]) + + # Statistical parameters + n_obs = contingency_table.sum().sum() + n_rows, n_cols = contingency_table.shape + min_degrees = min(n_rows - 1, n_cols - 1) + min_dim = min(n_rows, n_cols) + + # Chi2 test results + chi2_stat, p_val, _, _ = chi2_contingency(contingency_table) + + # Adjusted Cramer's V calculation + if min_degrees == 0 or min_dim <= 1: + adj_v = 0.0 + else: + std_v = np.sqrt(chi2_stat / (n_obs * min_degrees)) + max_v = np.sqrt(min_degrees / (min_dim - 1)) + adj_v = std_v / max_v + + # Store results + results['adj_cramers_v'].append(adj_v) + results['p_value'].append(p_val) + results['k_value'].append(min_dim) + results['features'].append(feature) + + except Exception as e: + print(f"Error processing {feature}: {str(e)}") + continue + + # Heatmap data preparation + v_matrix = pd.DataFrame([results['adj_cramers_v']], columns=results['features']) + k_matrix = pd.DataFrame([results['k_value']], columns=results['features']) + p_matrix = pd.DataFrame([results['p_value']], columns=results['features']) + + # Annotation formatting + annotations = ( + v_matrix.round(2).astype(str) + + "\n(k=" + k_matrix.astype(str) + ")" + + p_matrix.map(lambda p: " ✔" if p < 0.05 else " ✖") + ) + + # Visualization settings + plt.figure(figsize=(12, 6)) + ax = sns.heatmap( + v_matrix, + annot=annotations, + fmt="", + cmap="coolwarm", + cbar_kws={'label': 'Adjusted Cramer\'s V'}, + linewidths=0.5, + annot_kws={"size": 10, "color": "white"} + ) + + plt.title(f"Cramer's V Heatmap Association Strength: {dependent_var} vs Features\n(✔ = p < 0.05, ✖ = p ≥ 0.05)", pad=15) + plt.xticks(rotation=rotation) + plt.yticks([]) + plt.xlabel("Features") + plt.show() + +def pca_analysis( + df: pd.DataFrame, + n_components: int = 2, + normalize: bool = True, + rotation: int = 90, + figsize: tuple = (14, 8) +) -> None: + """ + Performs PCA analysis with variance explanation and annotated component loadings. + + Args: + df: Quantitative variables DataFrame + n_components: Number of principal components to keep + normalize: Whether to standardize variables (default True) + rotation: X-axis label rotation + figsize: Output figure dimensions + """ + # Check for duplicates + if df.columns.duplicated().any(): + dup_cols = df.columns[df.columns.duplicated()].tolist() + raise ValueError(f"Duplicate columns detected: {dup_cols}") + + df_clean = df.dropna() + + # Standardization + if normalize: + scaler = StandardScaler() + X = scaler.fit_transform(df_clean) + else: + X = df_clean.values + + # PCA computation + pca = PCA(n_components=n_components) + principal_components = pca.fit_transform(X) + + # Build components dataframe + loadings = pca.components_.T * np.sqrt(pca.explained_variance_) + components_df = pd.DataFrame( + loadings, + index=df_clean.columns, + columns=[f"PC{i+1}" for i in range(n_components)] + ) + + # Create visualization grid + fig = plt.figure(figsize=figsize) + gs = fig.add_gridspec(2, 1, height_ratios=[1, 2.5]) + + # Variance explanation plot + ax1 = fig.add_subplot(gs[0]) + explained_var = pca.explained_variance_ratio_ + cum_var = np.cumsum(explained_var) + ax1.plot(range(1, n_components+1), cum_var, 'o-', color='#2e5e87') + ax1.set_ylabel('Cumulative Explained Variance', fontsize=10) + ax1.set_ylim(0, 1.1) + ax1.set_xticks(range(1, n_components+1)) + + # Annotate variance percentage + for i, (ev, cv) in enumerate(zip(explained_var, cum_var)): + ax1.text(i+1, cv + 0.03, + f"{cv*100:.1f}%\n(cumul.)\n——\n{ev*100:.1f}%", + ha='center', + fontsize=8, + color='#174a6e') + + # Loadings heatmap + ax2 = fig.add_subplot(gs[1]) + heatmap = sns.heatmap( + components_df.T, + annot=components_df.T.round(2), + fmt="", + cmap='coolwarm', + cbar_kws={'label': 'Loading Value'}, + linewidths=0.5, + ax=ax2 + ) + + # Formatting + heatmap.set_yticklabels( + [f"PC{i+1}\n({ev*100:.1f}%)" for i, ev in enumerate(explained_var)], + rotation=0 + ) + plt.xticks(rotation=rotation) + plt.title(f"PCA Component Loadings (Normalized={normalize})", pad=20, fontsize=14) + plt.tight_layout() + plt.show() + + + +def compute_distances(df, tree_index): + """ + Computes distances between a specific tree and all other trees in the DataFrame. + + Args: + df: DataFrame containing tree data with 'geo_point_2d_a' (longitude) and 'geo_point_2d_b' (latitude) columns. + tree_index: Index of the specific tree in the DataFrame. + + Returns: + A pandas Series containing distances to all other trees from the specified tree. + Returns None if the tree_index is out of range or if necessary columns are missing. + """ + + # Safety checks + if not all(col in df.columns for col in ['geo_point_2d_a', 'geo_point_2d_b']): + print("Error: 'geo_point_2d_a' or 'geo_point_2d_b' columns not found in DataFrame.") + return None + + if tree_index not in df.index: + print(f"Error: tree_index {tree_index} is out of range.") + return None + + # Get coordinates of the chosen tree + tree_coords = (df.loc[tree_index, 'geo_point_2d_a'], df.loc[tree_index, 'geo_point_2d_b']) + + # Calculate distances to all other trees + distances = [] + for index, row in df.iterrows(): + other_tree_coords = (row['geo_point_2d_a'], row['geo_point_2d_b']) # Assuming latitude, longitude order + distance = geodesic(tree_coords, other_tree_coords).meters + distances.append(distance) + + return pd.Series(distances, index=df.index) # Use original index + + +def format_coefficient(value, limit=5): + threshold = 0.01 + digits_after_coma = 2 + while abs(value) < threshold: + threshold /= 10 + digits_after_coma += 1 + if digits_after_coma > limit: + digits_after_coma = 0 + # return f'{round(value,digits_after_coma):.{digits_after_coma}f}' + return f'{round(value,digits_after_coma)}' + + +# to drop categories inside of my categorical variable each time the count < 10 +def drop_infrequent_categories(df, column, threshold=10): + """Drops categories from a categorical variable if their count is below a threshold.""" + value_counts = df[column].value_counts() + categories_to_keep = value_counts[value_counts >= threshold].index + df[column] = df[column].astype('string').apply(lambda x: x if x in categories_to_keep else 'Other') return df \ No newline at end of file diff --git a/pandas_df_management.py b/pandas_df_management.py index c0e3136..494e71b 100644 --- a/pandas_df_management.py +++ b/pandas_df_management.py @@ -1,42 +1,42 @@ -# Find duplicates and only keep the rows with the maximum amount of values for the studied_features -# Duplicates finding will be based on the features_for_duplicates list -def remove_duplicates(original_dataframe, features_for_duplicates, studied_features): - if type(features_for_duplicates)!=list: - features_for_duplicates = [features_for_duplicates] - - dataframe_free_duplicates = original_dataframe.copy() - # Monitor amount of cleaned rows for each feature - removed_values_per_feature = {} - for feature in features_for_duplicates : - - # Create a list to store the indices of the rows to keep - rows_to_keep = [] - - # Iterate over the unique duplicate codes - for unique_value in dataframe_free_duplicates[dataframe_free_duplicates.duplicated(subset=[feature], keep=False)][feature].dropna().unique(): - # Get the subset of rows with the current duplicate code - duplicates_subset = dataframe_free_duplicates[dataframe_free_duplicates[feature] == unique_value] - - # Count the number of non-missing values for each row in the subset. 1 or more can be found - values_counts = duplicates_subset[studied_features].notna().sum(axis=1) - - # Find the index of the row with the maximum number of non-missing values. If several row are reaching the maximum number of non-missing values, - # then the first index will be arbitrarily chosen - max_values_index = values_counts.idxmax() - - # Find the index of the row with the maximum number of non-missing values. If several row are reaching the maximum number of non-missing values, - # then the first index will be arbitrarily chosen - rows_to_keep.append(max_values_index) - - # Get indices of non-duplicate rows - non_duplicate_indices = dataframe_free_duplicates[~dataframe_free_duplicates.duplicated(subset=[feature], keep=False)].index - - # Combine indices of non-duplicate rows and rows_to_keep - all_indices_to_keep = non_duplicate_indices.union(rows_to_keep) - - removed_values_per_feature[feature] = len(dataframe_free_duplicates) - len(all_indices_to_keep) - - # Create the final DataFrame - dataframe_free_duplicates = dataframe_free_duplicates.loc[all_indices_to_keep] - - return dataframe_free_duplicates, removed_values_per_feature +# Find duplicates and only keep the rows with the maximum amount of values for the studied_features +# Duplicates finding will be based on the features_for_duplicates list +def remove_duplicates(original_dataframe, features_for_duplicates, studied_features): + if type(features_for_duplicates)!=list: + features_for_duplicates = [features_for_duplicates] + + dataframe_free_duplicates = original_dataframe.copy() + # Monitor amount of cleaned rows for each feature + removed_values_per_feature = {} + for feature in features_for_duplicates : + + # Create a list to store the indices of the rows to keep + rows_to_keep = [] + + # Iterate over the unique duplicate codes + for unique_value in dataframe_free_duplicates[dataframe_free_duplicates.duplicated(subset=[feature], keep=False)][feature].dropna().unique(): + # Get the subset of rows with the current duplicate code + duplicates_subset = dataframe_free_duplicates[dataframe_free_duplicates[feature] == unique_value] + + # Count the number of non-missing values for each row in the subset. 1 or more can be found + values_counts = duplicates_subset[studied_features].notna().sum(axis=1) + + # Find the index of the row with the maximum number of non-missing values. If several row are reaching the maximum number of non-missing values, + # then the first index will be arbitrarily chosen + max_values_index = values_counts.idxmax() + + # Find the index of the row with the maximum number of non-missing values. If several row are reaching the maximum number of non-missing values, + # then the first index will be arbitrarily chosen + rows_to_keep.append(max_values_index) + + # Get indices of non-duplicate rows + non_duplicate_indices = dataframe_free_duplicates[~dataframe_free_duplicates.duplicated(subset=[feature], keep=False)].index + + # Combine indices of non-duplicate rows and rows_to_keep + all_indices_to_keep = non_duplicate_indices.union(rows_to_keep) + + removed_values_per_feature[feature] = len(dataframe_free_duplicates) - len(all_indices_to_keep) + + # Create the final DataFrame + dataframe_free_duplicates = dataframe_free_duplicates.loc[all_indices_to_keep] + + return dataframe_free_duplicates, removed_values_per_feature diff --git a/strings_management.py b/strings_management.py index c1eb54f..01354c0 100644 --- a/strings_management.py +++ b/strings_management.py @@ -1,219 +1,219 @@ -import nltk -from nltk.corpus import stopwords -import re -import numpy as np -import pandas as pd -from deep_translator import GoogleTranslator -from googletrans import Translator -from concurrent.futures import ThreadPoolExecutor, as_completed - -def select_key_words( - text, - keep_between_separators=True, - regex_pattern=r'[a-zA-Z\u00C0-\u017F]+', # Default regex pattern: keeps words with letters and accents - stopword_languages=['english'], # List of languages for stopwords - separator=",", # Default separator - keep_duplicates = False # Return unique list -): - if pd.isna(text): - return np.nan - else: - # Compile the custom regex pattern - word_pattern = re.compile(regex_pattern, re.UNICODE) - - # Build the stop words set based on the provided languages - stop_words = set() - for language in stopword_languages: - try: - stop_words.update(stopwords.words(language)) - except OSError: - print(f"Warning: Could not load stopwords for language '{language}'. " - f"Ensure the NLTK stopword corpus is downloaded using 'nltk.download('stopwords')'.") - - # Mode 1: Keep elements separated by the custom separator as single keywords - if keep_between_separators: - # Split the text using the provided separator while keeping the segments - segments = re.split(rf'({re.escape(separator)})', text.lower()) # Split using custom separator - segments = [seg.strip() for seg in segments if seg.strip() and seg != separator] # Clean empty segments - - # Apply regex to extract words and clean each segment - filtered_segments = [ - ' '.join(word_pattern.findall(seg)) - for seg in segments - ] - - if keep_duplicates: - # Remove duplicates while preserving order - seen = set() - ordered_segments = [seg for seg in filtered_segments if seg not in seen and not seen.add(seg) and len(seg)>0] - else: - ordered_segments = [seg for seg in filtered_segments if len(seg)>0] - - return ordered_segments - - # Mode 2: Extract individual words (ignoring the separator) - else: - # Extract words using the regex pattern - keywords = word_pattern.findall(text.lower()) - - if not keep_duplicates: - # Remove stop words while preserving the order and remove duplicates - seen = set() - filtered_words = [word for word in keywords if word not in stop_words and word not in seen and not seen.add(word) and len(word)>0] - else: - filtered_words = [word for word in keywords if word not in stop_words and len(word)>0] - - - return filtered_words - -# Function to count occurrences of the first ingredient in a string -def count_words_occurrences(text, keep_duplicates=False): - if pd.isna(text): - result = np.nan - else: - result = {} - for word in select_key_words(text, keep_duplicates=keep_duplicates): - count = text.lower().count(word.lower()) - result[word] = count - return result - - -def translate_word(word, source_language="auto", target_language="en", translator_lib='deep_translator'): - """ - Traduction d'un mot en utilisant soit `deep_translator`, soit `googletrans`. - Args: - word (str): Texte à traduire - source_language (str): Langue source (auto par défaut) - target_language (str): Langue cible (en par défaut) - translator_lib (str): Utilise 'deep_translator' ou 'googletrans'. - - Returns: - dict: 'original' contient le mot d'origine, 'translated' la traduction ou None en cas d'erreur. - """ - try: - translated = None - error_in_text = None - - if translator_lib == 'deep_translator': - # Traduction avec deep_translator - translated = GoogleTranslator(source=source_language, target=target_language).translate(word) - # Vérification si le résultat contient des erreurs textuelles inattendues - # Adapté pour détecter les messages comme "Error 500..." ou similaires - if "Error" in translated or "error" in translated.lower(): - error_in_text = f"Erreur détectée dans la réponse traduite : '{translated}'" - raise ValueError(error_in_text) - - elif translator_lib == 'googletrans': - # Traduction avec googletrans - from googletrans import Translator - translator = Translator() - result = translator.translate(word, src=source_language, dest=target_language) - translated = result.text.lower() - - else: - raise ValueError("Bibliothèque non supportée. Utilisez 'deep_translator' ou 'googletrans'.") - - return {"original": word, "translated": translated} - - except Exception as e: - if e==error_in_text: - raise ValueError(error_in_text) - # Gestion explicite des erreurs avec diagnostic - else: - raise ValueError(f"Other Error: {e}") - # return {"original": word, "translated": None, "error": str(e)} - - - - -def translate_all_words_parallel(words, source_language="auto", target_language="en", max_workers=5, translator_lib="deep_translator"): - """ - Translates a list of words in parallel, returning a dictionary where - the key is the original word, and the value is the translated word. - - Args: - words (list): The list of words to be translated. - target_language (str): Target language for translation (default is English "en"). - max_workers (int): Maximum number of threads for parallel processing. - - Returns: - dict: A dictionary with original words as keys and translations as values. - """ - translations = {} # Dictionary to store the translations - with ThreadPoolExecutor(max_workers=max_workers) as executor: - # Submit parallel tasks for translation - future_to_word = {executor.submit(translate_word, word, source_language, target_language, translator_lib): word for word in words} - - print(future_to_word) - # Process the completed tasks - for future in as_completed(future_to_word): - result = future.result() # Get the result of the translation - if result.get("translated"): # If translation is successful - translations[result["original"]] = result["translated"] - else: - # If there's an error, store None or the error message as the translation - translations[result["original"]] = None - - return translations # Return all translations - - -def translate_all_words_batch(words, source_language="auto", target_language="en"): - """ - Translates a list of words in a batch, returning a dictionary where the key - is the original word, and the value is the translated word. - This function uses the translate_batch function of deep_translator for better performance - on large datasets. - - Args: - words (list): The list of words to be translated. - source_language (str): Source language for translation (default is "auto"). - target_language (str): Target language for translation (default is English "en"). - - Returns: - dict: A dictionary with original words as keys and translations as values. - """ - translations = {} - - try: - # Use GoogleTranslator's translate_batch method from deep_translator - translated_batch = GoogleTranslator(source=source_language, target=target_language).translate_batch(words) - - # Match original words with their translations - for original, translated in zip(words, translated_batch): - translations[original] = translated - - except Exception as e: - # Handle exceptions and assign None to the words that failed to translate - for word in words: - translations[word] = None - print(f"An error occurred during batch translation: {e}") - - return translations - -# # other way to translate -# from googletrans import Translator -# import asyncio - -# async def translate_word_async(word, src_language='auto', target_language='en'): -# """ -# Asynchronously translates a single word or phrase using googletrans. -# """ -# translator = Translator() -# try: -# # Await translation for the specific word -# translation = await translator.translate(word, src=src_language, dest=target_language) -# return translation.text -# except Exception as e: -# print(f"Error while translating '{word}': {e}") -# return word # Return the original word in case of an error - -# async def translate_all_words_async(words, src_language='auto', target_language='en'): -# """ -# Asynchronously translates a list of words as fast as possible. -# """ -# # Create a list of translation tasks -# tasks = [translate_word_async(word, src_language, target_language) for word in words] - -# # Run all tasks concurrently and wait for their completion -# results = await asyncio.gather(*tasks) +import nltk +from nltk.corpus import stopwords +import re +import numpy as np +import pandas as pd +from deep_translator import GoogleTranslator +from googletrans import Translator +from concurrent.futures import ThreadPoolExecutor, as_completed + +def select_key_words( + text, + keep_between_separators=True, + regex_pattern=r'[a-zA-Z\u00C0-\u017F]+', # Default regex pattern: keeps words with letters and accents + stopword_languages=['english'], # List of languages for stopwords + separator=",", # Default separator + keep_duplicates = False # Return unique list +): + if pd.isna(text): + return np.nan + else: + # Compile the custom regex pattern + word_pattern = re.compile(regex_pattern, re.UNICODE) + + # Build the stop words set based on the provided languages + stop_words = set() + for language in stopword_languages: + try: + stop_words.update(stopwords.words(language)) + except OSError: + print(f"Warning: Could not load stopwords for language '{language}'. " + f"Ensure the NLTK stopword corpus is downloaded using 'nltk.download('stopwords')'.") + + # Mode 1: Keep elements separated by the custom separator as single keywords + if keep_between_separators: + # Split the text using the provided separator while keeping the segments + segments = re.split(rf'({re.escape(separator)})', text.lower()) # Split using custom separator + segments = [seg.strip() for seg in segments if seg.strip() and seg != separator] # Clean empty segments + + # Apply regex to extract words and clean each segment + filtered_segments = [ + ' '.join(word_pattern.findall(seg)) + for seg in segments + ] + + if keep_duplicates: + # Remove duplicates while preserving order + seen = set() + ordered_segments = [seg for seg in filtered_segments if seg not in seen and not seen.add(seg) and len(seg)>0] + else: + ordered_segments = [seg for seg in filtered_segments if len(seg)>0] + + return ordered_segments + + # Mode 2: Extract individual words (ignoring the separator) + else: + # Extract words using the regex pattern + keywords = word_pattern.findall(text.lower()) + + if not keep_duplicates: + # Remove stop words while preserving the order and remove duplicates + seen = set() + filtered_words = [word for word in keywords if word not in stop_words and word not in seen and not seen.add(word) and len(word)>0] + else: + filtered_words = [word for word in keywords if word not in stop_words and len(word)>0] + + + return filtered_words + +# Function to count occurrences of the first ingredient in a string +def count_words_occurrences(text, keep_duplicates=False): + if pd.isna(text): + result = np.nan + else: + result = {} + for word in select_key_words(text, keep_duplicates=keep_duplicates): + count = text.lower().count(word.lower()) + result[word] = count + return result + + +def translate_word(word, source_language="auto", target_language="en", translator_lib='deep_translator'): + """ + Traduction d'un mot en utilisant soit `deep_translator`, soit `googletrans`. + Args: + word (str): Texte à traduire + source_language (str): Langue source (auto par défaut) + target_language (str): Langue cible (en par défaut) + translator_lib (str): Utilise 'deep_translator' ou 'googletrans'. + + Returns: + dict: 'original' contient le mot d'origine, 'translated' la traduction ou None en cas d'erreur. + """ + try: + translated = None + error_in_text = None + + if translator_lib == 'deep_translator': + # Traduction avec deep_translator + translated = GoogleTranslator(source=source_language, target=target_language).translate(word) + # Vérification si le résultat contient des erreurs textuelles inattendues + # Adapté pour détecter les messages comme "Error 500..." ou similaires + if "Error" in translated or "error" in translated.lower(): + error_in_text = f"Erreur détectée dans la réponse traduite : '{translated}'" + raise ValueError(error_in_text) + + elif translator_lib == 'googletrans': + # Traduction avec googletrans + from googletrans import Translator + translator = Translator() + result = translator.translate(word, src=source_language, dest=target_language) + translated = result.text.lower() + + else: + raise ValueError("Bibliothèque non supportée. Utilisez 'deep_translator' ou 'googletrans'.") + + return {"original": word, "translated": translated} + + except Exception as e: + if e==error_in_text: + raise ValueError(error_in_text) + # Gestion explicite des erreurs avec diagnostic + else: + raise ValueError(f"Other Error: {e}") + # return {"original": word, "translated": None, "error": str(e)} + + + + +def translate_all_words_parallel(words, source_language="auto", target_language="en", max_workers=5, translator_lib="deep_translator"): + """ + Translates a list of words in parallel, returning a dictionary where + the key is the original word, and the value is the translated word. + + Args: + words (list): The list of words to be translated. + target_language (str): Target language for translation (default is English "en"). + max_workers (int): Maximum number of threads for parallel processing. + + Returns: + dict: A dictionary with original words as keys and translations as values. + """ + translations = {} # Dictionary to store the translations + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # Submit parallel tasks for translation + future_to_word = {executor.submit(translate_word, word, source_language, target_language, translator_lib): word for word in words} + + print(future_to_word) + # Process the completed tasks + for future in as_completed(future_to_word): + result = future.result() # Get the result of the translation + if result.get("translated"): # If translation is successful + translations[result["original"]] = result["translated"] + else: + # If there's an error, store None or the error message as the translation + translations[result["original"]] = None + + return translations # Return all translations + + +def translate_all_words_batch(words, source_language="auto", target_language="en"): + """ + Translates a list of words in a batch, returning a dictionary where the key + is the original word, and the value is the translated word. + This function uses the translate_batch function of deep_translator for better performance + on large datasets. + + Args: + words (list): The list of words to be translated. + source_language (str): Source language for translation (default is "auto"). + target_language (str): Target language for translation (default is English "en"). + + Returns: + dict: A dictionary with original words as keys and translations as values. + """ + translations = {} + + try: + # Use GoogleTranslator's translate_batch method from deep_translator + translated_batch = GoogleTranslator(source=source_language, target=target_language).translate_batch(words) + + # Match original words with their translations + for original, translated in zip(words, translated_batch): + translations[original] = translated + + except Exception as e: + # Handle exceptions and assign None to the words that failed to translate + for word in words: + translations[word] = None + print(f"An error occurred during batch translation: {e}") + + return translations + +# # other way to translate +# from googletrans import Translator +# import asyncio + +# async def translate_word_async(word, src_language='auto', target_language='en'): +# """ +# Asynchronously translates a single word or phrase using googletrans. +# """ +# translator = Translator() +# try: +# # Await translation for the specific word +# translation = await translator.translate(word, src=src_language, dest=target_language) +# return translation.text +# except Exception as e: +# print(f"Error while translating '{word}': {e}") +# return word # Return the original word in case of an error + +# async def translate_all_words_async(words, src_language='auto', target_language='en'): +# """ +# Asynchronously translates a list of words as fast as possible. +# """ +# # Create a list of translation tasks +# tasks = [translate_word_async(word, src_language, target_language) for word in words] + +# # Run all tasks concurrently and wait for their completion +# results = await asyncio.gather(*tasks) # return results \ No newline at end of file