From 8b017005c563510d0eb7130950d033161d2c7195 Mon Sep 17 00:00:00 2001 From: tyler Date: Tue, 21 Jul 2026 00:54:51 -0700 Subject: [PATCH] pylance standardization --- changelog.md | 1 + flares.py | 275 ++++++++++++------ src/analysis/crossgroupbrainimage.py | 28 +- src/analysis/crossgroupstats.py | 74 +++-- src/analysis/exporttocsv.py | 59 ++-- src/analysis/intergroupbrainimage.py | 79 +++-- .../intergroupfunctionalconnectivity.py | 23 +- src/analysis/intergroupstats.py | 71 +++-- src/analysis/participantbrain.py | 39 ++- src/analysis/participantfoldchannels.py | 2 +- .../participantfunctionalconnectivity.py | 39 ++- src/analysis/participantimage.py | 36 ++- src/shared/flaresbasewidget.py | 85 ++++-- src/shared/shareddata.py | 23 +- src/window/about.py | 9 +- src/window/terminal.py | 23 +- src/window/updateoptodes.py | 6 +- src/window/userguide.py | 2 +- src/window/viewerlauncher.py | 2 +- src/window/welcome.py | 9 +- 20 files changed, 608 insertions(+), 277 deletions(-) diff --git a/changelog.md b/changelog.md index 2ec877e..65025ec 100644 --- a/changelog.md +++ b/changelog.md @@ -31,6 +31,7 @@ - Fixed a crucial bug where short channels were not being processed and filtered the same way as long channels before being used as regressors - Fixed a crucial bug where short channels were being presented to the design matrix as normal long channels - Fixed a crucial bug where long channels could be interpolated from short channels. Short channels are still potentially interpolated from long channels. See [this link](https://git.research.dezeeuw.ca/tyler/flares/issues/80) for more information regarding this issue. +- Decreased unnecessary processing time when fOLDing channels by an order of magnitude - Added a welcome message when the terminal is opened, resized the terminal, and added more commands diff --git a/flares.py b/flares.py index a8918ab..8801ee2 100644 --- a/flares.py +++ b/flares.py @@ -8,12 +8,13 @@ License: GPL-3.0 # Built-in imports import os +from pathlib import Path import sys import platform import threading import logging from io import BytesIO -from typing import Any, Optional, cast, Literal, Union +from typing import Any, Optional, Sequence, cast, Literal, Union from itertools import compress from copy import deepcopy from multiprocessing import Queue, Pool @@ -1765,29 +1766,6 @@ def _check_load_fold(fold_files, atlas): -def fold_channel_specificity_normal(raw, fold_files=None, atlas="Juelich", interpolate=False): - """Return the landmarks and specificity a channel is sensitive to. - - Parameters - - """ # noqa: E501 - _validate_type(raw, BaseRaw, "raw") - - reference_locations = generate_montage_locations() - - fold_tbl = _check_load_fold(fold_files, atlas) - - chan_spec = list() - for cidx in range(len(raw.ch_names)): - tbl = _source_detector_fold_table( - raw, cidx, reference_locations, fold_tbl, interpolate - ) - chan_spec.append(tbl.reset_index(drop=True)) - - return chan_spec - - - def resource_path(relative_path): """ Get absolute path to resource regardless of running directly or packaged using PyInstaller @@ -1958,35 +1936,42 @@ def resource_path(relative_path): -def fold_channels(raw: BaseRaw, p_name: str, progress_queue=None) -> dict[str, list[dict[str, Any]]]: +def fold_channels(raw: BaseRaw, p_name: str, atlas: str='Brodmann', progress_queue=None) -> dict[str, list[dict[str, Any]]]: """Runs in background process. Does only heavy math/lookup. Returns data instead of a static image. """ if getattr(sys, 'frozen', False): - set_config('MNE_NIRS_FOLD_PATH', resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary")) + fold_dir = resource_path("./mne_data/fOLD/fOLD-public-master/Supplementary") else: path = os.path.expanduser("~") + "/mne_data/fOLD/fOLD-public-master/Supplementary" - set_config('MNE_NIRS_FOLD_PATH', resource_path(path)) + fold_dir = resource_path(path) + + set_config('MNE_NIRS_FOLD_PATH', fold_dir) hbo_channel_names = cast(list[str], getattr(raw.copy().pick(picks='hbo'), "ch_names")) - # Store clean, picklable data lists instead of complex DataFrames - channel_results = {} + _validate_type(raw, BaseRaw, "raw") + reference_locations = generate_montage_locations() + + fold_tbl = _check_load_fold(fold_files=fold_dir, atlas=atlas) + + channel_results = {} step_idx = 0 - for channel_name in hbo_channel_names: - channel_data = raw.copy().pick(picks=channel_name) - output = cast(list[DataFrame], fold_channel_specificity_normal(channel_data, interpolate=True, atlas='Brodmann')) + for cidx, channel_name in enumerate(hbo_channel_names): + tbl = _source_detector_fold_table( + raw, cidx, reference_locations, fold_tbl, interpolate=True + ) channel_results[channel_name] = [] - for df_data in output: - # Extract just raw primitive types so they transfer over process channels flawlessly - for _, row in df_data.iterrows(): - channel_results[channel_name].append({ - 'Landmark': str(row['Landmark']), - 'Specificity': float(row['Specificity']) - }) + + for _, row in tbl.iterrows(): + channel_results[channel_name].append({ + 'Landmark': str(row['Landmark']), + 'Specificity': float(row['Specificity']) + }) + step_idx += 1 if progress_queue is not None: progress_queue.put((p_name, step_idx)) @@ -2180,7 +2165,7 @@ def plot_3d_evoked_array( return brain -def aggregate_fnirs_group_geometry(raw_list): +def aggregate_fnirs_group_geometry(raw_list: Sequence[BaseRaw | None]) -> BaseRaw: """ Averages fNIRS geometry across participants in two tiers: 1. Average by Channel Pairing (S_D). @@ -2246,7 +2231,15 @@ def aggregate_fnirs_group_geometry(raw_list): -def brain_3d_visualization(raw_haemo, df_cha, selected_event, t_or_theta: Literal['t', 'theta'] = 'theta', show_optodes: Literal['sensors', 'labels', 'none', 'all'] = 'all', show_text: bool = True, brain_bounds: float = 1.0) -> None: +def brain_3d_visualization( + raw_haemo: BaseRaw | None, + df_cha: DataFrame | None, + selected_event: str | None, + t_or_theta: Literal["t", "theta"] = "theta", + show_optodes: Literal["sensors", "labels", "none", "all"] = "all", + show_text: bool = True, + brain_bounds: float | tuple[float, float] | Sequence[float] = 1.0, +) -> None: clim = dict(kind="value", pos_lims=(0, brain_bounds/2, brain_bounds)) @@ -2582,7 +2575,14 @@ def plot_2d_3d_contrasts_between_groups( -def plot_fir_model_results(df, raw_haemo, dm, selected_event, l_bound, u_bound): +def plot_fir_model_results( + df: DataFrame, + raw_haemo: BaseRaw | None, + dm: DataFrame | None, + selected_event: str | None, + l_bound: float, + u_bound: float, +) -> None: df["isActivity"] = [f"{selected_event}" in n for n in df["Condition"]] @@ -2801,11 +2801,19 @@ def load_snirf(file_path: str) -> tuple[BaseRaw, Figure]: -def run_roi_second_level_analysis(df_roi_all, df_cha_all=None, raw_haemo=None, - p_threshold=0.05, min_subjects=5, - correction_method='fdr_bh', target_chroma='hbo', - graph_bounds=None, roi_config=None, - threshold_topo=False): # Added parameter +def run_roi_second_level_analysis( + df_roi_all: DataFrame, + df_cha_all: DataFrame | None = None, + raw_haemo: BaseRaw | None = None, + p_threshold: float = 0.05, + min_subjects: int = 5, + correction_method: str | None = "fdr_bh", + target_chroma: str = "hbo", + graph_bounds: float | None = None, + roi_config: str | Path | None = None, + threshold_topo: bool = False, +) -> DataFrame: + """ Perform group-level ROI analysis, prints stats to console, plots the ROI bar chart, and dynamically plots isolated channel-level group topography maps based on a JSON config. @@ -3064,13 +3072,24 @@ def clean_subject_id(path_or_id): -def run_cross_group_second_level_analysis(df_roi_all, file_paths_a, file_paths_b, - group_a_name="Group A", group_b_name="Group B", - df_cha_all=None, raw_haemo=None, - p_threshold=0.05, min_subjects=3, - correction_method='fdr_bh', target_chroma='hbo', - selected_event=None, graph_bounds=None, - roi_config=None, threshold_topo=False): +def run_cross_group_second_level_analysis( + df_roi_all: DataFrame, + file_paths_a: list[str], + file_paths_b: list[str], + group_a_name: str = "Group A", + group_b_name: str = "Group B", + df_cha_all: DataFrame | None = None, + raw_haemo: Any = None, + p_threshold: float = 0.05, + min_subjects: int = 3, + correction_method: str | None = "fdr_bh", + target_chroma: str = "hbo", + selected_event: str | None = None, + graph_bounds: tuple[float, float] | list[float] | None = None, + roi_config: Path | str | None = None, + threshold_topo: bool = False, +) -> DataFrame: + """ Perform cross-group independent statistical analyses (Group A vs Group B), renders a grouped bar chart with significance brackets, and plots a group-contrast topography map. @@ -3331,11 +3350,21 @@ def run_cross_group_second_level_analysis(df_roi_all, file_paths_a, file_paths_b -def run_cross_group_laterality_analysis(df_roi_all_a, df_roi_all_b, roi_pairs, condition, - group_a_name="Group A", group_b_name="Group B", - target_chroma='hbo', min_subjects=3, - p_threshold=0.05, correction_method=None, - roi_contra_label=None, roi_ipsi_label=None): +def run_cross_group_laterality_analysis( + df_roi_all_a: DataFrame, + df_roi_all_b: DataFrame, + roi_pairs: tuple[str, str] | None, + condition: str | None, + group_a_name: str = "Group A", + group_b_name: str = "Group B", + target_chroma: str = "hbo", + min_subjects: int = 3, + p_threshold: float = 0.05, + correction_method: str | None = None, + roi_contra_label: str | None = None, + roi_ipsi_label: str | None = None, +) -> DataFrame: + """ Compare LATERALITY between two independent groups of subjects (e.g. a control group vs. a target group), using Welch's t-test on each @@ -3582,11 +3611,20 @@ def run_cross_group_laterality_analysis(df_roi_all_a, df_roi_all_b, roi_pairs, c -def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_name, roi_json_path, - group_a_name="Group A", group_b_name="Group B", - target_chroma='hbo', min_subjects=3, - p_threshold=0.05, correction_method='fdr_bh', - weighted=True): +def run_cross_group_contrast_analysis( + df_contrasts_a: DataFrame, + df_contrasts_b: DataFrame, + contrast_name: str, + roi_json_path: str | Path | None, + group_a_name: str = "Group A", + group_b_name: str = "Group B", + target_chroma: str = "hbo", + min_subjects: int = 3, + p_threshold: float = 0.05, + correction_method: str = "fdr_bh", + weighted: bool = True, +) -> DataFrame: + """ Compare a JOINT-FIT TASK CONTRAST (e.g. '2.0_vs_3.0'), aggregated to ROI level, between two independent groups. This is the cross-group analog @@ -3665,10 +3703,10 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n if df_a_filt.empty: print(f"[ERROR] Contrast '{contrast_name}' not found anywhere in {group_a_name}'s data.") - return pd.DataFrame() + return DataFrame() if df_b_filt.empty: print(f"[ERROR] Contrast '{contrast_name}' not found anywhere in {group_b_name}'s data.") - return pd.DataFrame() + return DataFrame() roi_a = aggregate_channel_contrasts_to_roi(df_a_filt, roi_json_path, weighted=weighted) roi_b = aggregate_channel_contrasts_to_roi(df_b_filt, roi_json_path, weighted=weighted) @@ -3679,7 +3717,7 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n if roi_a.empty or roi_b.empty: print(f"[ERROR] No ROI-aggregated values produced for one or both groups " f"(check regions.json channel names against this montage).") - return pd.DataFrame() + return DataFrame() all_rois = sorted(set(roi_a['ROI'].unique()) | set(roi_b['ROI'].unique())) results = [] @@ -3798,10 +3836,17 @@ def run_cross_group_contrast_analysis(df_contrasts_a, df_contrasts_b, contrast_n -def run_roi_paired_contrast_analysis(df_roi_all, roi_pairs, condition, - target_chroma='hbo', min_subjects=5, - p_threshold=0.05, correction_method=None, - roi_a_label=None, roi_b_label=None): +def run_roi_paired_contrast_analysis( + df_roi_all: DataFrame, + roi_pairs: Sequence[tuple[str, str]] | list[list[str]], + condition: str, + target_chroma: str = 'hbo', + min_subjects: int = 5, + p_threshold: float = 0.05, + correction_method: str | None = None, + roi_a_label: str | None = None, + roi_b_label: str | None = None, +) -> DataFrame: """ Paired within-subject ROI contrast (e.g. contralateral minus ipsilateral motor ROI), as a companion to run_roi_second_level_analysis rather than a @@ -4012,7 +4057,11 @@ def run_roi_paired_contrast_analysis(df_roi_all, roi_pairs, condition, -def aggregate_channel_contrasts_to_roi(df_contrasts, roi_json_path, weighted=True): +def aggregate_channel_contrasts_to_roi( + df_contrasts: DataFrame, + roi_json_path: str | Path | None, + weighted: bool = True +) -> DataFrame: """ Combine already-computed per-channel CONTRAST results (e.g. your '2.0_vs_3.0' rows from contrasts.csv / contrast_results) into @@ -5519,13 +5568,13 @@ def process_participant(file_path, progress_callback=None): if BAD_CHANNELS_HANDLING != "None" and not FOLDING_BYP: raw, fig_dropped, fig_raw_before, bad_channels = mark_bads(raw, bad_sci, bad_snr, bad_psp, bad_coeff_var, bad_amplitude_range, bad_noise, bad_disp) if fig_dropped and fig_raw_before is not None: - fig_individual["fig2"] = fig_dropped - fig_individual["fig3"] = fig_raw_before + fig_individual["Bad Channels by Method"] = fig_dropped + fig_individual["Bad Channels Data"] = fig_raw_before if bad_channels: if BAD_CHANNELS_HANDLING == "Interpolate": raw, fig_raw_after, fig_compare = interpolate_fNIRS_bads_weighted_average(raw, max_dist=MAX_DIST, min_neighbors=MIN_NEIGHBORS) - fig_individual["fig4"] = fig_raw_after - fig_individual["Compare"] = fig_compare + fig_individual["Data after Interpolating Bad Channels"] = fig_raw_after + fig_individual["Bad Channels Interpolation Results"] = fig_compare elif BAD_CHANNELS_HANDLING == "Remove": raw = remove_bad_channels(raw, bad_channels) if progress_callback: progress_callback(13) @@ -5542,7 +5591,7 @@ def process_participant(file_path, progress_callback=None): if TDDR and not FOLDING_BYP: raw_od = temporal_derivative_distribution_repair(raw_od) fig_raw_od_tddr = raw_od.plot(duration=raw.times[-1], n_channels=raw.info['nchan'], title="After TDDR (Motion Correction)", show=False) - fig_individual["TDDR"] = fig_raw_od_tddr + fig_individual["Temporal Derivative Distribution Repair"] = fig_raw_od_tddr if progress_callback: progress_callback(15) logger.info("Step 15 Completed.") @@ -5556,7 +5605,7 @@ def process_participant(file_path, progress_callback=None): # Step 17: Haemoglobin Concentration raw_haemo = beer_lambert_law(raw_od, ppf=calculate_dpf(file_path)) fig_raw_haemo_bll = raw_haemo.plot(duration=raw_haemo.times[-1], n_channels=raw_haemo.info['nchan'], title="HbO and HbR Signals", show=False) - fig_individual["BLL"] = fig_raw_haemo_bll + fig_individual["Modified Beer Lambert Law"] = fig_raw_haemo_bll if progress_callback: progress_callback(17) logger.info("Step 17 Completed.") @@ -5564,15 +5613,15 @@ def process_participant(file_path, progress_callback=None): if ENHANCE_NEGATIVE_CORRELATION and not FOLDING_BYP: raw_haemo = enhance_negative_correlation(raw_haemo) fig_raw_haemo_enc = raw_haemo.plot(duration=raw_haemo.times[-1], n_channels=raw_haemo.info['nchan'], title="Enhance Negative Correlation", show=False) - fig_individual["ENC"] = fig_raw_haemo_enc + fig_individual["Enhance Negative Correlation"] = fig_raw_haemo_enc if progress_callback: progress_callback(18) logger.info("Step 18 Completed.") # Step 19: Filter if FILTER and not FOLDING_BYP: raw_haemo, fig_filter, fig_raw_haemo_filter = filter_the_data(raw_haemo) - fig_individual["filter1"] = fig_filter - fig_individual["filter2"] = fig_raw_haemo_filter + fig_individual["Filter_1"] = fig_filter + fig_individual["Filter_2"] = fig_raw_haemo_filter if progress_callback: progress_callback(19) logger.info("Step 19 Completed.") @@ -5580,7 +5629,7 @@ def process_participant(file_path, progress_callback=None): if not FOLDING_BYP: events, event_dict = events_from_annotations(raw_haemo) fig_events = plot_events(events, event_id=event_dict, sfreq=raw_haemo.info["sfreq"], show=False) - fig_individual["events"] = fig_events + fig_individual["Events"] = fig_events if progress_callback: progress_callback(20) logger.info("Step 20 Completed.") @@ -5652,7 +5701,11 @@ def sanitize_paths_for_pickle(raw_haemo, epochs): epochs._raw._filenames = [str(p) for p in epochs._raw._filenames] -def functional_connectivity_spectral_epochs(epochs, n_lines, vmin): +def functional_connectivity_spectral_epochs( + epochs: DataFrame | None, + n_lines: int, + vmin: float, +) -> None: # will crash without this load epochs.load_data() @@ -5691,7 +5744,11 @@ def functional_connectivity_spectral_epochs(epochs, n_lines, vmin): -def functional_connectivity_spectral_time(epochs, n_lines, vmin): +def functional_connectivity_spectral_time( + epochs: DataFrame | None, + n_lines: int, + vmin: float, +) -> None: # will crash without this load epochs.load_data() @@ -5735,7 +5792,12 @@ def functional_connectivity_spectral_time(epochs, n_lines, vmin): -def functional_connectivity_envelope(epochs, n_lines, vmin): +def functional_connectivity_envelope( + epochs: DataFrame | None, + n_lines: int, + vmin: float, +) -> None: + # will crash without this load epochs.load_data() @@ -5765,7 +5827,12 @@ def functional_connectivity_envelope(epochs, n_lines, vmin): ) -def functional_connectivity_betas(raw_hbo, n_lines, vmin, event_name=None): +def functional_connectivity_betas( + raw_hbo: BaseRaw, + n_lines: int, + vmin: float, + event_name: str | None = None, +) -> None: raw_hbo = raw_hbo.copy().pick(picks="hbo") onsets = raw_hbo.annotations.onset @@ -5966,7 +6033,15 @@ def get_single_subject_beta_corr(raw_hbo, event_name=None, config=None): return corr_matrix, raw_hbo.ch_names -def run_group_functional_connectivity(haemo_dict, config_dict, selected_paths, event_name, n_lines, vmin): +def run_group_functional_connectivity( + haemo_dict: dict[str | Path, BaseRaw], + config_dict: dict[str, Any], + selected_paths: list[str], + event_name: str | None, + n_lines: int, + vmin: float, +) -> None: + """Aggregates multiple participants and triggers the plot.""" all_z_matrices = [] common_names = None @@ -6067,4 +6142,28 @@ def run_group_functional_connectivity(haemo_dict, config_dict, selected_paths, e sig_avg_r, common_names, n_lines=n_lines, title=f"Group Connectivity: {event_name if event_name else 'All Events'}", vmin=vmin, vmax=1.0, colormap='hot' - ) \ No newline at end of file + ) + + +def sparks_csv_export( + haemo_obj: BaseRaw, + save_path: str, +) -> None: + + 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) diff --git a/src/analysis/crossgroupbrainimage.py b/src/analysis/crossgroupbrainimage.py index b8fb60b..26b461e 100644 --- a/src/analysis/crossgroupbrainimage.py +++ b/src/analysis/crossgroupbrainimage.py @@ -1,20 +1,28 @@ """ Filename: crossgroupbrainimage.py Description: Logic for the Cross-Group Brain & Image analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in Imports +from pathlib import Path +from typing import Any, cast + # External library imports +from mne.io.base import BaseRaw + import pandas as pd +from pandas import DataFrame from flares import aggregate_fnirs_group_geometry, plot_2d_3d_contrasts_between_groups from src.shared.flaresbasewidget import CrossGroupUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "show_optodes", @@ -52,7 +60,15 @@ PARAMETERIZED_INDEXES = { class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + 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], + ) -> None: + super().__init__("CrossGroupBrainImage") self.setWindowTitle(f"Cross-Group Brain & Image Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict @@ -70,8 +86,9 @@ class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget): if request is None: return - (selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = request + (selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, raw_params) = request + param_values = cast(dict[int | str, dict[str, Any]], raw_params) # Build group-level contrast DataFrames def concat_group_contrasts(file_paths: list[str], event: str | None) -> pd.DataFrame: @@ -102,8 +119,11 @@ class CrossGroupBrainImageWidget(CrossGroupUIMixin, FlaresBaseWidget): if len(all_raw_objs) > 1: processed_raw = aggregate_fnirs_group_geometry(all_raw_objs) + elif len(all_raw_objs) == 1 and all_raw_objs[0] is not None: + processed_raw = all_raw_objs[0].copy() + processed_raw.pick(picks="hbo") # type: ignore else: - processed_raw = all_raw_objs[0].copy().pick(picks="hbo") + processed_raw = None # Visualizations for idx in selected_indexes: diff --git a/src/analysis/crossgroupstats.py b/src/analysis/crossgroupstats.py index 90fe79c..2cb7b9f 100644 --- a/src/analysis/crossgroupstats.py +++ b/src/analysis/crossgroupstats.py @@ -1,20 +1,28 @@ """ Filename: crossgroupstats.py Description: Cross-Group stats analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in imports +from pathlib import Path +from typing import Any, cast + # External library imports import pandas as pd +from pandas import DataFrame + +from mne.io.base import BaseRaw from flares import run_cross_group_contrast_analysis, run_cross_group_laterality_analysis, run_cross_group_second_level_analysis from src.shared.flaresbasewidget import CrossGroupUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "p_threshold", @@ -136,7 +144,18 @@ DESCRIPTION = """0. Raw ROI Comparison (run_cross_group_second_level_analysis) class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict, json_location): + + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + 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], + json_location: str | Path + ) -> None: + super().__init__("CrossGroupStats") self.setWindowTitle(f"Cross-Group Stats Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict @@ -144,7 +163,7 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): self.df_ind_dict = df_ind_dict self.design_matrix_dict = design_matrix_dict self.contrast_results_dict = contrast_results_dict - self.group_dict = group_dict + # self.group_dict = group_dict self.json_location = json_location self.setup_cross_group_ui(["0 (Raw ROI Comparison)", "1 (Laterality Comparison)", "2 (Contrast Comparison)",], placeholder_text=DESCRIPTION) @@ -155,23 +174,18 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): if request is None: return - (selected_event, file_paths_a, file_paths_b, all_selected_paths, selected_indexes, param_values,) = request + (selected_event, file_paths_a, file_paths_b, _, selected_indexes, raw_params) = request - if isinstance(self.df_ind_dict, dict): - # Filter out empty entries and concatenate - valid_dfs = [df for df in self.df_ind_dict.values() if isinstance(df, pd.DataFrame) and not df.empty] - if valid_dfs: - df_ind_combined = pd.concat(valid_dfs, ignore_index=True) - else: - df_ind_combined = pd.DataFrame() + param_values = cast(dict[int | str, dict[str, Any]], raw_params) + + valid_dfs = [df for df in self.df_ind_dict.values() if not df.empty] + if valid_dfs: + df_ind_combined = pd.concat(valid_dfs, ignore_index=True) else: - df_ind_combined = self.df_ind_dict - - if isinstance(self.cha_dict, dict): - valid_chas = [df for df in self.cha_dict.values() if isinstance(df, pd.DataFrame) and not df.empty] - cha_combined = pd.concat(valid_chas, ignore_index=True) if valid_chas else pd.DataFrame() - else: - cha_combined = self.cha_dict + df_ind_combined = pd.DataFrame() + + valid_chas = [df for df in self.cha_dict.values() if not df.empty] + cha_combined = pd.concat(valid_chas, ignore_index=True) if valid_chas else pd.DataFrame() sample_path = file_paths_a[0] p_haemo = self.haemo_dict.get(sample_path) @@ -213,8 +227,8 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): min_subjects = params.get("min_subjects", 3) correction_method = params.get("correction_method", "None") target_chroma = params.get("target_chroma", "hbo") - roi_a = params.get("roi_a", "").strip() - roi_b = params.get("roi_b", "").strip() + roi_a: str = params.get("roi_a", "").strip() + roi_b: str = params.get("roi_b", "").strip() if not roi_a or not roi_b: print("Both a contralateral and ipsilateral ROI name must be specified.") @@ -225,14 +239,19 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): # Build each group's dataframe directly from the dict using # the file-path lists as keys - no ID cleaning/matching needed. - def _build_group_df(file_paths, dict_source): + def _build_group_df( + file_paths: list[str], + dict_source: dict[str, DataFrame] + ) -> DataFrame: + valid_dfs = [ dict_source[fp] for fp in file_paths - if fp in dict_source and isinstance(dict_source[fp], pd.DataFrame) - and not dict_source[fp].empty + if fp in dict_source and not dict_source[fp].empty ] + return pd.concat(valid_dfs, ignore_index=True) if valid_dfs else pd.DataFrame() + df_roi_a = _build_group_df(file_paths_a, self.df_ind_dict) df_roi_b = _build_group_df(file_paths_b, self.df_ind_dict) @@ -271,8 +290,13 @@ class CrossGroupStatsWidget(CrossGroupUIMixin, FlaresBaseWidget): # directly from contrast_results_dict, keyed by file path - # same dict-key approach as the laterality patch, avoids # any ID-string matching. - def _build_group_contrast_df(file_paths, contrast_dict, name): - all_rows = [] + def _build_group_contrast_df( + file_paths: list[str], + contrast_dict: dict[str, dict[str, pd.DataFrame]], + name: str, + ) -> pd.DataFrame: + + all_rows: list[DataFrame] = [] for fp in file_paths: condition_dfs = contrast_dict.get(fp) if condition_dfs is None: diff --git a/src/analysis/exporttocsv.py b/src/analysis/exporttocsv.py index e1812b6..79ba358 100644 --- a/src/analysis/exporttocsv.py +++ b/src/analysis/exporttocsv.py @@ -1,6 +1,7 @@ """ 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 @@ -8,39 +9,51 @@ License: GPL-3.0 # Built-in imports import os +from pathlib import Path +from typing import Any # External library imports -import numpy as np -import pandas as pd +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, cha_dict, df_ind, design_matrix, group, contrast_results_dict): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + 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], + ) -> None: + super().__init__("ExportToCSV") self.setWindowTitle(f"Export To CSV Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict self.cha_dict = cha_dict - self.df_ind = df_ind - self.design_matrix = design_matrix - self.group = group - self.contrast_results_dict = contrast_results_dict + # self.df_ind = df_ind_dict + # self.design_matrix = design_matrix_dict + # self.contrast_results_dict = contrast_results_dict + # self.group = group_dict self.setup_csv_ui(["0 (Export Data to CSV)", "1 (CSV for SPARKS)",]) def process_request(self): - # TODO: Move this into flares for the call? selected_display_names = self._get_checked_items(self.participant_dropdown) - selected_file_paths = [] + 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: + if display_name == expected_display: selected_file_paths.append(fp) break @@ -52,7 +65,6 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget): QMessageBox.warning(self, "Selection Missing", "Please select at least one participant and one export type.") return - # 2. ASK ONCE: Select Output Directory output_dir = QFileDialog.getExistingDirectory(self, "Select Output Folder for CSV Exports") if not output_dir: @@ -78,29 +90,11 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget): 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") - if haemo_obj is not None: - 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) - success_count += 1 + sparks_csv_export(haemo_obj, save_path) + success_count += 1 else: print(f"No method defined for index {idx}") @@ -119,5 +113,4 @@ class ExportToCSVWidget(CSVUIMixin, FlaresBaseWidget): # mode=EventUpdateMode.WRITE_JSON, # caller="Video Alignment Tool" # ) - # win.show() - + # win.show() \ No newline at end of file diff --git a/src/analysis/intergroupbrainimage.py b/src/analysis/intergroupbrainimage.py index 193a96e..0b6c567 100644 --- a/src/analysis/intergroupbrainimage.py +++ b/src/analysis/intergroupbrainimage.py @@ -1,20 +1,29 @@ """ Filename: intergroupbrainimage.py Description: Logic for the Inter-Group Brain & Image analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in Imports +from pathlib import Path +from typing import Any, cast + # External library imports import pandas as pd +from pandas import DataFrame + +from mne import Annotations +from mne.io.base import BaseRaw from flares import aggregate_fnirs_group_geometry, plot_fir_model_results, brain_3d_visualization from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME +from mne.io import BaseRaw - -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "lower_bound", @@ -74,15 +83,24 @@ PARAMETERIZED_INDEXES = { class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + 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] + ) -> None: + super().__init__("InterGroupBrainImage") self.setWindowTitle(f"Inter-Group Brain & Image Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict - self.cha = cha - self.df_ind = df_ind - self.design_matrix = design_matrix - self.contrast_results = contrast_results - self.group = group + self.cha_dict = cha_dict + self.df_ind_dict = df_ind_dict + self.design_matrix_dict = design_matrix_dict + self.contrast_results_dict = contrast_results_dict + # self.group_dict = group_dict self.setup_inter_group_ui(["0 (GLM Results)", "1 (Significance)", "2 (Brain Activity Visualization)",]) @@ -92,35 +110,45 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): if request is None: return - (selected_event, selected_file_paths, selected_indexes, param_values,) = request + (selected_event, selected_file_paths, selected_indexes, raw_params) = request + + param_values = cast(dict[int | str, dict[str, Any]], raw_params) all_cha = pd.DataFrame() for file_path in selected_file_paths: haemo_obj = self.haemo_dict.get(file_path) + if haemo_obj is None: + continue + if selected_event: - participant_events = set(haemo_obj.annotations.description) + raw_annotations = getattr(haemo_obj, "annotations", None) + + if raw_annotations is not None: + annotations = cast(Annotations, raw_annotations) + descriptions = cast(list[str], list(annotations.description)) + participant_events: set[str] = set(descriptions) + else: + participant_events: set[str] = set() + if selected_event not in participant_events: print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") continue - - if haemo_obj is None: - continue - cha_df = self.cha.get(file_path) + cha_df = self.cha_dict.get(file_path) if cha_df is not None: all_cha = pd.concat([all_cha, cha_df], ignore_index=True) # Pass the necessary arguments to each method file_path = selected_file_paths[0] p_haemo = self.haemo_dict.get(file_path) - p_design_matrix = self.design_matrix.get(file_path) + p_design_matrix = self.design_matrix_dict.get(file_path) df_group = pd.DataFrame() if selected_file_paths: for file_path in selected_file_paths: - df = self.df_ind.get(file_path) + df = self.df_ind_dict.get(file_path) if df is not None: df_group = pd.concat([df_group, df], ignore_index=True) @@ -147,9 +175,9 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): print(f"Missing parameters for index {idx}, skipping.") continue - all_contrasts = [] + all_contrasts: list[DataFrame] = [] for fp in selected_file_paths: - condition_dfs = self.contrast_results.get(fp, {}) + condition_dfs = self.contrast_results_dict.get(fp, {}) if selected_event in condition_dfs: df = condition_dfs[selected_event].copy() df["ID"] = fp @@ -159,7 +187,8 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): print("No contrast data found for selected participants and event.") return - df_contrasts = pd.concat(all_contrasts, ignore_index=True) + # TODO: look at intergroupstats and figure out what to do + _ = pd.concat(all_contrasts, ignore_index=True) #flares.run_second_level_analysis(df_contrasts, p_haemo, p_val, graph_bounds) elif idx == 2: @@ -173,13 +202,15 @@ class InterGroupBrainImageWidget(InterGroupUIMixin, FlaresBaseWidget): print(f"Missing parameters for index {idx}, skipping.") continue - raw_list = [self.haemo_dict.get(fp) for fp in selected_file_paths] + all_raw_objs = [self.haemo_dict.get(fp) for fp in selected_file_paths if self.haemo_dict.get(fp)] - if len(selected_file_paths) > 1: - print(f"Aggregating geometry for {len(selected_file_paths)} participants...") - processed_raw = aggregate_fnirs_group_geometry(raw_list) + if len(all_raw_objs) > 1: + processed_raw = aggregate_fnirs_group_geometry(all_raw_objs) + elif len(all_raw_objs) == 1 and all_raw_objs[0] is not None: + processed_raw = all_raw_objs[0].copy() + processed_raw.pick(picks="hbo") # type: ignore else: - processed_raw = raw_list[0].copy().pick(picks="hbo") + processed_raw = None brain_3d_visualization(processed_raw, all_cha, selected_event, t_or_theta=t_or_theta, show_optodes=show_optodes, show_text=show_text, brain_bounds=brain_bounds) diff --git a/src/analysis/intergroupfunctionalconnectivity.py b/src/analysis/intergroupfunctionalconnectivity.py index 3ba87d1..1567e03 100644 --- a/src/analysis/intergroupfunctionalconnectivity.py +++ b/src/analysis/intergroupfunctionalconnectivity.py @@ -1,20 +1,27 @@ """ Filename: intergroupfunctionalconnectivity.py Description: Logic for the Inter-Group Functional Connectivity analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in imports +from pathlib import Path +from typing import Any, cast + # External library imports from PySide6.QtWidgets import QMessageBox +from mne.io.base import BaseRaw + from flares import run_group_functional_connectivity from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "n_lines", @@ -34,11 +41,17 @@ PARAMETERIZED_INDEXES = { class InterGroupFunctionalConnectivityWidget(InterGroupUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, group, config_dict): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + group_dict: dict[str, str], + config_dict: dict[str, str] + ) -> None: + super().__init__("InterGroupFunctionalConnectivity") self.setWindowTitle(f"Inter-Group Functional Connectivity Viewer [BETA] - {APP_NAME.upper()}") self.haemo_dict = haemo_dict - self.group = group + #self.group_dict = group_dict self.config_dict = config_dict QMessageBox.warning(self, f"Warning - {APP_NAME.upper()}", f"Functional Connectivity is still in development and the results should currently be taken with a grain of salt. " @@ -52,7 +65,9 @@ class InterGroupFunctionalConnectivityWidget(InterGroupUIMixin, FlaresBaseWidget if request is None: return - (selected_event, selected_file_paths, selected_indexes, param_values,) = request + (selected_event, selected_file_paths, selected_indexes, raw_params) = request + + param_values = cast(dict[int | str, dict[str, Any]], raw_params) for idx in selected_indexes: if idx == 0: diff --git a/src/analysis/intergroupstats.py b/src/analysis/intergroupstats.py index 96acec0..6f101c5 100644 --- a/src/analysis/intergroupstats.py +++ b/src/analysis/intergroupstats.py @@ -1,20 +1,29 @@ """ Filename: intergroupstats.py Description: Logic for the Inter-Group Stats analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in imports +from pathlib import Path +from typing import Any, cast + # External library imports import pandas as pd +from pandas import DataFrame + +from mne import Annotations +from mne.io.base import BaseRaw from flares import run_roi_paired_contrast_analysis, run_roi_second_level_analysis, aggregate_channel_contrasts_to_roi from src.shared.flaresbasewidget import InterGroupUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "p_threshold", @@ -148,40 +157,64 @@ DESCRIPTION = """0. ROI vs. Zero (run_roi_second_level_analysis) class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, cha, df_ind, design_matrix, contrast_results, group, json_location): + + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + 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], + json_location: str | Path + ) -> None: + super().__init__("InterGroupStats") self.setWindowTitle(f"Inter-Group Stats Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict - self.cha = cha - self.df_ind = df_ind - self.design_matrix = design_matrix - self.contrast_results = contrast_results - self.group = group + self.cha_dict = cha_dict + self.df_ind_dict = df_ind_dict + self.design_matrix_dict = design_matrix_dict + self.contrast_results_dict = contrast_results_dict + self.group_dict = group_dict self.json_location = json_location self.setup_inter_group_ui(["0 (ROI vs. Zero)", "1 (Paired ROI Contrast)", "2 (Joint Contrast, ROI-Aggregated)"], placeholder_text=DESCRIPTION) + def process_request(self): - request = self.get_common_request_data(PARAMETERIZED_INDEXES, self.json_location, self.contrast_results) + request = self.get_common_request_data(PARAMETERIZED_INDEXES, self.json_location, self.contrast_results_dict) if request is None: return - (selected_event, selected_file_paths, selected_indexes, param_values,) = request + (selected_event, selected_file_paths, selected_indexes, raw_params) = request - all_cha = pd.DataFrame() + param_values = cast(dict[int | str, dict[str, Any]], raw_params) + + all_cha = DataFrame() for file_path in selected_file_paths: haemo_obj = self.haemo_dict.get(file_path) + if haemo_obj is None: + continue + if selected_event: - participant_events = set(haemo_obj.annotations.description) + raw_annotations = getattr(haemo_obj, "annotations", None) + + if raw_annotations is not None: + annotations = cast(Annotations, raw_annotations) + descriptions = cast(list[str], list(annotations.description)) + participant_events: set[str] = set(descriptions) + else: + participant_events: set[str] = set() + if selected_event not in participant_events: print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") continue - if haemo_obj is None: - continue + - cha_df = self.cha.get(file_path) + cha_df = self.cha_dict.get(file_path) if cha_df is not None: all_cha = pd.concat([all_cha, cha_df], ignore_index=True) @@ -189,10 +222,10 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget): p_haemo = self.haemo_dict.get(file_path) # Concatenate individual ROI stats (df_ind) for all chosen subjects - df_group = pd.DataFrame() + df_group = DataFrame() if selected_file_paths: for file_path in selected_file_paths: - df = self.df_ind.get(file_path) + df = self.df_ind_dict.get(file_path) if df is not None: df_group = pd.concat([df_group, df], ignore_index=True) @@ -226,7 +259,7 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget): print(f"No ROI data matches the condition '{selected_event}'.") continue - all_cha_filtered = pd.DataFrame() + all_cha_filtered = DataFrame() if not all_cha.empty: if selected_event and 'Condition' in all_cha.columns: all_cha_filtered = all_cha[all_cha['Condition'] == selected_event] @@ -304,9 +337,9 @@ class InterGroupStatsWidget(InterGroupUIMixin, FlaresBaseWidget): continue - all_contrasts = [] + all_contrasts: list[DataFrame] = [] for fp in selected_file_paths: - condition_dfs = self.contrast_results.get(fp) + condition_dfs = self.contrast_results_dict.get(fp) if condition_dfs is None: print(f" [MISSING] '{fp}' not found in contrast_results.") continue diff --git a/src/analysis/participantbrain.py b/src/analysis/participantbrain.py index 8682cbb..9312a37 100644 --- a/src/analysis/participantbrain.py +++ b/src/analysis/participantbrain.py @@ -1,18 +1,28 @@ """ Filename: participantbrain.py Description: Logic for the Participant Brain analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in imports +from pathlib import Path +from typing import Any, cast + # External library imports +from mne import Annotations +from pandas import DataFrame + +from mne.io.base import BaseRaw + from flares import brain_3d_visualization, brain_landmarks_3d from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "show_optodes", @@ -57,7 +67,12 @@ PARAMETERIZED_INDEXES = { class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, cha_dict): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + cha_dict: dict[str, DataFrame], + ) -> None: + super().__init__("ParticipantBrain") self.setWindowTitle(f"Participant Brain Viewer - {APP_NAME.upper()}") self.haemo_dict = haemo_dict @@ -72,21 +87,31 @@ class ParticipantBrainViewerWidget(ParticipantUIMixin, FlaresBaseWidget): if request is None: return - (selected_event, selected_file_paths, selected_indexes, param_values,) = request + (selected_event, selected_file_paths, selected_indexes, raw_params) = request + + param_values = cast(dict[int | str, dict[str, Any]], raw_params) # 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 + if selected_event: - participant_events = set(haemo_obj.annotations.description) + raw_annotations = getattr(haemo_obj, "annotations", None) + + if raw_annotations is not None: + annotations = cast(Annotations, raw_annotations) + descriptions = cast(list[str], list(annotations.description)) + participant_events: set[str] = set(descriptions) + else: + participant_events: set[str] = set() + if selected_event not in participant_events: print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") continue - if haemo_obj is None: - raise Exception("How did we get here?") - cha = self.cha_dict.get(file_path) for idx in selected_indexes: diff --git a/src/analysis/participantfoldchannels.py b/src/analysis/participantfoldchannels.py index e768d52..eb34165 100644 --- a/src/analysis/participantfoldchannels.py +++ b/src/analysis/participantfoldchannels.py @@ -71,7 +71,7 @@ def single_participant_worker(file_path, raw_data, result_queue, progress_queue) try: from flares import fold_channels # Perform the heavy fold_channels logic - channel_results = fold_channels(raw_data, p_name, progress_queue) + channel_results = fold_channels(raw=raw_data, p_name=p_name, progress_queue=progress_queue) # Hand back results and signal completion result_queue.put({file_path: channel_results}) diff --git a/src/analysis/participantfunctionalconnectivity.py b/src/analysis/participantfunctionalconnectivity.py index 02e50b3..8313387 100644 --- a/src/analysis/participantfunctionalconnectivity.py +++ b/src/analysis/participantfunctionalconnectivity.py @@ -1,20 +1,30 @@ """ Filename: participantfunctionalconnectivity.py Description: Logic for the Participant Functional Connectivity analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +# Built-in Imports +from pathlib import Path +from typing import Any, cast + # External library imports from PySide6.QtWidgets import QMessageBox +from pandas import DataFrame + +from mne import Annotations +from mne.io.base import BaseRaw + from flares import functional_connectivity_betas, functional_connectivity_envelope, functional_connectivity_spectral_epochs, functional_connectivity_spectral_time from src.shared.flaresbasewidget import ParticipantUIMixin, FlaresBaseWidget from src.shared.shareddata import APP_NAME -PARAMETERIZED_INDEXES = { +PARAMETERIZED_INDEXES: dict[int, list[dict[str, Any]]] = { 0: [ { "key": "n_lines", @@ -79,7 +89,12 @@ PARAMETERIZED_INDEXES = { class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidget): - def __init__(self, haemo_dict, epochs_dict): + def __init__( + self, + haemo_dict: dict[str | Path, BaseRaw], + epochs_dict: dict[str, DataFrame], + ) -> None: + super().__init__("ParticipantFunctionalConnectivity") self.setWindowTitle(f"Participant Functional Connectivity Viewer [BETA] - {APP_NAME.upper()}") self.haemo_dict = haemo_dict @@ -97,22 +112,32 @@ class ParticipantFunctionalConnectivityWidget(ParticipantUIMixin, FlaresBaseWidg if request is None: return - (selected_event, selected_file_paths, selected_indexes, param_values,) = request + (selected_event, selected_file_paths, selected_indexes, raw_params) = request + + param_values = cast(dict[int | str, dict[str, Any]], raw_params) # Pass the necessary arguments to each method for file_path in selected_file_paths: haemo_obj = self.haemo_dict.get(file_path) epochs_obj = self.epochs_dict.get(file_path) + if haemo_obj is None: + continue + if selected_event: - participant_events = set(haemo_obj.annotations.description) + raw_annotations = getattr(haemo_obj, "annotations", None) + + if raw_annotations is not None: + annotations = cast(Annotations, raw_annotations) + descriptions = cast(list[str], list(annotations.description)) + participant_events: set[str] = set(descriptions) + else: + participant_events: set[str] = set() + if selected_event not in participant_events: print(f"Skipping {self.participant_map[file_path]}: Event '{selected_event}' not found.") continue - if haemo_obj is None: - raise Exception("How did we get here?") - for idx in selected_indexes: if idx == 0: diff --git a/src/analysis/participantimage.py b/src/analysis/participantimage.py index adc6aeb..682d80d 100644 --- a/src/analysis/participantimage.py +++ b/src/analysis/participantimage.py @@ -1,17 +1,20 @@ """ Filename: participantimage.py Description: Logic for the Participant Image analysis window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in Imports -import os +import os.path as op from pathlib import Path from datetime import datetime # External library imports +from mne.io.base import BaseRaw + from PySide6.QtWidgets import QGridLayout, QHBoxLayout, QMessageBox, QPushButton, QScrollArea, QWidget, QVBoxLayout, QLabel from PySide6.QtCore import Qt, QSize from PySide6.QtGui import QPixmap @@ -21,7 +24,13 @@ from src.shared.shareddata import APP_NAME class ParticipantImageViewerWidget(FlaresBaseWidget): - def __init__(self, haemo_dict, fig_bytes_dict): + + def __init__( + self, + haemo_dict: dict[str, BaseRaw], + fig_bytes_dict: dict[str, dict[str, bytes]] + ) -> None: + super().__init__("ParticipantImage") self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose) self.setWindowTitle(f"Participant Image Viewer - {APP_NAME.upper()}") @@ -29,12 +38,12 @@ class ParticipantImageViewerWidget(FlaresBaseWidget): self.fig_bytes_dict = fig_bytes_dict # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" - self.participant_dropdown_items = [] # "Participant 1 (filename)" + self.participant_map: dict[str, str] = {} + self.participant_dropdown_items: list[str] = [] for i, file_path in enumerate(self.haemo_dict.keys(), start=1): short_label = f"Participant {i}" - display_label = f"{short_label} ({os.path.basename(file_path)})" + display_label = f"{short_label} ({op.basename(file_path)})" self.participant_map[file_path] = short_label self.participant_dropdown_items.append(display_label) @@ -87,23 +96,24 @@ class ParticipantImageViewerWidget(FlaresBaseWidget): selected_display_names = self._get_checked_items(self.participant_dropdown) # Map from display names back to file paths - selected_file_paths = [] + selected_file_paths: list[str] = [] for display_name in selected_display_names: # Find file_path by matching display name for fp, short_label in self.participant_map.items(): - expected_display = f"{short_label} ({os.path.basename(fp)})" + expected_display = f"{short_label} ({Path(fp).name})" if display_name == expected_display: - selected_file_paths.append(fp) + selected_file_paths.append(str(fp)) break selected_labels = self._get_checked_items(self.image_index_dropdown) row, col = 0, 0 for file_path in selected_file_paths: - fig_list = self.fig_bytes_dict.get(file_path, []) - participant_label = self.participant_map[file_path] + fig_map: dict[str, bytes] = self.fig_bytes_dict.get(file_path, {}) + participant_label: str = self.participant_map.get(file_path, "Unknown") + for label in selected_labels: - fig_bytes = fig_list.get(label) + fig_bytes: bytes | None = fig_map.get(label) if not fig_bytes: continue @@ -149,7 +159,7 @@ class ParticipantImageViewerWidget(FlaresBaseWidget): for display_name in selected_display_names: # Match display name to file path for file_path, short_label in self.participant_map.items(): - expected_display = f"{short_label} ({os.path.basename(file_path)})" + expected_display = f"{short_label} ({op.basename(file_path)})" if display_name == expected_display: fig_dict = self.fig_bytes_dict.get(file_path, {}) for label in selected_image_labels: @@ -157,7 +167,7 @@ class ParticipantImageViewerWidget(FlaresBaseWidget): continue fig_bytes = fig_dict[label] timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"{os.path.basename(file_path)}_{label}_{timestamp}.png" + filename = f"{op.basename(file_path)}_{label}_{timestamp}.png" output_path = save_dir / filename with open(output_path, "wb") as f: f.write(fig_bytes) diff --git a/src/shared/flaresbasewidget.py b/src/shared/flaresbasewidget.py index b46da8c..67f6fbc 100644 --- a/src/shared/flaresbasewidget.py +++ b/src/shared/flaresbasewidget.py @@ -9,6 +9,8 @@ License: GPL-3.0 import os import json +from pathlib import Path +from typing import Sequence, Any from PySide6.QtWidgets import QApplication, QComboBox, QDialog, QGridLayout, QHBoxLayout, QLabel, QLineEdit, QListView, QMessageBox, QPushButton, QScrollArea, QVBoxLayout, QWidget, QFrame, QSpinBox from PySide6.QtGui import QStandardItemModel, QStandardItem, QPixmap, QIntValidator, QDoubleValidator from PySide6.QtCore import QEvent, QSize, Qt @@ -807,7 +809,11 @@ class FlaresBaseWidget(QWidget): self.image_index_dropdown = None - def _create_multiselect_dropdown(self, items): + def _create_multiselect_dropdown( + self, + items: Sequence[str] + ) -> FullClickComboBox: + combo = FullClickComboBox() combo.setView(QListView()) model = QStandardItemModel() @@ -874,7 +880,11 @@ class FlaresBaseWidget(QWidget): # checked.append(item.text()) # return checked - def _get_checked_items(self, combo=None): + def _get_checked_items( + self, + combo: QComboBox | None = None + ) -> list[str]: + target = combo if combo is not None else getattr(self, 'participant_dropdown', None) if target is None or target.model() is None: @@ -897,7 +907,10 @@ class FlaresBaseWidget(QWidget): return checked_items - def update_participant_dropdown_label(self, combo=None): + def update_participant_dropdown_label( + self, + combo: QComboBox | int | None = None +) -> None: """ Handles label updates for ANY participant dropdown. If 'combo' is None, it defaults to the standard self.participant_dropdown. @@ -1142,7 +1155,13 @@ class FlaresBaseWidget(QWidget): class CrossGroupUIMixin: - def setup_cross_group_ui(self, index_texts, placeholder_text=""): + participant_map: dict[str, str] + + def setup_cross_group_ui( + self, + index_texts: Sequence[str], + placeholder_text: str = "" + ) -> None: self.group_to_paths = {} for file_path, group_name in self.group_dict.items(): @@ -1293,7 +1312,13 @@ class CrossGroupUIMixin: return file_paths - def get_common_request_data(self, parameterized_indexes, json_location=None, contrast_dfs=None): + def get_common_request_data( + self, + parameterized_indexes: dict[int, list[dict[str, Any]]], + json_location: str | Path | None = None, + contrast_dfs: dict[str, dict[str, Any]] | None = None, + ) -> tuple[str | None, list[str], list[str], list[str], list[int], dict[str, Any]] | None: + selected_event = self.event_dropdown.currentText() if selected_event == "": selected_event = None @@ -1412,11 +1437,14 @@ class CrossGroupUIMixin: class CSVUIMixin: - def setup_csv_ui(self, index_texts): - + def setup_csv_ui( + self, + index_texts: Sequence[str] + ) -> None: + # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" - self.participant_dropdown_items = [] # "Participant 1 (filename)" + self.participant_map: dict[str, str] = {} # file_path -> "Participant 1" + self.participant_dropdown_items: list[str] = [] # "Participant 1 (filename)" for i, file_path in enumerate(self.haemo_dict.keys(), start=1): short_label = f"Participant {i}" @@ -1428,12 +1456,12 @@ class CSVUIMixin: self.top_bar = QHBoxLayout() self.layout.addLayout(self.top_bar) - self.participant_dropdown = self._create_multiselect_dropdown(self.participant_dropdown_items) + self.participant_dropdown: FullClickComboBox = self._create_multiselect_dropdown(self.participant_dropdown_items) self.participant_dropdown.currentIndexChanged.connect(self.update_participant_dropdown_label) self.index_texts = index_texts - self.image_index_dropdown = self._create_multiselect_dropdown(self.index_texts) + self.image_index_dropdown: FullClickComboBox = self._create_multiselect_dropdown(self.index_texts) self.image_index_dropdown.currentIndexChanged.connect(self.update_image_index_dropdown_label) self.submit_button = QPushButton("Submit") @@ -1456,13 +1484,20 @@ class CSVUIMixin: self.showMaximized() + class InterGroupUIMixin: - def setup_inter_group_ui(self, index_texts, placeholder_text=""): + + def setup_inter_group_ui( + self, + index_texts: Sequence[str], + placeholder_text: str = "" + ) -> None: + self.show_all_events = True self._updating_checkstates = False # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" + self.participant_map: dict[str, str] = {} # file_path -> "Participant 1" self.participant_dropdown_items = [] # "Participant 1 (filename)" for i, file_path in enumerate(self.haemo_dict.keys(), start=1): @@ -1525,7 +1560,13 @@ class InterGroupUIMixin: self.thumb_size = QSize(280, 180) self.showMaximized() - def get_common_request_data(self, parameterized_indexes, json_location=None, contrast_dfs=None): + def get_common_request_data( + self, + parameterized_indexes: dict[int, list[dict[str, Any]]], + json_location: str | Path | None = None, + contrast_dfs: dict[str, dict[str, Any]] | None = None, + ) -> tuple[str | None, list[str], list[int], dict[str, Any]] | None: + selected_event = self.event_dropdown.currentText() if selected_event == "": selected_event = None @@ -1570,7 +1611,7 @@ class InterGroupUIMixin: dynamic_rois = [] # 1. Check for the JSON file and parse ROI names - if os.path.exists(json_location): + if json_location is not None and os.path.exists(json_location): try: with open(json_location, 'r', encoding='utf-8') as f: regions_data = json.load(f) @@ -1645,9 +1686,13 @@ class InterGroupUIMixin: ) class ParticipantUIMixin: - def setup_participant_ui(self, index_texts): + def setup_participant_ui( + self, + index_texts: Sequence[str] + ) -> None: + # Create mappings: file_path -> participant label and dropdown display text - self.participant_map = {} # file_path -> "Participant 1" + self.participant_map: dict[str, str] = {} # file_path -> "Participant 1" self.participant_dropdown_items = [] # "Participant 1 (filename)" for i, file_path in enumerate(self.haemo_dict.keys(), start=1): @@ -1694,7 +1739,11 @@ class ParticipantUIMixin: self.showMaximized() - def get_common_request_data(self, parameterized_indexes): + def get_common_request_data( + self, + parameterized_indexes: dict[int, list[dict[str, Any]]] + ) -> tuple[str | None, list[str], list[int], dict[str, Any]] | None: + selected_event = self.event_dropdown.currentText() if selected_event == "": selected_event = None diff --git a/src/shared/shareddata.py b/src/shared/shareddata.py index b00086d..6c7d1ad 100644 --- a/src/shared/shareddata.py +++ b/src/shared/shareddata.py @@ -1,22 +1,27 @@ """ Filename: shareddata.py -Description: Shared constants and methods for FLARES +Description: Shared constants and methods other files depend on +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ -import sys +# Built-in imports import os +import sys import platform + CURRENT_VERSION = "1.5.0" APP_NAME = "flares" +APP_NAME_EXPANDED = "fNIRS Lightweight Analysis, Research, & Evaluation Suite" API_URL = f"https://git.research.dezeeuw.ca/api/v1/repos/tyler/{APP_NAME}/releases" API_URL_SECONDARY = f"https://git.research2.dezeeuw.ca/api/v1/repos/tyler/{APP_NAME}/releases" PLATFORM_NAME = platform.system().lower() -CHANGELOG_URL = "https://git.research.dezeeuw.ca/tyler/flares/raw/branch/main/changelog_major.md" -WIKI_URL = "https://git.research.dezeeuw.ca/tyler/flares/wiki" +CHANGELOG_URL = f"https://git.research.dezeeuw.ca/tyler/{APP_NAME}/raw/branch/main/changelog_major.md" +WIKI_URL = f"https://git.research.dezeeuw.ca/tyler/{APP_NAME}/wiki" + PIPELINE_STAGES = [ "Preprocessing", @@ -49,15 +54,11 @@ PIPELINE_STAGES = [ "Finishing Up" ] -def resource_path(relative_path): + +def resource_path(relative_path: str) -> str: """ Get absolute path to resource regardless of running directly or packaged using PyInstaller """ - if hasattr(sys, '_MEIPASS'): - # PyInstaller bundle path - base_path = sys._MEIPASS - else: - base_path = os.path.abspath(".") - + base_path = getattr(sys, "_MEIPASS", os.path.abspath(".")) return os.path.join(base_path, relative_path) \ No newline at end of file diff --git a/src/window/about.py b/src/window/about.py index 5a34af2..8c1fad6 100644 --- a/src/window/about.py +++ b/src/window/about.py @@ -1,6 +1,7 @@ """ Filename: about.py -Description: About window for FLARES +Description: About window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 @@ -9,7 +10,7 @@ License: GPL-3.0 from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel from PySide6.QtCore import Qt -from src.shared.shareddata import APP_NAME, CURRENT_VERSION +from src.shared.shareddata import APP_NAME, APP_NAME_EXPANDED, CURRENT_VERSION class AboutWindow(QWidget): """ @@ -19,14 +20,14 @@ class AboutWindow(QWidget): parent (QWidget, optional): Parent widget of this window. Defaults to None. """ - def __init__(self, parent=None): + def __init__(self, parent: QWidget | None = None) -> None: super().__init__(parent, Qt.WindowType.Window) self.setWindowTitle(f"About {APP_NAME.upper()}") self.resize(250, 100) layout = QVBoxLayout() label = QLabel(f"About {APP_NAME.upper()}", self) - label2 = QLabel("fNIRS Lightweight Analysis, Research, & Evaluation Suite", self) + label2 = QLabel(f"{APP_NAME_EXPANDED}", self) label3 = QLabel(f"{APP_NAME.upper()} is licensed under the GPL-3.0 licence. For more information, visit https://www.gnu.org/licenses/gpl-3.0.en.html", self) label4 = QLabel(f"Version v{CURRENT_VERSION}") diff --git a/src/window/terminal.py b/src/window/terminal.py index a8f8697..9bdf8e3 100644 --- a/src/window/terminal.py +++ b/src/window/terminal.py @@ -1,21 +1,24 @@ """ Filename: terminal.py -Description: Terminal window for FLARES +Description: Terminal window +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ +from typing import Any, Callable + from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit from PySide6.QtCore import Qt from src.shared.shareddata import API_URL, API_URL_SECONDARY, APP_NAME, CURRENT_VERSION, PLATFORM_NAME from src.window.about import AboutWindow -from updater import LocalPendingUpdateCheckThread, UpdateManager +from updater import UpdateManager class TerminalWindow(QWidget): - def __init__(self, parent=None): + def __init__(self, parent: QWidget | None = None) -> None: super().__init__(parent, Qt.WindowType.Window) self.setWindowTitle(f"Terminal - {APP_NAME.upper()}") self.resize(320, 180) @@ -30,7 +33,7 @@ class TerminalWindow(QWidget): layout.addWidget(self.input_line) self.setLayout(layout) - self.commands = { + self.commands: dict[str, Callable[..., Any]] = { "hello": self.cmd_hello, "help": self.cmd_help, "version": self.cmd_version, @@ -68,22 +71,22 @@ class TerminalWindow(QWidget): self.output_area.append(f"[Unknown command] '{command_name}'") - def cmd_hello(self, *args): + def cmd_hello(self, *args: Any) -> str: return "Hello from the terminal!" - def cmd_help(self, *args): + def cmd_help(self, *args: Any) -> str: return f"Available commands: {', '.join(self.commands.keys())}" - def cmd_version(self, *args): + def cmd_version(self, *args: Any) -> str: return f"{APP_NAME.upper()} is running version {CURRENT_VERSION}." - def cmd_about(self, *args): + def cmd_about(self, *args: Any) -> None: self.about = AboutWindow(self) self.about.show() - def cmd_update(self, *args): + def cmd_update(self, *args: Any) -> str: main_win = self.parent() - if main_win is None: + if not isinstance(main_win, QWidget): return "[Error] Main window context not found." self.updater = UpdateManager( diff --git a/src/window/updateoptodes.py b/src/window/updateoptodes.py index 8710f9c..30e421d 100644 --- a/src/window/updateoptodes.py +++ b/src/window/updateoptodes.py @@ -15,9 +15,9 @@ import numpy as np from PySide6.QtWidgets import QWidget, QVBoxLayout, QLabel, QHBoxLayout, QMessageBox, QLineEdit, QPushButton, QFileDialog from PySide6.QtCore import Qt -from mne.io import read_raw_snirf -from mne_nirs.io import write_raw_snirf -from mne.channels import make_dig_montage +from mne.io import read_raw_snirf #type: ignore +from mne_nirs.io import write_raw_snirf #type: ignore +from mne.channels import make_dig_montage #type: ignore from src.shared.shareddata import APP_NAME diff --git a/src/window/userguide.py b/src/window/userguide.py index 4a7c522..4c258aa 100644 --- a/src/window/userguide.py +++ b/src/window/userguide.py @@ -20,7 +20,7 @@ class UserGuideWindow(QWidget): parent (QWidget, optional): Parent widget of this window. Defaults to None. """ - def __init__(self, parent=None): + def __init__(self, parent: QWidget | None = None) -> None: super().__init__(parent, Qt.WindowType.Window) self.setWindowTitle(f"User Guide - {APP_NAME.upper()}") self.resize(250, 100) diff --git a/src/window/viewerlauncher.py b/src/window/viewerlauncher.py index baa55e2..bc9a0c7 100644 --- a/src/window/viewerlauncher.py +++ b/src/window/viewerlauncher.py @@ -40,7 +40,7 @@ class ViewerLauncherWidget(QWidget): ("Cross-Group Stats Viewer", CrossGroupStatsWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict, json_location], True), ("Inter-Group Brain and Image Viewer", InterGroupBrainImageWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), ("Cross-Group Brain and Image Viewer", CrossGroupBrainImageWidget, [haemo_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True), - ("Export To CSV Viewer", ExportToCSVWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, group_dict, contrast_results_dict], True) + ("Export To CSV Viewer", ExportToCSVWidget, [haemo_dict, cha_dict, df_ind_dict, design_matrix_dict, contrast_results_dict, group_dict], True) ] layout = QVBoxLayout(self) diff --git a/src/window/welcome.py b/src/window/welcome.py index 8f770a9..f3d28ac 100644 --- a/src/window/welcome.py +++ b/src/window/welcome.py @@ -1,6 +1,7 @@ """ Filename: welcome.py Description: Welcome dialog for FLARES +Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 @@ -9,13 +10,13 @@ License: GPL-3.0 from PySide6.QtWidgets import QTextBrowser, QVBoxLayout, QLabel, QDialog, QHBoxLayout, QPushButton from PySide6.QtGui import QDesktopServices, QIcon from PySide6.QtCore import QUrl -from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest +from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkReply from src.shared.shareddata import APP_NAME, CURRENT_VERSION, CHANGELOG_URL, resource_path class WelcomeDialog(QDialog): - def __init__(self, parent=None, direct=True, first=False): + def __init__(self, parent: QDialog | None = None, direct: bool = True, first: bool = False): super().__init__(parent) self.setWindowTitle(f"What's New - {APP_NAME.upper()}") self.setMinimumSize(550, 450) @@ -64,10 +65,10 @@ class WelcomeDialog(QDialog): self.network_manager.get(QNetworkRequest(QUrl(CHANGELOG_URL))) - def _on_download_complete(self, reply): + def _on_download_complete(self, reply: QNetworkReply) -> None: """Processes the downloaded markdown and drops it into the view frame.""" if reply.error() == reply.NetworkError.NoError: - raw_bytes = reply.readAll() + raw_bytes = reply.readAll().data() # Convert raw bytes to standard text string markdown_text = str(raw_bytes, encoding='utf-8')