Rules#

    use pamoja_codec::JsonCodec;
    use pamoja_core::{Receive, Transport};
    use pamoja_kit::Edge;
    use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
    use pamoja_profile::{Action, RuleEngine, Rules};

    // A rule is a file: the topic it watches, the line a reading crosses, the release
    // band that stops it firing over and over, and what to do on the way down and on the
    // way back. Two rules watch one bed here: one waters it when it dries past 30 and
    // stops once it is wetter than 35, and one raises an alarm when it is soaked past 60.
    let file = r#"{ "rules": [
  { "name": "water-when-dry",
    "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
    "then": [ { "drive": "bed-valve", "on": true },
              { "publish": "garden/bed-1/valve", "payload": "open" } ],
    "otherwise": [ { "drive": "bed-valve", "on": false },
                   { "publish": "garden/bed-1/valve", "payload": "closed" } ] },
  { "name": "flood-alarm",
    "when": { "topic": "garden/bed-1/moisture", "above": 60.0, "hysteresis": 5.0 },
    "then": [ { "publish": "garden/alarm", "payload": "waterlogged" } ] }
] }"#;

    // Three parties on one broker: the node that reads the bed, the engine that holds the
    // valve, and a watcher on the topics the rules publish to.
    let broker = LoopbackBroker::new();
    let mut probe = LoopbackTransport::new(broker.clone());
    let mut watcher = LoopbackTransport::new(broker.clone());
    probe.connect().await?;
    watcher.connect().await?;
    watcher.subscribe("garden/bed-1/valve").await?;
    watcher.subscribe("garden/alarm").await?;

    let valve = Valve::default();
    let mut engine = RuleEngine::new(
        Rules::from_json(file)?,
        LoopbackTransport::new(broker),
        JsonCodec,
    )
    .with_actuator("bed-valve", valve.clone());
    engine.connect().await?;
    println!(
        "watches   {}, and drives {}",
        engine.topics().join(", "),
        engine.actuators().join(", ")
    );

    // The bed dries out, is watered, and floods. A rule fires only as its condition sets
    // or clears, and the readings in between change nothing. At 65 two rules fire on one
    // reading, in the order the file lists them.
    for reading in [42.0f32, 31.0, 28.0, 33.0, 65.0, 50.0] {
        probe
            .send_text("garden/bed-1/moisture", &reading.to_string())
            .await?;
        let fired = engine.step().await?.expect("the link is up");
        if fired.is_empty() {
            println!("{reading:<10}nothing fired");
        }
        for one in &fired {
            let edge = match one.edge {
                Edge::Set => "set",
                Edge::Cleared => "cleared",
            };
            let actions: Vec<String> = one
                .actions
                .iter()
                .map(|action| match action {
                    Action::Drive { actuator, on } => {
                        format!("drive {actuator} {}", if *on { "on" } else { "off" })
                    }
                    Action::Publish { topic, payload } => format!("publish {payload} to {topic}"),
                })
                .collect();
            if actions.is_empty() {
                println!("{reading:<10}{} {edge}, with nothing to do", one.rule);
            } else {
                println!("{reading:<10}{} {edge}: {}", one.rule, actions.join(", "));
            }
        }
    }

    // The watcher heard every message the rules published, in the order they went out.
    let mut heard = Vec::new();
    for _ in 0..3 {
        let message = watcher.recv().await?.expect("a message");
        heard.push(message.text().expect("words").to_owned());
    }
    println!("heard     {}", heard.join(", "));
    let switches = valve.switches.lock().expect("valve lock").clone();
    let state = if switches.last() == Some(&true) {
        "on"
    } else {
        "off"
    };
    println!(
        "valve     switched {} times, and it is {state}",
        switches.len()
    );

use pamoja_profile::RuleEvaluator;

// A program that moves its own messages hands each reading to an evaluator, the
// engine's deciding half on its own, and carries out what it says.
let mut evaluator = RuleEvaluator::new(Rules::from_json(file)?)?;

// A reading that is not a number, such as the NaN a failed probe reports, is refused
// on a watched topic rather than leaving every rule as it was with nothing to say why.
match evaluator.evaluate("garden/bed-1/moisture", f32::NAN) {
    Ok(_) => println!("a reading of NaN was judged, which should never happen"),
    Err(error) => println!("refused   {error}"),
}

