""" Filename: exporttocsv.py Description: Logic for the Export To CSV analysis window Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in imports import os from typing import Any # External library imports from mne import Epochs from pandas import DataFrame from mne.io.base import BaseRaw from PySide6.QtWidgets import QFileDialog, QMessageBox from flares import sparks_csv_export from src.shared.flaresbasewidget import CSVUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget): def __init__( self, haemo_dict: dict[str, BaseRaw], epochs_dict: dict[str, Epochs], cha_dict: dict[str, DataFrame], df_ind_dict: dict[str, DataFrame], design_matrix_dict: dict[str, DataFrame], contrast_results_dict: dict[str, dict[str, Any]], group_dict: dict[str, str], config_dict: dict[str, dict[str, Any]], fir_feature_dict: dict[str, dict[str, Any]], qc_dict: dict[str, dict[str, Any]], ) -> None: super().__init__("ExportToCSV") self.setWindowTitle(f"Export To CSV Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict self.epochs_dict = epochs_dict self.cha_dict = cha_dict self.df_ind_dict = df_ind_dict self.design_matrix = design_matrix_dict self.contrast_results_dict = contrast_results_dict self.group_dict = group_dict self.config_dict = config_dict self.fir_feature_dict = fir_feature_dict self.qc_dict = qc_dict self.setup_csv_ui(["0 (Export Data to CSV)", "1 (CSV for SPARKS)", "2 (Export Configuration to CSV)", "3 (Paragraph of Configuration)", "4 (Export FIR Waveform Features to CSV)", "5 (Export Quality Control Metrics to CSV)", "6 (Export Master Consolidated Matrix [All Participants into 1 CSV])" ]) def process_request(self): selected_display_names = self._get_checked_items(self.participant_dropdown) selected_file_paths: list[str] = [] for display_name in selected_display_names: for fp, short_label in self.participant_map.items(): expected_display = f"{short_label} ({os.path.basename(fp)})" if display_name == expected_display: selected_file_paths.append(fp) break selected_indexes = [ int(s.split(" ")[0]) for s in self._get_checked_items(self.image_index_dropdown) ] if not selected_file_paths or not selected_indexes: QMessageBox.warning(self, "Selection Missing", "Please select at least one participant and one export type.") return output_dir = QFileDialog.getExistingDirectory(self, "Select Output Folder for CSV Exports") if not output_dir: print("Export cancelled: No folder selected.") return success_count = 0 # Pass the necessary arguments to each method for file_path in selected_file_paths: base_filename = os.path.splitext(os.path.basename(file_path))[0] haemo_obj = self.haemo_dict.get(file_path) if haemo_obj is None: continue cha = self.cha_dict.get(file_path) for idx in selected_indexes: try: if idx == 0: save_path = os.path.join(output_dir, f"{base_filename}_exported.csv") if cha is not None: cha.to_csv(save_path) success_count += 1 elif idx == 1: # SPARKS Export save_path = os.path.join(output_dir, f"{base_filename}_sparks.csv") sparks_csv_export(haemo_obj, save_path) success_count += 1 elif idx == 2: formatted_data = { os.path.basename(path.replace("\\", "/")): inner_dict for path, inner_dict in self.config_dict.items() } df = DataFrame(formatted_data) df.index.name = "Parameter" save_path = os.path.join(output_dir, f"{APP_NAME}_configuration.csv") df.to_csv(save_path) success_count += 1 elif idx == 3: first_params = next(iter(self.config_dict.values())) magic_string = self.gen_magic_str(first_params) self.placeholder_label.setText(magic_string) elif idx == 4: # FIR Waveform Features Export fir_data = self.fir_feature_dict.get(file_path) if fir_data: names = fir_data.get("feature_names", []) vals = fir_data.get("features", []) chans = fir_data.get("feature_channels", []) fir_df = DataFrame({ "Feature_Name": names, "Channel": chans if len(chans) == len(names) else ["N/A"] * len(names), "Value": vals }) save_path = os.path.join(output_dir, f"{base_filename}_fir_features.csv") fir_df.to_csv(save_path, index=False) success_count += 1 elif idx == 5: # Quality Control (QC) Metrics Export qc_data = self.qc_dict.get(file_path) if qc_data and isinstance(qc_data, dict): qc_df = DataFrame(list(qc_data.items()), columns=["Metric", "Value"]) save_path = os.path.join(output_dir, f"{base_filename}_qc_metrics.csv") qc_df.to_csv(save_path, index=False) success_count += 1 elif idx == 6: # Master Consolidated Matrix (1 single CSV combining all selected participants) save_path = os.path.join(output_dir, f"{APP_NAME}_master_consolidated.csv") if not os.path.exists(save_path): master_rows: list[dict[str, Any]] = [] for fp in selected_file_paths: abs_path = os.path.abspath(fp) grp = self.group_dict.get(fp, "Unknown") row: dict[str, Any] = {"Participant": abs_path, "Group": grp} # QC Metrics qc_info = self.qc_dict.get(fp, {}) if isinstance(qc_info, dict): for mk, mv in qc_info.items(): row[f"QC_{mk}"] = mv # FIR Features fir_info = self.fir_feature_dict.get(fp, {}) print("1") if isinstance(fir_info, dict): print("2") f_names = fir_info.get("feature_names", []) f_vals = fir_info.get("features", []) print("3") if len(f_names) == len(f_vals): print("4") for fn, fv in zip(f_names, f_vals): row[f"FIR_{fn}"] = fv master_rows.append(row) if master_rows: df_master = DataFrame(master_rows) df_master.to_csv(save_path, index=False) success_count += 1 else: print(f"No method defined for index {idx}") except Exception as e: print(f"Failed to export {file_path} (Type {idx}): {e}") # 4. Final Notification if success_count > 0: QMessageBox.information(self, "Export Complete", f"Successfully saved {success_count} CSV files to:\n{output_dir}") # # If SPARKS export was included, show the Event Window once at the end # if 1 in selected_indexes: # win = UpdateEventsWindow( # parent=self, # mode=EventUpdateMode.WRITE_JSON, # caller="Video Alignment Tool" # ) # win.show() def gen_magic_str(self, all_params: dict[str, str]) -> str: magic_str = "To start, the data was loaded into the application. " if all_params['DOWNSAMPLE']: magic_str += f"The data was downsampled to {all_params['DOWNSAMPLE_FREQUENCY']}hz. " else: magic_str += "The data was not downsampled and was retained at its original sampling frequency. " if all_params['TRIM']: magic_str += f"The data was then trimmed to only keep {all_params['SECONDS_TO_KEEP']} seconds before the first event, as the earlier data with no events would not be needed. " else: magic_str += "The data was not trimmed and was retained at its original length. " if all_params['OPTODE_PLACEMENT']: magic_str += f"A visualization check of the optode locations in 3D space was performed to ensure that they were at the correct location. " if all_params['SHORT_CHANNELS']: magic_str += f"A short channel(s) was specified with the maximum length of a short channel being {all_params['SHORT_CHANNELS_THRESHOLD']} metres. Channels that were over this distance but under {all_params['LONG_CHANNELS_THRESHOLD']} metres were retained. " else: magic_str += f"No short channel was specified. Channels between the lengths of {all_params['SHORT_CHANNELS_THRESHOLD']} and {all_params['LONG_CHANNELS_THRESHOLD']} meters were used with all other channles being discarded. " if all_params['HEART_RATE']: magic_str += f"The heart rate of the participant was attempted to be calculated. {all_params['SECONDS_TO_STRIP_HR']} seconds were ignored from the start of the file to avoid motion artifacts. " magic_str += f"The heart rate had hard limits of {all_params['MAX_LOW_HR']} to {all_params['MAX_HIGH_HR']} and was smoothed across {all_params['SMOOTHING_WINDOW_HR']} samples. " else: magic_str += "The heart rate of the participants was not attempted to be calculated. " if all_params['SCI']: magic_str += f"The Scalp Coupling Index was calculated over {all_params['SCI_TIME_WINDOW']} second windows. The threshold value was {all_params['SCI_THRESHOLD']}. " else: magic_str += "Scalp Coupling Index was not used. " if all_params['SNR']: magic_str += f"The Signal to Noise Ratio was calculated. The threshold value was set to {all_params['SNR_THRESHOLD']}db. " else: magic_str += "The Signal to Noise Ratio was not used. " if all_params['PSP']: magic_str += f"Peak Spectral Power was calculated over {all_params['PSP_TIME_WINDOW']} second windows. The threshold value was {all_params['PSP_THRESHOLD']}. " else: magic_str += "Peak Spectral Power was not used. " if all_params['COEFF_VAR']: magic_str += f"The Coefficient of Variation was calculated. The threshold value was set to {all_params['COEFF_VAR_THRESHOLD']}. " else: magic_str += "The Coefficient of Variation was not used. " if all_params['MAD']: magic_str += f"Median Absolute Deviation was calculated. The threshold value was set to {all_params['MAD_THRESHOLD']}. " else: magic_str += "Median Absolute Deviation was not used. " if all_params['PSD_NOISE']: magic_str += f"Power Spectral Density Noise was calculated with a target frequency divisor of {all_params['TARGET_FREQ_DIV']} and a decibal limit of {all_params['DB_LIMIT']}. " else: magic_str += f"Power Spectral Density Noise was not used. " magic_str += "More coming soon." return magic_str