WindowAnalysisResult#

class neurodent.visualization.WindowAnalysisResult(result: DataFrame, animal_id: str | None = None, genotype: str | None = None, channel_names: list[str] | None = None, assume_from_number=False, bad_channels_dict: dict[str, list[str]] = {}, suppress_short_interval_error=False, lof_scores_dict: dict[str, dict] = {}) → None[source]#

Bases: AnimalFeatureParser

Wrapper for output of windowed analysis. Has useful functions like group-wise and global averaging, filtering, and saving

Parameters:

  • result (DataFrame)

  • animal_id (str)

  • genotype (str)

  • channel_names (list[str])

  • bad_channels_dict (dict[str, list[str]])

  • lof_scores_dict (dict[str, dict])

__init__(result: DataFrame, animal_id: str | None = None, genotype: str | None = None, channel_names: list[str] | None = None, assume_from_number=False, bad_channels_dict: dict[str, list[str]] = {}, suppress_short_interval_error=False, lof_scores_dict: dict[str, dict] = {}) → None[source]#
Parameters:

  • result (pd.DataFrame) – Result comes from AnimalOrganizer.compute_windowed_analysis()

  • animal_id (str, optional) – Identifier for the animal where result was computed from. Defaults to None.

  • genotype (str, optional) – Genotype of animal. Defaults to None.

  • channel_names (list[str], optional) – List of channel names. Defaults to None.

  • assume_channels (bool, optional) – If true, assumes channel names according to AnimalFeatureParser.DEFAULT_CHNUM_TO_NAME. Defaults to False.

  • bad_channels_dict (dict[str, list[str]], optional) – Dictionary of channels to reject for each recording session. Defaults to {}.

  • suppress_short_interval_error (bool, optional) – If True, suppress ValueError for short intervals between timestamps. Useful for aggregated WARs with large window sizes. Defaults to False.

  • lof_scores_dict (dict[str, dict])

Return type:

None

copy()[source]#

Create a deep copy of the WindowAnalysisResult object.

Returns:

A deep copy of the current instance with all attributes copied.

Return type:

WindowAnalysisResult

reorder_and_pad_channels(target_channels: list[str], use_abbrevs: bool = True, inplace: bool = True) → DataFrame[source]#

Reorder and pad channels to match a target channel list.

This method ensures that the data has a consistent channel order and structure by reordering existing channels and padding missing channels with NaNs.

Parameters:

  • target_channels (list[str]) – List of target channel names to match

  • use_abbrevs (bool, optional) – If True, target channel names are read as channel abbreviations instead of channel names. Defaults to True.

  • inplace (bool, optional) – If True, modify the result in place. Defaults to True.

Returns:

DataFrame with reordered and padded channels

Return type:

pd.DataFrame

read_sars_spikes(sars: list[SpikeAnalysisResult | FrequencyDomainSpikeAnalysisResult], read_mode: Literal['sa', 'mne'] = 'sa', inplace=True)[source]#

Integrate spike analysis results into WAR by adding nspike/lognspike features.

This method extracts spike timing information from spike detection results and bins them according to the WAR’s time windows, adding spike count features to each row.

Parameters:

  • sars (list[SpikeAnalysisResult | FrequencyDomainSpikeAnalysisResult]) – List of SpikeAnalysisResult or FrequencyDomainSpikeAnalysisResult objects. One result per recording session (animalday).

  • read_mode (Literal['sa', 'mne'] (default: 'sa')) – Mode for extracting spike data: - “sa”: Read from SortingAnalyzer objects (result_sas attribute) - “mne”: Read from MNE RawArray objects (result_mne attribute)

  • inplace (default: True) – If True, modifies self.result and returns self. If False, returns a new WindowAnalysisResult.

Returns:

WAR object with added spike features (nspike, lognspike).

  • If inplace=True: returns self with modified result DataFrame

  • If inplace=False: returns new WAR object with enhanced result DataFrame

Return type:

WindowAnalysisResult

Notes

  • The number of sars must match the number of unique animaldays in self.result

  • Spikes are binned into time windows matching the existing WAR fragments

  • nspike: array of spike counts per channel for each time window

  • lognspike: log-transformed spike counts using core.log_transform()

Example

