tyler/flares
2
0
Fork 0
Code Issues 28 Pull Requests Actions Packages Projects Releases 12 Wiki Activity

further variable changes

Browse Source
This commit is contained in:
tyler
2026-01-23 11:25:01 -08:00
parent 9779a63a9c
commit 1b78f1904d
3 changed files with 239 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

318
main.py
View File

@@ -22,6 +22,7 @@ import subprocess
from pathlib import Path, PurePosixPath
from datetime import datetime
from multiprocessing import Process, current_process, freeze_support, Manager
from enum import Enum, auto
# External library imports
import numpy as np
@@ -215,9 +216,10 @@ SECTIONS = [
{
"title": "Other",
"params": [
{"name": "TIME_WINDOW_START", "default": "0", "type": int, "help": "Where to start averaging the fir model bins. Only affects the significance and contrast images."},
{"name": "TIME_WINDOW_END", "default": "15", "type": int, "help": "Where to end averaging the fir model bins. Only affects the significance and contrast images."},
{"name": "MAX_WORKERS", "default": 4, "type": int, "help": "Number of files to be processed at once. Lowering this value may help on underpowered systems."},
{"name": "TIME_WINDOW_START", "default": 0, "type": int, "help": "Where to start averaging the fir model bins. Only affects the significance and contrast images."},
{"name": "TIME_WINDOW_END", "default": 15, "type": int, "help": "Where to end averaging the fir model bins. Only affects the significance and contrast images."},
{"name": "MAX_WORKERS", "default": 4, "type": str, "help": "Number of files to be processed at once. Setting this to a small integer value may help on underpowered systems. Remove the value to use an automatic amount."},
{"name": "VERBOSITY", "default": False, "type": bool, "help": "True will log lots of debugging information to the log file. False will only log required data."},
]
},
]
@@ -766,15 +768,23 @@ class UpdateOptodesWindow(QWidget):
write_raw_snirf(raw, save_path)
class EventUpdateMode(Enum):
WRITE_SNIRF = auto() # destructive
WRITE_JSON = auto() # non-destructive
class UpdateEventsWindow(QWidget):
def __init__(self, parent=None):
def __init__(self, parent=None, mode=EventUpdateMode.WRITE_SNIRF, caller=None):
super().__init__(parent, Qt.WindowType.Window)
self.mode = mode
self.caller = caller or self.__class__.__name__
self.setWindowTitle("Update event markers")
self.resize(760, 200)
print("INIT MODE:", mode)
self.label_file_a = QLabel("SNIRF file:")
self.line_edit_file_a = QLineEdit()
self.line_edit_file_a.setReadOnly(True)
@@ -1075,120 +1085,139 @@ class UpdateEventsWindow(QWidget):
QMessageBox.warning(self, "No SNIRF file", "Please select a SNIRF file.")
return
boris_obs = self.boris_data["observations"][selected_obs]
# --- Extract videos + delays ---
files = boris_obs.get("file", {})
offsets = boris_obs.get("media_info", {}).get("offset", {})
videos = {}
for key, path in files.items():
if path: # only include videos that exist
delay = offsets.get(key, 0.0) # default 0 if missing
videos[key] = {"file": path, "delay": delay}
base_name = os.path.splitext(os.path.basename(file_a))[0]
suggested_name = f"{base_name}_{suffix}.snirf"
# Open save dialog
save_path, _ = QFileDialog.getSaveFileName(
self,
"Save SNIRF File As",
suggested_name,
"SNIRF Files (*.snirf)"
)
if self.mode == EventUpdateMode.WRITE_SNIRF:
# Open save dialog for SNIRF
base_name = os.path.splitext(os.path.basename(file_a))[0]
suggested_name = f"{base_name}_{suffix}.snirf"
save_path, _ = QFileDialog.getSaveFileName(
self,
"Save SNIRF File As",
suggested_name,
"SNIRF Files (*.snirf)"
)
if not save_path:
print("SNIRF save cancelled.")
return
if not save_path.lower().endswith(".snirf"):
save_path += ".snirf"
if not save_path:
print("Save cancelled.")
return
try:
raw = read_raw_snirf(file_a, preload=True)
if not save_path.lower().endswith(".snirf"):
save_path += ".snirf"
# --- Align BORIS events to SNIRF ---
boris_events = boris_obs.get("events", [])
onsets, durations, descriptions = [], [], []
open_events = {} # label -> list of start times
label_counts = {}
used_times = set()
sfreq = raw.info['sfreq']
min_shift = 1.0 / sfreq
max_attempts = 10
try:
raw = read_raw_snirf(snirf_path, preload=True)
for event in boris_events:
if not isinstance(event, list) or len(event) < 3:
continue
event_time = event[0]
label = event[2]
count = label_counts.get(label, 0) + 1
label_counts[label] = count
onsets = []
durations = []
descriptions = []
if label not in open_events:
open_events[label] = []
open_events = {} # label -> list of start times
label_counts = {}
if count % 2 == 1:
open_events[label].append(event_time)
else:
if open_events[label]:
start_time = open_events[label].pop(0)
duration = event_time - start_time
if duration <= 0:
continue
used_times = set()
sfreq = raw.info['sfreq'] # sampling frequency in Hz
min_shift = 1.0 / sfreq
max_attempts = 10
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
continue
for event in boris_events:
if not isinstance(event, list) or len(event) < 3:
continue
event_time = event[0]
label = event[2]
count = label_counts.get(label, 0) + 1
label_counts[label] = count
if label not in open_events:
open_events[label] = []
if count % 2 == 1:
# Odd occurrence = start event
open_events[label].append(event_time)
else:
# Even occurrence = end event
if open_events[label]:
matched_start = open_events[label].pop(0)
duration = event_time - matched_start
if duration <= 0:
print(f"Warning: Duration for {label} is non-positive ({duration}). Skipping.")
continue
shifted_start = matched_start + time_shift
adjusted_time = shifted_start
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
onsets.append(adjusted_time)
durations.append(duration)
descriptions.append(label)
# Handle unmatched starts
for label, starts in open_events.items():
for start_time in starts:
adjusted_time = start_time + time_shift
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
if attempts == max_attempts:
print(f"Warning: Couldn't find unique time for {label} @ {matched_start}s. Skipping.")
continue
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
print(f"Adding event: {label} @ {adjusted_time:.3f}s for {duration:.3f}s")
onsets.append(adjusted_time)
durations.append(duration)
durations.append(0.0)
descriptions.append(label)
else:
print(f"Warning: Unmatched end for label '{label}' at {event_time:.3f}s. Skipping.")
# Optionally warn about any unmatched starts left open
for label, starts in open_events.items():
for start_time in starts:
shifted_start = start_time + time_shift
new_annotations = Annotations(onset=onsets, duration=durations, description=descriptions)
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
QMessageBox.information(self, "Success", "SNIRF file updated with aligned BORIS events.")
adjusted_time = shifted_start
attempts = 0
while round(adjusted_time, 6) in used_times and attempts < max_attempts:
adjusted_time += min_shift
attempts += 1
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to update SNIRF file:\n{e}")
if attempts == max_attempts:
print(f"Warning: Couldn't find unique time for unmatched start {label} @ {start_time}s. Skipping.")
continue
elif self.mode == EventUpdateMode.WRITE_JSON:
# Open save dialog for JSON
base_name = os.path.splitext(os.path.basename(file_a))[0]
suggested_name = f"{base_name}_{suffix}_alignment.json"
save_path, _ = QFileDialog.getSaveFileName(
self,
"Save Event Alignment JSON As",
suggested_name,
"JSON Files (*.json)"
)
if not save_path:
print("JSON save cancelled.")
return
if not save_path.lower().endswith(".json"):
save_path += ".json"
adjusted_time = round(adjusted_time, 6)
used_times.add(adjusted_time)
# Build JSON dict
json_data = {
"observation": selected_obs,
"snirf_anchor": {"label": snirf_label, "time": snirf_anchor_time},
"boris_anchor": {"label": boris_label, "time": boris_anchor_time},
"time_shift": time_shift,
"videos": videos
}
print(f"Warning: Unmatched start for label '{label}' at {start_time:.3f}s. Adding with duration 0.")
onsets.append(adjusted_time)
durations.append(0.0)
descriptions.append(label)
new_annotations = Annotations(onset=onsets, duration=durations, description=descriptions)
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
QMessageBox.information(self, "Success", "SNIRF file updated with aligned BORIS events.")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to update SNIRF file:\n{e}")
# Write JSON
try:
with open(save_path, "w", encoding="utf-8") as f:
json.dump(json_data, f, indent=4)
QMessageBox.information(self, "Success", f"Event alignment saved to:\n{save_path}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to write JSON:\n{e}")
def update_optode_positions(self, file_a, file_b, save_path):
@@ -1221,6 +1250,47 @@ class UpdateEventsWindow(QWidget):
write_raw_snirf(raw, save_path)
def _apply_events_to_snirf(self, raw, new_annotations, save_path):
raw.set_annotations(new_annotations)
write_raw_snirf(raw, save_path)
def _write_event_mapping_json(
self,
file_a,
file_b,
selected_obs,
snirf_anchor,
boris_anchor,
time_shift,
mapped_events,
save_path
):
import json
from datetime import datetime
import os
payload = {
"source": {
"called_from": self.caller,
"snirf_file": os.path.basename(file_a),
"boris_file": os.path.basename(file_b),
"observation": selected_obs
},
"alignment": {
"snirf_anchor": snirf_anchor,
"boris_anchor": boris_anchor,
"time_shift_seconds": time_shift
},
"events": mapped_events,
"created_at": datetime.utcnow().isoformat() + "Z"
}
with open(save_path, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
return save_path
class ProgressBubble(QWidget):
"""
A clickable widget displaying a progress bar made of colored rectangles and a label.
@@ -2506,7 +2576,7 @@ class ExportDataAsCSVViewerWidget(QWidget):
self.index_texts = [
"0 (Export Data to CSV)",
# "1 (second image)",
"1 (CSV for SPARKS)",
# "2 (third image)",
# "3 (fourth image)",
]
@@ -2658,7 +2728,6 @@ class ExportDataAsCSVViewerWidget(QWidget):
# Pass the necessary arguments to each method
for file_path in selected_file_paths:
haemo_obj = self.haemo_dict.get(file_path)
if haemo_obj is None:
continue
@@ -2692,10 +2761,63 @@ class ExportDataAsCSVViewerWidget(QWidget):
QMessageBox.critical(self, "Error", f"Failed to update SNIRF file:\n{e}")
elif idx == 1:
try:
suggested_name = f"{file_path}_sparks.csv"
# Open save dialog
save_path, _ = QFileDialog.getSaveFileName(
self,
"Save SNIRF File As",
suggested_name,
"CSV Files (*.csv)"
)
if not save_path:
print("Save cancelled.")
return
if not save_path.lower().endswith(".csv"):
save_path += ".csv"
# Save the CSV here
raw = haemo_obj
data, times = raw.get_data(return_times=True)
ann_col = np.full(times.shape, "", dtype=object)
if raw.annotations is not None and len(raw.annotations) > 0:
for onset, duration, desc in zip(
raw.annotations.onset,
raw.annotations.duration,
raw.annotations.description
):
mask = (times >= onset) & (times < onset + duration)
ann_col[mask] = desc
df = pd.DataFrame(data.T, columns=raw.ch_names)
df.insert(0, "annotation", ann_col)
df.insert(0, "time", times)
df.to_csv(save_path, index=False)
QMessageBox.information(self, "Success", "CSV file has been saved.")
win = UpdateEventsWindow(
parent=self,
mode=EventUpdateMode.WRITE_JSON,
caller="Video Alignment Tool"
)
win.show()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to update SNIRF file:\n{e}")
else:
print(f"No method defined for index {idx}")
class ClickableLabel(QLabel):
def __init__(self, full_pixmap: QPixmap, thumbnail_pixmap: QPixmap):
super().__init__()
@@ -4321,7 +4443,7 @@ class MainApplication(QMainWindow):
def update_event_markers(self):
if self.events is None or not self.events.isVisible():
self.events = UpdateEventsWindow(self)
self.events = UpdateEventsWindow(self, EventUpdateMode.WRITE_SNIRF, "Manual SNIRF Edit")
self.events.show()
def open_file_dialog(self):