Function process_rrd_session

depot/mapper/src/main.rs:342–444

Process an RRD session file by extracting its contents. Extracts poses, LiDAR, camera, and GPS data from the .rrd file and writes them to an extracted/ directory alongside it.

(
    state: SharedState,
    rrd_path: PathBuf,
)

/// Extracts poses, LiDAR, camera, and GPS data from the .rrd file
/// and writes them to an extracted/ directory alongside it.
async fn process_rrd_session(
    state: SharedState,
    rrd_path: PathBuf,
) -> Result<(), MapperError> {
    // Check if already skipped (in-memory state, persisted to maps volume)
    {
        let s = state.read().await;
        if s.skipped_sessions.contains(&rrd_path) {
            debug!(path = %rrd_path.display(), "Previously skipped, not reprocessing");
            return Ok(());
        }
    }

    info!(path = %rrd_path.display(), "Processing RRD session");

    // Determine output directory
    let extract_dir = if rrd_path.file_name().map(|n| n == "session.rrd").unwrap_or(false) {
        rrd_path.parent().unwrap().join("extracted")
    } else {
        rrd_path.with_extension("extracted")
    };

    // Check if already extracted on disk
    if extract_dir.exists() {
        debug!(path = %rrd_path.display(), "Already extracted, skipping");
        return Ok(());
    }

    // Check if file is still being written (wait for stable mtime)
    let metadata = tokio::fs::metadata(&rrd_path).await?;
    let mtime = metadata.modified()?;
    tokio::time::sleep(Duration::from_secs(2)).await;
    let new_metadata = tokio::fs::metadata(&rrd_path).await?;
    if new_metadata.modified()? != mtime {
        debug!(path = %rrd_path.display(), "File still being written, skipping");
        return Ok(());
    }

    // Extract data from RRD file
    let rrd_path_clone = rrd_path.clone();
    let extract_dir_clone = extract_dir.clone();
    let extraction_result = tokio::task::spawn_blocking(move || {
        let result = rrd_extractor::extract_from_rrd(&rrd_path_clone)?;

        // If we have LiDAR data, write extracted data
        if !result.lidar_frames.is_empty() {
            rrd_extractor::write_extracted_data(&result, &extract_dir_clone)
                .map_err(|e| MapperError::ProcessingFailed(e.to_string()))?;
        }

        Ok::<_, MapperError>(result)
    })
    .await
    .map_err(|e| MapperError::ProcessingFailed(format!("Task join error: {}", e)))??;

    // No LiDAR data — record skip in state (persisted to writable maps volume)
    if extraction_result.lidar_frames.is_empty() {
        warn!(path = %rrd_path.display(), "No LiDAR data in RRD, marking as skipped");