教程：使用Theta扫描管道的端到端研究工作流¶
Note
阅读时间：约30-35分钟
难度：初级到中级
前置条件：对导航轨迹的基本理解
本教程演示如何使用ThetaSweepPipeline对实验轨迹数据进行完整的端到端分析，无需深入了解CANN实现细节。
1. 管道介绍¶
1.1 什么是管道？¶
管道 为完整的分析提供了一个高级接口，无需实现细节：
[ ]:
# Traditional approach (manual)
1. Load data
2. Create networks
3. Run simulation
4. Compute analyses
5. Generate visualizations
6. Save results
# Pipeline approach (automated)
pipeline = ThetaSweepPipeline(trajectory_data, times)
results = pipeline.run() # Everything done!
1.2 为什么使用管道？¶
对于实验神经科学家： - 无需理解CANN数学原理 - 专注于您的数据和研究问题 - 可重复、标准化的分析
对于计算研究人员： - 快速原型设计 - 参数扫描和批处理 - 一致的输出格式
1.3 ThetaSweepPipeline 概述¶
ThetaSweepPipeline 实现完整的theta扫描分析：
输入： - 轨迹数据（位置随时间的变化） - 时序信息
自动处理： - 方向细胞网络模拟 - 网格细胞网络模拟 - Theta调制计算 - 群体活动分析
输出： - 轨迹分析图 - Theta扫描动画 - 群体活动可视化 - 用于自定义分析的原始模拟数据
2. 快速开始：基本管道用法¶
2.1 最小示例¶
最简单的用法 - 仅包含轨迹和时间：
[ ]:
import numpy as np
from canns.pipeline import ThetaSweepPipeline # :cite:p:`chu2024firing,ji2025systems`
# Example: Load your experimental trajectory
# positions: shape (n_steps, 2) - [x, y] coordinates
# times: shape (n_steps,) - timestamps in seconds
positions = np.load('my_trajectory.npy') # Your data
times = np.load('my_times.npy')
# Run complete analysis (one line!)
pipeline = ThetaSweepPipeline(
trajectory_data=positions,
times=times
)
results = pipeline.run(output_dir="results/")
print(f"Animation saved to: {results['animation_path']}")
print(f"Analysis plots in: results/")
就这样！管道会自动处理所有事情。
2.2 理解输出结果¶
运行后，你将在输出目录中找到：
results/
├── trajectory_analysis.png # 轨迹路径和统计信息
├── population_activity_hd.png # 方向细胞活动
├── population_activity_gc.png # 网格细胞活动
├── theta_sweep_animation.gif # 完整动力学动画
└── simulation_data.npz # 用于自定义分析的原始数据
2.3 快速数据格式检查¶
你的轨迹数据应该是：
[ ]:
# positions: (n_steps, 2) array
print(f"Position shape: {positions.shape}") # Should be (N, 2)
print(f"Position range X: [{positions[:,0].min()}, {positions[:,0].max()}]")
print(f"Position range Y: [{positions[:,1].min()}, {positions[:,1].max()}]")
# times: (n_steps,) array
print(f"Times shape: {times.shape}") # Should be (N,)
print(f"Duration: {times[-1] - times[0]:.2f}s")
print(f"Mean dt: {np.mean(np.diff(times)):.4f}s")
常见问题： - 位置单位不是米？相应地缩放 - 时间单位不是秒？先转换 - 非均匀采样？管道自动处理
3. 加载外部轨迹数据¶
3.1 从 CSV 文件加载¶
[ ]:
import pandas as pd
# Load from CSV
df = pd.read_csv('trajectory.csv')
# Extract positions and times
positions = df[['x', 'y']].values # (n_steps, 2)
times = df['time'].values # (n_steps,)
# Run pipeline
pipeline = ThetaSweepPipeline(positions, times)
results = pipeline.run()
CSV 格式示例： .. code-block:: csv
time,x,y 0.000,0.5,0.5 0.001,0.501,0.502 0.002,0.503,0.505 …
3.2 从 MATLAB 文件导入¶
[ ]:
from scipy.io import loadmat
# Load MATLAB file
data = loadmat('trajectory.mat')
positions = data['positions'] # Already (n_steps, 2)
times = data['times'].flatten() # Flatten if needed
pipeline = ThetaSweepPipeline(positions, times)
results = pipeline.run()
3.3 来自跟踪软件¶
跟踪系统的常见格式：
[ ]:
# DeepLabCut output
import pandas as pd
dlc_data = pd.read_csv('tracking_output.csv', header=[0,1,2])
x = dlc_data[('bodypart1', 'x')].values
y = dlc_data[('bodypart1', 'y')].values
positions = np.column_stack([x, y])
# Bonsai output
bonsai_data = pd.read_csv('bonsai_tracking.csv')
positions = bonsai_data[['X', 'Y']].values / 100 # Convert cm to m
# Custom tracking
# Always ensure: positions in meters, times in seconds
3.4 合成测试数据¶
用于测试或演示：
[ ]:
def create_test_trajectory(n_steps=1000, dt=0.002):
"""Create smooth test trajectory"""
times = np.linspace(0, (n_steps-1)*dt, n_steps)
# Circular trajectory with some noise
t_param = np.linspace(0, 4*np.pi, n_steps)
radius = 0.5
center = np.array([0.75, 0.75])
x = center[0] + radius * np.cos(t_param)
y = center[1] + radius * np.sin(t_param)
# Add small noise
noise = np.random.normal(0, 0.01, (n_steps, 2))
positions = np.column_stack([x, y]) + noise
return positions, times
# Test the pipeline
positions, times = create_test_trajectory()
pipeline = ThetaSweepPipeline(positions, times, env_size=1.5)
results = pipeline.run(output_dir="test_results/")
4. 高级自定义¶
4.1 环境配置¶
调整环境参数：
[ ]:
pipeline = ThetaSweepPipeline(
trajectory_data=positions,
times=times,
env_size=2.0, # Environment size (meters)
dt=0.001, # Simulation time step (seconds)
)
何时调整：
- env_size: 匹配你的实验竞技场 (1m, 2m, 等)
- dt: 对于非常快的运动使用更细的时间分辨率
4.2 网络参数¶
自定义方向和网格细胞网络:
[ ]:
pipeline = ThetaSweepPipeline(
trajectory_data=positions,
times=times,
# Direction cell network
direction_cell_params={
'num': 100, # Number of direction cells
'adaptation_strength': 15.0, # SFA :cite:p:`mi2014spike,li2025dynamics` strength
'noise_strength': 0.0, # Activity noise
},
# Grid cell network
grid_cell_params={
'num_gc_x': 100, # Grid cells per dimension
'adaptation_strength': 8.0, # SFA strength
'mapping_ratio': 5, # Grid spacing control
'phase_offset': 1.0/20, # Theta sweep magnitude
},
)
参数效果：
- 更高的 adaptation_strength：更强的θ振荡
- 更大的 mapping_ratio：更小的网格间距
- 更大的 phase_offset：更大的θ扫描
4.3 Theta调制设置¶
控制θ节律参数：
[ ]:
pipeline = ThetaSweepPipeline(
trajectory_data=positions,
times=times,
theta_params={
'theta_strength_hd': 1.0, # HD cell modulation strength
'theta_strength_gc': 0.5, # Grid cell modulation strength
'theta_cycle_len': 100.0, # Cycle length (ms)
},
)
生物学对应关系：
- theta_cycle_len=100ms → 10 Hz theta频率
- theta_cycle_len=125ms → 8 Hz theta频率
4.4 动画和可视化¶
自定义输出可视化：
[ ]:
results = pipeline.run(
output_dir="custom_results/",
save_animation=True, # Generate animation
save_plots=True, # Save analysis plots
animation_fps=10, # Animation frame rate
animation_n_step=20, # Sample every N frames
verbose=True, # Print progress
)
性能提示：
- 降低 animation_fps 以获得更小的文件
- 提高 animation_n_step 以加快生成速度
- 设置 save_animation=False 跳过动画生成（更快）
5. 完整的研究示例¶
5.1 现实场景¶
分析一个记录的导航会话：
[ ]:
import numpy as np
import pandas as pd
from canns.pipeline import ThetaSweepPipeline # :cite:p:`chu2024firing,ji2025systems`
# Load experimental data
print("Loading experimental trajectory...")
df = pd.read_csv('experiment_2024_session_3.csv')
# Extract and preprocess
positions = df[['x_cm', 'y_cm']].values / 100 # Convert cm to meters
times = df['timestamp_ms'].values / 1000 # Convert ms to seconds
# Data quality check
print(f"Trajectory: {len(positions)} samples")
print(f"Duration: {times[-1] - times[0]:.2f}s")
print(f"Position range: X[{positions[:,0].min():.2f}, {positions[:,0].max():.2f}]m, "
f"Y[{positions[:,1].min():.2f}, {positions[:,1].max():.2f}]m")
# Remove any NaN values (tracking failures)
valid = ~np.isnan(positions).any(axis=1)
positions = positions[valid]
times = times[valid]
print(f"Valid samples: {len(positions)}")
5.2 运行综合分析¶
[ ]:
# Configure pipeline with experimental parameters
pipeline = ThetaSweepPipeline(
trajectory_data=positions,
times=times,
env_size=1.5, # 1.5m x 1.5m arena
direction_cell_params={
'num': 100,
'adaptation_strength': 15.0, # SFA :cite:p:`mi2014spike,li2025dynamics`
'noise_strength': 0.05, # Small noise for realism
},
grid_cell_params={
'num_gc_x': 100,
'adaptation_strength': 8.0,
'mapping_ratio': 5,
},
theta_params={
'theta_strength_hd': 1.0,
'theta_strength_gc': 0.5,
'theta_cycle_len': 100.0, # 10 Hz
},
)
# Run analysis
print("\nRunning theta sweep analysis...")
results = pipeline.run(
output_dir="experiment_analysis/",
save_animation=True,
save_plots=True,
animation_fps=15,
verbose=True,
)
print("\n✅ Analysis complete!")
print(f"📊 Results saved to: experiment_analysis/")
print(f"🎬 Animation: {results['animation_path']}")
5.3 自定义后处理¶
访问原始仿真数据进行自定义分析：
[ ]:
# Load simulation results
sim_data = results['simulation_data']
# Available data
dc_activity = sim_data['dc_activity'] # Direction cell firing
gc_activity = sim_data['gc_activity'] # Grid cell firing
theta_phase = sim_data['theta_phase'] # Theta phase over time
internal_pos = sim_data['internal_position'] # Decoded position
# Example analysis: Phase precession :cite:p:`ji2025phase,o1993phase`
import matplotlib.pyplot as plt
# Find high activity periods
activity_threshold = gc_activity.mean() + gc_activity.std()
high_activity = gc_activity.max(axis=1) > activity_threshold
# Plot phase vs time for high activity
plt.figure(figsize=(12, 4))
plt.scatter(times[high_activity], theta_phase[high_activity],
c=gc_activity[high_activity].max(axis=1),
cmap='viridis', s=10, alpha=0.6)
plt.xlabel('Time (s)')
plt.ylabel('Theta Phase (rad)')
plt.title('Phase Precession During High Grid Cell Activity')
plt.colorbar(label='Max Activity')
plt.savefig('experiment_analysis/phase_precession.png', dpi=150)
plt.show()
# Compare decoded vs actual position
plt.figure(figsize=(10, 5))
plt.subplot(1, 2, 1)
plt.plot(positions[:,0], positions[:,1], 'k-', alpha=0.5, label='Actual')
plt.plot(internal_pos[:,0], internal_pos[:,1], 'r-', alpha=0.7, label='Decoded')
plt.xlabel('X Position (m)')
plt.ylabel('Y Position (m)')
plt.title('Position Tracking Accuracy')
plt.legend()
plt.axis('equal')
plt.subplot(1, 2, 2)
error = np.linalg.norm(positions - internal_pos, axis=1)
plt.plot(times, error, 'b-', alpha=0.7)
plt.xlabel('Time (s)')
plt.ylabel('Position Error (m)')
plt.title('Decoding Error Over Time')
plt.tight_layout()
plt.savefig('experiment_analysis/position_accuracy.png', dpi=150)
plt.show()
print(f"\n📈 Mean position error: {error.mean():.3f}m")
print(f"📈 Max position error: {error.max():.3f}m")
5.4 批量处理多个会话¶
处理多个实验会话：
[ ]:
import glob
from pathlib import Path
# Find all session files
session_files = glob.glob('experiments/session_*.csv')
print(f"Found {len(session_files)} sessions to process")
# Process each session
for session_file in session_files:
session_name = Path(session_file).stem
print(f"\nProcessing {session_name}...")
# Load data
df = pd.read_csv(session_file)
positions = df[['x_cm', 'y_cm']].values / 100
times = df['timestamp_ms'].values / 1000
# Run pipeline
pipeline = ThetaSweepPipeline(positions, times, env_size=1.5)
output_dir = f"batch_results/{session_name}/"
results = pipeline.run(
output_dir=output_dir,
save_animation=False, # Skip animation for speed
save_plots=True,
verbose=False,
)
print(f" ✓ Complete: {output_dir}")
print("\n🎉 Batch processing complete!")
6. 后续步骤¶
恭喜！您已经学会了如何使用 ThetaSweepPipeline 进行端到端的 theta 扫描分析。
主要要点¶
管道简化工作流 - 复杂模型的一行代码分析
灵活的数据加载 - CSV、MATLAB、跟踪软件
自动输出 - 图表、动画和原始数据
可定制参数 - 需要时获得完全控制
批处理 - 高效地分析多个会话
何时使用管道¶
适合以下情况： - 没有编码专业知识的实验神经科学家 - 快速原型设计和探索性分析 - 标准化处理多个数据集 - 生成出版质量的图表 - 教学和演示
考虑手动方法的情况： - 需要非标准模型架构 - 实现新的分析方法 - 需要对每一步进行细粒度控制 - 扩展管道功能
管道功能总结¶
功能 | 基本用法 | 高级用法 |
|---------|————-|----------------|
| 数据输入 | trajectory_data, times | 多种格式加载器 |
| 环境 | 自动检测 | 可定制的 env_size、dt |
| 网络 | 默认参数 | 完整参数字典 |
| Theta | 默认 10 Hz | 自定义频率、强度 |
| 输出 | 标准图表 | 原始数据 + 自定义分析 |
| 批处理 | 单个会话 | 多会话处理 |
继续学习¶
扩展管道¶
想要修改或扩展管道？
检查源代码：
canns.pipeline.theta_sweep.ThetaSweepPipeline
2. 继承和自定义： .. code-block:: python
from canns.pipeline import ThetaSweepPipeline
- class MyCustomPipeline(ThetaSweepPipeline):
- def custom_analysis(self):
# 在这里添加您的分析 pass
贡献：通过 GitHub pull requests 提交增强功能
获取帮助¶
示例：查看
examples/pipeline/目录社区：GitHub 上的讨论和问答
最佳实践¶
数据质量第一：在管道之前清理跟踪数据
从简单开始：最初使用默认参数
验证输出：检查轨迹图的合理性
文档参数：保存配置以实现可重现性
版本控制：跟踪用于每项分析的管道版本
感谢您完成 CANN 教程系列！您现在拥有 CANN 建模、脑启发式学习和实际研究工作流的全面知识。