pamoja_profile/
node.rs

1//! The ready-to-run node a profile assembles around real components.
2
3use core::time::Duration;
4
5use pamoja_codec::Codec;
6use pamoja_core::{Actuator, Result, Sensor, Transport};
7use pamoja_power::PowerMode;
8
9use crate::{Controller, Profile, Reaction};
10
11/// An actuator that accepts and ignores commands.
12///
13/// Profiles that only observe - such as a well-level monitor - have no output to
14/// drive. [`Node::monitor`] wires this in their place so a node has one uniform
15/// shape whether or not it switches an actuator.
16#[derive(Clone, Copy, Debug, Default)]
17pub struct NoActuator;
18
19impl Actuator for NoActuator {
20    type Command = bool;
21
22    async fn apply(&mut self, _command: bool) -> Result<()> {
23        Ok(())
24    }
25}
26
27/// A profile assembled around the components that make it run.
28///
29/// The node is the thin shell that ties a [`Profile`]'s decision logic to real I/O.
30/// Each [`tick`](Node::tick) reads the sensor, runs the profile's
31/// [`Controller`](crate::Controller), drives the actuator when the controller calls
32/// for it, and publishes the reading over the transport with the supplied codec. The
33/// control math lives in `pamoja-kit`, the power schedule in `pamoja-power`, and the
34/// wire format in the codec, so the node adds composition, not behavior.
35///
36/// Readings are real-world `f32` units (degrees, percent, litres), the form the
37/// `pamoja-kit` controllers expect; a driver is responsible for calibrating raw
38/// counts into those units before the node sees them.
39///
40/// # Examples
41///
42/// ```
43/// use pamoja_codec::CborCodec;
44/// use pamoja_core::{Actuator, Result, Sensor, Transport};
45/// use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
46/// use pamoja_profile::{Node, Profile};
47///
48/// struct Probe(f32);
49/// impl Sensor for Probe {
50///     type Reading = f32;
51///     async fn read(&mut self) -> Result<f32> {
52///         Ok(self.0)
53///     }
54/// }
55///
56/// struct Cooler;
57/// impl Actuator for Cooler {
58///     type Command = bool;
59///     async fn apply(&mut self, _on: bool) -> Result<()> {
60///         Ok(())
61///     }
62/// }
63///
64/// # async fn run() -> Result<()> {
65/// let broker = LoopbackBroker::new();
66/// let mut link = LoopbackTransport::new(broker);
67/// link.connect().await?;
68///
69/// // A warm fridge, assembled straight from its profile.
70/// let mut node = Node::new(Profile::vaccine_fridge_monitor(), Probe(9.0), Cooler, link, CborCodec);
71/// let reaction = node.tick().await?;
72/// assert_eq!(reaction.actuator, Some(true)); // the cooler runs
73/// assert!(reaction.alert.is_some()); // and 9 C is a spoilage excursion
74/// # Ok(())
75/// # }
76/// ```
77pub struct Node<S, A, T, C> {
78    profile: Profile,
79    controller: Controller,
80    sensor: S,
81    actuator: A,
82    transport: T,
83    codec: C,
84}
85
86impl<S, A, T, C> Node<S, A, T, C> {
87    /// Assembles a node from a profile and the components that drive it.
88    ///
89    /// # Arguments
90    ///
91    /// * `profile` - the profile to assemble; its policy becomes the node's controller.
92    /// * `sensor` - the source of readings.
93    /// * `actuator` - the output the controller switches.
94    /// * `transport` - the link readings are published over; expected to be connected.
95    /// * `codec` - the wire format readings are encoded with.
96    ///
97    /// # Returns
98    ///
99    /// A node ready to [`tick`](Node::tick).
100    pub fn new(profile: Profile, sensor: S, actuator: A, transport: T, codec: C) -> Self {
101        let controller = profile.controller();
102        Self {
103            profile,
104            controller,
105            sensor,
106            actuator,
107            transport,
108            codec,
109        }
110    }
111
112    /// Returns the profile this node was assembled from.
113    ///
114    /// # Returns
115    ///
116    /// A reference to the node's [`Profile`].
117    pub fn profile(&self) -> &Profile {
118        &self.profile
119    }
120
121    /// Returns the power mode and wait interval for the next cycle.
122    ///
123    /// This assembles the profile's [`PowerSchedule`](crate::PowerSchedule) into a
124    /// `pamoja-power` governor: as the battery drains the interval stretches, and a
125    /// charging panel eases the node back toward its active cadence. The node never
126    /// sleeps; the caller waits the
127    /// returned [`Duration`] before the next [`tick`](Node::tick), so timing stays
128    /// outside the node and the decision logic remains synchronous and testable.
129    ///
130    /// # Arguments
131    ///
132    /// * `soc` - the battery state of charge in `[0.0, 1.0]`.
133    /// * `charging` - whether the panel is currently delivering charge.
134    ///
135    /// # Returns
136    ///
137    /// The [`PowerMode`] to run in and how long to wait before the next cycle.
138    pub fn schedule(&self, soc: f32, charging: bool) -> (PowerMode, Duration) {
139        let plan = self.profile.power.plan();
140        let mode = plan.mode_while_charging(soc, charging);
141        (mode, plan.interval_for(mode))
142    }
143}
144
145impl<S, T, C> Node<S, NoActuator, T, C> {
146    /// Assembles a node for a profile that observes without driving an output.
147    ///
148    /// Wires a [`NoActuator`] in place of a real output, so a monitoring profile such
149    /// as [`well_level`](Profile::well_level) reads and publishes with the same shape
150    /// as a controlling one.
151    ///
152    /// # Arguments
153    ///
154    /// * `profile` - the profile to assemble.
155    /// * `sensor` - the source of readings.
156    /// * `transport` - the link readings are published over; expected to be connected.
157    /// * `codec` - the wire format readings are encoded with.
158    ///
159    /// # Returns
160    ///
161    /// A node ready to [`tick`](Node::tick), with no output to switch.
162    pub fn monitor(profile: Profile, sensor: S, transport: T, codec: C) -> Self {
163        Node::new(profile, sensor, NoActuator, transport, codec)
164    }
165}
166
167impl<S, A, T, C> Node<S, A, T, C>
168where
169    S: Sensor<Reading = f32>,
170    A: Actuator<Command = bool>,
171    T: Transport,
172    C: Codec<f32>,
173{
174    /// Runs one read-decide-act-publish cycle.
175    ///
176    /// Reads the sensor, evaluates the profile's controller, applies the resulting
177    /// command to the actuator when the controller calls for one, and publishes the
178    /// reading to the profile's topic.
179    ///
180    /// # Returns
181    ///
182    /// The [`Reaction`] the controller produced: the actuator setting that was
183    /// applied (if any) and any alert the reading raised.
184    ///
185    /// # Errors
186    ///
187    /// Returns [`Error::Io`](pamoja_core::Error::Io) or
188    /// [`Error::Closed`](pamoja_core::Error::Closed) if the sensor read or the
189    /// actuator command fails, [`Error::Codec`](pamoja_core::Error::Codec) if the
190    /// reading cannot be encoded, and [`Error::Transport`](pamoja_core::Error::Transport)
191    /// or [`Error::Closed`](pamoja_core::Error::Closed) if the publish fails.
192    pub async fn tick(&mut self) -> Result<Reaction> {
193        let reading = self.sensor.read().await?;
194        let reaction = self.controller.evaluate(reading);
195        if let Some(on) = reaction.actuator {
196            self.actuator.apply(on).await?;
197        }
198        let payload = self.codec.encode(&reading)?;
199        self.transport.send(&self.profile.topic, &payload).await?;
200        Ok(reaction)
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use std::sync::{Arc, Mutex};
207
208    use pamoja_codec::CborCodec;
209    use pamoja_core::Error;
210    use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
211    use pamoja_power::PowerMode;
212
213    use super::*;
214
215    // A sensor that plays back a fixed list of readings, then reports closed.
216    struct ScriptedSensor {
217        readings: std::vec::IntoIter<f32>,
218    }
219
220    impl ScriptedSensor {
221        fn new(readings: Vec<f32>) -> Self {
222            Self {
223                readings: readings.into_iter(),
224            }
225        }
226    }
227
228    impl Sensor for ScriptedSensor {
229        type Reading = f32;
230
231        async fn read(&mut self) -> Result<f32> {
232            self.readings.next().ok_or(Error::Closed)
233        }
234    }
235
236    // An actuator that records every command it is given.
237    #[derive(Clone)]
238    struct Recording(Arc<Mutex<Vec<bool>>>);
239
240    impl Actuator for Recording {
241        type Command = bool;
242
243        async fn apply(&mut self, on: bool) -> Result<()> {
244            self.0.lock().expect("commands lock").push(on);
245            Ok(())
246        }
247    }
248
249    async fn connected_pair(filter: &str) -> (LoopbackTransport, LoopbackTransport) {
250        let broker = LoopbackBroker::new();
251        let mut gateway = LoopbackTransport::new(broker.clone());
252        let mut link = LoopbackTransport::new(broker);
253        gateway.connect().await.expect("gateway connect");
254        link.connect().await.expect("link connect");
255        gateway.subscribe(filter).await.expect("subscribe");
256        (gateway, link)
257    }
258
259    #[tokio::test]
260    async fn tick_reads_actuates_and_publishes() {
261        let (mut gateway, link) = connected_pair("cold-chain/#").await;
262        let commands = Arc::new(Mutex::new(Vec::new()));
263        let mut node = Node::new(
264            Profile::vaccine_fridge_monitor(),
265            ScriptedSensor::new(vec![9.0]),
266            Recording(commands.clone()),
267            link,
268            CborCodec,
269        );
270
271        let reaction = node.tick().await.expect("tick");
272        assert_eq!(reaction.actuator, Some(true));
273        assert!(reaction.alert.is_some());
274        assert_eq!(*commands.lock().expect("commands"), vec![true]);
275
276        let message = gateway.recv().await.expect("recv").expect("a reading");
277        assert_eq!(message.topic, "cold-chain/fridge/temperature");
278        let reading: f32 = CborCodec.decode(&message.payload).expect("decode");
279        assert_eq!(reading, 9.0);
280    }
281
282    #[tokio::test]
283    async fn monitor_publishes_without_an_actuator() {
284        let (mut gateway, link) = connected_pair("water/#").await;
285        let mut node = Node::monitor(
286            Profile::well_level(),
287            ScriptedSensor::new(vec![3.2]),
288            link,
289            CborCodec,
290        );
291
292        let reaction = node.tick().await.expect("tick");
293        assert_eq!(reaction.actuator, None);
294
295        let message = gateway.recv().await.expect("recv").expect("a reading");
296        assert_eq!(message.topic, "water/well/level");
297        let reading: f32 = CborCodec.decode(&message.payload).expect("decode");
298        assert_eq!(reading, 3.2);
299    }
300
301    #[test]
302    fn schedule_follows_state_of_charge() {
303        // The schedule reads only the profile, so the components can be placeholders.
304        let node = Node::monitor(Profile::vaccine_fridge_monitor(), (), (), ());
305        assert_eq!(node.schedule(0.9, false).0, PowerMode::Active);
306        assert_eq!(node.schedule(0.1, false).0, PowerMode::Critical);
307        // A charging panel eases off by one mode.
308        assert_eq!(node.schedule(0.1, true).0, PowerMode::Saver);
309        assert_eq!(node.schedule(0.9, false).1, Duration::from_secs(60));
310    }
311}
pamoja_profile/node.rs

pamoja_profile/
node.rs
