Multimodal API

Alignment and cross-modal analysis between Miniscope calcium imaging and electrophysiology.

How to pass arguments: same pattern as other pipelines — use MultimodalPipeline.run(...) with keyword arguments; see Getting started section 5a and Pipelines.

Multimodal pipeline

ace_neuro.pipelines.multimodal.MultimodalPipeline

Orchestrates ephys + miniscope analysis and multimodal alignment.

After :meth:run, these instance attributes are populated (None if a stage did not apply):

• :attr:ephys_pipeline — :class:EphysPipeline instance used for this run
• :attr:miniscope_pipeline — :class:MiniscopePipeline instance used
• :attr:t_ca_im — aligned calcium frame times from TTL sync
• :attr:low_confidence_periods — sync quality mask from alignment
• :attr:ephys_idx_all_TTL_events — ephys sample indices for TTL events
• :attr:ephys_idx_ca_events — ephys indices at calcium events (if ca_events)
• :attr:ca_frame_num_of_ephys_idx — per-frame mapping (if TTL indices exist)
• :attr:ca_events_phases_ephys — phase samples for CA events (ephys band)
• :attr:ca_events_phases_miniscope — phase samples for CA events (miniscope)
• :attr:phase_hist_ephys / :attr:phase_bin_edges_ephys — histogram of ephys phases
• :attr:phase_hist_miniscope / :attr:phase_bin_edges_miniscope — histogram of miniscope phases

Source code in src/ace_neuro/pipelines/multimodal.py