// A topic no rule watches is not judged at all, so even a NaN there says nothing.
if evaluator
    .evaluate("garden/bed-2/moisture", f32::NAN)?
    .is_empty()
{
    println!(
        "ignored   no rule watches garden/bed-2/moisture, so even a NaN there is not judged"
    );
}

// A file no engine could run is refused as it loads, with the rule and the reason.
for edited in [
    file.replace("garden/bed-1/moisture", "garden/+/moisture"),
    file.replace("\"flood-alarm\"", "\"water-when-dry\""),
] {
    match Rules::from_json(&edited).and_then(RuleEvaluator::new) {
        Ok(_) => println!("a file no engine could run was accepted, which should never happen"),
        Err(error) => println!("refused   {error}"),
    }
}

// With no release band, readings that hover at the line set and clear the rule on
// every sample, and each edge switches the valve. The band of 5 holds it through them.
let fires = |text: &str| -> std::result::Result<usize, Box<dyn Error>> {
    let mut judge = RuleEvaluator::new(Rules::from_json(text)?)?;
    let mut count = 0;
    for reading in [29.9, 30.1, 29.8, 30.2] {
        count += judge.evaluate("garden/bed-1/moisture", reading)?.len();
    }
    Ok(count)
};
let bare = fires(&file.replace("\"hysteresis\": 5.0", "\"hysteresis\": 0.0"))?;
let banded = fires(file)?;
println!(
    "chatter   4 readings hovering at 30 fire the rule {bare} times with no release band, {banded} with a band of 5"
);

import { LoopbackBroker } from '@pamoja/loopback'
import { type RuleAction, RuleActionKind, RuleEngine } from '@pamoja/profile'

// A rule is a file: the topic it watches, the line a reading crosses, the release band
// that stops it firing over and over, and what to do on the way down and on the way back.
// Two rules watch one bed here: one waters it when it dries past 30 and stops once it is
// wetter than 35, and one raises an alarm when it is soaked past 60.
const file = `{ "rules": [
  { "name": "water-when-dry",
    "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
    "then": [ { "drive": "bed-valve", "on": true },
              { "publish": "garden/bed-1/valve", "payload": "open" } ],
    "otherwise": [ { "drive": "bed-valve", "on": false },
                   { "publish": "garden/bed-1/valve", "payload": "closed" } ] },
  { "name": "flood-alarm",
    "when": { "topic": "garden/bed-1/moisture", "above": 60.0, "hysteresis": 5.0 },
    "then": [ { "publish": "garden/alarm", "payload": "waterlogged" } ] }
] }`

// Says what one action does, in the words the output uses.
const described = (action: RuleAction) =>
  action.kind === RuleActionKind.Drive
    ? `drive ${action.actuator} ${action.on ? 'on' : 'off'}`
    : `publish ${action.payload} to ${action.topic}`

async function main() {
  // Three parties on one broker: the node that reads the bed, the engine that holds the
  // valve, and a watcher on the topics the rules publish to.
  const broker = new LoopbackBroker()
  const probe = broker.link()
  const watcher = broker.link()
  const link = broker.link()
  await probe.connect()
  await watcher.connect()
  await link.connect()
  await watcher.subscribe('garden/bed-1/valve')
  await watcher.subscribe('garden/alarm')

  // The engine runs the file off its link: it listens on every topic a rule watches,
  // switches the outputs it was given by the names the file uses, and publishes over the
  // same link. Here the valve is a list of the settings it was given.
  const valve: boolean[] = []
  const engine = new RuleEngine(file, link, { actuators: { 'bed-valve': (on) => valve.push(on) } })
  await engine.listen()
  console.log(`watches   ${engine.topics.join(', ')}, and drives ${engine.actuators.join(', ')}`)

  // The bed dries out, is watered, and floods. A rule fires only as its condition sets or
  // clears, and the readings in between change nothing. At 65 two rules fire on one
  // reading, in the order the file lists them.
  for (const reading of [42, 31, 28, 33, 65, 50]) {
    await probe.send('garden/bed-1/moisture', String(reading))
    const fired = (await engine.step())!
    const at = String(reading).padEnd(10)
    if (fired.length === 0) {
      console.log(`${at}nothing fired`)
    }
    for (const one of fired) {
      if (one.actions.length === 0) {
        console.log(`${at}${one.rule} ${one.edge}, with nothing to do`)
      } else {
        console.log(`${at}${one.rule} ${one.edge}: ${one.actions.map(described).join(', ')}`)
      }
    }
  }

  // The watcher heard every message the rules published, in the order they went out.
  const heard: string[] = []
  for (let i = 0; i < 3; i += 1) {
    heard.push((await watcher.recv())!.text!)
  }
  console.log(`heard     ${heard.join(', ')}`)
  console.log(`valve     switched ${valve.length} times, and it is ${valve[valve.length - 1] ? 'on' : 'off'}`)
  return { heard, valve, engine }
}

