Create app.py

app.py

@@ -0,0 +1,319 @@
+import gradio as gr
+import plotly.graph_objects as go
+import numpy as np
+from pathlib import Path
+
+# -----------------------------
+# Configuration
+# -----------------------------
+
+FREQ_BANDS = {
+    'Delta': (1, 4),
+    'Theta': (4, 8),
+    'Alpha': (8, 12),
+    'Low_Beta': (12, 20),
+    'High_Beta': (20, 30),
+    'Low_Gamma': (30, 50),
+    'High_Gamma': (50, 100)
+}
+
+condition_labels = {
+    'bima_activity_on': 'Bima Activity (ON)',
+    'bima_activity_off': 'Bima Activity (OFF)',
+    'hands_move_on': 'Hands Move (ON)',
+    'hands_move_off': 'Hands Move (OFF)',
+    'rest_eyes_closed_on': 'Rest Eyes Closed (ON)',
+    'rest_eyes_closed_off': 'Rest Eyes Closed (OFF)',
+    'rest_eyes_open_on': 'Rest Eyes Open (ON)',
+    'rest_eyes_open_off': 'Rest Eyes Open (OFF)'
+}
+
+condition_colors = {
+    'bima_activity': 'orange',
+    'hands_move': 'red',
+    'rest_eyes_closed': 'blue',
+    'rest_eyes_open': 'green'
+}
+
+band_colors = {
+    'Delta': 'lightsalmon', 'Theta': 'wheat', 'Alpha': 'mediumpurple',
+    'Low_Beta': 'skyblue', 'High_Beta': 'lightcoral',
+    'Low_Gamma': 'lightgreen', 'High_Gamma': 'plum'
+}
+
+METHOD_DISPLAY = {"multitaper": "Multitaper", "welch": "Welch"}
+
+
+# -----------------------------
+# PSD Analyzer Class
+# -----------------------------
+
+class PSDAnalyzer:
+    def __init__(self, data_file):
+        data_file = Path(data_file)
+        if not data_file.exists():
+            raise FileNotFoundError(f"Data file not found: {data_file}")
+
+        loaded = np.load(data_file, allow_pickle=True)
+        self.all_data = loaded['data'].tolist()
+
+        # Extract metadata
+        self.subjects_list = sorted({d['subject'] for d in self.all_data})
+        self.conditions_list = sorted({f"{d['condition']}_{d['med_state']}" for d in self.all_data})
+        self.region_names = self.all_data[0]['region_names']
+
+        # Precompute traces
+        self.traces = self._precompute_traces()
+
+        print(f"✅ Loaded {len(self.all_data)} subject-condition combinations")
+        print(f"Subjects: {len(self.subjects_list)} | Conditions: {len(self.conditions_list)} | Regions: {len(self.region_names)}")
+
+    def _precompute_traces(self):
+        """Precompute traces grouped by region → condition → method."""
+        traces = {
+            region: {
+                cond: {'multitaper': [], 'welch': [], 'subjects': []}
+                for cond in self.conditions_list
+            }
+            for region in self.region_names
+        }
+
+        for entry in self.all_data:
+            subject = entry['subject']
+            cond_key = f"{entry['condition']}_{entry['med_state']}"
+            for result in entry['results']:
+                region_name = result['region_name']
+                for method in ['multitaper', 'welch']:
+                    psd_data = result['psd_dual'].get(method)
+                    if psd_data is not None:
+                        freqs, psd = psd_data['freqs'], psd_data['psd']
+                        traces[region_name][cond_key][method].append({'freqs': freqs, 'psd': psd, 'subject': subject})
+                        if subject not in traces[region_name][cond_key]['subjects']:
+                            traces[region_name][cond_key]['subjects'].append(subject)
+        return traces
+
+    def create_plot(self, region, method, selected_conditions, selected_subjects, log_scale, show_bands):
+        """Generate Plotly figure based on selections."""
+        fig = go.Figure()
+        freq_range = (1, 100)
+
+        # Handle no conditions selected
+        if not selected_conditions:
+            fig.add_annotation(text="Please select at least one condition", x=0.5, y=0.5, showarrow=False, xref="paper", yref="paper")
+            return fig
+
+        plotted_conditions = []
+
+        for condition in selected_conditions:
+            data = self.traces[region][condition][method]
+            if not data:
+                continue
+
+            # Filter frequency range
+            filtered_data = [
+                {k: v for k, v in item.items()}
+                for item in data
+                if np.any((item['freqs'] >= freq_range[0]) & (item['freqs'] <= freq_range[1]))
+            ]
+            if not filtered_data:
+                continue
+
+            # Determine subjects to plot
+            if "All Subjects" in selected_subjects:
+                freqs = filtered_data[0]['freqs']
+                mask = (freqs >= freq_range[0]) & (freqs <= freq_range[1])
+                freqs = freqs[mask]
+                psds = np.array([item['psd'][mask] for item in filtered_data])
+                mean_psd = np.mean(psds, axis=0)
+
+                base_cond = condition.rsplit('_', 1)[0]
+                color = condition_colors.get(base_cond, 'gray')
+                linestyle = 'solid' if 'on' in condition else 'dash'
+
+                fig.add_trace(go.Scatter(
+                    x=freqs, y=mean_psd,
+                    mode='lines',
+                    name=f"{condition_labels.get(condition, condition)} (Mean, n={len(psds)})",
+                    line=dict(color=color, width=3, dash=linestyle)
+                ))
+            else:
+                available_subjects = [s for s in selected_subjects if s != "All Subjects"]
+                subject_data = [item for item in filtered_data if item['subject'] in available_subjects]
+                if not subject_data:
+                    continue
+
+                base_cond = condition.rsplit('_', 1)[0]
+                color = condition_colors.get(base_cond, 'gray')
+                linestyle = 'solid' if 'on' in condition else 'dash'
+
+                for item in subject_data:
+                    freqs, psd = item['freqs'], item['psd']
+                    mask = (freqs >= freq_range[0]) & (freqs <= freq_range[1])
+                    if not np.any(mask):
+                        continue
+                    fig.add_trace(go.Scatter(
+                        x=freqs[mask], y=psd[mask],
+                        mode='lines',
+                        name=f"{condition_labels.get(condition, condition)} - {item['subject']}",
+                        line=dict(color=color, width=2, dash=linestyle),
+                        opacity=0.8
+                    ))
+            plotted_conditions.append(condition)
+
+        # No valid data
+        if not plotted_conditions:
+            fig.add_annotation(
+                text="No data for selected conditions/subjects",
+                x=0.5, y=0.5, showarrow=False, xref="paper", yref="paper"
+            )
+            return fig
+
+        # Add frequency band visualization - Clean & non-overlapping
+        if show_bands:
+            # 1. Shaded background for each band
+            for band, (low, high) in FREQ_BANDS.items():
+                if low < freq_range[1] and high > freq_range[0]:
+                    fig.add_vrect(
+                        x0=low, x1=high,
+                        fillcolor=band_colors[band], opacity=0.15,
+                        layer="below", line_width=0
+                    )
+
+            # 2. Dotted lines at band edges
+            all_edges = sorted(set([low for low, _ in FREQ_BANDS.values()] + 
+                                   [high for _, high in FREQ_BANDS.values()]))
+            for edge in all_edges:
+                if freq_range[0] < edge < freq_range[1]:
+                    fig.add_vline(x=edge, line=dict(color="gray", width=1, dash="dot"), opacity=0.5)
+
+            # 3. Estimate Y position for band labels
+            try:
+                max_power = max(trace.y.max() for trace in fig.data if hasattr(trace, 'y'))
+                y_pos = max_power * 1.15 if not log_scale else max_power * 3  # higher in log space
+            except Exception:
+                y_pos = 1.1
+
+            # 4. Place band names at center, above the plot
+            for band, (low, high) in FREQ_BANDS.items():
+                center = (low + high) / 2
+                if freq_range[0] <= center <= freq_range[1]:
+                    fig.add_annotation(
+                        x=center,
+                        y=y_pos,
+                        text=band,
+                        showarrow=False,
+                        font=dict(size=10, color="dimgray"),
+                        xanchor="center",
+                        yanchor="bottom",
+                        opacity=0.9,
+                        xref="x",
+                        yref="y"
+                    )
+
+        # Final layout
+        yaxis_title = "Power (log)" if log_scale else "Power"
+        fig.update_layout(
+            title=f"PSD - {region} | {METHOD_DISPLAY[method]}",
+            xaxis_title="Frequency (Hz)",
+            yaxis_title=yaxis_title,
+            yaxis_type="log" if log_scale else "linear",
+            template="plotly_white",
+            height=650,
+            legend=dict(
+                y=0.99, yanchor="top", x=1.02, xanchor="left",
+                bgcolor="rgba(255,255,255,0.8)", font_size=11
+            ),
+            margin=dict(r=160, t=60, b=80),
+            hovermode='x unified'
+        )
+        fig.update_xaxes(
+            showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)',
+            showline=True, linewidth=1, linecolor='gray'
+        )
+        fig.update_yaxes(
+            showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)',
+            showline=True, linewidth=1, linecolor='gray'
+        )
+
+        return fig
+
+
+# -----------------------------
+# Gradio Interface
+# -----------------------------
+
+def create_app(analyzer: PSDAnalyzer):
+    with gr.Blocks(
+        theme=gr.themes.Soft(),
+        title="PSD Analysis Dashboard",
+        css=".big-plot { height: 700px; }"
+    ) as demo:
+        gr.Markdown("# 📊 Power Spectral Density Analysis Dashboard")
+
+        # TOP ROW: Data Selection
+        with gr.Row():
+            with gr.Column(scale=1):
+                subjects = gr.CheckboxGroup(
+                    choices=["All Subjects"] + analyzer.subjects_list,
+                    value=["All Subjects"],
+                    label="Subjects",
+                    interactive=True
+                )
+            with gr.Column(scale=1):
+                conditions = gr.CheckboxGroup(
+                    choices=analyzer.conditions_list,
+                    value=analyzer.conditions_list[:2],
+                    label="Conditions",
+                    interactive=True
+                )
+
+        # CENTER: BIG PSD PLOT
+        with gr.Row():
+            with gr.Column():
+                plot_output = gr.Plot(
+                    label="PSD Plot",
+                    elem_classes="big-plot"
+                )
+
+        # BOTTOM ROW: Display Options
+        with gr.Row():
+            with gr.Column():
+                log_scale = gr.Checkbox(value=True, label="Use Log Scale (Y-axis)")
+            with gr.Column():
+                show_bands = gr.Checkbox(value=True, label="Show Frequency Bands")
+
+        # BOTTOM ROW: Method (Left) and Region (Right)
+        with gr.Row():
+            with gr.Column(min_width=200):
+                method = gr.Radio(
+                    choices=["multitaper", "welch"],
+                    value="multitaper",
+                    label="PSD Method",
+                    interactive=True
+                )
+            with gr.Column(min_width=200):
+                region = gr.Dropdown(
+                    choices=analyzer.region_names,
+                    value=analyzer.region_names[0],
+                    label="Brain Region",
+                    interactive=True
+                )
+
+        # Update plot on any input change
+        inputs = [region, method, conditions, subjects, log_scale, show_bands]
+        for comp in inputs:
+            comp.change(fn=analyzer.create_plot, inputs=inputs, outputs=plot_output)
+        demo.load(fn=analyzer.create_plot, inputs=inputs, outputs=plot_output)
+
+    return demo
+
+
+# -----------------------------
+# Launch App
+# -----------------------------
+
+if __name__ == "__main__":
+    DATA_PATH = "psd_voxel_all_data.npz" 
+    analyzer = PSDAnalyzer(DATA_PATH)
+    app = create_app(analyzer)
+    app.launch(share=True, show_error=True)