class MultimodalPipeline:
    """Orchestrates ephys + miniscope analysis and multimodal alignment.

    After :meth:`run`, these instance attributes are populated (``None`` if a
    stage did not apply):

    - :attr:`ephys_pipeline` — :class:`EphysPipeline` instance used for this run
    - :attr:`miniscope_pipeline` — :class:`MiniscopePipeline` instance used
    - :attr:`t_ca_im` — aligned calcium frame times from TTL sync
    - :attr:`low_confidence_periods` — sync quality mask from alignment
    - :attr:`ephys_idx_all_TTL_events` — ephys sample indices for TTL events
    - :attr:`ephys_idx_ca_events` — ephys indices at calcium events (if ``ca_events``)
    - :attr:`ca_frame_num_of_ephys_idx` — per-frame mapping (if TTL indices exist)
    - :attr:`ca_events_phases_ephys` — phase samples for CA events (ephys band)
    - :attr:`ca_events_phases_miniscope` — phase samples for CA events (miniscope)
    - :attr:`phase_hist_ephys` / :attr:`phase_bin_edges_ephys` — histogram of ephys phases
    - :attr:`phase_hist_miniscope` / :attr:`phase_bin_edges_miniscope` — histogram of miniscope phases
    """

    def __init__(self) -> None:
        self.miniscope_pipeline = MiniscopePipeline()
        self.ephys_pipeline = EphysPipeline()
        self.t_ca_im: np.ndarray | None = None
        self.low_confidence_periods: Any = None
        self.ephys_idx_all_TTL_events: Any = None
        self.ephys_idx_ca_events: Any = None
        self.ca_frame_num_of_ephys_idx: Any = None
        self.ca_events_phases_ephys: Any = None
        self.ca_events_phases_miniscope: Any = None
        self.phase_hist_ephys: Any = None
        self.phase_bin_edges_ephys: Any = None
        self.phase_hist_miniscope: Any = None
        self.phase_bin_edges_miniscope: Any = None

    def run(
        self,
        line_num: int,
        project_path: str | Path | None = None,
        data_path: str | Path | None = None,
        # ephys parameters
        channel_name: str = 'PFCLFPvsCBEEG',
        remove_artifacts: bool = False,
        filter_type: str | None = None,
        filter_range: list[float] = [0.5, 4],
        plot_channel: bool = False,
        plot_spectrogram: bool = False,
        plot_phases: bool = False,
        logging_level: str = "CRITICAL",

        # miniscope parameters
        miniscope_filenames: list[str] = [],
        # preprocessing parameters
        crop: bool = True,
        crop_coords: list[int] | tuple[int, int, int, int] | None = None,
        detrend_method: str = 'median',
        df_over_f: bool = False,
        # if df_over_f = True
        secs_window: float = 5,
        quantile_min: float = 8,
        df_over_f_method: str = 'delta_f_over_sqrt_f',

        # processing parameters
        parallel: bool = False,
        n_processes: int = 6,
        apply_motion_correction: bool = True,
        inspect_motion_correction: bool = True,
        plot_params: bool = False,
        run_CNMFE: bool = True,
        save_estimates: bool = True,
        save_CNMFE_estimates_filename: str = 'estimates.hdf5',
        save_CNMFE_params: bool = False,

        # post-processing parameters
        remove_components_with_gui: bool = True,
        find_calcium_events: bool = True,
        derivative_for_estimates: str = 'first',
        event_height: float = 5,
        compute_miniscope_phase: bool = True,
        filter_miniscope_data: bool = True,
        n: int = 2,
        cut: list[float] = [0.1, 1.5],
        ftype: str = 'butter',
        btype: str = 'bandpass',
        inline: bool = False,
        compute_miniscope_spectrogram: bool = True,
        window_length: float = 30,
        window_step: float = 3,
        freq_lims: list[float] = [0, 15],
        time_bandwidth: float = 23,

        # multimodal parameters
        delete_TTLs: bool = True,
        fix_TTL_gaps: bool = False,
        only_experiment_events: bool = True,
        all_TTL_events: bool = True,
        ca_events: bool = False,
        time_range: list[float] | None = None,
        headless: bool = False
    ) -> None:
        """Run the complete multimodal analysis pipeline.

        Executes both ephys and miniscope pipelines, synchronizes their
        timestamps via TTL events, and performs phase-locked calcium event
        analysis.

        Args:
            line_num: Experiment line number in experiments.csv.
            channel_name: Ephys channel name to analyze.
            remove_artifacts: If True, remove ephys artifacts.
            filter_type: Ephys filter type ('butter', 'fir') or None.
            filter_range: [low, high] bandpass cutoffs for ephys.
            plot_channel: If True, plot ephys time series.
            plot_spectrogram: If True, plot ephys spectrogram.
            plot_phases: If True, plot phase histograms.
            logging_level: Verbosity level.
            miniscope_filenames: List of movie files to load.
            crop: If True, crop the movie.
            crop_coords: Crop coordinates as (x0, y0, x1, y1) tuple/list.
                If None, reads from analysis_parameters.csv or opens the GUI.
            detrend_method: 'median' or 'linear' detrending.
            df_over_f: If True, compute DF/F.
            parallel: If True, use multiprocessing.
            n_processes: Number of parallel processes.
            apply_motion_correction: If True, correct motion.
            run_CNMFE: If True, run source extraction.
            delete_TTLs: If True, remove dropped frame TTLs.
            fix_TTL_gaps: If True, interpolate missing TTLs.
            only_experiment_events: If True, keep only experiment events.
            all_TTL_events: If True, process all TTL events.
            ca_events: If True, include calcium event analysis.
            time_range: Optional [start, end] time range to analyze.
            headless: If True, disable all GUI interactions.
        """
        self.t_ca_im = None
        self.low_confidence_periods = None
        self.ephys_idx_all_TTL_events = None
        self.ephys_idx_ca_events = None
        self.ca_frame_num_of_ephys_idx = None
        self.ca_events_phases_ephys = None
        self.ca_events_phases_miniscope = None
        self.phase_hist_ephys = None
        self.phase_bin_edges_ephys = None
        self.phase_hist_miniscope = None
        self.phase_bin_edges_miniscope = None

        self.ephys_pipeline = EphysPipeline()
        try:
            self.ephys_pipeline.run(
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                channel_name=channel_name,
                remove_artifacts=remove_artifacts,
                filter_type=filter_type,
                filter_range=filter_range,
                plot_channel=plot_channel,
                plot_spectrogram=plot_spectrogram,
                plot_phases=plot_phases,
                logging_level=logging_level,
                headless=headless,
            )
        except Exception as e:
            raise PipelineExecutionError(
                "Multimodal run failed during ephys sub-pipeline.",
                stage="run_ephys_pipeline",
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                hint="Check ephys input paths and channel parameters before multimodal sync.",
            ) from e

        self.miniscope_pipeline = MiniscopePipeline()
        try:
            self.miniscope_pipeline.run(
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                filenames=miniscope_filenames,
                crop=crop,
                crop_coords=crop_coords,
                detrend_method=detrend_method,
                df_over_f=df_over_f,
                secs_window=secs_window,
                quantile_min=quantile_min,
                df_over_f_method=df_over_f_method,
                parallel=parallel,
                n_processes=n_processes,
                apply_motion_correction=apply_motion_correction,
                inspect_motion_correction=inspect_motion_correction,
                plot_params=plot_params,
                run_CNMFE=run_CNMFE,
                save_estimates=save_estimates,
                save_CNMFE_estimates_filename=save_CNMFE_estimates_filename,
                save_CNMFE_params=save_CNMFE_params,
                remove_components_with_gui=remove_components_with_gui,
                find_calcium_events=find_calcium_events,
                derivative_for_estimates=derivative_for_estimates,
                event_height=event_height,
                compute_miniscope_phase=compute_miniscope_phase,
                filter_miniscope_data=filter_miniscope_data,
                n=n,
                cut=cut,
                ftype=ftype,
                btype=btype,
                inline=inline,
                compute_miniscope_spectrogram=compute_miniscope_spectrogram,
                window_length=window_length,
                window_step=window_step,
                freq_lims=freq_lims,
                time_bandwidth=time_bandwidth,
                headless=headless,
            )
        except Exception as e:
            raise PipelineExecutionError(
                "Multimodal run failed during miniscope sub-pipeline.",
                stage="run_miniscope_pipeline",
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                hint="Check miniscope inputs and CNMF-E parameters before multimodal sync.",
            ) from e

        channel_object = self.ephys_pipeline.ephys_data_manager.get_channel(channel_name)
        frame_rate = self.miniscope_pipeline.miniscope_data_manager.fr
        ca_events_idx = self.miniscope_pipeline.miniscope_data_manager.ca_events_idx
        miniscope_phases = self.miniscope_pipeline.miniscope_data_manager.miniscope_phases

        try:
            t_ca_im, low_confidence_periods, channel_object, miniscope_dm = sync_neuralynx_miniscope_timestamps(
                channel_object,
                self.miniscope_pipeline.miniscope_data_manager,
                self.ephys_pipeline.ephys_data_manager,
                delete_TTLs=delete_TTLs,
                fix_TTL_gaps=fix_TTL_gaps,
                only_experiment_events=only_experiment_events,
            )
        except Exception as e:
            raise PipelineExecutionError(
                "Timestamp synchronization failed in multimodal pipeline.",
                stage="sync_timestamps",
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                hint="Validate TTL events and ensure both pipelines produced aligned timing metadata.",
            ) from e

        print("\nSuccess! Setting changed variables.")
        self.miniscope_pipeline.miniscope_data_manager = miniscope_dm
        self.ephys_pipeline.ephys_data_manager.channels[channel_name] = channel_object

        self.t_ca_im = t_ca_im
        self.low_confidence_periods = low_confidence_periods

        try:
            ephys_idx_all_TTL_events, ephys_idx_ca_events = find_ephys_idx_of_TTL_events(
                t_ca_im,
                channel_object,
                frame_rate,
                all_TTL_events=all_TTL_events,
                ca_events_idx=ca_events_idx if ca_events else None,
            )
        except Exception as e:
            raise PipelineExecutionError(
                "Failed to map TTL events into ephys indices.",
                stage="map_ttl_to_ephys_indices",
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                hint="Check frame-rate metadata and TTL event quality.",
            ) from e
        self.ephys_idx_all_TTL_events = ephys_idx_all_TTL_events
        self.ephys_idx_ca_events = ephys_idx_ca_events

        if ephys_idx_all_TTL_events is not None:
            self.ca_frame_num_of_ephys_idx = find_ca_movie_frame_num_of_ephys_idx(channel_object, ephys_idx_all_TTL_events)

        if ephys_idx_ca_events is not None:
            try:
                self.ca_events_phases_ephys = ephys_phase_ca_events(ephys_idx_ca_events, channel_object, neurons='all')
                self.ca_events_phases_miniscope = miniscope_phase_ca_events(ca_events_idx, miniscope_phases, neurons='all')
            except Exception as e:
                raise PipelineExecutionError(
                    "Failed to compute event-locked phases.",
                    stage="compute_phase_locked_events",
                    line_num=line_num,
                    project_path=project_path,
                    data_path=data_path,
                    hint="Ensure phase arrays and calcium event indices are valid and non-empty.",
                ) from e

        hist1, bin_edges1 = None, None
        hist2, bin_edges2 = None, None

        if ephys_idx_ca_events is not None:
            if self.ca_events_phases_ephys is not None:
                res1 = phase_ca_events_histogram(self.ca_events_phases_ephys)
                hist1, bin_edges1 = res1[0], res1[1]

            if self.ca_events_phases_miniscope is not None:
                res2 = phase_ca_events_histogram(self.ca_events_phases_miniscope)
                hist2, bin_edges2 = res2[0], res2[1]

        self.phase_hist_ephys = hist1
        self.phase_bin_edges_ephys = bin_edges1
        self.phase_hist_miniscope = hist2
        self.phase_bin_edges_miniscope = bin_edges2

