Device profiles#

use pamoja_codec::{Codec, JsonCodec};
use pamoja_core::{Receive, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_profile::{Node, Profile};

// A profile is a file. This one ships in the catalog under profiles/: it holds a
// brooder at 32 C by switching a heat lamp, says what it reads, and says how a
// dashboard draws it.
let text = std::fs::read_to_string("profiles/brooder-heater.json")?;
let profile = Profile::from_json(&text)?;
let reads = profile
    .reads
    .clone()
    .expect("a catalog profile says what it reads");
println!(
    "profile   {} reads {} in {} and reports on {}",
    profile.name, reads.quantity, reads.unit, profile.topic
);
let topic = profile.topic.clone();
let element = profile
    .presentation
    .clone()
    .expect("a presentation")
    .elements[0]
    .clone();

// A node is the profile and the parts that make it run: a sensor, an output, and a
// link. A morning of readings stands in for the probe, and a dashboard listens on the
// same broker.
let broker = LoopbackBroker::new();
let mut link = LoopbackTransport::new(broker.clone());
let mut dashboard = LoopbackTransport::new(broker);
link.connect().await?;
dashboard.connect().await?;
dashboard.subscribe(&topic).await?;
let morning = Morning(vec![27.5, 31.8, 32.6, 32.1, 31.4]);
let mut node = Node::new(profile, morning, Lamp::default(), link, JsonCodec)?;

// Each tick reads, decides, switches the lamp, and publishes the reading. The lamp
// comes on at 31.5 C or below and goes off at 32.5 C or above, and in between it stays
// as it was; a reading more than 4 C from 32 raises an alert as well.
let mut was = false;
for _ in 0..5 {
    let tick = node.tick().await?;
    let on = tick.reaction.actuator == Some(true);
    let change = match (was, on) {
        (false, true) => "lamp on",
        (true, false) => "lamp off",
        (true, true) => "lamp stays on",
        (false, false) => "lamp stays off",
    };
    let alert = tick
        .reaction
        .alert
        .map(|alert| format!(", alert {}", alert.kind()))
        .unwrap_or_default();
    println!("{:<10}{change}{alert}", format!("{} C", tick.reading));
    was = on;
}

// The dashboard heard every reading the node published.
let mut heard = Vec::new();
for _ in 0..5 {
    let message = dashboard.recv().await?.expect("a reading");
    let reading: f32 = JsonCodec.decode(&message.payload)?;
    heard.push(reading.to_string());
}
println!("heard     {} on {topic}", heard.join(", "));

// Between ticks the node waits as long as its battery allows: often on a healthy
// charge, sparingly on a low one. run() does this until stopped, waiting each interval.
for charge in [0.8, 0.3, 0.1] {
    let (mode, wait) = node.schedule(charge, false);
    println!(
        "battery   at {:.0}% it runs {mode:?} and waits {} s",
        charge * 100.0,
        wait.as_secs()
    );
}

// The same file says how a dashboard draws the node.
let [low, high] = element.band.expect("a band");
println!(
    "draws     {} in {} on a {}, safe from {low} to {high}",
    element.key,
    element.unit,
    element.viz.name()
);

use pamoja_profile::Alert;

// A level warns before a tank or a well runs dry. The shipped well profile counts
// 0.5 m as dry and warns once the last fall puts dry six samples away or nearer.
let mut well = Profile::well_level().controller()?;
for depth in [5.0, 4.4, 3.8] {
    match well.evaluate(depth).alert {
        Some(Alert::RunningOut { samples }) => {
            println!("well      {depth} m: dry in {samples} samples at this rate, RunningOut")
        }
        _ => println!("well      {depth} m: no warning yet"),
    }
}

// A surge warns when a reading moves too far in one sample. The shipped flood sensor
// warns when a river rises more than 0.3 m between two readings.
let mut river = Profile::flood_sensor().controller()?;
for gauge in [1.2, 1.35, 1.9] {
    match river.evaluate(gauge).alert {
        Some(Alert::ChangingFast { rate }) => {
            println!("river     {gauge} m: up {rate:.2} m in one sample, ChangingFast")
        }
        _ => println!("river     {gauge} m: no warning"),
    }
}

use pamoja_profile::{BoxedPolicy, Params, Policy, PolicyRegistry, Reaction};

/// A policy of the program's own: the lamp on below the setpoint, and a condition of
/// its own when the chicks are chilled.
struct BrooderGuard {
    setpoint: f32,
    chilled_below: f32,
}

impl Policy for BrooderGuard {
    type Reading = f32;
    type Command = bool;

    fn evaluate(&mut self, reading: &f32) -> Reaction {
        let chilled = *reading < self.chilled_below;
        Reaction {
            actuator: Some(*reading < self.setpoint),
            alert: chilled.then_some(Alert::Custom {
                code: "Chilled",
                value: *reading,
            }),
        }
    }
}

// A manifest may name a control kind the library never shipped, with its parameters
// beside it. The program registers the code that decides it under that name, and the
// node runs whichever kind the file names.
let guarded =
    Profile::from_json(&text.replace("\"kind\": \"setpoint\"", "\"kind\": \"brooder_guard\""))?;
let registry = PolicyRegistry::new().register("brooder_guard", |params: &Params| {
    let setpoint = params.number("setpoint").unwrap_or(32.0) as f32;
    let band = params.number("safe_band").unwrap_or(4.0) as f32;
    Ok(Box::new(BrooderGuard {
        setpoint,
        chilled_below: setpoint - band,
    }) as BoxedPolicy)
});
println!(
    "custom    {} is decided by the program's own code, registered under its name",
    guarded.control.kind()
);
let mut link = LoopbackTransport::new(LoopbackBroker::new());
link.connect().await?;
let mut node = Node::resolve(
    guarded,
    &registry,
    Morning(vec![27.5]),
    Lamp::default(),
    link,
    JsonCodec,
)?;
let tick = node.tick().await?;
let lamp = if tick.reaction.actuator == Some(true) {
    "on"
} else {
    "off"
};
let alert = tick.reaction.alert.map(Alert::kind).unwrap_or("none");
println!(
    "{:<10}lamp {lamp}, alert {alert}",
    format!("{} C", tick.reading)
);

use pamoja_security::DeviceIdentity;

// A profile can arrive over a link as well as from a disk: on an MQTT topic the gateway
// publishes to, or as the body of an HTTP response. The gateway signs what it sends and
// each node holds only the gateway's public key, so a profile is checked before it is
// read, and one from anywhere else never runs.
let gateway = DeviceIdentity::from_seed(&[7u8; 32]);
let trusted = gateway.public();
let broker = LoopbackBroker::new();
let mut uplink = LoopbackTransport::new(broker.clone());
let mut downlink = LoopbackTransport::new(broker);
uplink.connect().await?;
downlink.connect().await?;
let fleet = "fleet/brooders/profile";
downlink.subscribe(fleet).await?;

uplink
    .send(fleet, &gateway.sign_message(text.as_bytes()))
    .await?;
let message = downlink.recv().await?.expect("a profile");
let signed = trusted.verify_message(&message.payload)?;
let delivered = Profile::from_json(std::str::from_utf8(signed)?)?;
println!(
    "network   {} arrived on {fleet}, signed by the gateway, and loads",
    delivered.name
);

let stranger = DeviceIdentity::from_seed(&[9u8; 32]);
uplink
    .send(fleet, &stranger.sign_message(text.as_bytes()))
    .await?;
let message = downlink.recv().await?.expect("a profile");
if trusted.verify_message(&message.payload).is_err() {
    println!("network   one signed by any other key is refused before it is read");
}

// A probe that fails reports a reading that is not a number. The controller raises
// it rather than going quiet, and the lamp holds its state; what off means for the
// chicks is the node's call.
let profile = Profile::from_json(&text)?;
let mut controller = profile.controller()?;
controller.evaluate(27.5);
let failed = controller.evaluate(f32::NAN);
if let Some(alert) = failed.alert {
    let holds = if failed.actuator == Some(true) {
        "on"
    } else {
        "off"
    };
    println!(
        "probe     a reading of NaN raises {}, and the lamp holds {holds}",
        alert.kind()
    );
}

// A controller built again for each reading forgets the lamp was on, so inside the
// deadband it switches the lamp off.
let first = profile.controller()?.evaluate(27.5).actuator;
let then = profile.controller()?.evaluate(31.8).actuator;
if first == Some(true) && then == Some(false) {
    println!(
        "fresh     built again for each reading, the controller turns the lamp off at 31.8 C"
    );
}

// A manifest no node could run is refused as it loads, with the reason. So is a
// misspelled field, with the one it was probably meant to be and where it sits,
// rather than leaving the default in its place without a word.
for edited in [
    text.replace("\"hysteresis\": 0.5", "\"hysteresis\": 0.0"),
    text.replace("\"saver_secs\": 600", "\"saver_secs\": 60"),
    text.replace("\"saver_below\"", "\"saver_bellow\""),
] {
    match Profile::from_json(&edited) {
        Ok(_) => {
            println!("a manifest no node could run was accepted, which should never happen")
        }
        Err(error) => println!("refused   {error}"),
    }
}

// A kind the library does not ship loads with its parameters, but no built-in
// controller decides it, so a node without a registry that knows it is refused
// rather than running one that never switches the lamp.
let unknown =
    Profile::from_json(&text.replace("\"kind\": \"setpoint\"", "\"kind\": \"brooder_guard\""))?;
match unknown.controller() {
    Ok(_) => println!("a custom kind ran without its policy, which should never happen"),
    Err(error) => println!("refused   {error}"),
}

import { readFileSync } from 'node:fs'
import { LoopbackBroker } from '@pamoja/loopback'
import { Node, Profile } from '@pamoja/profile'

// A profile is a file. This one ships in the catalog under profiles/: it holds a brooder at
// 32 C by switching a heat lamp, says what it reads, and says how a dashboard draws it.
const text = readFileSync('profiles/brooder-heater.json', 'utf8')
const profile = Profile.fromJson(text)

async function example(): Promise<boolean> {
  const reads = profile.reads!
  console.log(`profile   ${profile.name} reads ${reads.quantity} in ${reads.unit} and reports on ${profile.topic}`)

  // A node is the profile and the parts that make it run: a sensor, an output, and a link.
  // A morning of readings stands in for the probe, and a dashboard listens on the same
  // broker.
  const broker = new LoopbackBroker()
  const link = broker.link()
  const dashboard = broker.link()
  await link.connect()
  await dashboard.connect()
  await dashboard.subscribe(profile.topic)
  const morning = [27.5, 31.8, 32.6, 32.1, 31.4]
  let lamp = false
  const node = new Node({
    profile,
    read: () => morning.shift()!,
    drive: (on) => {
      lamp = on
    },
    link,
  })

  // Each tick reads, decides, switches the lamp, and publishes the reading. The lamp comes on
  // at 31.5 C or below and goes off at 32.5 C or above, and in between it stays as it was; a
  // reading more than 4 C from 32 raises an alert as well.
  let was = false
  for (let at = 0; at < 5; at += 1) {
    const { reading, reaction } = await node.tick()
    const on = reaction.actuator === true
    const change = on ? (was ? 'lamp stays on' : 'lamp on') : was ? 'lamp off' : 'lamp stays off'
    const alert = reaction.alert ? `, alert ${reaction.alert.kind}` : ''
    console.log(`${`${reading} C`.padEnd(10)}${change}${alert}`)
    was = on
  }

  // The dashboard heard every reading the node published.
  const heard: number[] = []
  for (let at = 0; at < 5; at += 1) {
    heard.push((await dashboard.recv())!.number!)
  }
  console.log(`heard     ${heard.join(', ')} on ${profile.topic}`)

  // Between ticks the node waits as long as its battery allows: often on a healthy charge,
  // sparingly on a low one. run() does this until stopped, waiting each interval.
  for (const charge of [0.8, 0.3, 0.1]) {
    const { mode, waitMs } = node.schedule(charge)
    console.log(`battery   at ${(charge * 100).toFixed(0)}% it runs ${mode} and waits ${waitMs / 1000} s`)
  }

  // The same file says how a dashboard draws the node.
  const element = profile.presentation!.elements[0]
  const [low, high] = element.band!
  console.log(`draws     ${element.key} in ${element.unit} on a ${element.viz}, safe from ${low} to ${high}`)
  return lamp
}

import { AlertKind } from '@pamoja/profile'

function kinds(): void {
  // A level warns before a tank or a well runs dry. The shipped well profile counts 0.5 m
  // as dry and warns once the last fall puts dry six samples away or nearer.
  const well = Profile.wellLevel().controller()
  for (const depth of [5.0, 4.4, 3.8]) {
    const alert = well.evaluate(depth).alert
    if (alert?.kind === AlertKind.RunningOut) {
      console.log(`well      ${depth} m: dry in ${alert.samples} samples at this rate, RunningOut`)
    } else {
      console.log(`well      ${depth} m: no warning yet`)
    }
  }

  // A surge warns when a reading moves too far in one sample. The shipped flood sensor warns
  // when a river rises more than 0.3 m between two readings.
  const river = Profile.floodSensor().controller()
  for (const gauge of [1.2, 1.35, 1.9]) {
    const alert = river.evaluate(gauge).alert
    if (alert?.kind === AlertKind.ChangingFast) {
      console.log(`river     ${gauge} m: up ${alert.rate?.toFixed(2)} m in one sample, ChangingFast`)
    } else {
      console.log(`river     ${gauge} m: no warning`)
    }
  }
}

import { type Policy, PolicyRegistry } from '@pamoja/profile'

// A policy of the program's own: the lamp on below the setpoint, and a condition of its own
// when the chicks are chilled.
function brooderGuard(params: Record<string, number | boolean | string>): Policy {
  const setpoint = Number(params.setpoint ?? 32)
  const chilledBelow = setpoint - Number(params.safe_band ?? 4)
  return {
    evaluate: (reading) => ({
      actuator: reading < setpoint,
      alert: reading < chilledBelow ? { kind: AlertKind.Custom, code: 'Chilled', value: reading } : undefined,
    }),
  }
}

async function custom(): Promise<void> {
  // A manifest may name a control kind the library never shipped, with its parameters
  // beside it. The program registers the code that decides it under that name, and the node
  // runs whichever kind the file names.
  const guarded = Profile.fromJson(text.replace('"kind": "setpoint"', '"kind": "brooder_guard"'))
  const registry = new PolicyRegistry().register('brooder_guard', brooderGuard)
  console.log(`custom    ${guarded.control.customKind} is decided by the program's own code, registered under its name`)
  const link = new LoopbackBroker().link()
  await link.connect()
  let lamp = false
  const node = new Node({
    profile: guarded,
    policy: registry,
    read: () => 27.5,
    drive: (on) => {
      lamp = on
    },
    link,
  })
  const { reading, reaction } = await node.tick()
  const alert = reaction.alert?.code ?? reaction.alert?.kind ?? 'none'
  console.log(`${`${reading} C`.padEnd(10)}lamp ${lamp ? 'on' : 'off'}, alert ${alert}`)
}

import { DeviceIdentity, verifyMessage } from '@pamoja/security'

async function network(): Promise<Profile> {
  // A profile can arrive over a link as well as from a disk: on an MQTT topic the gateway
  // publishes to, or as the body of an HTTP response. The gateway signs what it sends and each
  // node holds only the gateway's public key, so a profile is checked before it is read, and
  // one from anywhere else never runs.
  const gateway = DeviceIdentity.fromSeed(Buffer.alloc(32, 7))
  const trusted = gateway.publicKey()
  const broker = new LoopbackBroker()
  const uplink = broker.link()
  const downlink = broker.link()
  await uplink.connect()
  await downlink.connect()
  const fleet = 'fleet/brooders/profile'
  await downlink.subscribe(fleet)

  await uplink.send(fleet, gateway.signMessage(text))
  const signed = verifyMessage(trusted, (await downlink.recv())!.payload)
  const delivered = Profile.fromJson(signed!.toString('utf8'))
  console.log(`network   ${delivered.name} arrived on ${fleet}, signed by the gateway, and loads`)

  const stranger = DeviceIdentity.fromSeed(Buffer.alloc(32, 9))
  await uplink.send(fleet, stranger.signMessage(text))
  if (verifyMessage(trusted, (await downlink.recv())!.payload) === null) {
    console.log('network   one signed by any other key is refused before it is read')
  }
  return delivered
}

function wrong(): void {
  // A probe that fails reports a reading that is not a number. The controller raises it
  // rather than going quiet, and the lamp holds its state; what off means for the chicks is
  // the node's call.
  const controller = profile.controller()
  controller.evaluate(27.5)
  const failed = controller.evaluate(Number.NaN)
  if (failed.alert) {
    const holds = failed.actuator === true ? 'on' : 'off'
    console.log(`probe     a reading of NaN raises ${failed.alert.kind}, and the lamp holds ${holds}`)
  }

  // A controller built again for each reading forgets the lamp was on, so inside the
  // deadband it switches the lamp off.
  const first = profile.controller().evaluate(27.5).actuator
  const then = profile.controller().evaluate(31.8).actuator
  if (first === true && then === false) {
    console.log('fresh     built again for each reading, the controller turns the lamp off at 31.8 C')
  }

  // A manifest no node could run is refused as it loads, with the reason. So is a misspelled
  // field, with the one it was probably meant to be and where it sits, rather than leaving
  // the default in its place without a word.
  for (const edited of [
    text.replace('"hysteresis": 0.5', '"hysteresis": 0.0'),
    text.replace('"saver_secs": 600', '"saver_secs": 60'),
    text.replace('"saver_below"', '"saver_bellow"'),
  ]) {
    try {
      Profile.fromJson(edited)
      console.log('a manifest no node could run was accepted, which should never happen')
    } catch (error) {
      console.log(`refused   ${(error as Error).message}`)
    }
  }

  // A kind the library does not ship loads with its parameters, but no built-in controller
  // decides it, so a node without a registry that knows it is refused rather than running
  // one that never switches the lamp.
  const unknown = Profile.fromJson(text.replace('"kind": "setpoint"', '"kind": "brooder_guard"'))
  try {
    unknown.controller()
    console.log('a custom kind ran without its policy, which should never happen')
  } catch (error) {
    console.log(`refused   ${(error as Error).message}`)
  }
}

from pathlib import Path

from pamoja.loopback import LoopbackBroker
from pamoja.profile import Node, Profile

# A profile is a file. This one ships in the catalog under profiles/: it holds a brooder at
# 32 C by switching a heat lamp, says what it reads, and says how a dashboard draws it.
text = Path("profiles/brooder-heater.json").read_text(encoding="utf-8")
profile = Profile.from_json(text)
print(
    f"profile   {profile.name} reads {profile.reads.quantity} in {profile.reads.unit} "
    f"and reports on {profile.topic}"
)


async def example() -> bool:
    # A node is the profile and the parts that make it run: a sensor, an output, and a link.
    # A morning of readings stands in for the probe, and a dashboard listens on the same
    # broker.
    broker = LoopbackBroker()
    link = broker.link()
    dashboard = broker.link()
    await link.connect()
    await dashboard.connect()
    await dashboard.subscribe(profile.topic)
    morning = [27.5, 31.8, 32.6, 32.1, 31.4]
    lamp = []
    node = Node(profile, read=lambda: morning.pop(0), drive=lamp.append, link=link)

    # Each tick reads, decides, switches the lamp, and publishes the reading. The lamp comes
    # on at 31.5 C or below and goes off at 32.5 C or above, and in between it stays as it
    # was; a reading more than 4 C from 32 raises an alert as well.
    was = False
    for _ in range(5):
        tick = await node.tick()
        on = tick.reaction.actuator is True
        if on:
            change = "lamp stays on" if was else "lamp on"
        else:
            change = "lamp off" if was else "lamp stays off"
        alert = f", alert {tick.reaction.alert.kind}" if tick.reaction.alert else ""
        print(f"{f'{tick.reading:g} C':<10}{change}{alert}")
        was = on

    # The dashboard heard every reading the node published.
    heard = [f"{(await dashboard.recv()).number:g}" for _ in range(5)]
    print(f"heard     {', '.join(heard)} on {profile.topic}")

    # Between ticks the node waits as long as its battery allows: often on a healthy charge,
    # sparingly on a low one. run() does this until stopped, waiting each interval.
    for charge in [0.8, 0.3, 0.1]:
        mode, seconds = node.schedule(charge)
        print(f"battery   at {charge * 100:.0f}% it runs {mode} and waits {seconds:g} s")

    # The same file says how a dashboard draws the node.
    element = profile.presentation.elements[0]
    low, high = element.band
    print(
        f"draws     {element.key} in {element.unit} on a {element.viz}, "
        f"safe from {low:g} to {high:g}"
    )
    return lamp[-1]


lamp_on = asyncio.run(example())

from pamoja.profile import AlertKind

# A level warns before a tank or a well runs dry. The shipped well profile counts 0.5 m as
# dry and warns once the last fall puts dry six samples away or nearer.
well = Profile.well_level().controller()
for depth in [5.0, 4.4, 3.8]:
    alert = well.evaluate(depth).alert
    if alert and alert.kind == AlertKind.RUNNING_OUT:
        print(f"well      {depth:g} m: dry in {alert.samples} samples at this rate, RunningOut")
    else:
        print(f"well      {depth:g} m: no warning yet")

# A surge warns when a reading moves too far in one sample. The shipped flood sensor warns
# when a river rises more than 0.3 m between two readings.
river = Profile.flood_sensor().controller()
for gauge in [1.2, 1.35, 1.9]:
    alert = river.evaluate(gauge).alert
    if alert and alert.kind == AlertKind.CHANGING_FAST:
        print(f"river     {gauge:g} m: up {alert.rate:.2f} m in one sample, ChangingFast")
    else:
        print(f"river     {gauge:g} m: no warning")

from pamoja.profile import CustomAlert, Decision, PolicyRegistry


class BrooderGuard:
    """A policy of the program's own: the lamp on below the setpoint, and a condition of its
    own when the chicks are chilled."""

    def __init__(self, params: dict) -> None:
        self.setpoint = params.get("setpoint", 32.0)
        self.chilled_below = self.setpoint - params.get("safe_band", 4.0)

    def evaluate(self, reading: float) -> Decision:
        chilled = reading < self.chilled_below
        alert = CustomAlert("Chilled", reading) if chilled else None
        return Decision(actuator=reading < self.setpoint, alert=alert)


async def custom() -> None:
    # A manifest may name a control kind the library never shipped, with its parameters
    # beside it. The program registers the code that decides it under that name, and the
    # node runs whichever kind the file names.
    guarded = Profile.from_json(text.replace('"kind": "setpoint"', '"kind": "brooder_guard"'))
    registry = PolicyRegistry().register("brooder_guard", BrooderGuard)
    print(
        f"custom    {guarded.control.custom_kind} is decided by the program's own code, "
        "registered under its name"
    )
    link = LoopbackBroker().link()
    await link.connect()
    lamp = []
    node = Node(guarded, read=lambda: 27.5, drive=lamp.append, link=link, policy=registry)
    tick = await node.tick()
    alert = tick.reaction.alert.code if tick.reaction.alert else "none"
    print(f"{f'{tick.reading:g} C':<10}lamp {'on' if lamp[-1] else 'off'}, alert {alert}")


asyncio.run(custom())

from pamoja.security import DeviceIdentity, verify_message


async def network() -> Profile:
    # A profile can arrive over a link as well as from a disk: on an MQTT topic the gateway
    # publishes to, or as the body of an HTTP response. The gateway signs what it sends and
    # each node holds only the gateway's public key, so a profile is checked before it is
    # read, and one from anywhere else never runs.
    gateway = DeviceIdentity.from_seed(bytes([7]) * 32)
    trusted = gateway.public_key
    broker = LoopbackBroker()
    uplink = broker.link()
    downlink = broker.link()
    await uplink.connect()
    await downlink.connect()
    fleet = "fleet/brooders/profile"
    await downlink.subscribe(fleet)

    await uplink.send(fleet, gateway.sign_message(text.encode("utf-8")))
    signed = verify_message(trusted, (await downlink.recv()).payload)
    delivered = Profile.from_json(signed.decode("utf-8"))
    print(f"network   {delivered.name} arrived on {fleet}, signed by the gateway, and loads")

    stranger = DeviceIdentity.from_seed(bytes([9]) * 32)
    await uplink.send(fleet, stranger.sign_message(text.encode("utf-8")))
    if verify_message(trusted, (await downlink.recv()).payload) is None:
        print("network   one signed by any other key is refused before it is read")
    return delivered


delivered = asyncio.run(network())

from pamoja.core import PamojaError

# A probe that fails reports a reading that is not a number. The controller raises it
# rather than going quiet, and the lamp holds its state; what off means for the chicks is
# the node's call.
controller = profile.controller()
controller.evaluate(27.5)
failed = controller.evaluate(float("nan"))
if failed.alert:
    holds = "on" if failed.actuator is True else "off"
    print(f"probe     a reading of NaN raises {failed.alert.kind}, and the lamp holds {holds}")

# A controller built again for each reading forgets the lamp was on, so inside the deadband
# it switches the lamp off.
first = profile.controller().evaluate(27.5).actuator
then = profile.controller().evaluate(31.8).actuator
if first is True and then is False:
    print("fresh     built again for each reading, the controller turns the lamp off at 31.8 C")

# A manifest no node could run is refused as it loads, with the reason. So is a misspelled
# field, with the one it was probably meant to be and where it sits, rather than leaving the
# default in its place without a word.
for edited in [
    text.replace('"hysteresis": 0.5', '"hysteresis": 0.0'),
    text.replace('"saver_secs": 600', '"saver_secs": 60'),
    text.replace('"saver_below"', '"saver_bellow"'),
]:
    try:
        Profile.from_json(edited)
        print("a manifest no node could run was accepted, which should never happen")
    except PamojaError as error:
        print(f"refused   {error}")

# A kind the library does not ship loads with its parameters, but no built-in controller
# decides it, so a node without a registry that knows it is refused rather than running one
# that never switches the lamp.
unknown = Profile.from_json(text.replace('"kind": "setpoint"', '"kind": "brooder_guard"'))
try:
    unknown.controller()
    print("a custom kind ran without its policy, which should never happen")
except PamojaError as error:
    print(f"refused   {error}")

// A profile is a file. This one ships in the catalog under profiles/: it holds a
// brooder at 32 C by switching a heat lamp, says what it reads, and says how a
// dashboard draws it.
string text = File.ReadAllText("profiles/brooder-heater.json");
using var profile = Profile.FromJson(text);
Reads reads = profile.Reads!.Value;
Console.WriteLine($"profile   {profile.Name} reads {reads.Quantity} in {reads.Unit} and reports on {profile.Topic}");

// A node is the profile and the parts that make it run: a sensor, an output, and a
// link. A morning of readings stands in for the probe, and a dashboard listens on the
// same broker.
using var broker = new LoopbackBroker();
using var link = broker.Link();
using var dashboard = broker.Link();
await link.ConnectAsync();
await dashboard.ConnectAsync();
await dashboard.SubscribeAsync(profile.Topic);
var morning = new Queue<float>([27.5f, 31.8f, 32.6f, 32.1f, 31.4f]);
bool lamp = false;
using var node = new Node(
    profile,
    () => ValueTask.FromResult(morning.Dequeue()),
    link,
    drive: on =>
    {
        lamp = on;
        return ValueTask.CompletedTask;
    });

// Each tick reads, decides, switches the lamp, and publishes the reading. The lamp
// comes on at 31.5 C or below and goes off at 32.5 C or above, and in between it stays
// as it was; a reading more than 4 C from 32 raises an alert as well.
bool was = false;
for (int at = 0; at < 5; at++)
{
    Tick tick = await node.TickAsync();
    bool on = tick.Reaction.Actuator == true;
    string change = on ? (was ? "lamp stays on" : "lamp on") : was ? "lamp off" : "lamp stays off";
    string alert = tick.Reaction.Alert is { } raised ? $", alert {raised.Kind}" : string.Empty;
    Console.WriteLine(Invariant($"{Invariant($"{tick.Reading} C"),-10}{change}{alert}"));
    was = on;
}

// The dashboard heard every reading the node published.
var heard = new List<string>();
for (int at = 0; at < 5; at++)
{
    TransportMessage? message = await dashboard.ReceiveAsync(TimeSpan.FromSeconds(5));
    heard.Add(Invariant($"{message!.Number}"));
}

Console.WriteLine($"heard     {string.Join(", ", heard)} on {profile.Topic}");

// Between ticks the node waits as long as its battery allows: often on a healthy
// charge, sparingly on a low one. RunAsync does this until cancelled, waiting each
// interval.
foreach (float charge in new[] { 0.8f, 0.3f, 0.1f })
{
    (var mode, TimeSpan wait) = node.Schedule(charge);
    Console.WriteLine(Invariant($"battery   at {charge * 100:F0}% it runs {mode} and waits {wait.TotalSeconds} s"));
}

// The same file says how a dashboard draws the node.
ElementSpec element = profile.Presentation!.Elements[0];
string graphic = element.Viz.ToString().ToLowerInvariant();
Console.WriteLine(Invariant(
    $"draws     {element.Key} in {element.Unit} on a {graphic}, safe from {element.Band![0]} to {element.Band[1]}"));

// A level warns before a tank or a well runs dry. The shipped well profile counts
// 0.5 m as dry and warns once the last fall puts dry six samples away or nearer.
using (var wellLevel = Profile.WellLevel())
using (Controller well = wellLevel.Controller())
{
    foreach (float depth in new[] { 5.0f, 4.4f, 3.8f })
    {
        Console.WriteLine(well.Evaluate(depth).Alert is { Kind: AlertKind.RunningOut } alert
            ? Invariant($"well      {depth} m: dry in {alert.Samples} samples at this rate, RunningOut")
            : Invariant($"well      {depth} m: no warning yet"));
    }
}

// A surge warns when a reading moves too far in one sample. The shipped flood sensor
// warns when a river rises more than 0.3 m between two readings.
using var floodSensor = Profile.FloodSensor();
using (Controller river = floodSensor.Controller())
{
    foreach (float gauge in new[] { 1.2f, 1.35f, 1.9f })
    {
        Console.WriteLine(river.Evaluate(gauge).Alert is { Kind: AlertKind.ChangingFast } alert
            ? Invariant($"river     {gauge} m: up {alert.Rate:F2} m in one sample, ChangingFast")
            : Invariant($"river     {gauge} m: no warning"));
    }
}

/// <summary>
/// A policy of the program's own: the lamp on below the setpoint, and a condition of its
/// own when the chicks are chilled.
/// </summary>
private sealed class BrooderGuard(IReadOnlyDictionary<string, object> parameters) : IPolicy
{
    private readonly float _setpoint = Convert.ToSingle(parameters["setpoint"]);
    private readonly float _chilledBelow = Convert.ToSingle(parameters["setpoint"]) - Convert.ToSingle(parameters["safe_band"]);

    public Reaction Evaluate(float reading) => new(
        reading < _setpoint,
        reading < _chilledBelow ? new Alert(AlertKind.Custom, null, null, null, "Chilled", reading) : null);
}

private static async Task CustomAsync(string text)
{
    // A manifest may name a control kind the library never shipped, with its parameters
    // beside it. The program registers the code that decides it under that name, and the
    // node runs whichever kind the file names.
    using var guarded = Profile.FromJson(text.Replace("\"kind\": \"setpoint\"", "\"kind\": \"brooder_guard\""));
    var registry = new PolicyRegistry().Register("brooder_guard", parameters => new BrooderGuard(parameters));
    Console.WriteLine($"custom    {guarded.Control.CustomKind} is decided by the program's own code, registered under its name");
    using var broker = new LoopbackBroker();
    using var link = broker.Link();
    await link.ConnectAsync();
    bool lamp = false;
    using var node = new Node(
        guarded,
        () => ValueTask.FromResult(27.5f),
        link,
        drive: on =>
        {
            lamp = on;
            return ValueTask.CompletedTask;
        },
        policy: registry.Resolve(guarded));
    Tick tick = await node.TickAsync();
    string alert = tick.Reaction.Alert?.Code ?? tick.Reaction.Alert?.Kind.ToString() ?? "none";
    Console.WriteLine(Invariant($"{Invariant($"{tick.Reading} C"),-10}lamp {(lamp ? "on" : "off")}, alert {alert}"));
}

private static async Task<Profile> NetworkAsync(string text)
{
    // A profile can arrive over a link as well as from a disk: on an MQTT topic the
    // gateway publishes to, or as the body of an HTTP response. The gateway signs what it
    // sends and each node holds only the gateway's public key, so a profile is checked
    // before it is read, and one from anywhere else never runs.
    byte[] seed = new byte[DeviceIdentity.KeyLength];
    Array.Fill(seed, (byte)7);
    using var gateway = new DeviceIdentity(seed);
    byte[] trusted = gateway.PublicKey;
    using var broker = new LoopbackBroker();
    using var uplink = broker.Link();
    using var downlink = broker.Link();
    await uplink.ConnectAsync();
    await downlink.ConnectAsync();
    const string fleet = "fleet/brooders/profile";
    await downlink.SubscribeAsync(fleet);

    byte[] manifest = Encoding.UTF8.GetBytes(text);
    await uplink.SendAsync(fleet, gateway.SignMessage(manifest));
    TransportMessage? message = await downlink.ReceiveAsync(TimeSpan.FromSeconds(5));
    byte[]? signed = DeviceIdentity.VerifyMessage(trusted, message!.Payload);
    var delivered = Profile.FromJson(Encoding.UTF8.GetString(signed!));
    Console.WriteLine($"network   {delivered.Name} arrived on {fleet}, signed by the gateway, and loads");

    Array.Fill(seed, (byte)9);
    using var stranger = new DeviceIdentity(seed);
    await uplink.SendAsync(fleet, stranger.SignMessage(manifest));
    message = await downlink.ReceiveAsync(TimeSpan.FromSeconds(5));
    if (DeviceIdentity.VerifyMessage(trusted, message!.Payload) is null)
    {
        Console.WriteLine("network   one signed by any other key is refused before it is read");
    }

    return delivered;
}

// A probe that fails reports a reading that is not a number. The controller raises
// it rather than going quiet, and the lamp holds its state; what off means for the
// chicks is the node's call.
using (Controller controller = profile.Controller())
{
    controller.Evaluate(27.5f);
    Reaction failed = controller.Evaluate(float.NaN);
    if (failed.Alert is { } invalid)
    {
        string holds = failed.Actuator == true ? "on" : "off";
        Console.WriteLine($"probe     a reading of NaN raises {invalid.Kind}, and the lamp holds {holds}");
    }
}

// A controller built again for each reading forgets the lamp was on, so inside the
// deadband it switches the lamp off.
bool? first;
bool? then;
using (Controller once = profile.Controller())
{
    first = once.Evaluate(27.5f).Actuator;
}

using (Controller again = profile.Controller())
{
    then = again.Evaluate(31.8f).Actuator;
}

if (first == true && then == false)
{
    Console.WriteLine("fresh     built again for each reading, the controller turns the lamp off at 31.8 C");
}

// A manifest no node could run is refused as it loads, with the reason. So is a
// misspelled field, with the one it was probably meant to be and where it sits,
// rather than leaving the default in its place without a word.
foreach (string edited in new[]
{
    text.Replace("\"hysteresis\": 0.5", "\"hysteresis\": 0.0"),
    text.Replace("\"saver_secs\": 600", "\"saver_secs\": 60"),
    text.Replace("\"saver_below\"", "\"saver_bellow\""),
})
{
    try
    {
        using var accepted = Profile.FromJson(edited);
        Console.WriteLine("a manifest no node could run was accepted, which should never happen");
    }
    catch (PamojaException error)
    {
        Console.WriteLine($"refused   {error.Message}");
    }
}

// A kind the library does not ship loads with its parameters, but no built-in
// controller decides it, so a node without a registry that knows it is refused
// rather than running one that never switches the lamp.
using var unknown = Profile.FromJson(text.Replace("\"kind\": \"setpoint\"", "\"kind\": \"brooder_guard\""));
try
{
    using Controller inert = unknown.Controller();
    Console.WriteLine("a custom kind ran without its policy, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"refused   {error.Message}");
}