Create app.py

app.py

@@ -0,0 +1,121 @@
+import os
+import shutil
+from pathlib import Path
+import gradio as gr
+
+# Import your pipeline
+from ica_xtra import run_preprocessing_pipeline
+
+# Constants
+SUBJECT_ID = "sub-01"
+OUTPUT_DIR = Path("outputs")
+OUTPUT_DIR.mkdir(exist_ok=True)
+
+def preprocess_eeg(
+    eeg_file,
+    gpsc_file,
+    input_format="fif",
+    apply_highpass=True,
+    apply_lowpass=True,
+    apply_notch=True,
+    line_freq=60.0,
+    pre_ica_mad=3.5,
+    post_ica_mad=5.0,
+    interpolate_before_ica=False,
+    use_artifact_detection=True,
+):
+    # Clean output dir
+    subject_dir = OUTPUT_DIR / SUBJECT_ID
+    if subject_dir.exists():
+        shutil.rmtree(subject_dir)
+    subject_dir.mkdir(parents=True)
+
+    # Copy uploaded files
+    eeg_path = subject_dir / Path(eeg_file.name).name
+    gpsc_path = subject_dir / Path(gpsc_file.name).name
+    shutil.copy(eeg_file.name, eeg_path)
+    shutil.copy(gpsc_file.name, gpsc_path)
+
+    try:
+        run_preprocessing_pipeline(
+            subject=SUBJECT_ID,
+            input_path=str(eeg_path),
+            gpsc_file=str(gpsc_path),
+            base_output_path=str(OUTPUT_DIR),
+            input_format=input_format,
+            apply_highpass=apply_highpass,
+            apply_lowpass=apply_lowpass,
+            apply_notch=apply_notch,
+            line_freq=line_freq,
+            pre_ica_mad_threshold=pre_ica_mad,
+            post_ica_mad_threshold=post_ica_mad,
+            interpolate_before_ica=interpolate_before_ica,
+            use_artifact_detection_channels=use_artifact_detection,
+            append_subject_to_output=True,
+            plot=True,
+            log_to_file=True,
+            random_state=99,
+        )
+
+        # Gather outputs
+        cleaned_fif = subject_dir / f"{SUBJECT_ID}_eeg_ica_cleaned_raw.fif"
+        log_file = subject_dir / f"{SUBJECT_ID}_preproc_log.txt"
+        plot_dir = subject_dir / "plots"
+
+        files_to_return = [str(cleaned_fif), str(log_file)]
+        if plot_dir.exists():
+            for ext in ["*.png"]:
+                files_to_return.extend([str(p) for p in plot_dir.glob(ext)])
+
+        return files_to_return
+
+    except Exception as e:
+        error_log = subject_dir / "error.txt"
+        with open(error_log, "w") as f:
+            f.write(f"Preprocessing failed:\n{str(e)}")
+        return [str(error_log)]
+
+with gr.Blocks(theme=gr.themes.Base(), title="EEG Preprocessing") as demo:
+    gr.Markdown("# EEG Preprocessing Pipeline")
+    gr.Markdown("Upload EEG (.fif or .mff) and montage (.gpsc) files for automated cleaning.")
+
+    with gr.Row():
+        with gr.Column():
+            eeg_input = gr.File(label="EEG File (.fif or .mff)")
+            gpsc_input = gr.File(label="Montage File (.gpsc)")
+            format_dropdown = gr.Dropdown(choices=["fif", "mff"], value="fif", label="Input Format")
+
+            with gr.Accordion("Advanced Settings", open=False):
+                hp = gr.Checkbox(True, label="Apply Highpass (1.0 Hz)")
+                lp = gr.Checkbox(True, label="Apply Lowpass (100.0 Hz)")
+                notch = gr.Checkbox(True, label="Apply Notch Filter")
+                line_freq = gr.Number(60.0, label="Line Frequency (Hz)")
+                pre_mad = gr.Slider(2.0, 8.0, value=3.5, label="Pre-ICA MAD Threshold")
+                post_mad = gr.Slider(2.0, 8.0, value=5.0, label="Post-ICA MAD Threshold")
+                interp_before = gr.Checkbox(False, label="Interpolate Bad Channels Before ICA")
+                use_artifact = gr.Checkbox(True, label="Use Artifact Detection Channels")
+
+            run_btn = gr.Button("Run Preprocessing", variant="primary")
+
+        with gr.Column():
+            output_files = gr.Files(label="Download Results")
+
+    run_btn.click(
+        fn=preprocess_eeg,
+        inputs=[
+            eeg_input,
+            gpsc_input,
+            format_dropdown,
+            hp,
+            lp,
+            notch,
+            line_freq,
+            pre_mad,
+            post_mad,
+            interp_before,
+            use_artifact,
+        ],
+        outputs=output_files,
+    )
+
+demo.launch()