run(line_num, project_path=None, data_path=None, channel_name='PFCLFPvsCBEEG', remove_artifacts=False, filter_type=None, filter_range=[0.5, 4], plot_channel=False, plot_spectrogram=False, plot_phases=False, logging_level='CRITICAL', miniscope_filenames=[], crop=True, crop_coords=None, detrend_method='median', df_over_f=False, secs_window=5, quantile_min=8, df_over_f_method='delta_f_over_sqrt_f', parallel=False, n_processes=6, apply_motion_correction=True, inspect_motion_correction=True, plot_params=False, run_CNMFE=True, save_estimates=True, save_CNMFE_estimates_filename='estimates.hdf5', save_CNMFE_params=False, remove_components_with_gui=True, find_calcium_events=True, derivative_for_estimates='first', event_height=5, compute_miniscope_phase=True, filter_miniscope_data=True, n=2, cut=[0.1, 1.5], ftype='butter', btype='bandpass', inline=False, compute_miniscope_spectrogram=True, window_length=30, window_step=3, freq_lims=[0, 15], time_bandwidth=23, delete_TTLs=True, fix_TTL_gaps=False, only_experiment_events=True, all_TTL_events=True, ca_events=False, time_range=None, headless=False)

Run the complete multimodal analysis pipeline.