main()

import { RuleEvaluator } from '@pamoja/profile'

function wrong(): void {
  // A program that moves its own messages hands each reading to an evaluator, the engine's
  // deciding half on its own, and carries out what it says.
  const judge = RuleEvaluator.fromJson(file)

  // A reading that is not a number, such as the NaN a failed probe reports, is refused on a
  // watched topic rather than leaving every rule as it was with nothing to say why.
  try {
    judge.evaluate('garden/bed-1/moisture', Number.NaN)
    console.log('a reading of NaN was judged, which should never happen')
  } catch (error) {
    console.log(`refused   ${(error as Error).message}`)
  }

  // A topic no rule watches is not judged at all, so even a NaN there says nothing.
  if (judge.evaluate('garden/bed-2/moisture', Number.NaN).length === 0) {
    console.log('ignored   no rule watches garden/bed-2/moisture, so even a NaN there is not judged')
  }

  // A file no engine could run is refused as it loads, with the rule and the reason.
  for (const edited of [
    file.split('garden/bed-1/moisture').join('garden/+/moisture'),
    file.replace('"flood-alarm"', '"water-when-dry"'),
  ]) {
    try {
      RuleEvaluator.fromJson(edited)
      console.log('a file no engine could run was accepted, which should never happen')
    } catch (error) {
      console.log(`refused   ${(error as Error).message}`)
    }
  }

  // With no release band, readings that hover at the line set and clear the rule on every
  // sample, and each edge switches the valve. The band of 5 holds it through them.
  const fires = (text: string) => {
    const rules = RuleEvaluator.fromJson(text)
    let count = 0
    for (const reading of [29.9, 30.1, 29.8, 30.2]) {
      count += rules.evaluate('garden/bed-1/moisture', reading).length
    }
    return count
  }
  const bare = fires(file.split('"hysteresis": 5.0').join('"hysteresis": 0.0'))
  const banded = fires(file)
  console.log(
    `chatter   4 readings hovering at 30 fire the rule ${bare} times with no release band, ${banded} with a band of 5`,
  )
}

import asyncio

from pamoja.loopback import LoopbackBroker
from pamoja.profile import RuleActionKind, RuleEngine

# A rule is a file: the topic it watches, the line a reading crosses, the release band that
# stops it firing over and over, and what to do on the way down and on the way back. Two
# rules watch one bed here: one waters it when it dries past 30 and stops once it is wetter
# than 35, and one raises an alarm when it is soaked past 60.
file = """{ "rules": [
  { "name": "water-when-dry",
    "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
    "then": [ { "drive": "bed-valve", "on": true },
              { "publish": "garden/bed-1/valve", "payload": "open" } ],
    "otherwise": [ { "drive": "bed-valve", "on": false },
                   { "publish": "garden/bed-1/valve", "payload": "closed" } ] },
  { "name": "flood-alarm",
    "when": { "topic": "garden/bed-1/moisture", "above": 60.0, "hysteresis": 5.0 },
    "then": [ { "publish": "garden/alarm", "payload": "waterlogged" } ] }
] }"""

def described(action) -> str:
    """Says what one action does, in the words the output uses."""
    if action.kind == RuleActionKind.DRIVE:
        return f"drive {action.actuator} {'on' if action.on else 'off'}"
    return f"publish {action.payload} to {action.topic}"


