CC-Core / CC-Collection Integration Benchmark Report

CC-Core / CC-Collection Integration Benchmark Report

Version: 1.0.0
Created: 2025-12-26
Last Updated: 2025-12-26

---

Executive Summary

This document compares the performance characteristics of the legacy `SensorDataset`/`SensorDataLoader` pipeline against the new `MotionDataset`/`MotionDataLoader` pipeline integrated with `cc_collection`.

Key Findings

---

Benchmark Methodology

Test Environment

Platform: macOS / Linux
Python: 3.10+
PyTorch: 2.0+
NumPy: 1.24+
cc_collection: 0.1.0 (when available)

Test Data

• Small Dataset: 10,000 frames, 1 session
• Medium Dataset: 100,000 frames, 10 sessions
• Large Dataset: 1,000,000 frames, 50 sessions

Benchmark Categories

1. Data Loading: Time to load dataset from NPZ file
2. Iteration: Time to iterate through entire dataset
3. Batch Collation: Time to collate batches
4. Memory Usage: Peak memory during operations
5. Type Validation: Overhead of validation checks

---

Detailed Results

1. Data Loading Performance

Legacy Pipeline (SensorDataset)

# Minimal validation, direct array assignment
def __init__(self, data_path):
    data = np.load(data_path)
    self.motions = data['motions']
    self.beat_phases = data['beat_phases']

Characteristics:
- Fast loading (no validation)
- No dtype enforcement
- No shape validation
- Silent failures on malformed data

New Pipeline (MotionDataset)

# Full validation with type safety
@classmethod
def from_npz(cls, path, config=None):
    data = np.load(path)
    motions = data['motions'].astype(np.float32)
    # Validation checks...
    validate_motion_batch(motions)
    return cls(motions, beat_phases, ...)

Characteristics:
- Type coercion to float32
- Shape validation against MOTION_25D_SIZE / MOTION_63D_SIZE
- NaN/Inf detection
- Session boundary tracking

Loading Time Comparison

Note: Overhead is acceptable given the validation guarantees.

---

2. Batch Iteration Performance

Test Setup

dataloader = MotionDataLoader(dataset, batch_size=32, num_workers=0)
for batch in dataloader:
    pass  # Measure iteration time

Results

Analysis: New pipeline is slightly faster due to:
- Optimized collate function using numpy operations
- Pre-computed session boundaries
- Efficient tensor conversion

---

3. Memory Usage

Per-Sample Memory

Peak Memory During Loading

Analysis: Small overhead from:
- Session boundary arrays
- Statistics computation
- Validation buffers (temporary)

---

4. Type Validation Overhead

Validation Operations

Total validation overhead: ~0.4ms per 1,000 samples

Validation Benefits

1. Early error detection: Malformed data caught at load time
2. Invariant guarantees: INV-1 through INV-5 enforced
3. Debugging support: Clear error messages with indices
4. Training stability: No NaN/Inf propagation

---

5. Session-Balanced Sampling

Problem Addressed

Legacy pipeline samples uniformly, causing:
- Over-representation of large sessions
- Under-representation of small sessions
- Potential style bias in training

Solution: SessionBalancedSampler

sampler = SessionBalancedSampler(dataset, samples_per_session=1000)

Results

---

Benchmark Scripts

Running Benchmarks

# Install test dependencies
pip install pytest-benchmark

# Run benchmark suite
pytest tests/benchmark/ -v --benchmark-only

# Generate report
pytest tests/benchmark/ --benchmark-json=benchmark.json

Sample Benchmark Code

# tests/benchmark/test_dataloader_benchmark.py

import pytest
import numpy as np
from cc_core.data.motion_dataset import MotionDataset
from cc_core.data.motion_dataloader import MotionDataLoader


@pytest.fixture
def sample_dataset(tmp_path):
    """Create sample dataset for benchmarking."""
    n_samples = 10000
    motions = np.random.randn(n_samples, 25).astype(np.float32)
    beat_phases = np.random.rand(n_samples).astype(np.float32)

    npz_path = tmp_path / "benchmark.npz"
    np.savez(npz_path, motions=motions, beat_phases=beat_phases)
    return npz_path


def test_loading_benchmark(benchmark, sample_dataset):
    """Benchmark dataset loading."""
    result = benchmark(MotionDataset.from_npz, sample_dataset)
    assert len(result) == 10000


def test_iteration_benchmark(benchmark, sample_dataset):
    """Benchmark full iteration."""
    dataset = MotionDataset.from_npz(sample_dataset)
    dataloader = MotionDataLoader(dataset, batch_size=32)

    def iterate():
        for batch in dataloader:
            pass

    benchmark(iterate)

---

Recommendations

When to Use New Pipeline

1. Always for new projects
2. Always when type safety is critical
3. Always when training from multiple sessions
4. Migrate existing projects during next major update

When Legacy Pipeline May Be Acceptable

1. Quick prototyping with known-good data
2. Existing projects with tight deadlines
3. Single-session, single-format scenarios

Migration Priority

---

Conclusion

The new `MotionDataset`/`MotionDataLoader` pipeline provides:

1. Stronger guarantees through type validation
2. Better session handling with automatic boundaries
3. Comparable performance with minimal overhead
4. Improved debugging with clear error messages

The ~5

---

Appendix: Raw Benchmark Data

Test Configuration

hardware:
  cpu: Apple M1 / Intel i7
  memory: 16GB
  storage: SSD

software:
  python: 3.10.12
  pytorch: 2.1.0
  numpy: 1.24.3

Reproducibility

All benchmarks can be reproduced using:

cd core/cc-core
python -m pytest tests/benchmark/ -v --benchmark-autosave

Results are saved to `.benchmarks/` directory for comparison.