create_signal/
create_signal.rs

1// Build a signal by hand and read it back with wall-clock timestamps.
2//
3// The trick for making the domain timestamps line up with the PC clock is NOT
4// to move the origin to "now": the origin stays pinned at the Unix epoch, and
5// the current time is carried entirely by the integer domain *ticks*.  With a
6// tick resolution of 1/1000 s (one tick = one millisecond) a sample's absolute
7// time is  origin + tick/1000 s, so a tick equal to the current Unix time in
8// milliseconds reads back as the current wall-clock time.  Spacing the ticks
9// 200 apart (200 ms) then yields exactly 5 samples per second.
10
11use std::f64::consts::PI;
12use std::time::{Duration, SystemTime, UNIX_EPOCH};
13
14use opendaq::{
15    DataDescriptor, DataDescriptorBuilder, DataPacket, DataRule, Instance, Ratio, ReadTimeoutType,
16    SampleType, SignalConfig, StreamReader, StreamReaderOptions, TickConverter, Unit,
17};
18
19const SAMPLES_PER_SECOND: usize = 5;
20const TICK_RESOLUTION: Ratio = Ratio::new(1, 1000); // seconds per tick (1 ms)
21const TICKS_PER_SAMPLE: i64 = 200; // 1 / (5 samples/s * 1/1000 s/tick) = 200 ticks = 200 ms
22
23/// Render nanoseconds since the Unix epoch as a readable UTC time-of-day,
24/// e.g. "21:05:28.400000".
25fn time_of_day(unix_nanos: i128) -> String {
26    let nanos = unix_nanos.rem_euclid(86_400 * 1_000_000_000);
27    let seconds = nanos / 1_000_000_000;
28    let micros = (nanos % 1_000_000_000) / 1_000;
29    format!(
30        "{:02}:{:02}:{:02}.{micros:06}",
31        seconds / 3600,
32        seconds / 60 % 60,
33        seconds % 60
34    )
35}
36
37/// The current PC time as an integer tick count (milliseconds since the Unix
38/// epoch, i.e. since the domain origin).
39fn now_ticks() -> i64 {
40    SystemTime::now()
41        .duration_since(UNIX_EPOCH)
42        .expect("system clock is before 1970")
43        .as_millis() as i64
44}
45
46/// Send `samples` as one packet whose implicit domain ticks start at `offset`
47/// and advance by `TICKS_PER_SAMPLE` per sample.
48fn send_chunk(
49    signal: &SignalConfig,
50    domain_descriptor: &DataDescriptor,
51    value_descriptor: &DataDescriptor,
52    offset: i64,
53    samples: &[f64],
54) -> opendaq::Result<()> {
55    let count = samples.len();
56    let domain_packet = DataPacket::new(domain_descriptor, count, offset)?;
57    let packet = DataPacket::with_domain(&domain_packet, value_descriptor, count, 0)?;
58    packet.set_data::<f64>(samples)?;
59    signal.send_packet(&packet)
60}
61
62fn main() -> opendaq::Result<()> {
63    let instance = Instance::new()?;
64    let context = instance.context()?.expect("instance context");
65
66    let domain_descriptor = {
67        let builder = DataDescriptorBuilder::new()?;
68        builder.set_sample_type(SampleType::Int64)?;
69        builder.set_name("time")?;
70        // Origin stays at the epoch; the wall-clock time lives in the ticks.
71        builder.set_origin("1970-01-01T00:00:00Z")?;
72        builder.set_tick_resolution(TICK_RESOLUTION)?;
73        builder.set_unit(&Unit::new(-1, "s", "second", "time")?)?;
74        builder.set_rule(&DataRule::linear(TICKS_PER_SAMPLE, 0)?)?;
75        builder.build()?.expect("domain descriptor")
76    };
77    let value_descriptor = {
78        let builder = DataDescriptorBuilder::new()?;
79        builder.set_sample_type(SampleType::Float64)?;
80        builder.set_name("values")?;
81        builder.build()?.expect("value descriptor")
82    };
83
84    let domain_signal = SignalConfig::signal(&context, None, "time", None)?;
85    domain_signal.set_descriptor(&domain_descriptor)?;
86    let signal = SignalConfig::signal(&context, None, "values", None)?;
87    signal.set_descriptor(&value_descriptor)?;
88    signal.set_domain_signal(&domain_signal)?;
89
90    let reader = StreamReader::<f64, i64>::with_options(
91        &signal,
92        StreamReaderOptions {
93            timeout_type: ReadTimeoutType::Any,
94            ..Default::default()
95        },
96    )?;
97
98    // Stream a few seconds of a 5 Hz signal in real time.  Each one-second
99    // batch is one packet of 5 samples; the ticks stay contiguous across
100    // batches (batch k starts at start_tick + k*1000 ms), so the whole run is
101    // an unbroken 5 Hz stream anchored to when the program started.
102    let batches = 3;
103    let start_tick = now_ticks();
104    let to_timestamp = TickConverter::from_signal(&signal)?;
105    println!(
106        "Streaming {SAMPLES_PER_SECOND} samples/second, starting at {}",
107        time_of_day(to_timestamp.tick_to_unix_nanos(start_tick))
108    );
109    for batch in 0..batches {
110        let base_sample = batch * SAMPLES_PER_SECOND;
111        let offset = start_tick + base_sample as i64 * TICKS_PER_SAMPLE;
112        let samples: Vec<f64> = (0..SAMPLES_PER_SECOND)
113            .map(|i| (2.0 * PI * ((base_sample + i) as f64 / 25.0)).sin())
114            .collect();
115        send_chunk(
116            &signal,
117            &domain_descriptor,
118            &value_descriptor,
119            offset,
120            &samples,
121        )?;
122
123        let (values, ticks) = reader.read_with_domain(100, 1000)?;
124        for (value, tick) in values.iter().zip(&ticks) {
125            println!(
126                "  {}  ->  {value:.4}",
127                time_of_day(to_timestamp.tick_to_unix_nanos(*tick))
128            );
129        }
130        std::thread::sleep(Duration::from_secs(1));
131    }
132    Ok(())
133}
create_signal/create_signal.rs

create_signal/
create_signal.rs