async def main() -> tuple[list[str], list[bool], RuleEngine]:
    # Three parties on one broker: the node that reads the bed, the engine that holds the
    # valve, and a watcher on the topics the rules publish to.
    broker = LoopbackBroker()
    probe = broker.link()
    watcher = broker.link()
    link = broker.link()
    await probe.connect()
    await watcher.connect()
    await link.connect()
    await watcher.subscribe("garden/bed-1/valve")
    await watcher.subscribe("garden/alarm")

    # The engine runs the file off its link: it listens on every topic a rule watches,
    # switches the outputs it was given by the names the file uses, and publishes over the
    # same link. Here the valve is a list of the settings it was given.
    valve: list[bool] = []
    engine = RuleEngine(file, link, actuators={"bed-valve": valve.append})
    await engine.listen()
    print(f"watches   {', '.join(engine.topics)}, and drives {', '.join(engine.actuators)}")

    # The bed dries out, is watered, and floods. A rule fires only as its condition sets or
    # clears, and the readings in between change nothing. At 65 two rules fire on one
    # reading, in the order the file lists them.
    for reading in [42, 31, 28, 33, 65, 50]:
        await probe.send("garden/bed-1/moisture", str(reading))
        fired = await engine.step()
        at = f"{reading:<10}"
        if not fired:
            print(f"{at}nothing fired")
        for one in fired:
            if not one.actions:
                print(f"{at}{one.rule} {one.edge}, with nothing to do")
            else:
                print(f"{at}{one.rule} {one.edge}: {', '.join(described(a) for a in one.actions)}")

    # The watcher heard every message the rules published, in the order they went out.
    heard = [(await watcher.recv()).text for _ in range(3)]
    print(f"heard     {', '.join(heard)}")
    print(f"valve     switched {len(valve)} times, and it is {'on' if valve[-1] else 'off'}")
    return heard, valve, engine


heard, valve, engine = asyncio.run(main())

from pamoja.core import PamojaError
from pamoja.profile import RuleEvaluator

# A program that moves its own messages hands each reading to an evaluator, the engine's
# deciding half on its own, and carries out what it says.
judge = RuleEvaluator.from_json(file)

# A reading that is not a number, such as the NaN a failed probe reports, is refused on a
# watched topic rather than leaving every rule as it was with nothing to say why.
try:
    judge.evaluate("garden/bed-1/moisture", float("nan"))
    print("a reading of NaN was judged, which should never happen")
except PamojaError as error:
    print(f"refused   {error}")

# A topic no rule watches is not judged at all, so even a NaN there says nothing.
if not judge.evaluate("garden/bed-2/moisture", float("nan")):
    print("ignored   no rule watches garden/bed-2/moisture, so even a NaN there is not judged")

# A file no engine could run is refused as it loads, with the rule and the reason.
for edited in [
    file.replace("garden/bed-1/moisture", "garden/+/moisture"),
    file.replace('"flood-alarm"', '"water-when-dry"'),
]:
    try:
        RuleEvaluator.from_json(edited)
        print("a file no engine could run was accepted, which should never happen")
    except PamojaError as error:
        print(f"refused   {error}")


def fires(text: str) -> int:
    """Counts how often the rules fire as four readings hover at the line."""
    rules = RuleEvaluator.from_json(text)
    return sum(len(rules.evaluate("garden/bed-1/moisture", r)) for r in [29.9, 30.1, 29.8, 30.2])


# With no release band, readings that hover at the line set and clear the rule on every
# sample, and each edge switches the valve. The band of 5 holds it through them.
bare = fires(file.replace('"hysteresis": 5.0', '"hysteresis": 0.0'))
banded = fires(file)
print(
    f"chatter   4 readings hovering at 30 fire the rule {bare} times with no release band, "
    f"{banded} with a band of 5"
)

// A rule is a file: the topic it watches, the line a reading crosses, the release
// band that stops it firing over and over, and what to do on the way down and on
// the way back. Two rules watch one bed here: one waters it when it dries past 30
// and stops once it is wetter than 35, and one raises an alarm when it is soaked
// past 60.
const string file = """
    { "rules": [
      { "name": "water-when-dry",
        "when": { "topic": "garden/bed-1/moisture", "below": 30.0, "hysteresis": 5.0 },
        "then": [ { "drive": "bed-valve", "on": true },
                  { "publish": "garden/bed-1/valve", "payload": "open" } ],
        "otherwise": [ { "drive": "bed-valve", "on": false },
                       { "publish": "garden/bed-1/valve", "payload": "closed" } ] },
      { "name": "flood-alarm",
        "when": { "topic": "garden/bed-1/moisture", "above": 60.0, "hysteresis": 5.0 },
        "then": [ { "publish": "garden/alarm", "payload": "waterlogged" } ] }
    ] }
    """;

// Three parties on one broker: the node that reads the bed, the engine that holds
// the valve, and a watcher on the topics the rules publish to.
using var broker = new LoopbackBroker();
using LoopbackTransport probe = broker.Link();
using LoopbackTransport watcher = broker.Link();
using LoopbackTransport link = broker.Link();
await probe.ConnectAsync();
await watcher.ConnectAsync();
await link.ConnectAsync();
await watcher.SubscribeAsync("garden/bed-1/valve");
await watcher.SubscribeAsync("garden/alarm");