Executes both ephys and miniscope pipelines, synchronizes their timestamps via TTL events, and performs phase-locked calcium event analysis.

Parameters:

Name	Type	Description	Default
line_num	int	Experiment line number in experiments.csv.	required
channel_name	str	Ephys channel name to analyze.	'PFCLFPvsCBEEG'
remove_artifacts	bool	If True, remove ephys artifacts.	False
filter_type	str | None	Ephys filter type ('butter', 'fir') or None.	None
filter_range	list[float]	[low, high] bandpass cutoffs for ephys.	[0.5, 4]
plot_channel	bool	If True, plot ephys time series.	False
plot_spectrogram	bool	If True, plot ephys spectrogram.	False
plot_phases	bool	If True, plot phase histograms.	False
logging_level	str	Verbosity level.	'CRITICAL'
miniscope_filenames	list[str]	List of movie files to load.	[]
crop	bool	If True, crop the movie.	True
crop_coords	list[int] | tuple[int, int, int, int] | None	Crop coordinates as (x0, y0, x1, y1) tuple/list. If None, reads from analysis_parameters.csv or opens the GUI.	None
detrend_method	str	'median' or 'linear' detrending.	'median'
df_over_f	bool	If True, compute DF/F.	False
parallel	bool	If True, use multiprocessing.	False
n_processes	int	Number of parallel processes.	6
apply_motion_correction	bool	If True, correct motion.	True
run_CNMFE	bool	If True, run source extraction.	True
delete_TTLs	bool	If True, remove dropped frame TTLs.	True
fix_TTL_gaps	bool	If True, interpolate missing TTLs.	False
only_experiment_events	bool	If True, keep only experiment events.	True
all_TTL_events	bool	If True, process all TTL events.	True
ca_events	bool	If True, include calcium event analysis.	False
time_range	list[float] | None	Optional [start, end] time range to analyze.	None
headless	bool	If True, disable all GUI interactions.	False

Source code in src/ace_neuro/pipelines/multimodal.py

