bacnet_objects/database/
trend_poll.rs

1//! Object-owned trend polling schedule. The caller owns synchronization and timers.
2
3use std::collections::{HashMap, HashSet};
4use std::time::Duration;
5
6use bacnet_encoding::constructed::encode_log_multiple_record;
7use bacnet_encoding::primitives::encode_property_value;
8use bacnet_types::constructed::{
9    BACnetDeviceObjectPropertyReference, BACnetLogMultipleRecord, BACnetLogRecord, LogData,
10    LogDatum, LogValue,
11};
12use bacnet_types::enums::{
13    ErrorClass, ErrorCode, LoggingType, ObjectType, PropertyIdentifier as P,
14};
15use bacnet_types::error::Error;
16use bacnet_types::primitives::{Date, ObjectIdentifier, PropertyValue, Time};
17use bytes::BytesMut;
18use tracing::warn;
19
20use super::{LocalDevice, ObjectDatabase};
21use crate::device_reference::decode_reference;
22use crate::log_buffer::ANY_VALUE_MAX_OCTETS;
23use crate::traits::{BACnetObject, MonotonicClock};
24use alignment::{Alignment, DAY};
25
26mod alignment;
27
28/// Local maximum idle/configuration reconciliation delay and failure backoff.
29/// This is a scheduling policy, not a BACnet timing guarantee.
30const RECONCILE: Duration = Duration::from_millis(100);
31
32#[derive(PartialEq)]
33struct Configuration {
34    mode: Mode,
35    /// Log_DeviceObjectProperty as read, compared whole for scheduling
36    /// ownership.
37    reference: PropertyValue,
38    /// What each acquisition reads: the one reference of a Trend Log, or every
39    /// element of a Trend Log Multiple in array order.
40    members: Vec<Member>,
41}
42
43/// When a log acquires.
44#[derive(Debug, Clone, Copy, PartialEq)]
45enum Mode {
46    /// POLLED: every `interval` hundredths, counted from each acquisition.
47    Polled { interval: u32 },
48    /// POLLED with Align_Intervals (Clauses 12.25.27 and 12.30.14): at each
49    /// local time of day `offset` hundredths past a multiple of `interval`,
50    /// which divides a day. `offset` is Interval_Offset modulo the interval
51    /// (12.25.28 and 12.30.15).
52    Aligned { interval: u32, offset: u32 },
53    /// TRIGGERED with Trigger TRUE: one acquisition now (Clauses 12.25.29
54    /// and 12.30.16).
55    Triggered,
56}
57
58/// One monitored reference of a log.
59#[derive(Debug, Clone, Copy, PartialEq)]
60enum Member {
61    /// A Trend Log Multiple element naming object or device instance 4194303,
62    /// which Clause 12.30.11 treats as empty, or the unset reference of a
63    /// triggered Trend Log, which reads the same way (#1417).
64    Unspecified,
65    /// A property to read, at one element when `index` is present.
66    Reference {
67        target: ObjectIdentifier,
68        property: P,
69        index: Option<u32>,
70        /// The reference's optional Device member.
71        device: Option<ObjectIdentifier>,
72    },
73}
74
75struct Schedule {
76    configuration: Configuration,
77    last_success: Option<Duration>,
78    /// Latest failed attempt, independent of the last accepted sample.
79    retry_completed: Option<Duration>,
80    /// An aligned log's plan, kept on the Device clock.
81    alignment: Alignment,
82}
83
84impl Schedule {
85    fn new(configuration: Configuration) -> Self {
86        Self {
87            configuration,
88            last_success: None,
89            retry_completed: None,
90            alignment: Alignment::default(),
91        }
92    }
93
94    fn remaining(&self, now: Duration) -> Duration {
95        if let Some(failed) = self.retry_completed {
96            return RECONCILE.saturating_sub(now.saturating_sub(failed));
97        }
98        match (self.configuration.mode, self.last_success) {
99            (Mode::Polled { interval }, Some(success)) => {
100                hundredths(interval).saturating_sub(now.saturating_sub(success))
101            }
102            (Mode::Aligned { interval, offset }, _) => self.alignment.wait(now, interval, offset),
103            _ => Duration::ZERO,
104        }
105    }
106}
107
108fn hundredths(value: u32) -> Duration {
109    Duration::from_millis(u64::from(value) * 10)
110}
111
112#[derive(Default)]
113pub(super) struct TrendPollSchedule(HashMap<ObjectIdentifier, Schedule>);
114
115impl TrendPollSchedule {
116    pub(super) fn retire(&mut self, oid: &ObjectIdentifier) {
117        self.0.remove(oid);
118    }
119
120    pub(super) fn clear(&mut self) {
121        self.0.clear();
122    }
123}
124
125impl ObjectDatabase {
126    /// Poll every due local Trend Log or Trend Log Multiple synchronously and
127    /// return the next wait.
128    ///
129    /// Each acquisition reads every reference of the log: a Trend Log's one,
130    /// or each Log_DeviceObjectProperty element of a Trend Log Multiple, whose
131    /// record then carries one value per element in array order (#1203).
132    /// Only this database is read. A reference whose Device member names
133    /// another device (see [`ObjectDatabase::local_device`]) is never read
134    /// here, even when a same-numbered local object exists: its value is a
135    /// failure, PROPERTY / OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED. A Trend Log
136    /// Multiple element naming instance 4194303 is empty and fails with
137    /// PROPERTY / NO_PROPERTY_SPECIFIED. A reference with an array index reads
138    /// that element (#1205). A local read that fails yields the error a
139    /// ReadProperty of it would report: OBJECT / UNKNOWN_OBJECT for a missing
140    /// object, PROPERTY / PROPERTY_IS_NOT_AN_ARRAY for an index on a property
141    /// that isn't an array, or the read's own error, such as INVALID_ARRAY_INDEX
142    /// for an index past the end (#1183).
143    ///
144    /// Only POLLED logs with a nonzero Log_Interval and at least one reference
145    /// are polled. A TRIGGERED log whose Trigger reads TRUE makes one
146    /// acquisition, and the object clears Trigger once it accepts the record
147    /// (Clauses 12.25.29 and 12.30.16). A Trend Log Multiple without members
148    /// records no values for it, and a Trend Log without a reference records
149    /// PROPERTY / NO_PROPERTY_SPECIFIED, so Trigger never stays TRUE. With
150    /// Align_Intervals TRUE and a Log_Interval that divides a day, a POLLED
151    /// log acquires when the Device clock's time of day is Interval_Offset
152    /// (modulo the interval) past a multiple of the interval (Clauses
153    /// 12.25.27-28 and 12.30.14-15); the first acquisition waits for such a
154    /// boundary. Such a log is judged against the Device clock on every pass,
155    /// so a clock change moves its plan with it (see the `alignment` module).
156    ///
157    /// Every pass also lets each Trend Log, Trend Log Multiple and Event Log
158    /// look at its Start_Time / Stop_Time window, with or without a
159    /// monotonic clock, so one that opens or closes is recorded within a pass
160    /// even when no record arrives. An Event Log takes its records from the
161    /// application and the device's notifications, never from this poller,
162    /// but this pass is the one regular visit every log gets.
163    ///
164    /// The caller must hold exclusive database access for this whole call. The
165    /// bound monotonic clock drives scheduling; the shared Device clock
166    /// provides each actual acquisition timestamp. Without either clock an
167    /// attempt cannot succeed. A successful call to an object's insertion hook,
168    /// including its accepted disabled/count-only outcomes, starts the
169    /// configured interval.
170    ///
171    /// Log_Interval is in hundredths. Deadlines follow actual completion, without
172    /// catch-up bursts. Invalid timestamps and insertion errors preserve the last
173    /// success and retry after 100 ms. The returned wait is at most 100 ms so a
174    /// timer-driven caller also reconciles configuration changes and idle logs.
175    /// Slow synchronous work can make another object due; a minimum 1 ms yield
176    /// avoids a zero-delay loop. These bounds are local policy, not real-time
177    /// guarantees. Clock, object reads and insertion hooks must remain bounded.
178    pub fn poll_trend_logs(&mut self) -> Duration {
179        let mut logs = self.find_by_type(ObjectType::TREND_LOG);
180        logs.extend(self.find_by_type(ObjectType::TREND_LOG_MULTIPLE));
181        let trend_logs = logs.len();
182        logs.extend(self.find_by_type(ObjectType::EVENT_LOG));
183        // Every log looks at its window, Event Logs included, whether or not
184        // a monotonic clock drives acquisition.
185        for oid in &logs {
186            if let Some(log) = self.get_mut(oid) {
187                log.refresh_log_window_internal();
188            }
189        }
190        logs.truncate(trend_logs);
191        let Some(monotonic) = self.monotonic_clock.clone() else {
192            return RECONCILE;
193        };
194        let mut eligible = HashSet::new();
195        let local = self.local_device();
196        for oid in logs {
197            // Exclusive access prevents structural change after selection.
198            let Some(configuration) = self.get(&oid).and_then(configuration) else {
199                continue;
200            };
201            eligible.insert(oid);
202            if self
203                .trend_poll
204                .0
205                .get(&oid)
206                .is_none_or(|entry| entry.configuration != configuration)
207            {
208                self.trend_poll.0.insert(oid, Schedule::new(configuration));
209            }
210            self.poll_one(oid, local, &*monotonic);
211        }
212        self.trend_poll.0.retain(|oid, _| eligible.contains(oid));
213        let now = monotonic();
214        let remaining = self
215            .trend_poll
216            .0
217            .values()
218            .map(|entry| entry.remaining(now))
219            .min()
220            .unwrap_or(RECONCILE)
221            .min(RECONCILE);
222        if remaining.is_zero() {
223            Duration::from_millis(1)
224        } else {
225            remaining
226        }
227    }
228
229    /// Acquire for `oid`'s schedule entry when it is due.
230    fn poll_one(&mut self, oid: ObjectIdentifier, local: LocalDevice, monotonic: &MonotonicClock) {
231        let now = monotonic();
232        let entry = &self.trend_poll.0[&oid];
233        let mode = entry.configuration.mode;
234        // An aligned log looks at the Device clock on every pass, so its plan
235        // follows a clock change; the others wait for their deadline. Both
236        // wait out the retry delay after a failed attempt.
237        let waiting = match mode {
238            Mode::Aligned { .. } => entry
239                .retry_completed
240                .is_some_and(|failed| now.saturating_sub(failed) < RECONCILE),
241            _ => !entry.remaining(now).is_zero(),
242        };
243        if waiting {
244            return;
245        }
246        let frame = self
247            .clock_frame()
248            .filter(|frame| frame.is_valid_actual_datetime());
249        let entry = self.trend_poll.0.get_mut(&oid).unwrap();
250        let mut boundary = None;
251        if let Mode::Aligned { interval, offset } = mode {
252            // Without a clock there is no boundary to find; look again later.
253            let Some(frame) = frame else {
254                entry.retry_completed = Some(now);
255                return;
256            };
257            entry.retry_completed = None;
258            boundary = entry.alignment.due(now, &frame, interval, offset);
259            if boundary.is_none() {
260                return;
261            }
262        }
263        let accepted = frame.is_some_and(|frame| {
264            let values: Vec<LogValue> = self.trend_poll.0[&oid]
265                .configuration
266                .members
267                .iter()
268                .map(|member| self.acquire(local, member))
269                .collect();
270            let object = self.get_mut(&oid).unwrap();
271            let inserted = if oid.object_type() == ObjectType::TREND_LOG_MULTIPLE {
272                object.add_trend_multiple_record(BACnetLogMultipleRecord {
273                    date: frame.local_date,
274                    time: frame.local_time,
275                    log_data: LogData::Values(values),
276                })
277            } else {
278                object.add_trend_record(BACnetLogRecord {
279                    date: frame.local_date,
280                    time: frame.local_time,
281                    // A Trend Log has exactly one reference.
282                    log_datum: values
283                        .into_iter()
284                        .next()
285                        .map_or(LogDatum::NullValue, LogDatum::from),
286                    status_flags: None,
287                })
288            };
289            inserted
290                .inspect_err(|error| {
291                    warn!(object = %oid, %error, "trend-log record insertion failed");
292                })
293                .is_ok()
294        });
295        let completed = monotonic();
296        if accepted && mode == Mode::Triggered {
297            // Served: the next Trigger starts a fresh entry.
298            self.trend_poll.0.remove(&oid);
299            return;
300        }
301        let entry = self.trend_poll.0.get_mut(&oid).unwrap();
302        if !accepted {
303            entry.retry_completed = Some(completed);
304            return;
305        }
306        entry.last_success = Some(completed);
307        entry.retry_completed = None;
308        if let Some(boundary) = boundary {
309            entry.alignment.serve(boundary);
310        }
311    }
312
313    /// One reference's value, or the failure that stopped its read
314    /// (Clause 12.25 Log_Buffer; Clause 12.30.19 for each member).
315    fn acquire(&self, local: LocalDevice, member: &Member) -> LogValue {
316        let Member::Reference {
317            target,
318            property,
319            index,
320            device,
321        } = *member
322        else {
323            return failure(ErrorClass::PROPERTY, ErrorCode::NO_PROPERTY_SPECIFIED);
324        };
325        if !local.is_local(device) {
326            // Reading another device's property is not supported here.
327            return failure(
328                ErrorClass::PROPERTY,
329                ErrorCode::OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED,
330            );
331        }
332        let Some(target) = self.get(&target) else {
333            return failure(ErrorClass::OBJECT, ErrorCode::UNKNOWN_OBJECT);
334        };
335        // ReadProperty's gate, which the Averaging sampler shares: only an
336        // array has elements, whatever an object's read arm does with the
337        // index.
338        if index.is_some() && !target.is_array_property(property) {
339            return failure(ErrorClass::PROPERTY, ErrorCode::PROPERTY_IS_NOT_AN_ARRAY);
340        }
341        match target.read_property(property, index) {
342            Ok(value) => property_value_to_log_value(&value),
343            Err(Error::Protocol { class, code } | Error::Structured { class, code, .. }) => {
344                LogValue::Failure {
345                    error_class: class,
346                    error_code: code,
347                }
348            }
349            // An error with no BACnet class and code reports as ReadProperty
350            // would answer it.
351            Err(_) => failure(ErrorClass::SERVICES, ErrorCode::OTHER),
352        }
353    }
354}
355
356fn failure(class: ErrorClass, code: ErrorCode) -> LogValue {
357    LogValue::Failure {
358        error_class: u32::from(class.to_raw()),
359        error_code: u32::from(code.to_raw()),
360    }
361}
362
363/// What a log's properties ask of the poller now; `None` when nothing.
364fn configuration(object: &dyn BACnetObject) -> Option<Configuration> {
365    let read = |property| object.read_property(property, None).ok();
366    let logging_type = match read(P::LOGGING_TYPE) {
367        Some(PropertyValue::Enumerated(value)) => LoggingType::from_raw(value),
368        _ => LoggingType::POLLED,
369    };
370    let mode = match logging_type {
371        // An object without a Trigger row is never triggered.
372        LoggingType::TRIGGERED => {
373            (read(P::TRIGGER) == Some(PropertyValue::Boolean(true))).then_some(Mode::Triggered)?
374        }
375        LoggingType::POLLED => {
376            let Some(PropertyValue::Unsigned(raw)) = read(P::LOG_INTERVAL) else {
377                return None;
378            };
379            let interval = u32::try_from(raw).ok().filter(|value| *value > 0)?;
380            let offset = match read(P::INTERVAL_OFFSET) {
381                Some(PropertyValue::Unsigned(offset)) => (offset % u64::from(interval)) as u32,
382                _ => 0,
383            };
384            let aligned = read(P::ALIGN_INTERVALS) == Some(PropertyValue::Boolean(true));
385            // Clause 12.30.14 aligns only an interval that goes evenly into
386            // one of its clock periods; each of those goes evenly into a day,
387            // so going evenly into a day is the same test.
388            if aligned && DAY.is_multiple_of(interval) {
389                Mode::Aligned { interval, offset }
390            } else {
391                Mode::Polled { interval }
392            }
393        }
394        _ => return None,
395    };
396    let reference = read(P::LOG_DEVICE_OBJECT_PROPERTY)?;
397    let members = if object.object_identifier().object_type() == ObjectType::TREND_LOG_MULTIPLE {
398        let PropertyValue::List(elements) = &reference else {
399            return None;
400        };
401        elements.iter().map(member).collect::<Option<Vec<_>>>()?
402    } else {
403        match member(&reference)? {
404            // A log with no reference polls nothing. A Trigger on one is
405            // still served, with a failure, so it doesn't stay TRUE.
406            Member::Unspecified if mode != Mode::Triggered => return None,
407            member => vec![member],
408        }
409    };
410    // A Trigger is served even with no members, so it never stays TRUE.
411    if members.is_empty() && mode != Mode::Triggered {
412        return None;
413    }
414    Some(Configuration {
415        mode,
416        reference,
417        members,
418    })
419}
420
421/// One reference as a read serves it: a single
422/// BACnetDeviceObjectPropertyReference in its Clause 21 encoding (#1234).
423/// Anything else leaves the log unpolled. An unset reference, its object or
424/// Device at instance 4194303, names nothing: an empty Trend Log Multiple
425/// element (Clause 12.30.11), or a Trend Log without a reference (#1417).
426fn member(value: &PropertyValue) -> Option<Member> {
427    let reference: BACnetDeviceObjectPropertyReference = decode_reference(value).ok()?;
428    if reference.is_unset() {
429        return Some(Member::Unspecified);
430    }
431    Some(Member::Reference {
432        target: reference.object_identifier,
433        property: P::from_raw(reference.property_identifier),
434        index: reference.property_array_index,
435        device: reference.device_identifier,
436    })
437}
438
439/// A read value as the log stores it: one of the datatypes both record kinds
440/// name, or else the any-value alternative holding the value's own encoding,
441/// the bytes a ReadProperty of it would carry (Clauses 12.25.14 and
442/// 12.30.19).
443///
444/// An any-value over [`ANY_VALUE_MAX_OCTETS`] logs PROPERTY /
445/// VALUE_TOO_LONG instead, so every record stays small enough to page. A
446/// value no record could carry (one that fails to encode, a bit string with
447/// impossible padding, framed bytes whose tags don't balance) logs SERVICES
448/// / OTHER, so the log never refuses what the poller hands it.
449fn property_value_to_log_value(value: &PropertyValue) -> LogValue {
450    let value = match value {
451        PropertyValue::Real(v) => LogValue::RealValue(*v),
452        PropertyValue::Unsigned(v) => LogValue::UnsignedValue(*v),
453        PropertyValue::Signed(v) => LogValue::SignedValue(i64::from(*v)),
454        PropertyValue::Boolean(v) => LogValue::BooleanValue(*v),
455        PropertyValue::Enumerated(v) => LogValue::EnumValue(u64::from(*v)),
456        PropertyValue::BitString { unused_bits, data } => LogValue::BitstringValue {
457            unused_bits: *unused_bits,
458            data: data.clone(),
459        },
460        PropertyValue::Null => LogValue::NullValue,
461        other => {
462            let mut encoded = BytesMut::new();
463            if encode_property_value(&mut encoded, other).is_err() {
464                // A value the read could not have carried either.
465                return failure(ErrorClass::SERVICES, ErrorCode::OTHER);
466            }
467            if encoded.len() > ANY_VALUE_MAX_OCTETS {
468                return failure(ErrorClass::PROPERTY, ErrorCode::VALUE_TOO_LONG);
469            }
470            LogValue::AnyValue(encoded.to_vec())
471        }
472    };
473    if !matches!(
474        value,
475        LogValue::BitstringValue { .. } | LogValue::AnyValue(_)
476    ) {
477        return value;
478    }
479    // Only these two alternatives can hold something no record carries;
480    // encoding a one-member record tells.
481    let probe = BACnetLogMultipleRecord {
482        date: Date {
483            year: 0,
484            month: 1,
485            day: 1,
486            day_of_week: 1,
487        },
488        time: Time {
489            hour: 0,
490            minute: 0,
491            second: 0,
492            hundredths: 0,
493        },
494        log_data: LogData::Values(vec![value.clone()]),
495    };
496    match encode_log_multiple_record(&probe, &mut BytesMut::new()) {
497        Ok(()) => value,
498        Err(_) => failure(ErrorClass::SERVICES, ErrorCode::OTHER),
499    }
500}
501
502#[cfg(test)]
503mod tests;
bacnet_objects/database/trend_poll.rs

bacnet_objects/database/
trend_poll.rs
