File size: 4,351 Bytes
fa07286 2825a8a fa07286 2825a8a fa07286 2825a8a 66d9c69 2825a8a 66d9c69 60c9728 2825a8a 66d9c69 2825a8a fa07286 2825a8a 66d9c69 fa07286 2825a8a 66d9c69 fa07286 9f981e7 2825a8a 9f981e7 fa07286 2825a8a 66d9c69 2825a8a fa07286 2825a8a 66d9c69 fa07286 2825a8a 66d9c69 2825a8a fa07286 66d9c69 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
import os
import shutil
import zipfile
from pathlib import Path
import gradio as gr
from lcmv_class import LCMVSourceEstimator
# Built-in files (must be in repo root)
BUILT_IN_GPSC = Path("ghw280_from_egig.gpsc")
BUILT_IN_SRC = Path("fsaverage-vol-5mm-src.fif")
if not BUILT_IN_GPSC.is_file():
raise FileNotFoundError(f"Required montage file not found: {BUILT_IN_GPSC}")
if not BUILT_IN_SRC.is_file():
raise FileNotFoundError(f"Required source space file not found: {BUILT_IN_SRC}")
# Prepare fsaverage structure
FS_DIR = Path("derivatives/lcmv/fsaverage")
BEM_DIR = FS_DIR / "bem"
BEM_DIR.mkdir(parents=True, exist_ok=True)
EXPECTED_SRC = Path("derivatives/lcmv/fsaverage-vol-5mm-src.fif")
EXPECTED_SRC.parent.mkdir(parents=True, exist_ok=True)
if not EXPECTED_SRC.exists():
shutil.copy(BUILT_IN_SRC, EXPECTED_SRC)
# Fixed subject/task
SUBJECT_ID = "sub"
TASK = ""
def run_lcmv(
cleaned_fif,
run_difumo: bool = True,
reg: float = 0.01,
n_jobs: int = 1,
download_option: str = "full_zip", # "full_zip" or "difumo_only"
):
abs_base = Path(".").resolve()
subject_output = abs_base / "derivatives" / "lcmv" / f"{SUBJECT_ID}_{TASK}"
if subject_output.exists():
shutil.rmtree(subject_output)
subject_output.mkdir(parents=True)
# Copy input
fif_path = subject_output / "cleaned_input.fif"
shutil.copy(cleaned_fif.name, fif_path)
# Config
config = {
'project_base': str(abs_base),
'subject_id': SUBJECT_ID,
'task': TASK,
'ica_file_path': str(fif_path.relative_to(abs_base)),
'gpsc_file_path': str(abs_base / BUILT_IN_GPSC),
'reg': reg,
'n_jobs': n_jobs,
}
try:
os.environ['SUBJECTS_DIR'] = str(abs_base / "derivatives" / "lcmv")
estimator = LCMVSourceEstimator(config)
metadata = estimator.run_enhanced_computation()
if run_difumo:
difumo_config = {'dimension': 512, 'resolution_mm': 2}
estimator.run_difumo_extraction(difumo_config=difumo_config)
# ✅ Dual output logic
if download_option == "difumo_only":
difumo_file = subject_output / "difumo_time_courses.npy"
if difumo_file.exists():
return str(difumo_file)
else:
raise FileNotFoundError("DiFuMo file not found. Ensure 'Extract DiFuMo' is enabled.")
else: # full_zip
zip_path = abs_base / f"{SUBJECT_ID}_{TASK}_lcmv_output.zip"
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for root, _, files in os.walk(subject_output):
for file in files:
file_path = Path(root) / file
arcname = file_path.relative_to(abs_base)
zf.write(file_path, arcname)
return str(zip_path)
except Exception as e:
error_log = subject_output / "lcmv_error.log"
with open(error_log, "w") as f:
f.write(f"LCMV failed:\n{str(e)}")
return str(error_log)
# Gradio UI
with gr.Blocks(theme=gr.themes.Base(), title="LCMV Source Estimation") as demo:
gr.Markdown("# LCMV Source Estimation")
gr.Markdown("Upload a cleaned EEG file (output from ICA pipeline). Uses built-in `ghw280_from_egig.gpsc` and precomputed source space.")
with gr.Row():
with gr.Column():
fif_input = gr.File(label="Cleaned EEG (.fif)", file_types=[".fif"])
run_difumo = gr.Checkbox(True, label="Extract DiFuMo 512 time courses")
reg = gr.Number(0.01, label="LCMV Regularization (reg)")
n_jobs = gr.Number(1, label="Parallel Jobs (n_jobs)", precision=0)
download_option = gr.Radio(
choices=["full_zip", "difumo_only"],
value="full_zip",
label="Download Option",
info="• full_zip: All outputs (STC, metadata, plots, DiFuMo)\n• difumo_only: Only difumo_time_courses.npy (for connectivity)"
)
run_btn = gr.Button("Run LCMV", variant="primary")
with gr.Column():
output_file = gr.File(label="Download Output")
run_btn.click(
fn=run_lcmv,
inputs=[fif_input, run_difumo, reg, n_jobs, download_option],
outputs=output_file,
)
demo.launch() |