pamoja_profile/node.rs
1//! The ready-to-run node a profile assembles around real components.
2
3use core::convert::Infallible;
4use core::future::Future;
5use core::time::Duration;
6
7use pamoja_codec::Codec;
8use pamoja_core::{Actuator, Result, Sensor, Transport};
9use pamoja_power::PowerMode;
10
11use crate::{BoxedPolicy, Controller, Policy, PolicyRegistry, Profile, Reaction};
12
13/// An actuator that accepts and ignores commands.
14///
15/// Profiles that only observe - such as a well-level monitor - have no output to
16/// drive. [`Node::monitor`] wires this in their place so a node has one uniform
17/// shape whether or not it switches an actuator.
18#[derive(Clone, Copy, Debug, Default)]
19pub struct NoActuator;
20
21impl Actuator for NoActuator {
22 type Command = bool;
23
24 async fn apply(&mut self, _command: bool) -> Result<()> {
25 Ok(())
26 }
27}
28
29/// A profile assembled around the components that make it run.
30///
31/// The node is the thin shell that ties a [`Profile`]'s decision logic to real I/O.
32/// Each [`tick`](Node::tick) reads the sensor, runs the node's policy, drives the
33/// actuator when the policy calls for it, and publishes the reading over the
34/// transport with the supplied codec. The control math lives in `pamoja-kit`, the
35/// power schedule in `pamoja-power`, and the wire format in the codec, so the node
36/// adds composition, not behavior.
37///
38/// [`Node::new`] runs the profile's [`Controller`], which reads real-world `f32`
39/// units (degrees, percent, liters), the form the `pamoja-kit` controllers expect; a
40/// driver is responsible for calibrating raw counts into those units before the node
41/// sees them, or [`Sensor::map`] selects one. [`Node::with_policy`] runs a
42/// [`Policy`] of your own instead, over whatever the sensor reads and whatever the
43/// actuator takes.
44///
45/// # Examples
46///
47/// ```
48/// use pamoja_codec::CborCodec;
49/// use pamoja_core::{Actuator, Result, Sensor, Transport};
50/// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
51/// use pamoja_profile::{Node, Profile};
52///
53/// struct Probe(f32);
54/// impl Sensor for Probe {
55/// type Reading = f32;
56/// async fn read(&mut self) -> Result<f32> {
57/// Ok(self.0)
58/// }
59/// }
60///
61/// struct Cooler;
62/// impl Actuator for Cooler {
63/// type Command = bool;
64/// async fn apply(&mut self, _on: bool) -> Result<()> {
65/// Ok(())
66/// }
67/// }
68///
69/// # async fn run() -> Result<()> {
70/// let broker = LoopbackBroker::new();
71/// let mut link = LoopbackTransport::new(broker);
72/// link.connect().await?;
73///
74/// // A warm fridge, assembled straight from its profile.
75/// let mut node = Node::new(Profile::vaccine_fridge_monitor(), Probe(9.0), Cooler, link, CborCodec)?;
76/// let reaction = node.tick().await?.reaction;
77/// assert_eq!(reaction.actuator, Some(true)); // the cooler runs
78/// assert!(reaction.alert.is_some()); // and 9 C is a spoilage excursion
79/// # Ok(())
80/// # }
81/// ```
82pub struct Node<S, A, T, C, P = Controller> {
83 profile: Profile,
84 policy: P,
85 sensor: S,
86 actuator: A,
87 transport: T,
88 codec: C,
89 mode: Option<PowerMode>,
90}
91
92impl<S, A, T, C> Node<S, A, T, C> {
93 /// Assembles a node from a profile and the components that drive it.
94 ///
95 /// # Arguments
96 ///
97 /// * `profile` - the profile to assemble; its policy becomes the node's controller.
98 /// * `sensor` - the source of readings.
99 /// * `actuator` - the output the controller switches.
100 /// * `transport` - the link readings are published over; expected to be connected.
101 /// * `codec` - the wire format readings are encoded with.
102 ///
103 /// # Returns
104 ///
105 /// A node ready to [`tick`](Node::tick).
106 ///
107 /// # Errors
108 ///
109 /// Returns [`Error::Codec`](pamoja_core::Error::Codec) when the profile names a custom
110 /// control kind, which no built-in controller decides; [`Node::resolve`] assembles
111 /// one through a [`PolicyRegistry`] that registers it.
112 pub fn new(profile: Profile, sensor: S, actuator: A, transport: T, codec: C) -> Result<Self> {
113 let controller = profile.controller()?;
114 Ok(Node::with_policy(
115 profile, controller, sensor, actuator, transport, codec,
116 ))
117 }
118}
119
120impl<S, A, T, C> Node<S, A, T, C, BoxedPolicy> {
121 /// Assembles a node whose policy a registry resolves from the profile's control kind.
122 ///
123 /// A built-in kind resolves to its controller, and a custom kind to the policy the
124 /// factory registered under its name builds from the parameters beside it, so one
125 /// program runs any manifest its registry covers.
126 ///
127 /// # Arguments
128 ///
129 /// * `profile` - the profile to assemble.
130 /// * `registry` - what resolves the profile's control kind.
131 /// * `sensor` - the source of readings.
132 /// * `actuator` - the output the policy switches.
133 /// * `transport` - the link readings are published over; expected to be connected.
134 /// * `codec` - the wire format readings are encoded with.
135 ///
136 /// # Returns
137 ///
138 /// A node ready to [`tick`](Node::tick).
139 ///
140 /// # Errors
141 ///
142 /// Returns [`Error::Codec`](pamoja_core::Error::Codec) naming the kind when it is
143 /// custom and not registered, or whatever the kind's factory returns.
144 ///
145 /// # Examples
146 ///
147 /// ```
148 /// use pamoja_codec::JsonCodec;
149 /// use pamoja_core::{Error, Result, Sensor, Transport};
150 /// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
151 /// use pamoja_profile::{BoxedPolicy, Controller, NoActuator, Node, PolicyRegistry, Profile};
152 ///
153 /// struct Air(f32);
154 /// impl Sensor for Air {
155 /// type Reading = f32;
156 /// async fn read(&mut self) -> Result<f32> {
157 /// Ok(self.0)
158 /// }
159 /// }
160 ///
161 /// # async fn run() -> Result<()> {
162 /// let profile = Profile::from_json(r#"{
163 /// "name": "orchard-frost",
164 /// "reads": { "quantity": "temperature", "unit": "celsius" },
165 /// "topic": "orchard/air/temperature",
166 /// "control": { "kind": "frost_guard", "warn_below": 2.0 },
167 /// "power": { "active_secs": 60, "saver_secs": 300, "critical_secs": 900 }
168 /// }"#)?;
169 /// let registry = PolicyRegistry::new().register("frost_guard", |params| {
170 /// let warn_below = params.number("warn_below").ok_or(Error::Unsupported("warn_below"))?;
171 /// Ok(Box::new(Controller::setpoint(warn_below as f32, 0.5, false, 5.0)) as BoxedPolicy)
172 /// });
173 ///
174 /// let mut link = LoopbackTransport::new(LoopbackBroker::new());
175 /// link.connect().await?;
176 /// let mut node = Node::resolve(profile, ®istry, Air(1.0), NoActuator, link, JsonCodec)?;
177 /// assert_eq!(node.tick().await?.reaction.actuator, Some(true)); // cold: the heater comes on
178 /// # Ok(())
179 /// # }
180 /// ```
181 pub fn resolve(
182 profile: Profile,
183 registry: &PolicyRegistry,
184 sensor: S,
185 actuator: A,
186 transport: T,
187 codec: C,
188 ) -> Result<Self> {
189 let policy = registry.resolve(&profile.control)?;
190 Ok(Node::with_policy(
191 profile, policy, sensor, actuator, transport, codec,
192 ))
193 }
194}
195
196impl<S, A, T, C, P> Node<S, A, T, C, P> {
197 /// Assembles a node around a policy of your own.
198 ///
199 /// The profile still names the topic and the power schedule; the policy decides
200 /// each reading, so a node can read a driver's whole measurement and command an
201 /// actuator that takes more than on and off. A policy a [`PolicyRegistry`] resolved
202 /// from the profile's own control kind goes here too.
203 ///
204 /// # Arguments
205 ///
206 /// * `profile` - the profile to assemble.
207 /// * `policy` - what decides each reading.
208 /// * `sensor` - the source of readings.
209 /// * `actuator` - the output the policy commands.
210 /// * `transport` - the link readings are published over; expected to be connected.
211 /// * `codec` - the wire format readings are encoded with.
212 ///
213 /// # Returns
214 ///
215 /// A node ready to [`tick`](Node::tick).
216 ///
217 /// # Examples
218 ///
219 /// ```
220 /// use pamoja_codec::CborCodec;
221 /// use pamoja_core::{Actuator, Result, Sensor, Transport};
222 /// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
223 /// use pamoja_profile::{Alert, Node, Policy, Profile, Reaction};
224 /// use serde::{Deserialize, Serialize};
225 ///
226 /// // A driver reads two quantities at once; the policy decides on both.
227 /// #[derive(Clone, Copy, Serialize, Deserialize)]
228 /// struct Soil {
229 /// moisture: f32,
230 /// temperature: f32,
231 /// }
232 ///
233 /// struct Probe;
234 /// impl Sensor for Probe {
235 /// type Reading = Soil;
236 /// async fn read(&mut self) -> Result<Soil> {
237 /// Ok(Soil { moisture: 22.0, temperature: 1.5 })
238 /// }
239 /// }
240 ///
241 /// struct Valve;
242 /// impl Actuator for Valve {
243 /// type Command = bool;
244 /// async fn apply(&mut self, _open: bool) -> Result<()> {
245 /// Ok(())
246 /// }
247 /// }
248 ///
249 /// // Water only when the bed is dry and not about to freeze.
250 /// struct DripPolicy;
251 /// impl Policy for DripPolicy {
252 /// type Reading = Soil;
253 /// type Command = bool;
254 /// fn evaluate(&mut self, soil: &Soil) -> Reaction {
255 /// let frost = soil.temperature < 2.0;
256 /// Reaction {
257 /// actuator: Some(soil.moisture < 25.0 && !frost),
258 /// alert: frost.then_some(Alert::Custom { code: "FrostRisk", value: soil.temperature }),
259 /// }
260 /// }
261 /// }
262 ///
263 /// # async fn run() -> Result<()> {
264 /// let mut link = LoopbackTransport::new(LoopbackBroker::new());
265 /// link.connect().await?;
266 /// let profile = Profile::irrigation_node();
267 /// let mut node = Node::with_policy(profile, DripPolicy, Probe, Valve, link, CborCodec);
268 /// let reaction = node.tick().await?.reaction;
269 /// assert_eq!(reaction.actuator, Some(false)); // dry, but frost is closer
270 /// assert_eq!(reaction.alert.map(Alert::kind), Some("FrostRisk"));
271 /// # Ok(())
272 /// # }
273 /// ```
274 ///
275 /// [`PolicyRegistry`]: crate::PolicyRegistry
276 pub fn with_policy(
277 profile: Profile,
278 policy: P,
279 sensor: S,
280 actuator: A,
281 transport: T,
282 codec: C,
283 ) -> Self {
284 Self {
285 profile,
286 policy,
287 sensor,
288 actuator,
289 transport,
290 codec,
291 mode: None,
292 }
293 }
294
295 /// Returns the profile this node was assembled from.
296 ///
297 /// # Returns
298 ///
299 /// A reference to the node's [`Profile`].
300 pub fn profile(&self) -> &Profile {
301 &self.profile
302 }
303
304 /// Returns the policy deciding this node's readings.
305 ///
306 /// # Returns
307 ///
308 /// A reference to the node's policy.
309 pub fn policy(&self) -> &P {
310 &self.policy
311 }
312
313 /// The sensor the node reads, so a driver can be reconfigured between ticks.
314 ///
315 /// # Returns
316 ///
317 /// The sensor.
318 pub fn sensor_mut(&mut self) -> &mut S {
319 &mut self.sensor
320 }
321
322 /// The actuator the node drives, so its own state can be read or set directly.
323 ///
324 /// The control loop switches this on every [`tick`](Node::tick). Reaching it is for
325 /// what the loop does not do: reading back what a recording output captured, or
326 /// driving it by hand for a manual override.
327 ///
328 /// # Returns
329 ///
330 /// The actuator.
331 pub fn actuator_mut(&mut self) -> &mut A {
332 &mut self.actuator
333 }
334
335 /// The link readings are published over.
336 ///
337 /// A node on a link that comes and goes usually publishes through a
338 /// [`TransportLadder`](https://docs.rs/pamoja-ladder), which buffers a reading no
339 /// rung would take. Flushing that buffer once the link is back is the caller's, so
340 /// this is how the caller reaches it.
341 ///
342 /// # Returns
343 ///
344 /// The transport.
345 pub fn transport_mut(&mut self) -> &mut T {
346 &mut self.transport
347 }
348
349 /// Takes the node apart into the pieces it was assembled from.
350 ///
351 /// # Returns
352 ///
353 /// The profile, the policy, the sensor, the actuator, the transport, and the codec.
354 pub fn into_parts(self) -> (Profile, P, S, A, T, C) {
355 (
356 self.profile,
357 self.policy,
358 self.sensor,
359 self.actuator,
360 self.transport,
361 self.codec,
362 )
363 }
364
365 /// Returns the power mode and wait interval for the next cycle.
366 ///
367 /// This assembles the profile's [`PowerSchedule`](crate::PowerSchedule) into a
368 /// `pamoja-power` governor: as the battery drains the interval stretches, and a
369 /// charging panel eases the node back toward its active cadence. The node remembers
370 /// the mode it chose, so a charge hovering at a threshold keeps it in the lower mode
371 /// until the charge clears the schedule's hysteresis margin. The node never sleeps;
372 /// the caller waits the returned [`Duration`] before the next [`tick`](Node::tick),
373 /// so timing stays outside the node and the decision logic remains synchronous and
374 /// testable.
375 ///
376 /// # Arguments
377 ///
378 /// * `soc` - the battery state of charge in `[0.0, 1.0]`.
379 /// * `charging` - whether the panel is currently delivering charge.
380 ///
381 /// # Returns
382 ///
383 /// The [`PowerMode`] to run in and how long to wait before the next cycle.
384 pub fn schedule(&mut self, soc: f32, charging: bool) -> (PowerMode, Duration) {
385 let plan = self.profile.power.plan();
386 let mode = match self.mode {
387 Some(current) => plan.next_mode_while_charging(current, soc, charging),
388 None => plan.mode_while_charging(soc, charging),
389 };
390 self.mode = Some(mode);
391 (mode, plan.interval_for(mode))
392 }
393
394 /// Returns the power mode the last [`schedule`](Node::schedule) chose.
395 ///
396 /// # Returns
397 ///
398 /// The mode, or `None` before the first schedule.
399 pub fn power_mode(&self) -> Option<PowerMode> {
400 self.mode
401 }
402}
403
404impl<S, T, C> Node<S, NoActuator, T, C> {
405 /// Assembles a node for a profile that observes without driving an output.
406 ///
407 /// Wires a [`NoActuator`] in place of a real output, so a monitoring profile such
408 /// as [`well_level`](Profile::well_level) reads and publishes with the same shape
409 /// as a controlling one.
410 ///
411 /// # Arguments
412 ///
413 /// * `profile` - the profile to assemble.
414 /// * `sensor` - the source of readings.
415 /// * `transport` - the link readings are published over; expected to be connected.
416 /// * `codec` - the wire format readings are encoded with.
417 ///
418 /// # Returns
419 ///
420 /// A node ready to [`tick`](Node::tick), with no output to switch.
421 ///
422 /// # Errors
423 ///
424 /// Returns [`Error::Codec`](pamoja_core::Error::Codec) when the profile names a custom
425 /// control kind, as [`Node::new`] does.
426 pub fn monitor(profile: Profile, sensor: S, transport: T, codec: C) -> Result<Self> {
427 Node::new(profile, sensor, NoActuator, transport, codec)
428 }
429}
430
431impl<S, A, T, C, P> Node<S, A, T, C, P>
432where
433 S: Sensor,
434 P: Policy<Reading = S::Reading>,
435 P::Command: Clone,
436 A: Actuator<Command = P::Command>,
437 T: Transport,
438 C: Codec<S::Reading>,
439{
440 /// Runs one read-decide-act-publish cycle.
441 ///
442 /// Reads the sensor, evaluates the node's policy, applies the resulting command
443 /// to the actuator when the policy calls for one, and publishes the reading to
444 /// the profile's topic.
445 ///
446 /// # Returns
447 ///
448 /// The [`Tick`]: the reading, and the [`Reaction`] the policy produced, the command
449 /// that was applied (if any) and any alert the reading raised.
450 ///
451 /// # Errors
452 ///
453 /// Returns [`Error::Io`](pamoja_core::Error::Io) or
454 /// [`Error::Closed`](pamoja_core::Error::Closed) if the sensor read or the
455 /// actuator command fails, [`Error::Codec`](pamoja_core::Error::Codec) if the
456 /// reading cannot be encoded, and [`Error::Transport`](pamoja_core::Error::Transport)
457 /// or [`Error::Closed`](pamoja_core::Error::Closed) if the publish fails.
458 pub async fn tick(&mut self) -> Result<Tick<S::Reading, P::Command>> {
459 let reading = self.sensor.read().await?;
460 let reaction = self.policy.evaluate(&reading);
461 if let Some(command) = reaction.actuator.clone() {
462 self.actuator.apply(command).await?;
463 }
464 let payload = self.codec.encode(&reading)?;
465 self.transport.send(&self.profile.topic, &payload).await?;
466 Ok(Tick { reading, reaction })
467 }
468
469 /// Ticks, then waits the interval the battery's charge calls for, for as long as every
470 /// tick succeeds.
471 ///
472 /// The node keeps no clock of its own: `wait` is the runtime's sleep, such as
473 /// `tokio::time::sleep`, so a node runs on any executor, and a test hands it one that
474 /// returns at once. A program that carries on past a failed tick writes this loop
475 /// itself from [`tick`](Node::tick) and [`schedule`](Node::schedule).
476 ///
477 /// # Arguments
478 ///
479 /// * `battery` - reads the state of charge, from 0 to 1, and whether the panel is
480 /// charging, before each wait; `|| (1.0, false)` for a node on mains power.
481 /// * `wait` - waits a duration.
482 ///
483 /// # Returns
484 ///
485 /// Only an error: the loop runs until a tick fails.
486 ///
487 /// # Errors
488 ///
489 /// The first failed tick, with what [`tick`](Node::tick) returns; a sensor that has
490 /// ended reports [`Error::Closed`](pamoja_core::Error::Closed).
491 ///
492 /// # Examples
493 ///
494 /// ```
495 /// use core::time::Duration;
496 ///
497 /// use pamoja_codec::JsonCodec;
498 /// use pamoja_core::{Actuator, Error, Result, Sensor, Transport};
499 /// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
500 /// use pamoja_profile::{Node, Profile};
501 ///
502 /// // A morning of readings that ends, which ends the loop.
503 /// struct Morning(Vec<f32>);
504 /// impl Sensor for Morning {
505 /// type Reading = f32;
506 /// async fn read(&mut self) -> Result<f32> {
507 /// if self.0.is_empty() { Err(Error::Closed) } else { Ok(self.0.remove(0)) }
508 /// }
509 /// }
510 ///
511 /// struct Lamp;
512 /// impl Actuator for Lamp {
513 /// type Command = bool;
514 /// async fn apply(&mut self, _on: bool) -> Result<()> {
515 /// Ok(())
516 /// }
517 /// }
518 ///
519 /// # async fn run() -> Result<()> {
520 /// let mut link = LoopbackTransport::new(LoopbackBroker::new());
521 /// link.connect().await?;
522 /// let profile = Profile::vaccine_fridge_monitor();
523 /// let mut node = Node::new(profile, Morning(vec![4.0, 6.0]), Lamp, link, JsonCodec)?;
524 ///
525 /// let mut waited = Vec::new();
526 /// let ended = node
527 /// .run(|| (0.3, false), |interval: Duration| {
528 /// waited.push(interval);
529 /// async {}
530 /// })
531 /// .await;
532 /// assert!(matches!(ended, Err(Error::Closed)));
533 /// assert_eq!(waited, [Duration::from_secs(300), Duration::from_secs(300)]); // saver
534 /// # Ok(())
535 /// # }
536 /// ```
537 pub async fn run<W, F>(
538 &mut self,
539 mut battery: impl FnMut() -> (f32, bool),
540 mut wait: W,
541 ) -> Result<Infallible>
542 where
543 W: FnMut(Duration) -> F,
544 F: Future<Output = ()>,
545 {
546 loop {
547 self.tick().await?;
548 let (charge, charging) = battery();
549 let (_, interval) = self.schedule(charge, charging);
550 wait(interval).await;
551 }
552 }
553}
554
555/// One reading a node took, and what its policy decided about it.
556#[derive(Clone, Debug, PartialEq)]
557pub struct Tick<R = f32, C = bool> {
558 /// The reading, as the sensor returned it and the node published it.
559 pub reading: R,
560 /// What the policy decided: the command applied, if any, and any alert.
561 pub reaction: Reaction<C>,
562}
563
564#[cfg(test)]
565mod tests {
566 use std::sync::{Arc, Mutex};
567
568 use pamoja_codec::CborCodec;
569 use pamoja_core::Error;
570 use pamoja_core::Receive;
571 use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
572 use pamoja_power::PowerMode;
573
574 use super::*;
575 use crate::Alert;
576
577 // A sensor that plays back a fixed list of readings, then reports closed.
578 struct ScriptedSensor {
579 readings: std::vec::IntoIter<f32>,
580 }
581
582 impl ScriptedSensor {
583 fn new(readings: Vec<f32>) -> Self {
584 Self {
585 readings: readings.into_iter(),
586 }
587 }
588 }
589
590 impl Sensor for ScriptedSensor {
591 type Reading = f32;
592
593 async fn read(&mut self) -> Result<f32> {
594 self.readings.next().ok_or(Error::Closed)
595 }
596 }
597
598 // An actuator that records every command it is given.
599 #[derive(Clone)]
600 struct Recording<C = bool>(Arc<Mutex<Vec<C>>>);
601
602 impl<C: Send> Actuator for Recording<C> {
603 type Command = C;
604
605 async fn apply(&mut self, command: C) -> Result<()> {
606 self.0.lock().expect("commands lock").push(command);
607 Ok(())
608 }
609 }
610
611 async fn connected_pair(filter: &str) -> (LoopbackTransport, LoopbackTransport) {
612 let broker = LoopbackBroker::new();
613 let mut gateway = LoopbackTransport::new(broker.clone());
614 let mut link = LoopbackTransport::new(broker);
615 gateway.connect().await.expect("gateway connect");
616 link.connect().await.expect("link connect");
617 gateway.subscribe(filter).await.expect("subscribe");
618 (gateway, link)
619 }
620
621 #[tokio::test]
622 async fn tick_reads_actuates_and_publishes() {
623 let (mut gateway, link) = connected_pair("cold-chain/#").await;
624 let commands = Arc::new(Mutex::new(Vec::new()));
625 let mut node = Node::new(
626 Profile::vaccine_fridge_monitor(),
627 ScriptedSensor::new(vec![9.0]),
628 Recording(commands.clone()),
629 link,
630 CborCodec,
631 )
632 .expect("a built-in kind");
633
634 let reaction = node.tick().await.expect("tick").reaction;
635 assert_eq!(reaction.actuator, Some(true));
636 assert!(reaction.alert.is_some());
637 assert_eq!(*commands.lock().expect("commands"), vec![true]);
638
639 let message = gateway.recv().await.expect("recv").expect("a reading");
640 assert_eq!(message.topic, "cold-chain/fridge/temperature");
641 let reading: f32 = CborCodec.decode(&message.payload).expect("decode");
642 assert_eq!(reading, 9.0);
643 }
644
645 #[tokio::test]
646 async fn monitor_publishes_without_an_actuator() {
647 let (mut gateway, link) = connected_pair("water/#").await;
648 let mut node = Node::monitor(
649 Profile::well_level(),
650 ScriptedSensor::new(vec![3.2]),
651 link,
652 CborCodec,
653 )
654 .expect("a built-in kind");
655
656 let reaction = node.tick().await.expect("tick").reaction;
657 assert_eq!(reaction.actuator, None);
658
659 let message = gateway.recv().await.expect("recv").expect("a reading");
660 assert_eq!(message.topic, "water/well/level");
661 let reading: f32 = CborCodec.decode(&message.payload).expect("decode");
662 assert_eq!(reading, 3.2);
663 }
664
665 // A sensor reading two quantities at once, as a driver for a combined part does.
666 struct Climate(f32, f32);
667
668 impl Sensor for Climate {
669 type Reading = (f32, f32);
670
671 async fn read(&mut self) -> Result<(f32, f32)> {
672 Ok((self.0, self.1))
673 }
674 }
675
676 // A policy over the pair, commanding a fan duty rather than an on/off.
677 struct FanPolicy;
678
679 impl Policy for FanPolicy {
680 type Reading = (f32, f32);
681 type Command = u8;
682
683 fn evaluate(&mut self, &(temperature, humidity): &(f32, f32)) -> Reaction<u8> {
684 let duty = if temperature > 30.0 { 100 } else { 0 };
685 Reaction {
686 actuator: Some(duty),
687 alert: (humidity > 90.0).then_some(Alert::Custom {
688 code: "Muggy",
689 value: humidity,
690 }),
691 }
692 }
693 }
694
695 #[tokio::test]
696 async fn a_policy_of_your_own_reads_a_pair_and_commands_a_duty() {
697 let (mut gateway, link) = connected_pair("farm/#").await;
698 let duties = Arc::new(Mutex::new(Vec::new()));
699 let mut node = Node::with_policy(
700 Profile::irrigation_node(),
701 FanPolicy,
702 Climate(34.0, 95.0),
703 Recording::<u8>(duties.clone()),
704 link,
705 CborCodec,
706 );
707
708 let reaction = node.tick().await.expect("tick").reaction;
709 assert_eq!(reaction.actuator, Some(100));
710 assert_eq!(reaction.alert.map(Alert::kind), Some("Muggy"));
711 assert_eq!(*duties.lock().expect("duties"), vec![100]);
712 let _: &FanPolicy = node.policy();
713
714 let message = gateway.recv().await.expect("recv").expect("a reading");
715 assert_eq!(message.topic, "farm/irrigation/soil-moisture");
716 let reading: (f32, f32) = CborCodec.decode(&message.payload).expect("decode");
717 assert_eq!(reading, (34.0, 95.0));
718 }
719
720 #[tokio::test]
721 async fn a_node_runs_from_a_spawned_task() {
722 // The generic shape a maker's runner takes: any sensor and actuator, spawned on a
723 // multi-threaded runtime, which needs the whole tick to be Send.
724 fn spawn_node<S, A, T, C>(mut node: Node<S, A, T, C>) -> tokio::task::JoinHandle<()>
725 where
726 S: Sensor<Reading = f32> + Send + 'static,
727 A: Actuator<Command = bool> + Send + 'static,
728 T: Transport + Send + 'static,
729 C: Codec<f32> + Send + 'static,
730 {
731 tokio::spawn(async move {
732 for _ in 0..2 {
733 node.tick().await.expect("a tick");
734 }
735 })
736 }
737
738 let (mut gateway, link) = connected_pair("cold-chain/#").await;
739 let node = Node::new(
740 Profile::vaccine_fridge_monitor(),
741 ScriptedSensor::new(vec![9.0, 4.0]),
742 Recording(Arc::new(Mutex::new(Vec::new()))),
743 link,
744 CborCodec,
745 )
746 .expect("a built-in kind");
747 spawn_node(node).await.expect("the task finishes");
748 let first = gateway.recv().await.expect("recv").expect("a reading");
749 let reading: f32 = CborCodec.decode(&first.payload).expect("decode");
750 assert_eq!(reading, 9.0);
751 }
752
753 #[tokio::test]
754 async fn run_ticks_and_waits_the_battery_cadence_until_a_tick_fails() {
755 let (mut gateway, link) = connected_pair("cold-chain/#").await;
756 let commands = Arc::new(Mutex::new(Vec::new()));
757 let mut node = Node::new(
758 Profile::vaccine_fridge_monitor(),
759 ScriptedSensor::new(vec![9.0, 4.0, 6.0]),
760 Recording(commands.clone()),
761 link,
762 CborCodec,
763 )
764 .expect("a built-in kind");
765 let mut charges = [0.9, 0.3, 0.1].into_iter();
766 let mut waited = Vec::new();
767 let ended = node
768 .run(
769 || (charges.next().expect("a charge per tick"), false),
770 |interval| {
771 waited.push(interval);
772 async {}
773 },
774 )
775 .await;
776 assert!(matches!(ended, Err(Error::Closed)), "{ended:?}");
777 assert_eq!(
778 waited,
779 [
780 Duration::from_secs(60),
781 Duration::from_secs(300),
782 Duration::from_secs(900)
783 ]
784 );
785 assert_eq!(*commands.lock().expect("commands"), vec![true, false, true]);
786 let mut published: Vec<f32> = Vec::new();
787 for _ in 0..3 {
788 let message = gateway.recv().await.expect("recv").expect("a reading");
789 published.push(CborCodec.decode(&message.payload).expect("decode"));
790 }
791 assert_eq!(published, [9.0, 4.0, 6.0]);
792 assert_eq!(node.power_mode(), Some(PowerMode::Critical));
793 }
794
795 #[test]
796 fn schedule_follows_state_of_charge() {
797 // The schedule reads only the profile, so the components can be placeholders.
798 let mut node = Node::monitor(Profile::vaccine_fridge_monitor(), (), (), ()).unwrap();
799 assert_eq!(node.power_mode(), None);
800 assert_eq!(node.schedule(0.9, false).0, PowerMode::Active);
801 assert_eq!(node.schedule(0.1, false).0, PowerMode::Critical);
802 // A charging panel eases off by one mode.
803 assert_eq!(node.schedule(0.1, true).0, PowerMode::Saver);
804 assert_eq!(node.schedule(0.9, false).1, Duration::from_secs(60));
805 assert_eq!(node.power_mode(), Some(PowerMode::Active));
806 }
807
808 #[test]
809 fn a_charge_hovering_at_a_threshold_keeps_the_cadence() {
810 let mut node = Node::monitor(Profile::vaccine_fridge_monitor(), (), (), ()).unwrap();
811 let modes: Vec<PowerMode> = [0.52, 0.49, 0.51, 0.5, 0.53, 0.56]
812 .into_iter()
813 .map(|soc| node.schedule(soc, false).0)
814 .collect();
815 assert_eq!(
816 modes,
817 [
818 PowerMode::Active,
819 PowerMode::Saver,
820 PowerMode::Saver,
821 PowerMode::Saver,
822 PowerMode::Saver,
823 PowerMode::Active,
824 ]
825 );
826 }
827}