ACE-neuro: Ephys pipeline tutorial¶

Hands-on walkthrough of electrophysiology loading and analysis: metadata from your project CSVs, raw data under data_path, channel extraction, optional filtering and Hilbert phase, and plots via ChannelWorker.

Audience: You have a project directory with experiments.csv and analysis_parameters.csv, plus Neuralynx (or compatible) data under a shared raw-data root.

Published docs: after syncing notebooks (contributing), these render under the site Tutorials tab.

Prerequisites¶

• Python 3.10+ and a working ACE-neuro install. Recommended: conda env from linux_environment.yml, then pip install -e . --no-deps (see docs Getting started). A plain pip install -e . only works if CaImAn and scientific deps are already satisfied on your OS.
• project_path: folder that contains experiments.csv and analysis_parameters.csv at its top level.
• line_num: same experiment row in both CSVs.
• data_path: base folder for raw recordings; path fields in experiments.csv are resolved under this root.
• Ephys format: Built-in loaders include Neuralynx (.nev/.ncs) and ONIX RHS2116 (.raw); your ephys directory must match one of them or a custom EphysDataManager you registered.

Project layout: project_path vs data_path¶

Variable	Role
project_path	Experiment project (CSVs only live here).
data_path	Raw ephys tree (e.g. .ncs files).

Example:

my_project/                 ← project_path
  experiments.csv
  analysis_parameters.csv
shared_raw/                 ← data_path
  session_01/
    CSCxy.ncs

If you omit data_path when constructing ExperimentDataManager, the package falls back to an internal default under the repo — always pass data_path explicitly in real workflows.

In [ ]:

from pathlib import Path

# --- edit for your machine ---
project_path = Path("/path/to/your/project")
data_path = Path("/path/to/your/raw_data")
line_num = 96
channel_name = "PFCLFPvsCBEEG"

ex_csv = project_path / "experiments.csv"
ap_csv = project_path / "analysis_parameters.csv"
for label, p in ("experiments.csv", ex_csv), ("analysis_parameters.csv", ap_csv):
    if not p.is_file():
        raise FileNotFoundError(
            f"Missing {label} at {p}\n"
            f"project_path must be the directory that contains both CSVs (not the raw-data root)."
        )
print("OK: found both CSVs under project_path.")

from pathlib import Path # --- edit for your machine --- project_path = Path("/path/to/your/project") data_path = Path("/path/to/your/raw_data") line_num = 96 channel_name = "PFCLFPvsCBEEG" ex_csv = project_path / "experiments.csv" ap_csv = project_path / "analysis_parameters.csv" for label, p in ("experiments.csv", ex_csv), ("analysis_parameters.csv", ap_csv): if not p.is_file(): raise FileNotFoundError( f"Missing {label} at {p}\n" f"project_path must be the directory that contains both CSVs (not the raw-data root)." ) print("OK: found both CSVs under project_path.")

In [ ]:

from ace_neuro.shared.experiment_data_manager import ExperimentDataManager

edm = ExperimentDataManager(
    line_num,
    project_path=project_path,
    data_path=data_path,
    logging_level="INFO",
)
print("metadata id:", edm.metadata.get("id") if edm.metadata else None)
print("ephys directory (resolved):", edm.get_ephys_directory())

from ace_neuro.shared.experiment_data_manager import ExperimentDataManager edm = ExperimentDataManager( line_num, project_path=project_path, data_path=data_path, logging_level="INFO", ) print("metadata id:", edm.metadata.get("id") if edm.metadata else None) print("ephys directory (resolved):", edm.get_ephys_directory())

Table of contents¶

0. Setup — paths and CSVs (above)
1. Verify raw ephys for this line
2. Create EphysDataManager and import block
3. Process block → channels (artifacts + channel list)
4. Bandpass filter (filter_type, filter_range)
5. Hilbert phase
6. Plots via ChannelWorker
7. One-shot EphysPipeline.run
8. Troubleshooting

Step 1 — Verify raw data (file_downloader.verify_file_by_line)¶

Mirrors EphysPipeline.run. Ensures ephys files for this line_num exist under data_path.

In [ ]:

from ace_neuro.shared import file_downloader

experiments_csv = project_path / "experiments.csv"
file_downloader.verify_file_by_line(
    line_num=line_num,
    csv_path=experiments_csv,
    do_type="ephys",
    base_file_path=data_path,
)

from ace_neuro.shared import file_downloader experiments_csv = project_path / "experiments.csv" file_downloader.verify_file_by_line( line_num=line_num, csv_path=experiments_csv, do_type="ephys", base_file_path=data_path, )

Step 2 — Import ephys block (EphysDataManager.create)¶

Auto-selects backend from the folder layout. Processing is a separate step so you can see the object between import and channel extraction.

In [ ]:

from ace_neuro.ephys.ephys_data_manager import EphysDataManager

ephys_directory = edm.get_ephys_directory()
if ephys_directory is None:
    raise ValueError("ephys directory could not be determined from experiment metadata.")

ephys_dm = EphysDataManager.create(
    ephys_directory=ephys_directory,
    auto_import_ephys_block=True,
    auto_process_block=False,
    auto_compute_phases=False,
)

from ace_neuro.ephys.ephys_data_manager import EphysDataManager ephys_directory = edm.get_ephys_directory() if ephys_directory is None: raise ValueError("ephys directory could not be determined from experiment metadata.") ephys_dm = EphysDataManager.create( ephys_directory=ephys_directory, auto_import_ephys_block=True, auto_process_block=False, auto_compute_phases=False, )

