pamoja_dashboard/
fleet.rs

1//! A real fleet source: a project pushes readings in, the dashboard reads them out, and
2//! control commands queue back for the project to apply.
3//!
4//! The dashboard renders whatever implements [`StateSource`]. [`Mock`](crate::Mock) is the
5//! hardware-free demo; `Fleet` is the real one. Sensing in this SDK is async and the server
6//! is synchronous, so the project owns its own sampling loop - ticking its profiles and
7//! nodes on their power schedule - and pushes each result into a `Fleet` with the report
8//! methods. The dashboard reads the latest with [`snapshot`](StateSource::snapshot), and
9//! authenticated control [`Command`]s queue for the project to drain with
10//! [`take_commands`](Fleet::take_commands) and apply to real hardware, then reflect the
11//! result back through the report methods. The device stays authoritative; provisioning and
12//! actuation are also applied optimistically so the UI updates at once.
13//!
14//! A `Fleet` is cheap to [`Clone`] (it shares one inner state), so one handle drives the
15//! [`Server`](crate::Server) while another stays with the sampling loop.
16//!
17//! # Examples
18//!
19//! ```
20//! use pamoja_dashboard::{Fleet, LinkKind, Reading, Sensor, StateSource, Status};
21//!
22//! let fleet = Fleet::builder()
23//!     .org("clinic", "Kano clinic")
24//!     .group("clinic", "fridges", "Cold chain", LinkKind::Cellular)
25//!     .sensor(
26//!         "fridges",
27//!         Sensor::new("fridge-1", Reading::new("fridge_temp", 4.5, "celsius").with_band(2.0, 8.0)),
28//!     )
29//!     .build();
30//!
31//! // The sampling loop pushes a fresh reading; the dashboard sees it.
32//! fleet.report_reading(
33//!     "fridges",
34//!     "fridge-1",
35//!     Reading::new("fridge_temp", 9.2, "celsius").with_band(2.0, 8.0).with_status(Status::Alarm),
36//! );
37//! let mut handle = fleet.clone();
38//! assert_eq!(handle.snapshot().status, Status::Alarm);
39//! ```
40
41use std::collections::HashSet;
42use std::sync::{Arc, Mutex};
43use std::time::Instant;
44
45use crate::command::{Command, CommandError};
46use crate::source::StateSource;
47use crate::state::{EventRecord, Group, Link, LinkKind, Mode, Org, Reading, Sensor, State, Status};
48
49// How many recent values each sensor keeps for its sparkline.
50const HISTORY: usize = 32;
51
52struct Inner {
53    state: State,
54    commands: Vec<Command>,
55    started: Instant,
56    // The sensor element keys a client may add. `None` accepts any (the default); `Some`
57    // rejects a client `AddSensor` whose key is not listed, so a real device only takes the
58    // sensor types it can actually bind. Gateway-initiated discovery is never gated.
59    allowed: Option<HashSet<String>>,
60}
61
62/// A real fleet a project fills and the dashboard renders. Clone to share one between the
63/// serving layer and the sampling loop.
64#[derive(Clone)]
65pub struct Fleet {
66    inner: Arc<Mutex<Inner>>,
67}
68
69impl Fleet {
70    /// Starts building a fleet's initial structure.
71    ///
72    /// # Returns
73    ///
74    /// An empty [`FleetBuilder`].
75    pub fn builder() -> FleetBuilder {
76        FleetBuilder { orgs: Vec::new() }
77    }
78
79    /// Restores a fleet from a previously saved [`State`], for a gateway that persists its
80    /// fleet across restarts (save what [`snapshot`](StateSource::snapshot) returns, reload
81    /// it here on boot).
82    ///
83    /// # Arguments
84    ///
85    /// * `state` - the fleet structure and last readings to restore.
86    ///
87    /// # Returns
88    ///
89    /// A fleet holding the restored state.
90    pub fn from_state(state: State) -> Self {
91        Self {
92            inner: Arc::new(Mutex::new(Inner {
93                state,
94                commands: Vec::new(),
95                started: Instant::now(),
96                allowed: None,
97            })),
98        }
99    }
100
101    /// Restricts which sensor types a client may add, so a real device only accepts the
102    /// sensors it can bind a driver to.
103    ///
104    /// Without this, an `AddSensor` from the dashboard is always accepted (which suits the
105    /// hardware-free demo). With it, a client `AddSensor` whose element key is not listed is
106    /// rejected with [`CommandError::UnknownSensor`], and the dashboard reports that the
107    /// device does not support that sensor. Discovery through [`add_sensor`](Fleet::add_sensor)
108    /// is the device's own and stays unrestricted. Set this after building or restoring the
109    /// fleet; the keys are the element keys a deployment supports, the same ones its
110    /// presentation declares.
111    ///
112    /// # Arguments
113    ///
114    /// * `keys` - the sensor element keys a client may add, such as `"soil_moisture"`.
115    pub fn allow_sensors(&self, keys: impl IntoIterator<Item = impl Into<String>>) {
116        let mut inner = self.inner.lock().expect("fleet lock");
117        inner.allowed = Some(keys.into_iter().map(Into::into).collect());
118    }
119
120    /// Pushes a fresh reading for a sensor, appending it to the sensor's history.
121    ///
122    /// # Arguments
123    ///
124    /// * `group` - the sensor's group id.
125    /// * `sensor` - the sensor id.
126    /// * `reading` - the new reading (the caller sets its status and band).
127    pub fn report_reading(&self, group: &str, sensor: &str, reading: Reading) {
128        let mut inner = self.inner.lock().expect("fleet lock");
129        if let Some(target) = sensor_mut(&mut inner.state, group, sensor) {
130            let value = reading.value;
131            target.reading = reading;
132            target.history.push(value);
133            let len = target.history.len();
134            if len > HISTORY {
135                target.history.drain(0..len - HISTORY);
136            }
137        }
138        recompute(&mut inner.state);
139    }
140
141    /// Records a recent event for a sensor, newest first.
142    ///
143    /// # Arguments
144    ///
145    /// * `group` - the sensor's group id.
146    /// * `sensor` - the sensor id.
147    /// * `event` - the event to record.
148    pub fn report_event(&self, group: &str, sensor: &str, event: EventRecord) {
149        let mut inner = self.inner.lock().expect("fleet lock");
150        if let Some(target) = sensor_mut(&mut inner.state, group, sensor) {
151            target.events.insert(0, event);
152            target.events.truncate(8);
153        }
154        recompute(&mut inner.state);
155    }
156
157    /// Updates a group's link status (kind, signal strength, online).
158    ///
159    /// # Arguments
160    ///
161    /// * `group` - the group id.
162    /// * `link` - the new link status.
163    pub fn report_link(&self, group: &str, link: Link) {
164        let mut inner = self.inner.lock().expect("fleet lock");
165        if let Some(target) = group_mut(&mut inner.state, group) {
166            target.link = link;
167        }
168        recompute(&mut inner.state);
169    }
170
171    /// Updates a sensor's power mode and battery state of charge.
172    ///
173    /// # Arguments
174    ///
175    /// * `group` - the sensor's group id.
176    /// * `sensor` - the sensor id.
177    /// * `mode` - the work cadence the sensor's node is running at.
178    /// * `battery` - the state of charge in `[0.0, 1.0]`, or `None` if it has no battery.
179    pub fn report_power(&self, group: &str, sensor: &str, mode: Mode, battery: Option<f32>) {
180        let mut inner = self.inner.lock().expect("fleet lock");
181        if let Some(target) = sensor_mut(&mut inner.state, group, sensor) {
182            target.mode = mode;
183            target.battery = battery;
184        }
185    }
186
187    /// Drains the control commands queued since the last call, for the project to apply to
188    /// real hardware and persist, then reflect back through the report methods.
189    ///
190    /// # Returns
191    ///
192    /// The commands accepted since the last drain, in order.
193    pub fn take_commands(&self) -> Vec<Command> {
194        let mut inner = self.inner.lock().expect("fleet lock");
195        std::mem::take(&mut inner.commands)
196    }
197
198    /// Adds a group to an organization at runtime, so a gateway can surface a node the moment
199    /// it is discovered (a LoRa join, a new mesh neighbour). A no-op if the org is unknown.
200    ///
201    /// # Arguments
202    ///
203    /// * `org` - the organization id to add the group to.
204    /// * `group` - the group to add.
205    pub fn add_group(&self, org: &str, group: Group) {
206        self.mutate(Command::AddGroup {
207            org: org.to_owned(),
208            group,
209        });
210    }
211
212    /// Adds a sensor to a group at runtime, for a newly discovered sensor. A no-op if the
213    /// group is unknown.
214    ///
215    /// # Arguments
216    ///
217    /// * `group` - the group id to add the sensor to.
218    /// * `sensor` - the sensor to add.
219    pub fn add_sensor(&self, group: &str, sensor: Sensor) {
220        self.mutate(Command::AddSensor {
221            group: group.to_owned(),
222            sensor,
223            binding: None,
224        });
225    }
226
227    /// Removes a group by id at runtime, for a node that has gone away.
228    ///
229    /// # Arguments
230    ///
231    /// * `id` - the group id to remove.
232    pub fn remove_group(&self, id: &str) {
233        self.mutate(Command::RemoveGroup { id: id.to_owned() });
234    }
235
236    /// Removes a sensor by its `"groupId/sensorId"` path at runtime.
237    ///
238    /// # Arguments
239    ///
240    /// * `target` - the `"groupId/sensorId"` path to remove.
241    pub fn remove_sensor(&self, target: &str) {
242        self.mutate(Command::RemoveSensor {
243            target: target.to_owned(),
244        });
245    }
246
247    // Applies a structural change to the held state, the gateway-initiated counterpart of a
248    // dashboard command; it is not queued back to the gateway.
249    fn mutate(&self, command: Command) {
250        let mut inner = self.inner.lock().expect("fleet lock");
251        let _ = apply(&mut inner.state, &command);
252        recompute(&mut inner.state);
253    }
254}
255
256impl StateSource for Fleet {
257    fn snapshot(&mut self) -> State {
258        let mut inner = self.inner.lock().expect("fleet lock");
259        let uptime = inner.started.elapsed().as_secs();
260        inner.state.uptime_secs = Some(uptime);
261        inner.state.clone()
262    }
263
264    fn command(&mut self, command: &Command) -> Result<(), CommandError> {
265        let mut inner = self.inner.lock().expect("fleet lock");
266        // A real device only takes the sensor types it can bind; reject an add of anything else
267        // so the dashboard says so instead of showing a sensor that will never report.
268        if let Command::AddSensor { sensor, .. } = command {
269            if let Some(allowed) = &inner.allowed {
270                if !allowed.contains(&sensor.reading.key) {
271                    return Err(CommandError::UnknownSensor);
272                }
273            }
274        }
275        let outcome = apply(&mut inner.state, command);
276        if outcome.is_ok() {
277            inner.commands.push(command.clone());
278            recompute(&mut inner.state);
279        }
280        outcome
281    }
282}
283
284// Applies a command optimistically to the held state, mirroring the device's eventual
285// effect so the UI updates at once; the queued copy lets the project make it real.
286fn apply(state: &mut State, command: &Command) -> Result<(), CommandError> {
287    match command {
288        Command::Actuate { target, action } => {
289            let (group, sensor) = target.split_once('/').ok_or(CommandError::UnknownTarget)?;
290            let reading = sensor_mut(state, group, sensor)
291                .map(|s| &mut s.reading)
292                .ok_or(CommandError::UnknownTarget)?;
293            match &reading.actions {
294                Some(actions) if actions.iter().any(|a| a == action) => {
295                    reading.state = Some(format!("state.{action}"));
296                    Ok(())
297                }
298                Some(_) => Err(CommandError::InvalidAction),
299                None => Err(CommandError::Unsupported),
300            }
301        }
302        Command::AddGroup { org, group } => match org_mut(state, org) {
303            Some(target) => {
304                target.groups.push(group.clone());
305                Ok(())
306            }
307            None => Err(CommandError::UnknownTarget),
308        },
309        Command::RemoveGroup { id } => {
310            for org in &mut state.orgs {
311                org.groups.retain(|g| g.id != *id);
312            }
313            Ok(())
314        }
315        Command::AddSensor { group, sensor, .. } => match group_mut(state, group) {
316            Some(target) => {
317                target.sensors.push(sensor.clone());
318                Ok(())
319            }
320            None => Err(CommandError::UnknownTarget),
321        },
322        Command::RemoveSensor { target } => {
323            let (group_id, sensor_id) = target.split_once('/').unwrap_or(("", target));
324            for org in &mut state.orgs {
325                for group in &mut org.groups {
326                    if group.id == group_id {
327                        group.sensors.retain(|s| s.id != sensor_id);
328                    }
329                }
330            }
331            Ok(())
332        }
333    }
334}
335
336fn recompute(state: &mut State) {
337    for org in &mut state.orgs {
338        for group in &mut org.groups {
339            group.recompute_status();
340        }
341    }
342    state.recompute_status();
343}
344
345fn org_mut<'a>(state: &'a mut State, org: &str) -> Option<&'a mut Org> {
346    state.orgs.iter_mut().find(|o| o.id == org)
347}
348
349fn group_mut<'a>(state: &'a mut State, group: &str) -> Option<&'a mut Group> {
350    state
351        .orgs
352        .iter_mut()
353        .flat_map(|o| &mut o.groups)
354        .find(|g| g.id == group)
355}
356
357fn sensor_mut<'a>(state: &'a mut State, group: &str, sensor: &str) -> Option<&'a mut Sensor> {
358    state
359        .orgs
360        .iter_mut()
361        .flat_map(|o| &mut o.groups)
362        .filter(|g| g.id == group)
363        .flat_map(|g| &mut g.sensors)
364        .find(|s| s.id == sensor)
365}
366
367/// Builds a fleet's initial structure: organizations, their groups, and each group's
368/// sensors. Parents are referenced by id, so add an org before its groups and a group
369/// before its sensors.
370pub struct FleetBuilder {
371    orgs: Vec<Org>,
372}
373
374impl FleetBuilder {
375    /// Adds an organization.
376    ///
377    /// # Arguments
378    ///
379    /// * `id` - the stable organization id.
380    /// * `name` - the human-readable name.
381    ///
382    /// # Returns
383    ///
384    /// The builder, for chaining.
385    pub fn org(mut self, id: impl Into<String>, name: impl Into<String>) -> Self {
386        self.orgs.push(Org {
387            id: id.into(),
388            name: name.into(),
389            groups: Vec::new(),
390        });
391        self
392    }
393
394    /// Adds a group to an organization.
395    ///
396    /// # Arguments
397    ///
398    /// * `org` - the id of the organization to add the group to.
399    /// * `id` - the stable group id.
400    /// * `name` - the human-readable name.
401    /// * `kind` - the link the group reports over.
402    ///
403    /// # Returns
404    ///
405    /// The builder, for chaining.
406    pub fn group(
407        mut self,
408        org: &str,
409        id: impl Into<String>,
410        name: impl Into<String>,
411        kind: LinkKind,
412    ) -> Self {
413        if let Some(target) = self.orgs.iter_mut().find(|o| o.id == org) {
414            target.groups.push(Group {
415                id: id.into(),
416                name: name.into(),
417                link: Link {
418                    kind,
419                    strength: 4,
420                    online: true,
421                },
422                status: Status::Ok,
423                sensors: Vec::new(),
424                lat: None,
425                lon: None,
426            });
427        }
428        self
429    }
430
431    /// Adds a sensor to a group.
432    ///
433    /// # Arguments
434    ///
435    /// * `group` - the id of the group to add the sensor to.
436    /// * `sensor` - the sensor (build its reading with [`Reading`] and [`Sensor::new`]).
437    ///
438    /// # Returns
439    ///
440    /// The builder, for chaining.
441    pub fn sensor(mut self, group: &str, sensor: Sensor) -> Self {
442        for org in &mut self.orgs {
443            if let Some(target) = org.groups.iter_mut().find(|g| g.id == group) {
444                target.sensors.push(sensor);
445                break;
446            }
447        }
448        self
449    }
450
451    /// Finishes building, returning a [`Fleet`] ready to serve and report into.
452    ///
453    /// # Returns
454    ///
455    /// The assembled fleet.
456    pub fn build(mut self) -> Fleet {
457        let mut state = State {
458            orgs: std::mem::take(&mut self.orgs),
459            status: Status::Ok,
460            uptime_secs: None,
461            demo: false,
462        };
463        recompute(&mut state);
464        Fleet::from_state(state)
465    }
466}
467
468#[cfg(test)]
469mod tests {
470    use super::*;
471
472    fn fleet() -> Fleet {
473        Fleet::builder()
474            .org("clinic", "Kano clinic")
475            .group("clinic", "fridges", "Cold chain", LinkKind::Cellular)
476            .sensor(
477                "fridges",
478                Sensor::new("fridge-1", Reading::new("fridge_temp", 4.5, "celsius")),
479            )
480            .sensor(
481                "fridges",
482                Sensor::new(
483                    "valve",
484                    Reading::new("drip_valve", 0.0, "state")
485                        .with_state("state.closed")
486                        .with_actions(["open", "closed"]),
487                ),
488            )
489            .build()
490    }
491
492    #[test]
493    fn a_reported_reading_shows_in_the_snapshot_with_history() {
494        let fleet = fleet();
495        fleet.report_reading(
496            "fridges",
497            "fridge-1",
498            Reading::new("fridge_temp", 9.0, "celsius").with_status(Status::Alarm),
499        );
500        let mut handle = fleet.clone();
501        let state = handle.snapshot();
502        let sensor = &state.orgs[0].groups[0].sensors[0];
503        assert_eq!(sensor.reading.value, 9.0);
504        assert_eq!(sensor.history, vec![9.0]);
505        assert_eq!(
506            state.status,
507            Status::Alarm,
508            "the alarm reading lifts fleet status"
509        );
510    }
511
512    #[test]
513    fn an_actuate_command_updates_state_and_queues_for_the_project() {
514        let mut fleet = fleet();
515        fleet
516            .command(&Command::Actuate {
517                target: "fridges/valve".to_owned(),
518                action: "open".to_owned(),
519            })
520            .expect("valve accepts open");
521        let queued = fleet.take_commands();
522        assert_eq!(
523            queued.len(),
524            1,
525            "the command is queued for the project to apply"
526        );
527        let valve = sensor_after(&fleet, "fridges", "valve");
528        assert_eq!(valve.reading.state.as_deref(), Some("state.open"));
529        // Draining empties the queue.
530        assert!(fleet.take_commands().is_empty());
531    }
532
533    #[test]
534    fn an_invalid_actuate_is_refused_and_not_queued() {
535        let mut fleet = fleet();
536        assert_eq!(
537            fleet.command(&Command::Actuate {
538                target: "fridges/fridge-1".to_owned(),
539                action: "open".to_owned(),
540            }),
541            Err(CommandError::Unsupported)
542        );
543        assert!(fleet.take_commands().is_empty());
544    }
545
546    #[test]
547    fn provisioning_commands_change_the_structure() {
548        let mut fleet = fleet();
549        fleet
550            .command(&Command::AddSensor {
551                group: "fridges".to_owned(),
552                sensor: Sensor::new("fridge-2", Reading::new("fridge_temp", 5.0, "celsius")),
553                binding: None,
554            })
555            .expect("add sensor to a known group");
556        assert!(sensor_present(&fleet, "fridges", "fridge-2"));
557
558        fleet
559            .command(&Command::RemoveSensor {
560                target: "fridges/fridge-2".to_owned(),
561            })
562            .expect("remove the sensor");
563        assert!(!sensor_present(&fleet, "fridges", "fridge-2"));
564    }
565
566    #[test]
567    fn runtime_mutators_add_and_remove_for_discovery() {
568        let fleet = fleet();
569        fleet.add_group(
570            "clinic",
571            Group {
572                id: "ward".to_owned(),
573                name: "Ward".to_owned(),
574                link: Link {
575                    kind: LinkKind::Wifi,
576                    strength: 4,
577                    online: true,
578                },
579                status: Status::Ok,
580                sensors: Vec::new(),
581                lat: None,
582                lon: None,
583            },
584        );
585        fleet.add_sensor(
586            "ward",
587            Sensor::new("o2", Reading::new("oxygen_stock", 80.0, "percent")),
588        );
589        assert!(
590            sensor_present(&fleet, "ward", "o2"),
591            "discovered sensor shows"
592        );
593        fleet.remove_group("ward");
594        let mut handle = fleet.clone();
595        assert!(
596            !handle
597                .snapshot()
598                .orgs
599                .iter()
600                .flat_map(|o| &o.groups)
601                .any(|g| g.id == "ward"),
602            "removed group is gone"
603        );
604    }
605
606    #[test]
607    fn an_allow_list_rejects_unsupported_client_adds_but_not_discovery() {
608        let mut fleet = fleet();
609        fleet.allow_sensors(["fridge_temp"]);
610
611        // A supported type is accepted.
612        fleet
613            .command(&Command::AddSensor {
614                group: "fridges".to_owned(),
615                sensor: Sensor::new("f2", Reading::new("fridge_temp", 5.0, "celsius")),
616                binding: None,
617            })
618            .expect("a supported sensor is added");
619
620        // An unsupported type is rejected and not added.
621        assert_eq!(
622            fleet.command(&Command::AddSensor {
623                group: "fridges".to_owned(),
624                sensor: Sensor::new("w", Reading::new("wind_speed", 3.0, "meter_per_second")),
625                binding: None,
626            }),
627            Err(CommandError::UnknownSensor)
628        );
629        assert!(!sensor_present(&fleet, "fridges", "w"));
630
631        // Gateway-initiated discovery is the device's own and stays unrestricted.
632        fleet.add_sensor(
633            "fridges",
634            Sensor::new("disc", Reading::new("wind_speed", 3.0, "meter_per_second")),
635        );
636        assert!(sensor_present(&fleet, "fridges", "disc"));
637    }
638
639    #[test]
640    fn from_state_restores_a_saved_fleet() {
641        let mut original = fleet();
642        let saved = original.snapshot();
643        let mut restored = Fleet::from_state(saved.clone());
644        assert_eq!(restored.snapshot().orgs.len(), saved.orgs.len());
645    }
646
647    fn sensor_after(fleet: &Fleet, group: &str, sensor: &str) -> Sensor {
648        let mut handle = fleet.clone();
649        let state = handle.snapshot();
650        state
651            .orgs
652            .iter()
653            .flat_map(|o| &o.groups)
654            .filter(|g| g.id == group)
655            .flat_map(|g| &g.sensors)
656            .find(|s| s.id == sensor)
657            .expect("sensor")
658            .clone()
659    }
660
661    fn sensor_present(fleet: &Fleet, group: &str, sensor: &str) -> bool {
662        let mut handle = fleet.clone();
663        handle
664            .snapshot()
665            .orgs
666            .iter()
667            .flat_map(|o| &o.groups)
668            .filter(|g| g.id == group)
669            .flat_map(|g| &g.sensors)
670            .any(|s| s.id == sensor)
671    }
672}
pamoja_dashboard/fleet.rs

pamoja_dashboard/
fleet.rs
