telemetry_visualization/examples/simple_producer/src/main.rs

use api::client::telemetry::Telemetry;
use api::client::Client;
use api::data_type::DataType;
use api::data_value::DataValue;
use chrono::{TimeDelta, Utc};
use futures_util::future::join_all;
use num_traits::FloatConst;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::{sleep_until, Instant};
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    env_logger::init();

    let client = Arc::new(Client::connect("ws://[::1]:8080/backend")?);

    let time_offset = Telemetry::register(
        client.clone(),
        "simple_producer/time_offset".into(),
        DataType::Float64,
    )
    .await;

    let publish_offset = Telemetry::register(
        client.clone(),
        "simple_producer/publish_offset".into(),
        DataType::Float64,
    )
    .await;

    let await_offset = Telemetry::register(
        client.clone(),
        "simple_producer/await_offset".into(),
        DataType::Float64,
    )
    .await;

    let sin_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin".into(),
        DataType::Float32,
    )
    .await;
    let cos_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos".into(),
        DataType::Float64,
    )
    .await;
    let bool_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/bool".into(),
        DataType::Boolean,
    )
    .await;

    let sin2_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin2".into(),
        DataType::Float32,
    )
    .await;
    let cos2_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos2".into(),
        DataType::Float64,
    )
    .await;

    let sin3_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin3".into(),
        DataType::Float32,
    )
    .await;
    let cos3_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos3".into(),
        DataType::Float64,
    )
    .await;

    let sin4_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin4".into(),
        DataType::Float32,
    )
    .await;
    let cos4_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos4".into(),
        DataType::Float64,
    )
    .await;

    let sin5_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin5".into(),
        DataType::Float32,
    )
    .await;
    let cos5_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos5".into(),
        DataType::Float64,
    )
    .await;

    let sin6_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/sin6".into(),
        DataType::Float32,
    )
    .await;
    let cos6_tlm_handle = Telemetry::register(
        client.clone(),
        "simple_producer/cos6".into(),
        DataType::Float64,
    )
    .await;

    let cancellation_token = CancellationToken::new();
    {
        let cancellation_token = cancellation_token.clone();
        tokio::spawn(async move {
            let _ = tokio::signal::ctrl_c().await;
            cancellation_token.cancel();
        });
    }

    let start_time = Utc::now();
    let start_instant = Instant::now();
    let mut next_time = start_instant;
    let mut index = 0;
    let mut tasks = vec![];
    while !cancellation_token.is_cancelled() {
        next_time += Duration::from_millis(10);
        index += 1;
        sleep_until(next_time).await;
        let publish_time = start_time + TimeDelta::from_std(next_time - start_instant).unwrap();
        let actual_time = Instant::now();
        tasks.push(time_offset.publish(
            DataValue::Float64((actual_time - next_time).as_secs_f64()),
            Utc::now(),
        ));
        tasks.push(sin_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()),
            publish_time,
        ));
        tasks.push(bool_tlm_handle.publish(DataValue::Boolean(index % 1000 > 500), publish_time));
        tasks.push(sin2_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos2_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()),
            publish_time,
        ));
        tasks.push(sin3_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos3_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()),
            publish_time,
        ));
        tasks.push(sin4_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos4_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()),
            publish_time,
        ));
        tasks.push(sin5_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos5_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()),
            publish_time,
        ));
        tasks.push(sin6_tlm_handle.publish(
            DataValue::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()),
            publish_time,
        ));
        tasks.push(cos6_tlm_handle.publish(
            DataValue::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()),
            publish_time,
        ));

        tasks.push(publish_offset.publish(
            DataValue::Float64((Instant::now() - actual_time).as_secs_f64()),
            Utc::now(),
        ));

        // Join the tasks so they all run in parallel
        for task in join_all(tasks.drain(..)).await {
            task?;
        }

        tasks.push(await_offset.publish(
            DataValue::Float64((Instant::now() - actual_time).as_secs_f64()),
            Utc::now(),
        ));
    }

    Ok(())
}