Data Pipeline

from alpasim_grpc.utils.logs import async_read_pb_log

i = 0
async for log_entry in async_read_pb_log("path/to/rollout.asl"):
    print(log_entry)
    i += 1
    if i == 20:
        break

import asyncio
from pathlib import Path
from alpasim_grpc.utils.logs import async_read_pb_log
import pandas as pd

async def extract_trajectory(asl_path):
    """Extract ego trajectory from ASL log."""
    trajectory = []
    
    async for log_entry in async_read_pb_log(str(asl_path)):
        if log_entry.HasField('actor_poses'):
            ego_pose = log_entry.actor_poses.poses.get('EGO')
            if ego_pose:
                trajectory.append({
                    'timestamp_us': log_entry.actor_poses.timestamp_us,
                    'x': ego_pose.position.x,
                    'y': ego_pose.position.y,
                })
    
    return pd.DataFrame(trajectory)

async def process_all_rollouts(rollouts_dir):
    """Process all ASL logs in rollouts directory."""
    asl_files = Path(rollouts_dir).rglob('rollout.asl')
    
    tasks = [extract_trajectory(asl_file) for asl_file in asl_files]
    trajectories = await asyncio.gather(*tasks)
    
    return trajectories

# Run from Jupyter or async context
trajectories = await process_all_rollouts('tutorial/rollouts')

import pandas as pd
from pathlib import Path

def load_all_metrics(rollouts_dir):
    """Load all per-rollout metrics."""
    metrics_files = Path(rollouts_dir).rglob('metrics.parquet')
    
    dfs = []
    for metrics_file in metrics_files:
        df = pd.read_parquet(metrics_file)
        # Add rollout identifier
        df['rollout_path'] = str(metrics_file.parent)
        dfs.append(df)
    
    return pd.concat(dfs, ignore_index=True)

metrics_df = load_all_metrics('tutorial/rollouts')

# Filter to failed rollouts
failed = metrics_df[metrics_df['collision_at_fault'] == 1]
print(f"Found {len(failed)} failed rollouts")
print(failed[['rollout_path', 'dist_to_gt_trajectory', 'duration_frac_20s']])