Struct Node 

pub struct Node<S, A, T, C, P = Controller> { /* private fields */ }

A profile assembled around the components that make it run.

The node is the thin shell that ties a Profile’s decision logic to real I/O. Each tick reads the sensor, runs the node’s policy, drives the actuator when the policy calls for it, and publishes the reading over the transport with the supplied codec. The control math lives in pamoja-kit, the power schedule in pamoja-power, and the wire format in the codec, so the node adds composition, not behavior.

Node::new runs the profile’s Controller, which reads real-world f32 units (degrees, percent, liters), the form the pamoja-kit controllers expect; a driver is responsible for calibrating raw counts into those units before the node sees them, or [Sensor::map] selects one. Node::with_policy runs a Policy of your own instead, over whatever the sensor reads and whatever the actuator takes.

Examples

use pamoja_codec::CborCodec;
use pamoja_core::{Actuator, Result, Sensor, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_profile::{Node, Profile};

struct Probe(f32);
impl Sensor for Probe {
    type Reading = f32;
    async fn read(&mut self) -> Result<f32> {
        Ok(self.0)
    }
}

struct Cooler;
impl Actuator for Cooler {
    type Command = bool;
    async fn apply(&mut self, _on: bool) -> Result<()> {
        Ok(())
    }
}

let broker = LoopbackBroker::new();
let mut link = LoopbackTransport::new(broker);
link.connect().await?;

// A warm fridge, assembled straight from its profile.
let mut node = Node::new(Profile::vaccine_fridge_monitor(), Probe(9.0), Cooler, link, CborCodec)?;
let reaction = node.tick().await?.reaction;
assert_eq!(reaction.actuator, Some(true)); // the cooler runs
assert!(reaction.alert.is_some()); // and 9 C is a spoilage excursion

Implementations

impl<S, A, T, C> Node<S, A, T, C>

pub fn new( profile: Profile, sensor: S, actuator: A, transport: T, codec: C, ) -> Result<Self>

Assembles a node from a profile and the components that drive it.

Arguments

• profile - the profile to assemble; its policy becomes the node’s controller.
• sensor - the source of readings.
• actuator - the output the controller switches.
• transport - the link readings are published over; expected to be connected.
• codec - the wire format readings are encoded with.

Returns

A node ready to tick.

Errors

Returns Error::Codec when the profile names a custom control kind, which no built-in controller decides; Node::resolve assembles one through a PolicyRegistry that registers it.

impl<S, A, T, C> Node<S, A, T, C, BoxedPolicy>

pub fn resolve( profile: Profile, registry: &PolicyRegistry, sensor: S, actuator: A, transport: T, codec: C, ) -> Result<Self>

Assembles a node whose policy a registry resolves from the profile’s control kind.

A built-in kind resolves to its controller, and a custom kind to the policy the factory registered under its name builds from the parameters beside it, so one program runs any manifest its registry covers.

Arguments

• profile - the profile to assemble.
• registry - what resolves the profile’s control kind.
• sensor - the source of readings.
• actuator - the output the policy switches.
• transport - the link readings are published over; expected to be connected.
• codec - the wire format readings are encoded with.

Returns

A node ready to tick.

Errors

Returns Error::Codec naming the kind when it is custom and not registered, or whatever the kind’s factory returns.

Examples

use pamoja_codec::JsonCodec;
use pamoja_core::{Error, Result, Sensor, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_profile::{BoxedPolicy, Controller, NoActuator, Node, PolicyRegistry, Profile};

struct Air(f32);
impl Sensor for Air {
    type Reading = f32;
    async fn read(&mut self) -> Result<f32> {
        Ok(self.0)
    }
}

let profile = Profile::from_json(r#"{
    "name": "orchard-frost",
    "reads": { "quantity": "temperature", "unit": "celsius" },
    "topic": "orchard/air/temperature",
    "control": { "kind": "frost_guard", "warn_below": 2.0 },
    "power": { "active_secs": 60, "saver_secs": 300, "critical_secs": 900 }
}"#)?;
let registry = PolicyRegistry::new().register("frost_guard", |params| {
    let warn_below = params.number("warn_below").ok_or(Error::Unsupported("warn_below"))?;
    Ok(Box::new(Controller::setpoint(warn_below as f32, 0.5, false, 5.0)) as BoxedPolicy)
});

let mut link = LoopbackTransport::new(LoopbackBroker::new());
link.connect().await?;
let mut node = Node::resolve(profile, &registry, Air(1.0), NoActuator, link, JsonCodec)?;
assert_eq!(node.tick().await?.reaction.actuator, Some(true)); // cold: the heater comes on

impl<S, A, T, C, P> Node<S, A, T, C, P>

pub fn with_policy( profile: Profile, policy: P, sensor: S, actuator: A, transport: T, codec: C, ) -> Self

Assembles a node around a policy of your own.

The profile still names the topic and the power schedule; the policy decides each reading, so a node can read a driver’s whole measurement and command an actuator that takes more than on and off. A policy a PolicyRegistry resolved from the profile’s own control kind goes here too.

Arguments

• profile - the profile to assemble.
• policy - what decides each reading.
• sensor - the source of readings.
• actuator - the output the policy commands.
• transport - the link readings are published over; expected to be connected.
• codec - the wire format readings are encoded with.

Returns

A node ready to tick.

Examples

use pamoja_codec::CborCodec;
use pamoja_core::{Actuator, Result, Sensor, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_profile::{Alert, Node, Policy, Profile, Reaction};
use serde::{Deserialize, Serialize};

// A driver reads two quantities at once; the policy decides on both.
#[derive(Clone, Copy, Serialize, Deserialize)]
struct Soil {
    moisture: f32,
    temperature: f32,
}

struct Probe;
impl Sensor for Probe {
    type Reading = Soil;
    async fn read(&mut self) -> Result<Soil> {
        Ok(Soil { moisture: 22.0, temperature: 1.5 })
    }
}

struct Valve;
impl Actuator for Valve {
    type Command = bool;
    async fn apply(&mut self, _open: bool) -> Result<()> {
        Ok(())
    }
}

// Water only when the bed is dry and not about to freeze.
struct DripPolicy;
impl Policy for DripPolicy {
    type Reading = Soil;
    type Command = bool;
    fn evaluate(&mut self, soil: &Soil) -> Reaction {
        let frost = soil.temperature < 2.0;
        Reaction {
            actuator: Some(soil.moisture < 25.0 && !frost),
            alert: frost.then_some(Alert::Custom { code: "FrostRisk", value: soil.temperature }),
        }
    }
}

let mut link = LoopbackTransport::new(LoopbackBroker::new());
link.connect().await?;
let profile = Profile::irrigation_node();
let mut node = Node::with_policy(profile, DripPolicy, Probe, Valve, link, CborCodec);
let reaction = node.tick().await?.reaction;
assert_eq!(reaction.actuator, Some(false)); // dry, but frost is closer
assert_eq!(reaction.alert.map(Alert::kind), Some("FrostRisk"));

pub fn profile(&self) -> &Profile

Returns the profile this node was assembled from.

Returns

A reference to the node’s Profile.

pub fn policy(&self) -> &P

Returns the policy deciding this node’s readings.

Returns

A reference to the node’s policy.

pub fn sensor_mut(&mut self) -> &mut S

The sensor the node reads, so a driver can be reconfigured between ticks.

Returns

The sensor.

pub fn actuator_mut(&mut self) -> &mut A

The actuator the node drives, so its own state can be read or set directly.

The control loop switches this on every tick. Reaching it is for what the loop does not do: reading back what a recording output captured, or driving it by hand for a manual override.

Returns

The actuator.

pub fn transport_mut(&mut self) -> &mut T

The link readings are published over.

A node on a link that comes and goes usually publishes through a TransportLadder, which buffers a reading no rung would take. Flushing that buffer once the link is back is the caller’s, so this is how the caller reaches it.

Returns

The transport.

pub fn into_parts(self) -> (Profile, P, S, A, T, C)

Takes the node apart into the pieces it was assembled from.

Returns

The profile, the policy, the sensor, the actuator, the transport, and the codec.

pub fn schedule(&mut self, soc: f32, charging: bool) -> (PowerMode, Duration)

Returns the power mode and wait interval for the next cycle.

This assembles the profile’s PowerSchedule into a pamoja-power governor: as the battery drains the interval stretches, and a charging panel eases the node back toward its active cadence. The node remembers the mode it chose, so a charge hovering at a threshold keeps it in the lower mode until the charge clears the schedule’s hysteresis margin. The node never sleeps; the caller waits the returned Duration before the next tick, so timing stays outside the node and the decision logic remains synchronous and testable.

Arguments

• soc - the battery state of charge in [0.0, 1.0].
• charging - whether the panel is currently delivering charge.

Returns

The [PowerMode] to run in and how long to wait before the next cycle.

pub fn power_mode(&self) -> Option<PowerMode>

Returns the power mode the last schedule chose.

Returns

The mode, or None before the first schedule.

impl<S, T, C> Node<S, NoActuator, T, C>

pub fn monitor( profile: Profile, sensor: S, transport: T, codec: C, ) -> Result<Self>

Assembles a node for a profile that observes without driving an output.

Wires a NoActuator in place of a real output, so a monitoring profile such as well_level reads and publishes with the same shape as a controlling one.

Arguments

• profile - the profile to assemble.
• sensor - the source of readings.
• transport - the link readings are published over; expected to be connected.
• codec - the wire format readings are encoded with.

Returns

A node ready to tick, with no output to switch.

Errors

Returns Error::Codec when the profile names a custom control kind, as Node::new does.

impl<S, A, T, C, P> Node<S, A, T, C, P>

where S: Sensor, P: Policy<Reading = S::Reading>, P::Command: Clone, A: Actuator<Command = P::Command>, T: Transport, C: Codec<S::Reading>,

pub async fn tick(&mut self) -> Result<Tick<S::Reading, P::Command>>

Runs one read-decide-act-publish cycle.

Reads the sensor, evaluates the node’s policy, applies the resulting command to the actuator when the policy calls for one, and publishes the reading to the profile’s topic.

Returns

The Tick: the reading, and the Reaction the policy produced, the command that was applied (if any) and any alert the reading raised.

Errors

Returns Error::Io or Error::Closed if the sensor read or the actuator command fails, Error::Codec if the reading cannot be encoded, and Error::Transport or Error::Closed if the publish fails.

pub async fn run<W, F>( &mut self, battery: impl FnMut() -> (f32, bool), wait: W, ) -> Result<Infallible>

where W: FnMut(Duration) -> F, F: Future<Output = ()>,

Ticks, then waits the interval the battery’s charge calls for, for as long as every tick succeeds.

The node keeps no clock of its own: wait is the runtime’s sleep, such as tokio::time::sleep, so a node runs on any executor, and a test hands it one that returns at once. A program that carries on past a failed tick writes this loop itself from tick and schedule.

Arguments

• battery - reads the state of charge, from 0 to 1, and whether the panel is charging, before each wait; || (1.0, false) for a node on mains power.
• wait - waits a duration.

Returns

Only an error: the loop runs until a tick fails.

Errors

The first failed tick, with what tick returns; a sensor that has ended reports Error::Closed.

Examples

use core::time::Duration;

use pamoja_codec::JsonCodec;
use pamoja_core::{Actuator, Error, Result, Sensor, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_profile::{Node, Profile};

// A morning of readings that ends, which ends the loop.
struct Morning(Vec<f32>);
impl Sensor for Morning {
    type Reading = f32;
    async fn read(&mut self) -> Result<f32> {
        if self.0.is_empty() { Err(Error::Closed) } else { Ok(self.0.remove(0)) }
    }
}

struct Lamp;
impl Actuator for Lamp {
    type Command = bool;
    async fn apply(&mut self, _on: bool) -> Result<()> {
        Ok(())
    }
}

let mut link = LoopbackTransport::new(LoopbackBroker::new());
link.connect().await?;
let profile = Profile::vaccine_fridge_monitor();
let mut node = Node::new(profile, Morning(vec![4.0, 6.0]), Lamp, link, JsonCodec)?;

let mut waited = Vec::new();
let ended = node
    .run(|| (0.3, false), |interval: Duration| {
        waited.push(interval);
        async {}
    })
    .await;
assert!(matches!(ended, Err(Error::Closed)));
assert_eq!(waited, [Duration::from_secs(300), Duration::from_secs(300)]); // saver