clankers 0.1.6

//! End-to-end: `record_mcap` tapes configured topics into an MCAP file that
//! round-trips through inspect and replay.

use std::path::PathBuf;

use clankers::prelude::*;
use clankers::recording::{run_with_recording, McapRecorder, RecordingPlan};
use clankers::ros2::RosMessage;

#[test]
fn plan_resolution_respects_config_and_env() {
    // Default config: recording off.
    let ctx = RobotContext::new(ClankeRSConfig::default(), "/proj");
    assert!(RecordingPlan::resolve(&ctx, None, None).is_none());

    // Environment forces it on and picks the output path (what `clankers
    // record` sets).
    let plan = RecordingPlan::resolve(&ctx, Some("1"), Some("out/run.mcap")).unwrap();
    assert_eq!(plan.path, PathBuf::from("out/run.mcap"));

    // Config turns it on; path lands under work_dir/output_dir.
    let toml = r#"
[logging]
record_mcap = true
output_dir = "logs/"

[topics.input.camera]
name = "/plan/image"
type = "sensor_msgs/Image"

[topics.output.detections]
name = "/plan/detections"
"#;
    let ctx = RobotContext::new(ClankeRSConfig::parse(toml).unwrap(), "/proj");
    let plan = RecordingPlan::resolve(&ctx, None, None).unwrap();
    assert!(plan.path.starts_with("/proj/logs"));
    assert_eq!(
        plan.topics,
        vec![
            ("/plan/detections".to_string(), "clankeRS/Json".to_string()),
            ("/plan/image".to_string(), "sensor_msgs/Image".to_string()),
        ]
    );

    // Environment can also force recording OFF despite the config.
    assert!(RecordingPlan::resolve(&ctx, Some("0"), None).is_none());
}

#[tokio::test]
async fn records_configured_topics_and_replays_them() {
    let tmp = tempfile::tempdir().unwrap();
    let toml = r#"
[project]
name = "rec_node"

[topics.input.camera]
name = "/rec_rt/image"
type = "sensor_msgs/Image"

[topics.output.detections]
name = "/rec_rt/detections"
type = "clankeRS/DetectionArray"

[logging]
record_mcap = true
output_dir = "logs/"
"#;
    let ctx = RobotContext::new(ClankeRSConfig::parse(toml).unwrap(), tmp.path());
    let recorder = McapRecorder::start(&ctx)
        .await
        .unwrap()
        .expect("recording enabled by config");

    let node = RobotNode::new("rec_node").await.unwrap();
    let images = node
        .publish::<ImageMsg>("/rec_rt/image", QosProfile::sensor_data())
        .await
        .unwrap();
    let detections = node
        .publish::<DetectionArray>("/rec_rt/detections", QosProfile::default())
        .await
        .unwrap();

    for i in 0..3u8 {
        images
            .publish(ImageMsg::new(4, 4, vec![i; 4 * 4 * 3]))
            .await
            .unwrap();
    }
    detections
        .publish(DetectionArray {
            stamp_nanos: 42,
            frame_id: "camera".into(),
            detections: vec![],
        })
        .await
        .unwrap();

    let summary = recorder.finish().await.unwrap();
    assert_eq!(summary.messages, 4);
    assert!(summary.path.starts_with(tmp.path()));

    // The file inspects with the configured schema names...
    let log = McapLog::open(&summary.path).unwrap();
    let report = log.report();
    let mut names: Vec<_> = report.topics.iter().map(|t| t.name.as_str()).collect();
    names.sort();
    assert_eq!(names, ["/rec_rt/detections", "/rec_rt/image"]);
    assert!(report
        .topics
        .iter()
        .any(|t| t.schema.as_deref() == Some("sensor_msgs/Image")));

    // ...the payloads decode back to the published messages...
    let image_records = log.topic_messages("/rec_rt/image").unwrap();
    assert_eq!(image_records.len(), 3);
    let decoded = ImageMsg::deserialize(&image_records[0].data).unwrap();
    assert_eq!((decoded.width, decoded.height), (4, 4));

    // ...and the log replays like any other robot log.
    let replay = Replay::from_mcap(&summary.path).unwrap();
    let result = replay.run(|_msg| async { Ok(()) }).await.unwrap();
    assert_eq!(result.summary.input_messages, 4);
    assert_eq!(result.summary.dropped_messages, 0);
}

#[tokio::test]
async fn run_with_recording_wraps_a_node_future() {
    let tmp = tempfile::tempdir().unwrap();
    let toml = r#"
[project]
name = "wrapped_node"

[topics.output.status]
name = "/rec_wrap/status"

[logging]
record_mcap = true
output_dir = "logs/"
"#;
    let ctx = RobotContext::new(ClankeRSConfig::parse(toml).unwrap(), tmp.path());

    run_with_recording(ctx.clone(), async {
        let node = RobotNode::new("wrapped_node").await?;
        let status = node
            .publish::<DetectionArray>("/rec_wrap/status", QosProfile::default())
            .await?;
        status
            .publish(DetectionArray {
                stamp_nanos: 7,
                frame_id: "base".into(),
                detections: vec![],
            })
            .await?;
        Ok(())
    })
    .await
    .unwrap();

    let logs_dir = tmp.path().join("logs");
    let recorded: Vec<_> = std::fs::read_dir(&logs_dir)
        .expect("logs dir created")
        .filter_map(|e| e.ok())
        .map(|e| e.path())
        .filter(|p| p.extension().is_some_and(|ext| ext == "mcap"))
        .collect();
    assert_eq!(recorded.len(), 1, "expected one recording in {logs_dir:?}");

    let log = McapLog::open(&recorded[0]).unwrap();
    let messages = log.messages().unwrap();
    assert_eq!(messages.len(), 1);
    assert_eq!(messages[0].topic, "/rec_wrap/status");
}