>>> # After computing WAR and spike detection
>>> enhanced_war = war.read_sars_spikes(fdsar_list, read_mode="sa", inplace=False)
>>> enhanced_war.result['nspike']  # Spike counts per channel per window
read_mnes_spikes(raws: list[RawArray], inplace=True)[source]#

Extract spike features from MNE RawArray objects with spike annotations.

This method extracts spike timing from MNE annotations (where spikes are marked with channel-specific event labels) and bins them into WAR time windows.

Parameters:

  • raws (list[RawArray]) – List of MNE RawArray objects with spike annotations. One per recording session (animalday). Each should have annotations with channel names as event labels (e.g., ‘LMot’, ‘RMot’, etc.).

  • inplace (default: True) – If True, modifies self.result and returns self. If False, returns a new WindowAnalysisResult.

Returns:

WAR object with added spike features (nspike, lognspike).

Return type:

WindowAnalysisResult

Notes

  • Expects MNE annotations with channel names as event descriptions

  • Spike times are extracted from event onsets and binned to WAR windows

  • Channels not found in annotations will have empty spike arrays

  • Delegates to _read_from_spikes_all() for the actual binning logic

Example

>>> # From MNE spike annotations
>>> enhanced_war = war.read_mnes_spikes([mne_raw1, mne_raw2], inplace=False)
get_info()[source]#

Returns a formatted string with basic information about the WindowAnalysisResult object

get_result(features: list[str], exclude: list[str] = [], allow_missing=False)[source]#

Get windowed analysis result dataframe, with helpful filters

Parameters:

  • features (list[str]) – List of features to get from result

  • exclude (list[str], optional) – List of features to exclude from result; will override the features parameter. Defaults to [].

  • allow_missing (bool, optional) – If True, will return all requested features as columns regardless if they exist in result. Defaults to False.

Returns:

DataFrame with features in columns and windows in rows

Return type:

pd.DataFrame

get_groupavg_result(features: list[str], exclude: list[str] = [], df: DataFrame | None = None, groupby='animalday')[source]#

Group result and average within groups. Preserves data structure and shape for each feature.

Parameters:

  • features (list[str]) – List of features to get from result

  • exclude (list[str], optional) – List of features to exclude from result. Will override the features parameter. Defaults to [].

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • groupby (str, optional) – Feature or list of features to group by before averaging. Passed to the by parameter in pd.DataFrame.groupby(). Defaults to “animalday”.

Returns:

Result grouped by groupby and averaged for each group.

Return type:

pd.DataFrame

get_grouprows_result(features: list[str], exclude: list[str] = [], df: DataFrame | None = None, multiindex=['animalday', 'animal', 'genotype'], include=['duration', 'endfile'])[source]#
Parameters:

  • features (list[str])

  • exclude (list[str])

  • df (DataFrame | None)

get_filter_logrms_range(df: DataFrame | None = None, z_range=3, **kwargs)[source]#

Filter windows based on log(rms).

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • z_range (float, optional) – The z-score range to filter by. Values outside this range will be set to NaN.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_high_rms(df: DataFrame | None = None, max_rms=500, **kwargs)[source]#

Filter windows based on rms.

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • max_rms (float, optional) – The maximum rms value to filter by. Values above this will be set to NaN.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_low_rms(df: DataFrame | None = None, min_rms=30, **kwargs)[source]#

Filter windows based on rms.

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • min_rms (float, optional) – The minimum rms value to filter by. Values below this will be set to NaN.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_high_beta(df: DataFrame | None = None, max_beta_prop=0.4, **kwargs)[source]#

Filter windows based on beta power.

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • max_beta_prop (float, optional) – The maximum beta power to filter by. Values above this will be set to NaN. Defaults to 0.4.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_reject_channels(df: DataFrame | None = None, bad_channels: list[str] | None = None, use_abbrevs: bool | None = None, save_bad_channels: Literal['overwrite', 'union', None] = 'union', **kwargs)[source]#