def run(
    self,
    line_num: int,
    project_path: str | Path | None = None,
    data_path: str | Path | None = None,
    # ephys parameters
    channel_name: str = 'PFCLFPvsCBEEG',
    remove_artifacts: bool = False,
    filter_type: str | None = None,
    filter_range: list[float] = [0.5, 4],
    plot_channel: bool = False,
    plot_spectrogram: bool = False,
    plot_phases: bool = False,
    logging_level: str = "CRITICAL",

    # miniscope parameters
    miniscope_filenames: list[str] = [],
    # preprocessing parameters
    crop: bool = True,
    crop_coords: list[int] | tuple[int, int, int, int] | None = None,
    detrend_method: str = 'median',
    df_over_f: bool = False,
    # if df_over_f = True
    secs_window: float = 5,
    quantile_min: float = 8,
    df_over_f_method: str = 'delta_f_over_sqrt_f',

    # processing parameters
    parallel: bool = False,
    n_processes: int = 6,
    apply_motion_correction: bool = True,
    inspect_motion_correction: bool = True,
    plot_params: bool = False,
    run_CNMFE: bool = True,
    save_estimates: bool = True,
    save_CNMFE_estimates_filename: str = 'estimates.hdf5',
    save_CNMFE_params: bool = False,

    # post-processing parameters
    remove_components_with_gui: bool = True,
    find_calcium_events: bool = True,
    derivative_for_estimates: str = 'first',
    event_height: float = 5,
    compute_miniscope_phase: bool = True,
    filter_miniscope_data: bool = True,
    n: int = 2,
    cut: list[float] = [0.1, 1.5],
    ftype: str = 'butter',
    btype: str = 'bandpass',
    inline: bool = False,
    compute_miniscope_spectrogram: bool = True,
    window_length: float = 30,
    window_step: float = 3,
    freq_lims: list[float] = [0, 15],
    time_bandwidth: float = 23,

    # multimodal parameters
    delete_TTLs: bool = True,
    fix_TTL_gaps: bool = False,
    only_experiment_events: bool = True,
    all_TTL_events: bool = True,
    ca_events: bool = False,
    time_range: list[float] | None = None,
    headless: bool = False
) -> None:
    """Run the complete multimodal analysis pipeline.

    Executes both ephys and miniscope pipelines, synchronizes their
    timestamps via TTL events, and performs phase-locked calcium event
    analysis.

    Args:
        line_num: Experiment line number in experiments.csv.
        channel_name: Ephys channel name to analyze.
        remove_artifacts: If True, remove ephys artifacts.
        filter_type: Ephys filter type ('butter', 'fir') or None.
        filter_range: [low, high] bandpass cutoffs for ephys.
        plot_channel: If True, plot ephys time series.
        plot_spectrogram: If True, plot ephys spectrogram.
        plot_phases: If True, plot phase histograms.
        logging_level: Verbosity level.
        miniscope_filenames: List of movie files to load.
        crop: If True, crop the movie.
        crop_coords: Crop coordinates as (x0, y0, x1, y1) tuple/list.
            If None, reads from analysis_parameters.csv or opens the GUI.
        detrend_method: 'median' or 'linear' detrending.
        df_over_f: If True, compute DF/F.
        parallel: If True, use multiprocessing.
        n_processes: Number of parallel processes.
        apply_motion_correction: If True, correct motion.
        run_CNMFE: If True, run source extraction.
        delete_TTLs: If True, remove dropped frame TTLs.
        fix_TTL_gaps: If True, interpolate missing TTLs.
        only_experiment_events: If True, keep only experiment events.
        all_TTL_events: If True, process all TTL events.
        ca_events: If True, include calcium event analysis.
        time_range: Optional [start, end] time range to analyze.
        headless: If True, disable all GUI interactions.
    """
    self.t_ca_im = None
    self.low_confidence_periods = None
    self.ephys_idx_all_TTL_events = None
    self.ephys_idx_ca_events = None
    self.ca_frame_num_of_ephys_idx = None
    self.ca_events_phases_ephys = None
    self.ca_events_phases_miniscope = None
    self.phase_hist_ephys = None
    self.phase_bin_edges_ephys = None
    self.phase_hist_miniscope = None
    self.phase_bin_edges_miniscope = None

    self.ephys_pipeline = EphysPipeline()
    try:
        self.ephys_pipeline.run(
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            channel_name=channel_name,
            remove_artifacts=remove_artifacts,
            filter_type=filter_type,
            filter_range=filter_range,
            plot_channel=plot_channel,
            plot_spectrogram=plot_spectrogram,
            plot_phases=plot_phases,
            logging_level=logging_level,
            headless=headless,
        )
    except Exception as e:
        raise PipelineExecutionError(
            "Multimodal run failed during ephys sub-pipeline.",
            stage="run_ephys_pipeline",
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            hint="Check ephys input paths and channel parameters before multimodal sync.",
        ) from e

    self.miniscope_pipeline = MiniscopePipeline()
    try:
        self.miniscope_pipeline.run(
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            filenames=miniscope_filenames,
            crop=crop,
            crop_coords=crop_coords,
            detrend_method=detrend_method,
            df_over_f=df_over_f,
            secs_window=secs_window,
            quantile_min=quantile_min,
            df_over_f_method=df_over_f_method,
            parallel=parallel,
            n_processes=n_processes,
            apply_motion_correction=apply_motion_correction,
            inspect_motion_correction=inspect_motion_correction,
            plot_params=plot_params,
            run_CNMFE=run_CNMFE,
            save_estimates=save_estimates,
            save_CNMFE_estimates_filename=save_CNMFE_estimates_filename,
            save_CNMFE_params=save_CNMFE_params,
            remove_components_with_gui=remove_components_with_gui,
            find_calcium_events=find_calcium_events,
            derivative_for_estimates=derivative_for_estimates,
            event_height=event_height,
            compute_miniscope_phase=compute_miniscope_phase,
            filter_miniscope_data=filter_miniscope_data,
            n=n,
            cut=cut,
            ftype=ftype,
            btype=btype,
            inline=inline,
            compute_miniscope_spectrogram=compute_miniscope_spectrogram,
            window_length=window_length,
            window_step=window_step,
            freq_lims=freq_lims,
            time_bandwidth=time_bandwidth,
            headless=headless,
        )
    except Exception as e:
        raise PipelineExecutionError(
            "Multimodal run failed during miniscope sub-pipeline.",
            stage="run_miniscope_pipeline",
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            hint="Check miniscope inputs and CNMF-E parameters before multimodal sync.",
        ) from e

    channel_object = self.ephys_pipeline.ephys_data_manager.get_channel(channel_name)
    frame_rate = self.miniscope_pipeline.miniscope_data_manager.fr
    ca_events_idx = self.miniscope_pipeline.miniscope_data_manager.ca_events_idx
    miniscope_phases = self.miniscope_pipeline.miniscope_data_manager.miniscope_phases

    try:
        t_ca_im, low_confidence_periods, channel_object, miniscope_dm = sync_neuralynx_miniscope_timestamps(
            channel_object,
            self.miniscope_pipeline.miniscope_data_manager,
            self.ephys_pipeline.ephys_data_manager,
            delete_TTLs=delete_TTLs,
            fix_TTL_gaps=fix_TTL_gaps,
            only_experiment_events=only_experiment_events,
        )
    except Exception as e:
        raise PipelineExecutionError(
            "Timestamp synchronization failed in multimodal pipeline.",
            stage="sync_timestamps",
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            hint="Validate TTL events and ensure both pipelines produced aligned timing metadata.",
        ) from e

    print("\nSuccess! Setting changed variables.")
    self.miniscope_pipeline.miniscope_data_manager = miniscope_dm
    self.ephys_pipeline.ephys_data_manager.channels[channel_name] = channel_object

    self.t_ca_im = t_ca_im
    self.low_confidence_periods = low_confidence_periods

    try:
        ephys_idx_all_TTL_events, ephys_idx_ca_events = find_ephys_idx_of_TTL_events(
            t_ca_im,
            channel_object,
            frame_rate,
            all_TTL_events=all_TTL_events,
            ca_events_idx=ca_events_idx if ca_events else None,
        )
    except Exception as e:
        raise PipelineExecutionError(
            "Failed to map TTL events into ephys indices.",
            stage="map_ttl_to_ephys_indices",
            line_num=line_num,
            project_path=project_path,
            data_path=data_path,
            hint="Check frame-rate metadata and TTL event quality.",
        ) from e
    self.ephys_idx_all_TTL_events = ephys_idx_all_TTL_events
    self.ephys_idx_ca_events = ephys_idx_ca_events

    if ephys_idx_all_TTL_events is not None:
        self.ca_frame_num_of_ephys_idx = find_ca_movie_frame_num_of_ephys_idx(channel_object, ephys_idx_all_TTL_events)

    if ephys_idx_ca_events is not None:
        try:
            self.ca_events_phases_ephys = ephys_phase_ca_events(ephys_idx_ca_events, channel_object, neurons='all')
            self.ca_events_phases_miniscope = miniscope_phase_ca_events(ca_events_idx, miniscope_phases, neurons='all')
        except Exception as e:
            raise PipelineExecutionError(
                "Failed to compute event-locked phases.",
                stage="compute_phase_locked_events",
                line_num=line_num,
                project_path=project_path,
                data_path=data_path,
                hint="Ensure phase arrays and calcium event indices are valid and non-empty.",
            ) from e

    hist1, bin_edges1 = None, None
    hist2, bin_edges2 = None, None

    if ephys_idx_ca_events is not None:
        if self.ca_events_phases_ephys is not None:
            res1 = phase_ca_events_histogram(self.ca_events_phases_ephys)
            hist1, bin_edges1 = res1[0], res1[1]

        if self.ca_events_phases_miniscope is not None:
            res2 = phase_ca_events_histogram(self.ca_events_phases_miniscope)
            hist2, bin_edges2 = res2[0], res2[1]

    self.phase_hist_ephys = hist1
    self.phase_bin_edges_ephys = bin_edges1
    self.phase_hist_miniscope = hist2
    self.phase_bin_edges_miniscope = bin_edges2

