Update app.py
Browse files
app.py
CHANGED
|
@@ -42,18 +42,14 @@ def preprocess_eeg(
|
|
| 42 |
uploaded_file = Path(eeg_file.name)
|
| 43 |
|
| 44 |
if input_format == "mff":
|
| 45 |
-
# Validate: must be a ZIP file
|
| 46 |
if not zipfile.is_zipfile(uploaded_file):
|
| 47 |
raise ValueError("MFF input must be a ZIP file containing the .mff folder.")
|
| 48 |
-
# Extract directly into subject_dir as "sub.mff"
|
| 49 |
eeg_path = subject_dir / f"{SUBJECT_ID}.mff"
|
| 50 |
with zipfile.ZipFile(uploaded_file, 'r') as zf:
|
| 51 |
zf.extractall(eeg_path)
|
| 52 |
-
# Optional: validate
|
| 53 |
if not (eeg_path / "info.xml").exists():
|
| 54 |
raise ValueError("Invalid MFF: missing 'info.xml' in the ZIP.")
|
| 55 |
else:
|
| 56 |
-
# For FIF: copy the file into subject_dir with its original name
|
| 57 |
eeg_path = subject_dir / uploaded_file.name
|
| 58 |
shutil.copy(uploaded_file, eeg_path)
|
| 59 |
|
|
@@ -79,15 +75,42 @@ def preprocess_eeg(
|
|
| 79 |
random_state=99,
|
| 80 |
)
|
| 81 |
|
| 82 |
-
#
|
| 83 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 84 |
raw = mne.io.read_raw_fif(str(cleaned_fif), preload=False)
|
| 85 |
info_summary = subject_dir / f"{SUBJECT_ID}_info_summary.txt"
|
| 86 |
with open(info_summary, "w") as f:
|
| 87 |
f.write(f"=== EEG Info Summary for {SUBJECT_ID} ===\n")
|
| 88 |
f.write(str(raw.info))
|
| 89 |
|
| 90 |
-
# Create ZIP
|
| 91 |
zip_path = OUTPUT_DIR / f"{SUBJECT_ID}_preprocessing_output.zip"
|
| 92 |
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
| 93 |
for root, _, files in os.walk(subject_dir):
|
|
|
|
| 42 |
uploaded_file = Path(eeg_file.name)
|
| 43 |
|
| 44 |
if input_format == "mff":
|
|
|
|
| 45 |
if not zipfile.is_zipfile(uploaded_file):
|
| 46 |
raise ValueError("MFF input must be a ZIP file containing the .mff folder.")
|
|
|
|
| 47 |
eeg_path = subject_dir / f"{SUBJECT_ID}.mff"
|
| 48 |
with zipfile.ZipFile(uploaded_file, 'r') as zf:
|
| 49 |
zf.extractall(eeg_path)
|
|
|
|
| 50 |
if not (eeg_path / "info.xml").exists():
|
| 51 |
raise ValueError("Invalid MFF: missing 'info.xml' in the ZIP.")
|
| 52 |
else:
|
|
|
|
| 53 |
eeg_path = subject_dir / uploaded_file.name
|
| 54 |
shutil.copy(uploaded_file, eeg_path)
|
| 55 |
|
|
|
|
| 75 |
random_state=99,
|
| 76 |
)
|
| 77 |
|
| 78 |
+
# 🔧 FIX: Handle multiple cleaned FIF files
|
| 79 |
+
cleaned_fifs = list(subject_dir.glob("*_eeg_ica_cleaned_raw.fif"))
|
| 80 |
+
if not cleaned_fifs:
|
| 81 |
+
raise FileNotFoundError("No cleaned FIF file found after preprocessing.")
|
| 82 |
+
|
| 83 |
+
# Prefer the one with 'post_ica' in the name (final version)
|
| 84 |
+
final_fif = None
|
| 85 |
+
for f in cleaned_fifs:
|
| 86 |
+
if "post_ica" in f.name:
|
| 87 |
+
final_fif = f
|
| 88 |
+
break
|
| 89 |
+
if final_fif is None:
|
| 90 |
+
# If no post_ica file, use the most recently modified (fallback)
|
| 91 |
+
final_fif = max(cleaned_fifs, key=os.path.getmtime)
|
| 92 |
+
|
| 93 |
+
# Delete all cleaned FIFs except the final one
|
| 94 |
+
for f in cleaned_fifs:
|
| 95 |
+
if f != final_fif:
|
| 96 |
+
f.unlink()
|
| 97 |
+
|
| 98 |
+
# Rename to standard clean name
|
| 99 |
+
expected_name = subject_dir / f"{SUBJECT_ID}_eeg_ica_cleaned_raw.fif"
|
| 100 |
+
if final_fif != expected_name:
|
| 101 |
+
final_fif.rename(expected_name)
|
| 102 |
+
cleaned_fif = expected_name
|
| 103 |
+
else:
|
| 104 |
+
cleaned_fif = final_fif
|
| 105 |
+
|
| 106 |
+
# Load and save info summary
|
| 107 |
raw = mne.io.read_raw_fif(str(cleaned_fif), preload=False)
|
| 108 |
info_summary = subject_dir / f"{SUBJECT_ID}_info_summary.txt"
|
| 109 |
with open(info_summary, "w") as f:
|
| 110 |
f.write(f"=== EEG Info Summary for {SUBJECT_ID} ===\n")
|
| 111 |
f.write(str(raw.info))
|
| 112 |
|
| 113 |
+
# Create ZIP
|
| 114 |
zip_path = OUTPUT_DIR / f"{SUBJECT_ID}_preprocessing_output.zip"
|
| 115 |
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
| 116 |
for root, _, files in os.walk(subject_dir):
|