src.canns.pipeline

CANNs Pipeline Module

High-level pipelines for common analysis workflows, designed to make CANN models accessible to experimental researchers without requiring detailed knowledge of the underlying implementations.

Submodules

Classes

Pipeline

Abstract base class for CANNs pipelines.

ThetaSweepPipeline

High-level pipeline for theta sweep analysis of external trajectory data.

Functions

batch_process_trajectories(trajectory_list[, ...])

Process multiple trajectories in batch.

load_trajectory_from_csv(filepath[, x_col, y_col, ...])

Load trajectory data from CSV file and run theta sweep analysis.

Package Contents

class src.canns.pipeline.Pipeline

Bases: abc.ABC

Abstract base class for CANNs pipelines.

Pipelines orchestrate multi-step workflows (data preparation, model execution, visualization, etc.). This base class standardizes how we manage results and output directories so derived pipelines can focus on domain-specific logic.

get_results()

Return stored results or raise if the pipeline has not been executed.

has_results()

Check whether the pipeline has already produced results.

prepare_output_dir(output_dir, *, create=True)

Validate and optionally create the output directory for derived pipelines.

reset()

Reset stored state so the pipeline can be executed again cleanly.

abstractmethod run(*args, **kwargs)

Execute the pipeline and return a mapping of generated artifacts.

set_results(results)

Store pipeline results and return them for convenient chaining.

output_dir: pathlib.Path | None = None
results: dict[str, Any] | None = None
class src.canns.pipeline.ThetaSweepPipeline(trajectory_data, times=None, env_size=2.0, dt=0.001, direction_cell_params=None, grid_cell_params=None, theta_params=None, spatial_nav_params=None)[source]

Bases: src.canns.pipeline._base.Pipeline

High-level pipeline for theta sweep analysis of external trajectory data.

This pipeline abstracts the complex workflow of running CANN theta sweep models on experimental trajectory data, making it accessible to researchers who want to analyze neural responses without diving into implementation details.

Example

```python # Simple usage - just provide trajectory data pipeline = ThetaSweepPipeline(

trajectory_data=positions, # shape: (n_steps, 2) times=times # shape: (n_steps,)

)

results = pipeline.run(output_dir=”my_results/”) print(f”Animation saved to: {results[‘animation_path’]}”) ```

Initialize the theta sweep pipeline.

Parameters:

  • trajectory_data (numpy.ndarray) – Position coordinates with shape (n_steps, 2) for 2D trajectories

  • times (numpy.ndarray | None) – Optional time array with shape (n_steps,). If None, uniform time steps will be used

  • env_size (float) – Environment size (assumes square environment)

  • dt (float) – Simulation time step

  • direction_cell_params (dict[str, Any] | None) – Parameters for DirectionCellNetwork. If None, uses defaults

  • grid_cell_params (dict[str, Any] | None) – Parameters for GridCellNetwork. If None, uses defaults

  • theta_params (dict[str, Any] | None) – Parameters for theta modulation. If None, uses defaults

  • spatial_nav_params (dict[str, Any] | None) – Additional parameters for OpenLoopNavigationTask. If None, uses defaults

run(output_dir='theta_sweep_results', save_animation=True, save_plots=True, show_plots=False, animation_fps=10, animation_dpi=120, verbose=True)[source]

Run the complete theta sweep pipeline.

Parameters:

  • output_dir (str | pathlib.Path) – Directory to save output files

  • save_animation (bool) – Whether to save the theta sweep animation

  • save_plots (bool) – Whether to save analysis plots

  • show_plots (bool) – Whether to display plots interactively

  • animation_fps (int) – Frame rate for animation

  • animation_dpi (int) – DPI for animation output

  • verbose (bool) – Whether to print progress messages

Returns:

Dictionary containing paths to generated files and analysis data

Return type:

dict[str, Any]

direction_cell_params
direction_network = None
dt = 0.001
env_size = 2.0
grid_cell_params
grid_network = None
spatial_nav_params
spatial_nav_task = None
theta_params
times
trajectory_data
src.canns.pipeline.batch_process_trajectories(trajectory_list, output_base_dir='batch_results', **kwargs)[source]

Process multiple trajectories in batch.

Parameters:

  • trajectory_list (list) – List of (trajectory_data, times) tuples or trajectory_data arrays

  • output_base_dir (str) – Base directory for batch results

  • **kwargs – Additional parameters passed to ThetaSweepPipeline

Returns:

Dictionary mapping trajectory indices to results

Return type:

dict[str, dict[str, Any]]

src.canns.pipeline.load_trajectory_from_csv(filepath, x_col='x', y_col='y', time_col='time', **kwargs)[source]

Load trajectory data from CSV file and run theta sweep analysis.

Parameters:

  • filepath (str | pathlib.Path) – Path to CSV file

  • x_col (str) – Column name for x coordinates

  • y_col (str) – Column name for y coordinates

  • time_col (str | None) – Column name for time data (optional)

  • **kwargs – Additional parameters passed to ThetaSweepPipeline

Returns:

Dictionary containing analysis results and file paths

Return type:

dict[str, Any]