Alignment utilities

ace_neuro.multimodal.miniscope_ephys_alignment_utils.sync_neuralynx_miniscope_timestamps(channel, miniscope_dm, ephys_dm, delete_TTLs=True, fix_TTL_gaps=False, only_experiment_events=True)

Synchronize Neuralynx and miniscope timestamps.

This is a legacy wrapper. It delegates to the new MiniscopeDataManager sync_timestamps architecture.

Parameters:

Name	Type	Description	Default
channel	Channel	Channel object containing ephys events.	required
miniscope_dm	MiniscopeDataManager	MiniscopeDataManager with frame info and analysis params.	required
ephys_dm	EphysDataManager	EphysDataManager to extract TTLs from.	required
delete_TTLs	bool	If True, remove TTLs for dropped frames from analysis_params.	True
fix_TTL_gaps	bool	If True, interpolate missing TTL events.	False
only_experiment_events	bool	If True, remove TTL events from event list.	True

Returns:

Type	Description
Tuple[ndarray, ndarray, Channel, MiniscopeDataManager]	Tuple of (tCaIm, low_confidence_periods, channel, miniscope_dm)

Source code in src/ace_neuro/multimodal/miniscope_ephys_alignment_utils.py

def sync_neuralynx_miniscope_timestamps(
    channel: Channel, 
    miniscope_dm: 'MiniscopeDataManager', 
    ephys_dm: EphysDataManager, 
    delete_TTLs: bool = True, 
    fix_TTL_gaps: bool = False, 
    only_experiment_events: bool = True
) -> Tuple[np.ndarray, np.ndarray, Channel, 'MiniscopeDataManager']:
    """Synchronize Neuralynx and miniscope timestamps.

    This is a legacy wrapper. It delegates to the new MiniscopeDataManager
    sync_timestamps architecture.

    Args:
        channel: Channel object containing ephys events.
        miniscope_dm: MiniscopeDataManager with frame info and analysis params.
        ephys_dm: EphysDataManager to extract TTLs from.
        delete_TTLs: If True, remove TTLs for dropped frames from analysis_params.
        fix_TTL_gaps: If True, interpolate missing TTL events.
        only_experiment_events: If True, remove TTL events from event list.

    Returns:
        Tuple of (tCaIm, low_confidence_periods, channel, miniscope_dm)
    """
    print('Syncing calcium movie times using Data Manager...')

    # Let the data managers handle TTL extraction and alignment natively
    tCaIm, low_confidence_periods = miniscope_dm.sync_timestamps(
        ephys_dm=ephys_dm, 
        channel_name=channel.name,
        delete_TTLs=delete_TTLs,
        fix_TTL_gaps=fix_TTL_gaps
    )

    # Make an array of the Neuralynx events with the TTL events removed
    if only_experiment_events and isinstance(ephys_dm, __import__('ace_neuro.ephys.neuralynx_data_manager', fromlist=['NeuralynxDataManager']).NeuralynxDataManager):
        ttl_label_pattern = 'TTL Input on AcqSystem1_0 board 0 port 1 value (0x0001)'
        ttl_label_pattern_off = 'TTL Input on AcqSystem1_0 board 0 port 0 value (0x0000)'

        # Remove BOTH on and off pulses if they exist, assuming port 0 and 1 are used
        frame_acq_idx = __import__('numpy').char.startswith(channel.events['labels'].astype(str), 'TTL Input')
        experiment_event_idx = __import__('numpy').invert(frame_acq_idx)
        channel.events['labels'] = channel.events['labels'][experiment_event_idx]
        channel.events['timestamps'] = channel.events['timestamps'][experiment_event_idx]

    return tCaIm, low_confidence_periods, channel, miniscope_dm     