Filter channels to reject.

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • bad_channels (list[str]) – List of channels to reject. Can be either full channel names or abbreviations. The method will automatically detect which format is being used. If None, no filtering is performed.

  • use_abbrevs (bool, optional) – Override automatic detection. If True, channels are assumed to be channel abbreviations. If False, channels are assumed to be channel names. If None, channels are parsed to abbreviations and matched against self.channel_abbrevs.

  • save_bad_channels (Literal["overwrite", "union", None], optional) – How to save bad channels to self.bad_channels_dict. “overwrite”: Replace self.bad_channels_dict completely with bad channels applied to all sessions. “union”: Merge bad channels with existing self.bad_channels_dict for all sessions. None: Don’t save to self.bad_channels_dict. Defaults to “union”. Note: When using “overwrite” mode, the bad_channels parameter and bad_channels_dict parameter may conflict and overwrite each other’s bad channel definitions if both are provided.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_reject_channels_by_recording_session(df: DataFrame | None = None, bad_channels_dict: dict[str, list[str]] | None = None, use_abbrevs: bool | None = None, save_bad_channels: Literal['overwrite', 'union', None] = 'union', **kwargs)[source]#

Filter channels to reject for each recording session

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • bad_channels_dict (dict[str, list[str]]) – Dictionary of list of channels to reject for each recording session. Can be either full channel names or abbreviations. The method will automatically detect which format is being used. If None, the method will use the bad_channels_dict passed to the constructor.

  • use_abbrevs (bool, optional) – Override automatic detection. If True, channels are assumed to be channel abbreviations. If False, channels are assumed to be channel names. If None, channels are parsed to abbreviations and matched against self.channel_abbrevs.

  • save_bad_channels (Literal["overwrite", "union", None], optional) – How to save bad channels to self.bad_channels_dict. “overwrite”: Replace self.bad_channels_dict completely with bad_channels_dict. “union”: Merge bad_channels_dict with existing self.bad_channels_dict per session. None: Don’t save to self.bad_channels_dict. Defaults to “union”. Note: When using “overwrite” mode, the bad_channels parameter and bad_channels_dict parameter may conflict and overwrite each other’s bad channel definitions if both are provided.

Returns:

Boolean array of shape (M fragments, N channels). True = keep window, False = remove window

Return type:

np.ndarray

get_filter_morphological_smoothing(filter_mask: ndarray, smoothing_seconds: float, **kwargs) → ndarray[source]#

Apply morphological smoothing to a filter mask.

Parameters:

  • filter_mask (np.ndarray) – Input boolean mask of shape (n_windows, n_channels)

  • smoothing_seconds (float) – Time window in seconds for morphological operations

Returns:

Smoothed boolean mask

Return type:

np.ndarray

filter_morphological_smoothing(smoothing_seconds: float) → WindowAnalysisResult[source]#

Apply morphological smoothing to all data.

Parameters:

smoothing_seconds (float) – Time window in seconds for morphological operations

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_all(df: DataFrame | None = None, inplace=True, min_valid_channels=3, filters: list[callable] | None = None, morphological_smoothing_seconds: float | None = None, **kwargs)[source]#

Apply a list of filters to the data. Filtering should be performed before aggregation.

Parameters:

  • df (pd.DataFrame, optional) – If not None, this function will use this dataframe instead of self.result. Defaults to None.

  • inplace (bool, optional) – If True, modify the result in place. Defaults to True.

  • bad_channels (list[str], optional) – List of channels to reject. Defaults to None.

  • min_valid_channels (int, optional) – Minimum number of valid channels required per window. Defaults to 3.

  • filters (list[callable], optional) – List of filter functions to apply. Each function should return a boolean mask. If None, uses default filters: [get_filter_logrms_range, get_filter_high_rms, get_filter_low_rms, get_filter_high_beta]. Defaults to None.

  • morphological_smoothing_seconds (float, optional) – If provided, apply morphological opening/closing to smooth the filter mask. This removes isolated false positives/negatives along the time axis for each channel independently. The value specifies the time window in seconds for the morphological operations. Defaults to None.

  • save_bad_channels (Literal["overwrite", "union", None], optional) – How to save bad channels to self.bad_channels_dict. This parameter is passed to the filtering functions. Defaults to “union”. Note: When using “overwrite” mode, the bad_channels parameter and bad_channels_dict parameter may conflict and overwrite each other’s bad channel definitions if both are provided.

  • **kwargs – Additional keyword arguments to pass to filter functions.

Returns:

Filtered result

Return type:

WindowAnalysisResult

filter_logrms_range(z_range: float = 3) → WindowAnalysisResult[source]#

Filter based on log(rms) z-score range.

Parameters:

