One core.
Every language.
For the devices that
change lives.

use pamoja_codec::{decode_deltas, encode_deltas};
use pamoja_core::{Receive, Transport};
use pamoja_kit::Smoother;
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
use pamoja_security::{DeviceIdentity, PublicIdentity};
use pamoja_sensors::ds18b20::{temperature_from_celsius, Resolution, Scratchpad};

// The link. A loopback broker stands in for MQTT or CoAP, so this runs with no network
// and nothing listening. Point the node at a real transport and nothing below changes.
let broker = LoopbackBroker::new();
let mut node = LoopbackTransport::new(broker.clone());
let mut gateway = LoopbackTransport::new(broker);
node.connect().await?;
gateway.connect().await?;
let topic = "sensors/1/temperature";
gateway.subscribe(topic).await?;

// The device's identity is provisioned once and never leaves it. The gateway is told
// only the public half, which is how it recognizes this device later.
let device = DeviceIdentity::from_seed(&[7u8; 32]);
let known = PublicIdentity::from_bytes(&device.public().to_bytes()).expect("a valid key");
println!("gateway trusts device {}", known.fingerprint());

// A stand-in for the thermometer. On a running node these nine bytes arrive from the
// 1-Wire bus; here the library builds what a part at 25.0625 C would send.
let off_the_bus = Scratchpad::new(
    temperature_from_celsius(25.0625, Resolution::Bits12),
    Resolution::Bits12,
    75,
    -10,
)
.to_bytes();

// On the node. The part checksums every read, so a value mangled on a long run is an
// error rather than a plausible temperature a couple of degrees off.
let celsius = Scratchpad::parse(&off_the_bus)
    .expect("the thermometer's checksum matches")
    .temperature_celsius();
println!("read      {celsius:.4} C");

// Readings jitter, so smooth them, and send a batch rather than one at a time.
// Successive readings differ by very little, so the differences cost a fraction of
// what the readings would on a link that charges by the byte.
let mut smoother = Smoother::new(0.5);
let batch: Vec<i64> = [celsius, celsius + 0.5, celsius + 0.4]
    .into_iter()
    .map(|sample| (smoother.update(sample) * 100.0).round() as i64)
    .collect();
let packed = encode_deltas(&batch);
let (readings, bytes) = (batch.len(), packed.len());
println!("packed    {readings} readings into {bytes} bytes");

// Sign the batch and send it. The signature travels with the payload as one message,
// so there is nothing to keep together and split correctly at the far end.
let message = device.sign_message(&packed);
node.send(topic, &message)
    .await
    .expect("the node publishes");

// On the gateway. Verifying returns the payload, so a reading that was altered on the
// way, or signed by some other device, never reaches the code that unpacks it.
let received = gateway.recv().await?.expect("a message");
match known.verify_message(&received.payload) {
    Ok(payload) => {
        let readings = decode_deltas(payload).expect("a valid batch");
        println!("gateway   accepted {readings:?} in hundredths of a degree");
    }
    Err(error) => println!("gateway   rejected the reading: {error}"),
}

import { packSamples, unpackSamples } from '@pamoja/codec'
import { Smoother } from '@pamoja/kit'
import { LoopbackBroker } from '@pamoja/loopback'
import { DeviceIdentity, fingerprint, verifyMessage } from '@pamoja/security'
import { ds18b20 } from '@pamoja/sensors'

// The device's identity is provisioned once and never leaves it. The gateway is told only
// the public half, which is how it recognizes this device later.
const SEED = Buffer.alloc(32, 7)
const TOPIC = 'sensors/1/temperature'

async function main(): Promise<Buffer> {
  // The link. A loopback broker stands in for MQTT or CoAP, so this runs with no network
  // and nothing listening. Point the node at a real transport and nothing below changes.
  const broker = new LoopbackBroker()
  const node = broker.link()
  const gateway = broker.link()
  await node.connect()
  await gateway.connect()
  await gateway.subscribe(TOPIC)

  const device = DeviceIdentity.fromSeed(SEED)
  const known = device.publicKey()
  console.log(`gateway trusts device ${fingerprint(known)}`)

  // A stand-in for the thermometer. On a running node these nine bytes arrive from the
  // 1-Wire bus; here the library builds what a part at 25.0625 C would send.
  const offTheBus = ds18b20.buildScratchpad(25.0625, 12, 75, -10)

  // On the node. The part checksums every read, so a value mangled on a long run is an
  // error rather than a plausible temperature a couple of degrees off.
  const celsius = ds18b20.parseScratchpad(offTheBus).microCelsius / 1e6
  console.log(`read      ${celsius.toFixed(4)} C`)

  // Readings jitter, so smooth them, and send a batch rather than one at a time.
  // Successive readings differ by very little, so the differences cost a fraction of what
  // the readings would on a link that charges by the byte.
  const smoother = new Smoother(0.5)
  const batch = [celsius, celsius + 0.5, celsius + 0.4].map((sample) =>
    Math.round(smoother.update(sample) * 100),
  )
  const packed = packSamples(batch)
  console.log(`packed    ${batch.length} readings into ${packed.length} bytes`)

  // Sign the batch and send it. The signature travels with the payload as one message, so
  // there is nothing to keep together and split correctly at the far end.
  await node.send(TOPIC, device.signMessage(packed))

  // On the gateway. Verifying returns the payload, so a reading that was altered on the
  // way, or signed by some other device, never reaches the code that unpacks it.
  const received = await gateway.recv()
  const payload = verifyMessage(known, received!.payload)
  if (payload === null) {
    console.log('gateway   rejected the reading')
  } else {
    console.log(`gateway   accepted ${unpackSamples(payload).join(', ')} in hundredths of a degree`)
  }

  return received!.payload
}