ace_neuro.multimodal.miniscope_ephys_alignment_utils.find_ephys_idx_of_TTL_events(tCaIm, channel, frame_rate, ca_events_idx=None, all_TTL_events=True)

Finds the index of a calcium event in the Neuralynx timespace. If the miniscope class method to find the timing of calcium events has not been run yet, it runs that first. CHANNEL is the ephys channel with which to compare the timing of the ephys samples to the calcium event timing.

Source code in src/ace_neuro/multimodal/miniscope_ephys_alignment_utils.py

def find_ephys_idx_of_TTL_events(
    tCaIm: np.ndarray, 
    channel: Channel, 
    frame_rate: float, 
    ca_events_idx: Optional[Dict[int, np.ndarray]] = None, 
    all_TTL_events: bool = True
) -> Tuple[Optional[np.ndarray], Optional[Dict[int, np.ndarray]]]:
    """Finds the index of a calcium event in the Neuralynx timespace. If the miniscope class method to find the timing of calcium events has not been run yet, it runs that first.
    CHANNEL is the ephys channel with which to compare the timing of the ephys samples to the calcium event timing."""
    ephys_idx_all_TTL_events: Optional[np.ndarray] = None
    ephys_idx_ca_events_res: Optional[Dict[int, np.ndarray]] = None

    # Match up all calcium movie timestamps with their corresponding ephys timestamps.
    if all_TTL_events:
        print('Finding the indices of ephys timestamps that are closest to all calcium movie frame acquisition TTL events...')
        ephys_idx_all_TTL_events = np.empty(len(tCaIm),dtype=int)
        # Choose a number of indices after the last_index before which you are confident that the next index will be.
        # I am choosing the number of ephys indices during the time it takes for two calcium imaging frames.
        endPoint = round(int(channel.sampling_rate) * 2 / int(frame_rate))
        last_index = 0

        for k, CaIm_TTL_Event in enumerate(tCaIm):
            if k == 0:
                ephys_idx_all_TTL_events[k] = np.abs(channel.time_vector[last_index:] - CaIm_TTL_Event).argmin() + last_index
            elif len(channel.time_vector[last_index:]) - endPoint < 0:
                ephys_idx_all_TTL_events[k] = np.abs(channel.time_vector[last_index:] - CaIm_TTL_Event).argmin() + last_index
            else:
                ephys_idx_all_TTL_events[k] = np.abs(channel.time_vector[last_index:(last_index + endPoint)] - CaIm_TTL_Event).argmin() + last_index
            last_index = ephys_idx_all_TTL_events[k]

    # Look for the indices of the ephys timestamps that are closest to the calcium event (Neuralynx) timestamps.
    if ca_events_idx:
        print('Finding the indices of ephys timestamps that are closest to the calcium event (Neuralynx) timestamps...')
        ephys_idx_ca_events_res = {}
        for k in list(ca_events_idx.keys()):
            temp_list = []
            last_index = 0
            for j in range(len(ca_events_idx[k])):
                idx = np.abs(channel.time_vector[last_index:] - tCaIm[ca_events_idx[k][j]]).argmin() + last_index
                temp_list.append(idx)
                # Check to see if the gap between the calcium event time and the corresponding ephys timestamp is reasonable (within 1 frame's timestep).
                if np.abs(channel.time_vector[idx]-tCaIm[ca_events_idx[k][j]]) > (1/frame_rate):
                    # print('There are no ephys timestamps closer to the calcium event timestamp than the duration of a calcium movie frame!')
                    pass
                last_index = idx
            ephys_idx_ca_events_res[k] = np.array(temp_list)

    return ephys_idx_all_TTL_events, ephys_idx_ca_events_res

ace_neuro.multimodal.miniscope_ephys_alignment_utils.find_ca_movie_frame_num_of_ephys_idx(channel, ephys_idx_all_TTL_events)

Method to create an array the same size as obj.ephys[channel], where each element is the frame number of the corresponding calcium movie frame.

Source code in src/ace_neuro/multimodal/miniscope_ephys_alignment_utils.py

def find_ca_movie_frame_num_of_ephys_idx(
    channel: Channel, 
    ephys_idx_all_TTL_events: np.ndarray
) -> np.ndarray:
    """Method to create an array the same size as obj.ephys[channel], where each element is the frame number of the corresponding calcium movie frame."""
    ca_frame_num_of_ephys_idx = np.zeros(np.shape(channel.signal),dtype=int)

    # Assign a frame number to each element of ca_frame_num_of_ephys_idx. I'm not sure if the sample of obj.ephys that's closest to the TTL event should be paired with the preceding frame or not.
    for k, i in enumerate(ephys_idx_all_TTL_events[1:]):
        ca_frame_num_of_ephys_idx[i:ephys_idx_all_TTL_events[k]:-1] = k+1

    return ca_frame_num_of_ephys_idx