z_range (float) – Z-score range threshold. Defaults to 3.

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_high_rms(max_rms: float = 500) → WindowAnalysisResult[source]#

Filter out windows with RMS above threshold.

Parameters:

max_rms (float) – Maximum RMS threshold. Defaults to 500.

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_low_rms(min_rms: float = 50) → WindowAnalysisResult[source]#

Filter out windows with RMS below threshold.

Parameters:

min_rms (float) – Minimum RMS threshold. Defaults to 50.

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_high_beta(max_beta_prop: float = 0.4) → WindowAnalysisResult[source]#

Filter out windows with high beta power.

Parameters:

max_beta_prop (float) – Maximum beta power proportion. Defaults to 0.4.

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_reject_channels(bad_channels: list[str], use_abbrevs: bool | None = None) → WindowAnalysisResult[source]#

Filter out specified bad channels.

Parameters:

  • bad_channels (list[str]) – List of channel names to reject

  • use_abbrevs (bool, optional) – Whether to use abbreviations. Defaults to None.

Returns:

New filtered instance

Return type:

WindowAnalysisResult

filter_reject_channels_by_session(bad_channels_dict: dict[str, list[str]] | None = None, use_abbrevs: bool | None = None) → WindowAnalysisResult[source]#

Filter out bad channels by recording session.

Parameters:

  • bad_channels_dict (dict[str, list[str]], optional) – Dictionary mapping recording session identifiers to lists of bad channel names to reject. Session identifiers are in the format “{animal_id} {genotype} {day}” (e.g., “A10 WT Apr-01-2023”). Channel names can be either full names (e.g., “Left Auditory”) or abbreviations (e.g., “LAud”). If None, uses the bad_channels_dict from the constructor. Defaults to None.

  • use_abbrevs (bool, optional) – Override automatic channel name format detection. If True, channels are assumed to be abbreviations. If False, channels are assumed to be full names. If None, automatically detects format and converts to abbreviations for matching. Defaults to None.

Returns:

New filtered instance with bad channels masked as NaN for their

respective recording sessions

Return type:

WindowAnalysisResult

Examples