main()

import asyncio

from pamoja import sensors
from pamoja.codec import pack_samples, unpack_samples
from pamoja.kit import Smoother
from pamoja.loopback import LoopbackBroker
from pamoja.security import DeviceIdentity, fingerprint, verify_message

# The device's identity is provisioned once and never leaves it. The gateway is told only
# the public half, which is how it recognizes this device later.
SEED = bytes([7]) * 32
TOPIC = "sensors/1/temperature"


async def main() -> bytes:
    # The link. A loopback broker stands in for MQTT or CoAP, so this runs with no network
    # and nothing listening. Point the node at a real transport and nothing below changes.
    broker = LoopbackBroker()
    node = broker.link()
    gateway = broker.link()
    await node.connect()
    await gateway.connect()
    await gateway.subscribe(TOPIC)

    device = DeviceIdentity.from_seed(SEED)
    known = device.public_key
    print(f"gateway trusts device {fingerprint(known)}")

    # A stand-in for the thermometer. On a running node these nine bytes arrive from the
    # 1-Wire bus; here the library builds what a part at 25.0625 C would send.
    off_the_bus = sensors.ds18b20.build_scratchpad(25.0625, 12, 75, -10)

    # On the node. The part checksums every read, so a value mangled on a long run is an
    # error rather than a plausible temperature a couple of degrees off.
    celsius = sensors.ds18b20.parse_scratchpad(off_the_bus).micro_celsius / 1e6
    print(f"read      {celsius:.4f} C")

    # Readings jitter, so smooth them, and send a batch rather than one at a time.
    # Successive readings differ by very little, so the differences cost a fraction of
    # what the readings would on a link that charges by the byte.
    smoother = Smoother(0.5)
    batch = [
        round(smoother.update(sample) * 100)
        for sample in (celsius, celsius + 0.5, celsius + 0.4)
    ]
    packed = pack_samples(batch)
    print(f"packed    {len(batch)} readings into {len(packed)} bytes")

    # Sign the batch and send it. The signature travels with the payload as one message,
    # so there is nothing to keep together and split correctly at the far end.
    await node.send(TOPIC, device.sign_message(packed))

    # On the gateway. Verifying returns the payload, so a reading that was altered on the
    # way, or signed by some other device, never reaches the code that unpacks it.
    received = await gateway.recv()
    payload = verify_message(known, received.payload)
    if payload is None:
        print("gateway   rejected the reading")
    else:
        print(f"gateway   accepted {unpack_samples(payload)} in hundredths of a degree")

    return received.payload


message = asyncio.run(main())

byte[] seed = new byte[DeviceIdentity.KeyLength];
Array.Fill(seed, (byte)7);

// The link. A loopback broker stands in for MQTT or CoAP, so this runs with no
// network and nothing listening. Point the node at a real transport and nothing
// below changes.
using var broker = new LoopbackBroker();
using LoopbackTransport node = broker.Link();
using LoopbackTransport gateway = broker.Link();
await node.ConnectAsync();
await gateway.ConnectAsync();
await gateway.SubscribeAsync(Topic);

using var device = new DeviceIdentity(seed);
byte[] known = device.PublicKey;
Console.WriteLine($"gateway trusts device {DeviceIdentity.FingerprintOf(known)}");

// A stand-in for the thermometer. On a running node these nine bytes arrive from
// the 1-Wire bus; here the library builds what a part at 25.0625 C would send.
byte[] offTheBus = Ds18b20.BuildScratchpad(25.0625f, 12, 75, -10);

// On the node. The part checksums every read, so a value mangled on a long run is
// an error rather than a plausible temperature a couple of degrees off.
float celsius = Ds18b20.ParseScratchpad(offTheBus).MicroCelsius / 1e6f;
Console.WriteLine($"read      {celsius:F4} C");

// Readings jitter, so smooth them, and send a batch rather than one at a time.
// Successive readings differ by very little, so the differences cost a fraction of
// what the readings would on a link that charges by the byte.
using var smoother = new Smoother(0.5f);
long[] batch =
[
    .. new[] { celsius, celsius + 0.5f, celsius + 0.4f }
        .Select(sample => (long)Math.Round(smoother.Update(sample) * 100)),
];
byte[] packed = Codec.PackSamples(batch);
Console.WriteLine($"packed    {batch.Length} readings into {packed.Length} bytes");

// Sign the batch and send it. The signature travels with the payload as one
// message, so there is nothing to keep together and split correctly at the far end.
await node.SendAsync(Topic, device.SignMessage(packed));

// On the gateway. Verifying returns the payload, so a reading that was altered on
// the way, or signed by some other device, never reaches the code that unpacks it.
TransportMessage? received = await gateway.ReceiveAsync();
byte[]? payload = DeviceIdentity.VerifyMessage(known, received!.Payload);
if (payload is null)
{
    Console.WriteLine("gateway   rejected the reading");
}
else
{
    Console.WriteLine(
        $"gateway   accepted {string.Join(", ", Codec.UnpackSamples(payload))}"
        + " in hundredths of a degree");
}