Step 3 — process_ephys_block_to_channels¶

Builds Channel objects for each requested CSC name (channel_name must exist in metadata).

In [ ]:

ephys_dm.process_ephys_block_to_channels(
    remove_artifacts=True,
    channels=[channel_name],
)
ch0 = ephys_dm.get_channel(channel_name)
print("has signal:", ch0.signal is not None, "n:", len(ch0.signal) if ch0.signal is not None else None)

ephys_dm.process_ephys_block_to_channels( remove_artifacts=True, channels=[channel_name], ) ch0 = ephys_dm.get_channel(channel_name) print("has signal:", ch0.signal is not None, "n:", len(ch0.signal) if ch0.signal is not None else None)

Step 4 — Bandpass (filter_ephys)¶

EphysPipeline.run only filters when filter_type is not None. Here we call filter_ephys directly with replace_signal=False (pipeline default): filtered samples go to signal_filtered.

In [ ]:

ephys_dm.filter_ephys(channel_name, ftype="butter", cut=[0.5, 4.0], replace_signal=False)
ch = ephys_dm.get_channel(channel_name)
assert ch.signal_filtered is not None

ephys_dm.filter_ephys(channel_name, ftype="butter", cut=[0.5, 4.0], replace_signal=False) ch = ephys_dm.get_channel(channel_name) assert ch.signal_filtered is not None

Step 5 — Phase (compute_phases_all_channels)¶

Typically run after bandpass so phase reflects the band of interest.

In [ ]:

ephys_dm.compute_phases_all_channels()
ch = ephys_dm.get_channel(channel_name)

ephys_dm.compute_phases_all_channels() ch = ephys_dm.get_channel(channel_name)

Step 6 — ChannelWorker plots¶

plot_channel, plot_spectrogram, and plot_phases match the flags on EphysPipeline.run. Uncomment in a desktop session if your backend is interactive.

In [ ]:

from ace_neuro.ephys.channel_worker import ChannelWorker

cw = ChannelWorker(ch)
# cw.plot_channel(use_filtered=True)
# cw.plot_spectrogram(use_filtered=True, plot_events=False)
# cw.plot_phases()
print("Uncomment plot_* in a GUI-capable environment.")

from ace_neuro.ephys.channel_worker import ChannelWorker cw = ChannelWorker(ch) # cw.plot_channel(use_filtered=True) # cw.plot_spectrogram(use_filtered=True, plot_events=False) # cw.plot_phases() print("Uncomment plot_* in a GUI-capable environment.")

Step 7 — One-shot API¶

headless=True disables interactive pipeline plots; use earlier steps or set plot_* True with a GUI backend.

In [ ]:

from ace_neuro.pipelines.ephys import EphysPipeline

pipeline = EphysPipeline()
pipeline.run(
    line_num=line_num,
    project_path=project_path,
    data_path=data_path,
    channel_name=channel_name,
    remove_artifacts=True,
    filter_type="butter",
    filter_range=[0.5, 4.0],
    compute_phases=True,
    plot_channel=False,
    plot_spectrogram=False,
    plot_phases=False,
    headless=True,
)
ch = pipeline.ephys_data_manager.get_channel(channel_name)
print("filtered:", ch.signal_filtered is not None)

from ace_neuro.pipelines.ephys import EphysPipeline pipeline = EphysPipeline() pipeline.run( line_num=line_num, project_path=project_path, data_path=data_path, channel_name=channel_name, remove_artifacts=True, filter_type="butter", filter_range=[0.5, 4.0], compute_phases=True, plot_channel=False, plot_spectrogram=False, plot_phases=False, headless=True, ) ch = pipeline.ephys_data_manager.get_channel(channel_name) print("filtered:", ch.signal_filtered is not None)

Step 8 — Quick time-domain check (notebook)¶

Uses signal_filtered when present.

In [ ]:

import matplotlib.pyplot as plt
import numpy as np

ch = ephys_dm.get_channel(channel_name)
vec = ch.signal_filtered if ch.signal_filtered is not None else ch.signal
t = np.asarray(ch.time_vector)
dt = float(np.median(np.diff(t[:5000]))) if len(t) > 1 else 1.0
n = min(len(vec), int(30.0 / dt))
plt.figure(figsize=(12, 3))
plt.plot(t[:n], vec[:n])
plt.title(f"{channel_name} (~first 30 s)")
plt.xlabel("Time (s)")
plt.show()

import matplotlib.pyplot as plt import numpy as np ch = ephys_dm.get_channel(channel_name) vec = ch.signal_filtered if ch.signal_filtered is not None else ch.signal t = np.asarray(ch.time_vector) dt = float(np.median(np.diff(t[:5000]))) if len(t) > 1 else 1.0 n = min(len(vec), int(30.0 / dt)) plt.figure(figsize=(12, 3)) plt.plot(t[:n], vec[:n]) plt.title(f"{channel_name} (~first 30 s)") plt.xlabel("Time (s)") plt.show()

Troubleshooting¶

• CSVs not found: project_path must be the folder containing both files, not data_path.
• verify_file_by_line fails: Wrong data_path, missing downloads, or bad line — see data management guide / Box credentials.
• signal_filtered is None: No filter_type / no filter_ephys — filtering is opt-in.
• run_all_channels: Incomplete in the library; loop channels explicitly.