Filter specific channels per session using abbreviations: >>> bad_channels = { … “A10 WT Apr-01-2023”: [“LAud”, “RMot”], # Session 1: reject left auditory, right motor … “A10 WT Apr-02-2023”: [“LVis”] # Session 2: reject left visual only … } >>> filtered_war = war.filter_reject_channels_by_session(bad_channels, use_abbrevs=True)

Filter using full channel names: >>> bad_channels = { … “A12 KO May-15-2023”: [“Left Motor”, “Right Barrel”], … “A12 KO May-16-2023”: [“Left Auditory”, “Left Visual”, “Right Motor”] … } >>> filtered_war = war.filter_reject_channels_by_session(bad_channels, use_abbrevs=False)

Auto-detect channel format (recommended): >>> bad_channels = { … “A15 WT Jun-10-2023”: [“LMot”, “RBar”], # Will auto-detect as abbreviations … “A15 WT Jun-11-2023”: [“LAud”] … } >>> filtered_war = war.filter_reject_channels_by_session(bad_channels)

Note

  • Session identifiers must exactly match the “animalday” values in the result DataFrame

  • Available channel abbreviations: LAud, RAud, LVis, RVis, LHip, RHip, LBar, RBar, LMot, RMot

  • Channel names are case-insensitive and support various formats (e.g., “left aud”, “Left Auditory”)

  • If a session identifier is not found in bad_channels_dict, a warning is logged but processing continues

  • If a channel name is not recognized, a warning is logged but other channels are still processed

apply_filters(filter_config: dict | None = None, min_valid_channels: int = 3, morphological_smoothing_seconds: float | None = None) → WindowAnalysisResult[source]#

Apply multiple filters using configuration.

Parameters:

  • filter_config (dict, optional) – Dictionary of filter names and parameters. Available filters: ‘logrms_range’, ‘high_rms’, ‘low_rms’, ‘high_beta’, ‘reject_channels’, ‘reject_channels_by_session’, ‘morphological_smoothing’

  • min_valid_channels (int) – Minimum valid channels per window. Defaults to 3.

  • morphological_smoothing_seconds (float, optional) – Temporal smoothing window (deprecated, use config instead)

Returns:

New filtered instance

Return type:

WindowAnalysisResult

Examples

>>> config = {
...     'logrms_range': {'z_range': 3},
...     'high_rms': {'max_rms': 500},
...     'reject_channels': {'bad_channels': ['LMot', 'RMot']},
...     'morphological_smoothing': {'smoothing_seconds': 8.0}
... }
>>> filtered_war = war.apply_filters(config)
save_pickle_and_json(folder: str | Path, make_folder=True, filename: str | None = None, slugify_filename=False, save_abbrevs_as_chnames=False)[source]#

Archive window analysis result into the folder specified, as a pickle and json file.

Parameters:

  • folder (str | Path) – Destination folder to save results to

  • make_folder (bool, optional) – If True, create the folder if it doesn’t exist. Defaults to True.

  • filename (str, optional) – Name of the file to save. Defaults to “war”.

  • slugify_filename (bool, optional) – If True, slugify the filename (replace special characters). Defaults to False.

  • save_abbrevs_as_chnames (bool, optional) – If True, save the channel abbreviations as the channel names in the json file. Defaults to False.

get_bad_channels_by_lof_threshold(lof_threshold: float) → dict[source]#

Apply LOF threshold directly to stored scores to get bad channels.

Parameters:

lof_threshold (float) – Threshold for determining bad channels.

Returns:

Dictionary mapping animal days to lists of bad channel names.

Return type:

dict

get_lof_scores() → dict[source]#

Get LOF scores from this WAR.

Returns:

Dictionary mapping animal days to LOF score dictionaries.

Return type:

dict

evaluate_lof_threshold_binary(ground_truth_bad_channels: dict | None = None, threshold: float | None = None, evaluation_channels: list[str] | None = None) → tuple[source]#

Evaluate single threshold against ground truth for binary classification.

Parameters:

  • ground_truth_bad_channels (dict | None (default: None)) – Dict mapping animal-day to bad channel sets. If None, uses self.bad_channels_dict as ground truth.

  • threshold (float | None (default: None)) – LOF threshold to test

  • evaluation_channels (list[str] | None (default: None)) – Subset of channels to include in evaluation. If none, uses all channels.

Returns:

(y_true_list, y_pred_list) for sklearn.metrics.f1_score

Each element represents one channel from one animal-day

Return type:

tuple

classmethod load_pickle_and_json(folder_path=None, pickle_name=None, json_name=None)[source]#

Load WindowAnalysisResult from folder

Parameters:

  • folder_path (str, optional) – Path of folder containing .pkl and .json files. Defaults to None.

  • pickle_name (str, optional) – Name of the pickle file. Can be just the filename (e.g. “war.pkl”) or a path relative to folder_path (e.g. “subdir/war.pkl”). If None and folder_path is provided, expects exactly one .pkl file in folder_path. Defaults to None.

  • json_name (str, optional) – Name of the JSON file. Can be just the filename (e.g. “war.json”) or a path relative to folder_path (e.g. “subdir/war.json”). If None and folder_path is provided, expects exactly one .json file in folder_path. Defaults to None.

Raises:

  • ValueError – folder_path does not exist

  • ValueError – Expected exactly one pickle and one json file in folder_path (when pickle_name/json_name not specified)

  • FileNotFoundError – Specified pickle_name or json_name not found

Returns:

WindowAnalysisResult object

Return type:

result

aggregate_time_windows(groupby: list[str] | str = ['animalday', 'isday']) → None[source]#

Aggregate time windows into a single data point per groupby by averaging features. This reduces the number of rows in the result.

Parameters:

groupby (list[str] | str, optional) – Columns to group by. Defaults to [‘animalday’, ‘isday’], which groups by animalday (recording session) and isday (day/night).

Raises:

  • ValueError – groupby must be from [‘animalday’, ‘isday’]

  • ValueError – Columns in groupby not found in result

  • ValueError – Columns in groupby are not constant in groups

Return type:

None

add_unique_hash(nbytes: int | None = None)[source]#

Adds a hex hash to the animal ID to ensure uniqueness. This prevents collisions when, for example, multiple animals in ExperimentPlotter have the same animal ID.

Parameters:

nbytes (int, optional) – Number of bytes to generate. This is passed directly to secrets.token_hex(). Defaults to None, which generates 16 hex characters (8 bytes).