src.canns.pipeline.theta_sweep

Theta Sweep Pipeline for External Trajectory Analysis

This module provides a high-level pipeline for experimental scientists to analyze their trajectory data using CANN theta sweep models without needing to understand the underlying implementation details.

Classes

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.

Module Contents

class src.canns.pipeline.theta_sweep.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)

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)

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.theta_sweep.batch_process_trajectories(trajectory_list, output_base_dir='batch_results', **kwargs)

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.theta_sweep.load_trajectory_from_csv(filepath, x_col='x', y_col='y', time_col='time', **kwargs)

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]