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