"""
Scatter plot component for the visualization app.
"""
import logging
from typing import Any, Dict, List, Tuple
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import panel as pn
from bokeh.layouts import gridplot
from bokeh.models import BoxZoomTool, ColumnDataSource, DatetimeTickFormatter, HoverTool, Legend
from bokeh.plotting import figure
from scipy import stats
from scipy.stats import mannwhitneyu
from sklearn.metrics import silhouette_score
from sklearn.mixture import GaussianMixture
from itertools import combinations
from LCNE_patchseq_analysis.pipeline_util.s3 import get_public_url_cell_summary
from LCNE_patchseq_analysis.pipeline_util.s3 import load_mesh_from_s3
from LCNE_patchseq_analysis.data_util.mesh import trimesh_to_bokeh_data
from components.color_mapping import ColorMapping
from components.size_mapping import SizeMapping
# Set seaborn style
sns.set_context("paper")
logger = logging.getLogger(__name__)
# Define available color palettes
COLOR_PALETTES = [
"Viridis256",
"Plasma256",
"Magma256",
"Inferno256",
"Cividis256",
"Turbo256",
"Set3",
"Category10",
"Category20",
"Category20b",
"Category20c",
]
class ScatterPlot:
"""Handles scatter plot creation and updates."""
def __init__(self, df_meta: pd.DataFrame, data_holder: Any):
"""Initialize with metadata dataframe."""
self.df_meta = df_meta
self.color_mapping = ColorMapping(df_meta)
self.size_mapping = SizeMapping(df_meta)
self.data_holder = data_holder
# Add cell summary URLs to dataframe
self._add_cell_summary_urls()
self.controls = self.create_plot_controls(width=300)
self._latest_figures = {}
def _add_cell_summary_urls(self):
"""Add cell summary URLs to the dataframe."""
# Create a new column for cell summary URLs
self.df_meta["cell_summary_url"] = None
# Get URLs for each ephys_roi_id
for idx, row in self.df_meta.iterrows():
ephys_roi_id = str(int(row["ephys_roi_id"]))
try:
url = get_public_url_cell_summary(ephys_roi_id, if_check_exists=False)
self.df_meta.at[idx, "cell_summary_url"] = url
except Exception as e:
logger.warning(f"Could not get URL for ephys_roi_id {ephys_roi_id}: {e}")
self.df_meta.at[idx, "cell_summary_url"] = None
def create_plot_controls(self, width: int = 180) -> Dict[str, Any]:
"""Create the control widgets for the scatter plot."""
# Get numeric and categorical columns
numeric_cols = self.df_meta.select_dtypes(include=["number"]).columns.tolist()
categorical_cols = self.df_meta.select_dtypes(include=["object"]).columns.tolist()
available_cols = sorted(numeric_cols + categorical_cols)
# Append [valid N] to the available_cols for display purposes
available_cols = [f"{col} [valid {self.df_meta[col].count()}]" for col in available_cols]
all_cols = ["None"] + available_cols
controls = {
"x_axis_select": pn.widgets.Select(
name="X Axis",
options=all_cols,
value=[col for col in all_cols if "Y" in col][0],
sizing_mode="stretch_width",
),
"y_axis_select": pn.widgets.Select(
name="Y Axis",
options=all_cols,
value=[
col
for col in all_cols
if "ipfx_tau" in col
][0],
sizing_mode="stretch_width",
),
"color_col_select": pn.widgets.Select(
name="Color By",
options=all_cols,
value=[col for col in all_cols if "injection region" in col][0],
sizing_mode="stretch_width",
),
"color_palette_select": pn.widgets.Select(
name="Color Palette",
options=COLOR_PALETTES,
value="Viridis256",
sizing_mode="stretch_width",
),
"size_col_select": pn.widgets.Select(
name="Size By",
options=all_cols,
value="None",
sizing_mode="stretch_width",
),
"size_range_slider": pn.widgets.RangeSlider(
name="Size Range",
start=5,
end=40,
value=(10, 30),
step=1,
sizing_mode="stretch_width",
),
"size_gamma_slider": pn.widgets.FloatSlider(
name="Size Gamma",
start=0.1,
end=5,
value=1,
step=0.1,
sizing_mode="stretch_width",
),
"alpha_slider": pn.widgets.FloatSlider(
name="Alpha",
start=0.1,
end=1,
value=0.7,
step=0.1,
sizing_mode="stretch_width",
),
"width_slider": pn.widgets.IntSlider(
name="Width",
start=400,
end=1200,
value=800,
step=50,
sizing_mode="stretch_width",
),
"height_slider": pn.widgets.IntSlider(
name="Height",
start=400,
end=1200,
value=600,
step=50,
sizing_mode="stretch_width",
),
"bins_slider": pn.widgets.IntSlider(
name="Histogram bins",
start=10,
end=100,
value=50,
step=1,
sizing_mode="stretch_width",
),
"show_gmm": pn.widgets.Checkbox(
name="Show Gaussian Mixture Model",
value=True,
sizing_mode="stretch_width",
),
"show_linear_fit": pn.widgets.Checkbox(
name="Show Linear Fit",
value=True,
sizing_mode="stretch_width",
),
"n_components_x": pn.widgets.IntSlider(
name="Number of components (X)",
start=1,
end=5,
value=2,
step=1,
disabled=False,
sizing_mode="stretch_width",
),
"n_components_y": pn.widgets.IntSlider(
name="Number of components (Y)",
start=1,
end=5,
value=1,
step=1,
disabled=False,
sizing_mode="stretch_width",
),
"hist_height_slider": pn.widgets.IntSlider(
name="Distribution plot height",
start=50,
end=300,
value=150,
step=10,
sizing_mode="stretch_width",
),
"font_size_slider": pn.widgets.IntSlider(
name="Font Size",
start=10,
end=30,
value=15,
sizing_mode="stretch_width",
),
}
# Link the GMM checkbox to enable/disable the component sliders
def toggle_gmm_components(event):
controls["n_components_x"].disabled = not event.new
controls["n_components_y"].disabled = not event.new
controls["show_gmm"].param.watch(toggle_gmm_components, "value")
# Initialize the disabled state based on the initial checkbox value
controls["n_components_x"].disabled = not controls["show_gmm"].value
controls["n_components_y"].disabled = not controls["show_gmm"].value
return controls
def sync_controls_to_url(self):
"""Sync scatter plot controls to URL query parameters."""
location = pn.state.location
mapping = {
"x_axis_select": ("value", "scatter_x"),
"y_axis_select": ("value", "scatter_y"),
"color_col_select": ("value", "scatter_color"),
"color_palette_select": ("value", "scatter_palette"),
"size_col_select": ("value", "scatter_size"),
"size_range_slider": ("value", "scatter_size_range"),
"size_gamma_slider": ("value", "scatter_gamma"),
"alpha_slider": ("value", "scatter_alpha"),
"width_slider": ("value", "scatter_width"),
"height_slider": ("value", "scatter_height"),
"bins_slider": ("value", "scatter_bins"),
"show_gmm": ("value", "scatter_gmm"),
"show_linear_fit": ("value", "scatter_linear_fit"),
"n_components_x": ("value", "scatter_components_x"),
"n_components_y": ("value", "scatter_components_y"),
"hist_height_slider": ("value", "scatter_hist_height"),
"font_size_slider": ("value", "scatter_font_size"),
}
for control_name, (param_name, url_param) in mapping.items():
location.sync(self.controls[control_name], {param_name: url_param})
def create_tooltips(
self, x_col: str, y_col: str, color_col: str, size_col: str
) -> List[Tuple[str, str]]:
"""Create tooltips for the hover tool."""
tooltips = f"""
@Date_str, @{{injection region}}, @{{ephys_roi_id}},
@{{jem-id_cell_specimen}}
X = @{{{x_col}}} [{x_col}]
Y = @{{{y_col}}} [{y_col}]
Color = @{{{color_col}}} [{color_col}]
Size = @{{{size_col}}} [{size_col}]
"""
return tooltips
def create_marginal_histogram(
self,
data: pd.Series,
orientation: str,
width: int,
height: int,
alpha: float,
bins: int,
show_gmm: bool = False,
n_components: int = 1,
) -> figure:
"""Create a histogram for marginal distribution with optional GMM overlay."""
# Remove NaN values and convert to numeric
clean_data = pd.to_numeric(data, errors="coerce").dropna()
# If no valid data, create an empty plot
if clean_data.empty:
p = figure(
height=height,
width=width,
tools="",
toolbar_location=None,
x_range=(0, 1),
y_range=(0, 1),
)
p.text(
x=0.5,
y=0.5,
text=["No valid data"],
text_align="center",
text_baseline="middle",
)
return p
# Calculate histogram data (independent of orientation)
hist, edges = np.histogram(clean_data, bins=bins, density=True)
# Set axis ranges and quad parameters based on orientation
if orientation == "x":
x_range = (edges[0], edges[-1])
y_range = (0, hist.max() * 1.1)
else: # "y" orientation
x_range = (0, hist.max() * 1.1)
y_range = (edges[0], edges[-1])
# Create the figure
p = figure(
height=height,
width=width,
tools="",
toolbar_location=None,
x_range=x_range,
y_range=y_range,
)
# Plot the histogram using vbar/hbar (Bokeh's bar plot) instead of quad
if orientation == "x":
# Use vbar for x-orientation
p.vbar(
x=[(edges[i] + edges[i + 1]) / 2 for i in range(len(edges) - 1)],
top=hist,
width=(edges[1] - edges[0]) * 0.9, # Slightly narrower than bin width
fill_color="gray",
line_color="white",
alpha=0.9,
)
else: # "y" orientation
# Use hbar for y-orientation
p.hbar(
y=[(edges[i] + edges[i + 1]) / 2 for i in range(len(edges) - 1)],
right=hist,
height=(edges[1] - edges[0]) * 0.9, # Slightly narrower than bin width
fill_color="gray",
line_color="white",
alpha=0.9,
)
# Optional: Plot Gaussian Mixture Model overlay
if show_gmm:
gmm = GaussianMixture(n_components=n_components, random_state=42)
gmm.fit(clean_data.values.reshape(-1, 1))
domain = np.linspace(edges[0], edges[-1], 1000)
density = np.exp(gmm.score_samples(domain.reshape(-1, 1)))
# Calculate evaluation metrics
if n_components > 1:
labels = gmm.predict(clean_data.values.reshape(-1, 1))
silhouette = silhouette_score(clean_data.values.reshape(-1, 1), labels)
# Calculate BIC and AIC
bic = gmm.bic(clean_data.values.reshape(-1, 1))
else:
bic = np.nan
silhouette = np.nan
p.line(
*((domain, density) if orientation == "x" else (density, domain)),
line_color="black",
line_width=4,
alpha=0.9,
)
# Plot individual components
for i in range(n_components):
mean = gmm.means_[i][0]
std = np.sqrt(gmm.covariances_[i][0][0])
weight = gmm.weights_[i]
comp_density = (
weight
* np.exp(-0.5 * ((domain - mean) / std) ** 2)
/ (std * np.sqrt(2 * np.pi))
)
p.line(
*((domain, comp_density) if orientation == "x" else (comp_density, domain)),
line_color="black",
line_width=2,
alpha=0.9,
line_dash="dashed",
)
# Add metrics to the plot title
axis_to_show = p.xaxis if orientation == "x" else p.yaxis
axis_to_show.axis_label = f"Silhouette: {silhouette:.3f}, " f"BIC: {bic:.3f}"
# Font size
axis_to_show.axis_label_text_font_size = "10pt"
axis_to_show.major_label_text_font_size = "0pt"
# Hide axes and grid
axis_to_hide = p.xaxis if orientation == "y" else p.yaxis
axis_to_hide.visible = False
p.grid.visible = False
return p
def add_lc_mesh_overlay(self, p: figure, x_col: str, y_col: str) -> None:
"""Add LC mesh overlay to the plot based on axis column names.
Args:
p: Bokeh figure to add the mesh to
x_col: X-axis column name
y_col: Y-axis column name
"""
# Determine mesh direction based on column names
direction = None
if x_col.startswith("X ") and y_col.startswith("Y "):
direction = "sagittal"
elif x_col.startswith("Z "):
direction = "coronal"
if direction is not None:
try:
# Load and add LC mesh overlay
mesh = load_mesh_from_s3()
lc_mesh_bokeh = trimesh_to_bokeh_data(mesh, direction=direction)
mesh_source = ColumnDataSource(lc_mesh_bokeh)
p.patches(
source=mesh_source,
xs="xs",
ys="ys",
fill_alpha=0.3,
line_color=None,
fill_color="lightgray",
level="underlay",
nonselection_fill_alpha=0.3,
nonselection_line_alpha=0,
selection_fill_alpha=0.3,
selection_line_alpha=0,
muted_alpha=0.3,
)
except Exception as e:
logger.warning(f"Could not add LC mesh overlay: {e}")
def update_scatter_plot( # noqa: C901
self,
x_col: str,
y_col: str,
color_col: str,
color_palette: str,
size_col: str,
size_range: tuple,
size_gamma: float,
alpha: float,
width: int,
height: int,
font_size: int = 14,
bins: int = 30,
hist_height_slider: int = 100,
show_gmm: bool = False,
n_components_x: int = 2,
n_components_y: int = 1,
show_linear_fit: bool = True,
df_meta: pd.DataFrame = None,
) -> gridplot:
"""Update the scatter plot with new parameters."""
# Use provided dataframe if supplied, otherwise use instance df_meta
df_to_use = df_meta if df_meta is not None else self.df_meta
# Strip off [valid N] from the column name
x_col = x_col.split(" [valid ")[0]
y_col = y_col.split(" [valid ")[0]
color_col = color_col.split(" [valid ")[0]
size_col = size_col.split(" [valid ")[0]
# Create a new figure for the main scatter plot
p = figure(
x_axis_label=x_col,
y_axis_label=y_col,
tools="pan,wheel_zoom,box_zoom,reset,tap",
height=height,
width=width,
)
# Create ColumnDataSource from the dataframe
source = ColumnDataSource(df_to_use)
# If any column is Date, convert it to datetime
if x_col == "Date":
source.data[x_col] = pd.to_datetime(pd.Series(source.data[x_col]), errors="coerce")
p.xaxis.formatter = DatetimeTickFormatter(
years="%Y",
months="%Y-%m",
days="%Y-%m-%d",
)
# Create temporary color mapping for this specific dataframe
temp_color_mapping = ColorMapping(df_to_use)
# Determine color mapping
color = temp_color_mapping.determine_color_mapping(
color_col, color_palette, p, font_size=font_size
)
# Create temporary size mapping for this specific dataframe
temp_size_mapping = SizeMapping(df_to_use)
# Determine size mapping
size = temp_size_mapping.determine_size_mapping(
size_col, source, min_size=size_range[0], max_size=size_range[1], gamma=size_gamma
)
# Add scatter glyph using the data source
scatter_glyph = p.scatter(x=x_col, y=y_col, source=source, size=size, color=color, alpha=alpha)
# Add linear regression if requested and both columns are numeric
if show_linear_fit and x_col != "Date" and x_col != "None" and y_col != "None":
# Get clean numeric data
# Convert to numeric and drop rows where either x or y is NA
df_clean = df_to_use[[x_col, y_col]].apply(pd.to_numeric, errors="coerce").dropna()
x_data = df_clean[x_col]
y_data = df_clean[y_col]
# Only proceed if we have valid data
if not x_data.empty and not y_data.empty:
# Perform linear regression
slope, intercept, r_value, p_value, std_err = stats.linregress(x_data, y_data)
# Calculate fitted line points over dense grid for smooth CI curve
x_min, x_max = x_data.min(), x_data.max()
x_vals = np.linspace(x_min, x_max, 200)
y_fit = slope * x_vals + intercept
# Add confidence band for the regression line
n_points = len(x_data)
if n_points > 2:
residuals = y_data - (slope * x_data + intercept)
mse = np.sum(residuals ** 2) / (n_points - 2)
x_mean = x_data.mean()
Sxx = np.sum((x_data - x_mean) ** 2)
if Sxx > 0:
t_val = stats.t.ppf(0.975, n_points - 2)
se_fit = np.sqrt(
mse * (1 / n_points + (x_vals - x_mean) ** 2 / Sxx)
)
ci_upper = y_fit + t_val * se_fit
ci_lower = y_fit - t_val * se_fit
band_source = ColumnDataSource(
{
"x": np.concatenate([x_vals, x_vals[::-1]]),
"y": np.concatenate([ci_upper, ci_lower[::-1]]),
}
)
p.patch(
x="x",
y="y",
source=band_source,
fill_color="lightgray",
fill_alpha=0.3,
line_alpha=0,
level="underlay",
)
# Add fitted line
setting = (
{"line_width": 3, "line_dash": "solid"}
if p_value < 0.05
else {"line_width": 2, "line_dash": "dashed"}
)
line = p.line(x_vals, y_fit, line_color="black", **setting)
# Add legend with R² and p-value
legend_items = [
(f"Linear Fit (p = {p_value:.3e}, R² = {r_value**2:.3f})", [line]),
]
legend = Legend(
items=legend_items,
location="top_left",
label_text_font_size=f"{font_size-2}pt"
)
legend.click_policy = "hide"
p.add_layout(legend, 'above')
# Flip the y-axis if y_col is depth
if y_col == "Y (D --> V)":
p.y_range.flipped = True
# Add HoverTool with tooltips
tooltips = self.create_tooltips(x_col, y_col, color_col, size_col)
hovertool = HoverTool(
tooltips=tooltips,
attachment="right", # Fix tooltip to the right of the plot
formatters={"@Date": "datetime"},
renderers=[scatter_glyph], # Only apply hover to scatter points, not mesh patches
)
p.add_tools(hovertool)
# Define callback to update ephys_roi_id on point tap
def update_ephys_roi_id(attr, old, new):
if new:
selected_index = new[0]
ephys_roi_id = str(int(df_to_use.iloc[selected_index]["ephys_roi_id"]))
logger.info(f"Selected ephys_roi_id: {ephys_roi_id}")
# Update the data holder's ephys_roi_id
if hasattr(self, "data_holder"):
self.data_holder.ephys_roi_id_selected = ephys_roi_id
# Attach the callback to the selection changes
source.selected.on_change("indices", update_ephys_roi_id)
# Set the default tool activated on drag to be box zoom
p.toolbar.active_drag = p.select_one(BoxZoomTool)
# Set axis label font sizes
p.xaxis.axis_label_text_font_size = f"{font_size}pt"
p.yaxis.axis_label_text_font_size = f"{font_size}pt"
# Set major tick label font sizes
p.xaxis.major_label_text_font_size = f"{font_size*0.9}pt"
p.yaxis.major_label_text_font_size = f"{font_size*0.9}pt"
# Add LC mesh overlay if appropriate columns are selected
self.add_lc_mesh_overlay(p, x_col, y_col)
# Create marginal histograms
x_hist = None
try:
if x_col != "Date" and x_col != "None": # Skip histogram for Date column
x_hist = self.create_marginal_histogram(
df_to_use[x_col],
"x",
width=width,
height=hist_height_slider,
alpha=alpha,
bins=bins,
show_gmm=show_gmm,
n_components=n_components_x,
)
x_hist.x_range = p.x_range # Link x ranges
except Exception as e:
logger.warning(f"Could not create x histogram: {e}")
x_hist = None
y_hist = None
try:
if y_col != "Date" and y_col != "None": # Skip histogram for Date column
y_hist = self.create_marginal_histogram(
df_to_use[y_col],
"y",
width=hist_height_slider,
height=height,
alpha=alpha,
bins=bins,
show_gmm=show_gmm,
n_components=n_components_y,
)
y_hist.y_range = p.y_range # Link y ranges
except Exception as e:
logger.warning(f"Could not create y histogram: {e}")
y_hist = None
# Count non-NaN values grouped by "injection region"
count_non_nan = df_to_use.groupby("injection region")[[x_col, y_col]].count().T
count_non_nan.insert(0, "Total", count_non_nan.sum(axis=1))
count_non_nan.index = pd.Index(["X", "Y"], name="Valid N")
# Count NaN values (missing data) grouped by "injection region"
count_nan = df_to_use.groupby("injection region")[[x_col, y_col]].apply(lambda x: x.isna().sum()).T
count_nan.insert(0, "Total", count_nan.sum(axis=1))
count_nan.index = pd.Index(["X", "Y"], name="Missing Data")
# If the color column is categorical, generate violin plot and pairwise statistical tests
if color_col in self.df_meta.select_dtypes(include=["object"]).columns:
# --- Create marginalized histograms to compare across colors ---
# marginalized_histograms, _ = self.create_marginalized_histograms(
# df_to_use, y_col, color_col, color_palette, temp_color_mapping, p, font_size
# )
# --- Create violin plot to compare across injection regions ---
violin_plot = self.create_violin_plot(
df_to_use, y_col, color_col, color_palette, temp_color_mapping, p, font_size
)
# Perform pairwise statistical tests
# Drop NA for y_col and color_col for statistical tests
plot_df_for_stats = df_to_use[[y_col, color_col]].dropna() if (y_col != "Date" and y_col != "None" and color_col != "None") else pd.DataFrame()
pvalues_table = self.perform_pairwise_statistical_tests(plot_df_for_stats, y_col, color_col) if not plot_df_for_stats.empty else pn.pane.Markdown("**No statistical tests available**")
else:
violin_plot = pn.pane.Markdown("**Violin plot only available for categorical color columns**")
pvalues_table = pn.pane.Markdown("")
# Create grid layout
layout = pn.Row(
pn.Column(
gridplot(
[[y_hist, p], [None, x_hist]],
toolbar_location="right",
merge_tools=False,
toolbar_options={"logo": None},
),
pn.pane.Markdown(count_non_nan.to_markdown()),
pn.pane.Markdown(count_nan.to_markdown()),
),
pn.Column(
violin_plot,
pvalues_table,
pn.Spacer(height=20),
# marginalized_histograms,
),
)
# Store figures for export
self._latest_figures = {
"scatter_plot": p,
}
if x_hist is not None:
self._latest_figures["x_histogram"] = x_hist
if y_hist is not None:
self._latest_figures["y_histogram"] = y_hist
# Store violin plot if it's a matplotlib figure (not just a markdown pane)
if hasattr(violin_plot, 'object') and hasattr(violin_plot.object, 'savefig'):
self._latest_figures["violin_plot"] = violin_plot
return layout
def create_marginalized_histograms(
self,
df_to_use: pd.DataFrame,
y_col: str,
color_col: str,
color_palette: str,
temp_color_mapping: ColorMapping,
p: figure,
font_size: int
) -> Tuple[Any, Any]:
"""Create marginalized histograms to compare data across color groups.
Args:
df_to_use: DataFrame containing the data
y_col: Column name for y-axis variable
color_col: Column name for color grouping
color_palette: Color palette name
temp_color_mapping: ColorMapping instance for this data
p: Bokeh figure (used for color mapping extraction)
font_size: Font size for labels
Returns:
Tuple of (marginalized_histograms, None) - Creates KDE plots with mean±SEM
"""
# Prepare marginalized histogram using seaborn's histplot (KDE) for y_col by color_col
marginalized_histograms = pn.pane.Markdown("No marginalized histogram available.")
try:
if y_col != "Date" and y_col != "None" and color_col != "None":
fig, ax = plt.subplots(figsize=(4, 3.5), dpi=300)
# Drop NA for y_col and color_col
plot_df = df_to_use[[y_col, color_col]].dropna()
if not plot_df.empty:
# Extract color mapping from the scatter plot
color_palette_dict = None
color_mapping_result = temp_color_mapping.determine_color_mapping(
color_col, color_palette, p, font_size=font_size, if_add_color_bar=False
)
if isinstance(color_mapping_result, dict) and 'transform' in color_mapping_result:
color_mapper = color_mapping_result['transform']
if hasattr(color_mapper, 'factors') and hasattr(color_mapper, 'palette'):
color_palette_dict = dict(zip(color_mapper.factors, color_mapper.palette))
# Count number of samples per group (valid data)
group_counts = plot_df[color_col].value_counts().to_dict()
# Count missing data (NaN) per group from original dataframe
group_nan_counts = {}
for group in group_counts.keys():
# Get all rows for this group from original dataframe
group_mask = df_to_use[color_col] == group
# Count NaN values in y_col for this group
nan_count = df_to_use.loc[group_mask, y_col].isna().sum()
group_nan_counts[group] = nan_count
# Create a mapping from original group name to "group (n = valid, nan = missing)"
group_labels = {
group: f"{group} (n = {count}, missing {group_nan_counts.get(group, 0)})"
for group, count in group_counts.items()
}
# Add a new column for legend labels
plot_df["_legend_label"] = plot_df[color_col].map(group_labels)
sns.kdeplot(
data=plot_df,
x=y_col,
hue="_legend_label",
common_norm=False,
fill=False,
ax=ax,
palette=color_palette_dict if color_palette_dict is None else {
group_labels[group]: color_palette_dict[group] for group in group_labels if group in color_palette_dict
},
)
sns.despine(trim=True)
ax.set_xlabel(y_col)
# Compute mean ± SEM for each group and add as dot + errorbar
y_positions = [] # Track y positions for staggering
for i, (group, data_subset) in enumerate(plot_df.groupby(color_col)):
values = data_subset[y_col].dropna()
if len(values) > 0:
# Ensure values are numeric and convert to float
try:
numeric_values = pd.to_numeric(values, errors='coerce').dropna()
if len(numeric_values) > 0:
mean_val = float(np.mean(numeric_values))
# Use numpy's std with ddof=1 to calculate SEM manually
if len(numeric_values) > 1:
sem_val = float(np.std(numeric_values, ddof=1) / np.sqrt(len(numeric_values)))
else:
sem_val = 0.0
# Get color for this group
group_color = color_palette_dict.get(group, 'black') if color_palette_dict else 'black'
# Stagger y position slightly for each group
# Compute y position as 10% + i * 10% of the current ylim range
ylim = ax.get_ylim()
y_pos = ylim[0] + 0.05 * (ylim[1] - ylim[0]) + i * 0.05 * (ylim[1] - ylim[0])
y_positions.append(y_pos)
# Add dot for mean
ax.plot(mean_val, y_pos, 'o', color=group_color, markersize=4,
markeredgewidth=1)
# Add error bar for SEM
if sem_val > 0.0:
ax.errorbar(mean_val, y_pos, xerr=sem_val, color=group_color,
capsize=3, capthick=1.5, elinewidth=1.5, zorder=9)
except (ValueError, TypeError):
# Skip non-numeric data
continue
# Adjust y-axis limits to accommodate the error bars
current_ylim = ax.get_ylim()
if y_positions:
max_y_pos = max(y_positions)
ax.set_ylim(current_ylim[0], max(current_ylim[1], max_y_pos + 0.02))
# Move legend to top of the plot
y_lim = ax.get_ylim()
sns.move_legend(
ax,
loc="center left",
bbox_to_anchor=(1.01, 0.5),
ncol=1,
frameon=False,
fontsize="small",
title=color_col,
)
# Use Panel's matplotlib pane instead of manual base64 conversion
marginalized_histograms = pn.pane.Matplotlib(fig, dpi=300, tight=True, width=400)
# Close the figure to prevent memory leaks
plt.close(fig)
except Exception as e:
logger.warning(f"Could not create marginalized KDE histogram: {e}")
marginalized_histograms = pn.pane.Markdown("Marginalized histogram error.")
return marginalized_histograms, None
def perform_pairwise_statistical_tests(
self,
plot_df: pd.DataFrame,
y_col: str,
color_col: str
) -> Any:
"""Perform pairwise Mann-Whitney U tests between groups.
Args:
plot_df: DataFrame containing the cleaned data
y_col: Column name for y-axis variable
color_col: Column name for color grouping
Returns:
Panel markdown object with statistical test results
"""
pvalues_table = pn.pane.Markdown("**No statistical tests available**")
try:
# Perform pairwise Mann-Whitney U tests
pairwise_tests = {}
groups = list(plot_df[color_col].unique())
for group1, group2 in combinations(groups, 2):
data1 = pd.to_numeric(plot_df[plot_df[color_col] == group1][y_col], errors='coerce').dropna()
data2 = pd.to_numeric(plot_df[plot_df[color_col] == group2][y_col], errors='coerce').dropna()
if len(data1) > 0 and len(data2) > 0:
try:
statistic, p_value = mannwhitneyu(data1, data2, alternative='two-sided')
pairwise_tests[f"{group1} vs {group2}"] = p_value
except Exception as e:
logger.warning(f"Could not perform Mann-Whitney U test for {group1} vs {group2}: {e}")
pairwise_tests[f"{group1} vs {group2}"] = np.nan
# Create a table of p-values
if pairwise_tests:
pvalues_df = pd.DataFrame(list(pairwise_tests.items()), columns=['Comparison', 'p-value'])
pvalues_df['p-value'] = pvalues_df['p-value'].apply(lambda x: f"{x:.3e}" if not pd.isna(x) else "NaN")
pvalues_table = pn.pane.Markdown(
f"**Mann-Whitney U Test (pairwise comparisons)**\n\n{pvalues_df.to_markdown(index=False)}"
)
else:
pvalues_table = pn.pane.Markdown("**No pairwise comparisons available**")
except Exception as e:
logger.warning(f"Could not perform pairwise statistical tests: {e}")
pvalues_table = pn.pane.Markdown("**Statistical test error**")
return pvalues_table
def create_violin_plot(
self,
df_to_use: pd.DataFrame,
y_col: str,
color_col: str,
color_palette: str,
temp_color_mapping: ColorMapping,
p: figure,
font_size: int
) -> Any:
"""Create violin plot to compare data distributions across injection regions.
Args:
df_to_use: DataFrame containing the data
y_col: Column name for y-axis variable
color_col: Column name for color grouping (injection regions)
color_palette: Color palette name
temp_color_mapping: ColorMapping instance for this data
p: Bokeh figure (used for color mapping extraction)
font_size: Font size for labels
Returns:
Panel matplotlib object with violin plot
"""
violin_plot = pn.pane.Markdown("No violin plot available.")
try:
if y_col != "Date" and y_col != "None" and color_col != "None":
fig, ax = plt.subplots(figsize=(5, 4), dpi=300)
# Drop NA for y_col and color_col
plot_df = df_to_use[[y_col, color_col]].dropna()
if not plot_df.empty:
# Extract color mapping from the scatter plot
color_palette_dict = None
color_mapping_result = temp_color_mapping.determine_color_mapping(
color_col, color_palette, p, font_size=font_size, if_add_color_bar=False
)
if isinstance(color_mapping_result, dict) and 'transform' in color_mapping_result:
color_mapper = color_mapping_result['transform']
if hasattr(color_mapper, 'factors') and hasattr(color_mapper, 'palette'):
color_palette_dict = dict(zip(color_mapper.factors, color_mapper.palette))
# Count number of samples per group (valid data)
group_counts = plot_df[color_col].value_counts().to_dict()
# Count missing data (NaN) per group from original dataframe
group_nan_counts = {}
for group in group_counts.keys():
# Get all rows for this group from original dataframe
group_mask = df_to_use[color_col] == group
# Count NaN values in y_col for this group
nan_count = df_to_use.loc[group_mask, y_col].isna().sum()
group_nan_counts[group] = nan_count
# Convert y_col to numeric
plot_df[y_col] = pd.to_numeric(plot_df[y_col], errors='coerce')
plot_df = plot_df.dropna(subset=[y_col])
if not plot_df.empty:
# Get the order of groups to ensure consistency between violin plot and overlays
# Use sorted order to match seaborn's default behavior
groups_order = sorted(plot_df[color_col].unique())
# Create violin plot using seaborn with explicit order
sns.violinplot(
data=plot_df,
x=color_col,
y=y_col,
hue=color_col,
ax=ax,
palette=color_palette_dict,
inner="quart", # Show quartiles as inner elements
alpha=0.6,
cut=0, # No extension beyond the data range
order=groups_order, # Explicitly set the order
width=0.5,
legend=False
)
# Overlay raw data points with strip plot using the same order
sns.stripplot(
data=plot_df,
x=color_col,
y=y_col,
ax=ax,
color='black',
size=2,
alpha=0.5,
jitter=True,
order=groups_order # Use the same order
)
# Calculate and plot mean ± SEM for each group using the same order
groups = groups_order # Use the same order as seaborn plots
x_positions = np.arange(len(groups))
for i, group in enumerate(groups):
group_data = pd.to_numeric(plot_df[plot_df[color_col] == group][y_col], errors='coerce').dropna()
if len(group_data) > 0:
mean_val = float(np.mean(group_data))
if len(group_data) > 1:
sem_val = float(np.std(group_data, ddof=1) / np.sqrt(len(group_data)))
else:
sem_val = 0.0
# Plot mean as a larger point
group_color = color_palette_dict.get(group, 'black') if color_palette_dict else 'black'
ax.plot(i+0.45, mean_val, 'o', color=group_color, markersize=5,
markeredgecolor='black', markeredgewidth=1, zorder=10)
# Add error bar for SEM
if sem_val > 0.0:
ax.errorbar(i+0.45, mean_val, yerr=sem_val, color='black',
capsize=5, capthick=1, elinewidth=1, zorder=9,
fmt='none')
# Set x-axis labels with sample counts
group_labels_with_counts = [
f"{group}\n(n={group_counts.get(group, 0)}, missing {group_nan_counts.get(group, 0)})"
for group in groups
]
ax.set_xticks(x_positions)
ax.set_xticklabels(group_labels_with_counts, rotation=30, ha='right')
ax.set_ylabel(y_col)
ax.set_xlabel(color_col)
# Add grid and styling
sns.despine(trim=True)
# Adjust layout to prevent label cutoff
plt.tight_layout()
# Use Panel's matplotlib pane
violin_plot = pn.pane.Matplotlib(fig, dpi=300, tight=True, width=400)
# Close the figure to prevent memory leaks
plt.close(fig)
except Exception as e:
logger.warning(f"Could not create violin plot: {e}")
violin_plot = pn.pane.Markdown("Violin plot error.")
return violin_plot