// The engine runs the file off its link: it listens on every topic a rule watches,
// switches the outputs it was given by the names the file uses, and publishes over
// the same link. Here the valve is a list of the settings it was given.
var valve = new List<bool>();
using var engine = new RuleEngine(file, link, new Dictionary<string, Func<bool, ValueTask>>
{
    ["bed-valve"] = on =>
    {
        valve.Add(on);
        return ValueTask.CompletedTask;
    },
});
await engine.ListenAsync();
Console.WriteLine(
    $"watches   {string.Join(", ", engine.Topics)}, and drives {string.Join(", ", engine.Actuators)}");

static string Described(RuleAction action) => action.Kind == RuleActionKind.Drive
    ? $"drive {action.Actuator} {(action.On == true ? "on" : "off")}"
    : $"publish {action.Payload} to {action.Topic}";

// The bed dries out, is watered, and floods. A rule fires only as its condition
// sets or clears, and the readings in between change nothing. At 65 two rules fire
// on one reading, in the order the file lists them.
foreach (int reading in new[] { 42, 31, 28, 33, 65, 50 })
{
    await probe.SendAsync("garden/bed-1/moisture", reading.ToString(CultureInfo.InvariantCulture));
    IReadOnlyList<RuleFired> fired = (await engine.StepAsync(TimeSpan.FromSeconds(5)))!;
    string at = $"{reading,-10}";
    if (fired.Count == 0)
    {
        Console.WriteLine($"{at}nothing fired");
    }

    foreach (RuleFired one in fired)
    {
        string edge = one.Edge == Pamoja.Kit.Edge.Set ? "set" : "cleared";
        Console.WriteLine(one.Actions.Count == 0
            ? $"{at}{one.Rule} {edge}, with nothing to do"
            : $"{at}{one.Rule} {edge}: {string.Join(", ", one.Actions.Select(Described))}");
    }
}

// The watcher heard every message the rules published, in the order they went out.
var heard = new List<string>();
for (int i = 0; i < 3; i++)
{
    heard.Add((await watcher.ReceiveAsync())!.Text);
}

Console.WriteLine($"heard     {string.Join(", ", heard)}");
Console.WriteLine($"valve     switched {valve.Count} times, and it is {(valve[^1] ? "on" : "off")}");

// A program that moves its own messages hands each reading to an evaluator, the
// engine's deciding half on its own, and carries out what it says.
using var judge = RuleEvaluator.FromJson(file);

// A reading that is not a number, such as the NaN a failed probe reports, is
// refused on a watched topic rather than leaving every rule as it was with nothing
// to say why.
try
{
    judge.Evaluate("garden/bed-1/moisture", float.NaN);
    Console.WriteLine("a reading of NaN was judged, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"refused   {error.Message}");
}

// A topic no rule watches is not judged at all, so even a NaN there says nothing.
if (judge.Evaluate("garden/bed-2/moisture", float.NaN).Count == 0)
{
    Console.WriteLine("ignored   no rule watches garden/bed-2/moisture, so even a NaN there is not judged");
}

// A file no engine could run is refused as it loads, with the rule and the reason.
foreach (string edited in new[]
{
    file.Replace("garden/bed-1/moisture", "garden/+/moisture"),
    file.Replace("\"flood-alarm\"", "\"water-when-dry\""),
})
{
    try
    {
        using var accepted = RuleEvaluator.FromJson(edited);
        Console.WriteLine("a file no engine could run was accepted, which should never happen");
    }
    catch (PamojaException error)
    {
        Console.WriteLine($"refused   {error.Message}");
    }
}

// With no release band, readings that hover at the line set and clear the rule on
// every sample, and each edge switches the valve. The band of 5 holds it through
// them.
static int Fires(string text)
{
    using var rules = RuleEvaluator.FromJson(text);
    return new[] { 29.9f, 30.1f, 29.8f, 30.2f }
        .Sum(reading => rules.Evaluate("garden/bed-1/moisture", reading).Count);
}

int bare = Fires(file.Replace("\"hysteresis\": 5.0", "\"hysteresis\": 0.0"));
int banded = Fires(file);
Console.WriteLine(
    $"chatter   4 readings hovering at 30 fire the rule {bare} times with no release band, {banded} with a band of 5");