Serial framing#

use std::time::Duration;

use pamoja_hal::port::{SerialPort, Settings};
use pamoja_serial::cobs::{self, CobsDecoder};
use pamoja_serial::slip;

// A reading is a two-byte sequence number, most significant byte first, then its text.
fn reading(sequence: u16, text: &str) -> Vec<u8> {
    let mut payload = sequence.to_be_bytes().to_vec();
    payload.extend_from_slice(text.as_bytes());
    payload
}

// The line: 115200 baud, eight data bits, no parity, one stop bit. Ten bits a character.
let settings = Settings::new(115_200);
println!(
    "line         {settings}, {} bits a character, {:.2} us each",
    settings.bits_per_character(),
    settings.character_nanos() as f64 / 1_000.0
);

// The two ends of the cable with nothing plugged in. On a Raspberry Pi the gateway's end
// is SerialPort::open("/dev/serial0", settings) and nothing after this statement changes.
let (gateway, node) = SerialPort::pair(settings);

// A UART carries bytes, and nothing marks where a message ends, so the node frames each
// reading with COBS: zero becomes the one byte that ends a frame and never appears inside
// one, which matters here, since the sequence number is full of zeros.
let texts = ["wind=12.4", "wind=13.1", "wind=11.8"];
let mut sent = 0;
let mut frame = [0u8; cobs::max_encoded_len(32)];
for (sequence, text) in (1..).zip(texts) {
    let framed = cobs::encode(&reading(sequence, text), &mut frame)?;
    node.write(&frame[..framed])?;
    sent += framed;
}
println!(
    "node         {} readings of {} bytes, framed as {sent} bytes",
    texts.len(),
    reading(1, texts[0]).len()
);

// A read returns whatever has arrived, which is rarely one frame: here it is all three.
// The decoder splits the stream back into payloads at each delimiter.
let mut buffer = [0u8; 256];
let got = gateway.read(&mut buffer, Duration::from_millis(100))?;
println!("gateway      {got} bytes in one read");
let mut decoder: CobsDecoder<64> = CobsDecoder::new();
let mut payloads = Vec::new();
for &byte in &buffer[..got] {
    if let Ok(Some(payload)) = decoder.push(byte) {
        payloads.push(payload.to_vec());
    }
}
for payload in &payloads {
    let (sequence, text) = payload.split_at(2);
    let sequence = u16::from_be_bytes([sequence[0], sequence[1]]);
    println!("reading {sequence}    {}", String::from_utf8_lossy(text));
}

// What one frame costs on the wire at this speed, start and stop bits included.
let frame_len = sent / texts.len();
println!(
    "on the wire  {:.2} ms for a {frame_len}-byte frame at {settings}",
    settings.transfer_micros(frame_len) as f64 / 1_000.0
);

// The node restarts partway through a frame. As it comes back up it sends a lone
// delimiter, which closes off the half frame, so the gateway drops it rather than gluing
// it to the next one, and then it sends the reading again.
let framed = cobs::encode(&reading(4, "wind=12.9"), &mut frame)?;
node.write(&frame[..framed / 2])?;
node.write(&[cobs::DELIMITER])?;
node.write(&frame[..framed])?;
let got = gateway.read(&mut buffer, Duration::from_millis(100))?;
let mut dropped = 0;
let mut resent = Vec::new();
for &byte in &buffer[..got] {
    match decoder.push(byte) {
        Ok(Some(payload)) => resent.push(payload.to_vec()),
        Ok(None) => {}
        Err(_) => dropped += 1,
    }
}
println!(
    "restart      {dropped} frame cut short and dropped, then {}",
    String::from_utf8_lossy(&resent[0][2..])
);

// SLIP, the older framing, ends a frame with one reserved byte and escapes that byte and
// its own escape byte inside one. With no reserved bytes in a reading it costs a byte less
// than COBS; a payload full of them costs up to twice its length under SLIP, and never
// more than one byte in 254 over under COBS.
let first = reading(1, texts[0]);
let mut slip_frame = [0u8; slip::max_encoded_len(32)];
let slip_len = slip::encode(&first, &mut slip_frame)?;
let cobs_len = cobs::encode(&first, &mut frame)?;
println!(
    "framing      {} payload bytes: {slip_len} under SLIP, {cobs_len} under COBS",
    first.len()
);

// The node goes quiet. A read waits for the first byte up to its timeout; on a port with
// nothing plugged in it returns at once and counts the wait instead of sleeping through
// it, so a test of a silent node takes no time.
let before = gateway.waited_micros();
let got = gateway.read(&mut buffer, Duration::from_millis(500))?;
println!(
    "silence      {got} bytes in {} ms, counted and not slept",
    (gateway.waited_micros() - before) / 1_000
);

import { SerialPort } from '@pamoja/hal'
import { COBS_DELIMITER_BYTE, CobsDecoder, cobs, slip } from '@pamoja/serial'

// A reading is a two-byte sequence number, most significant byte first, then its text.
function reading(sequence: number, text: string): Buffer {
  const number = Buffer.alloc(2)
  number.writeUInt16BE(sequence)
  return Buffer.concat([number, Buffer.from(text)])
}

async function main() {
  // The line: 115200 baud, eight data bits, no parity, one stop bit. Ten bits a character.
  const settings = { baud: 115200 }
  const line = SerialPort.describe(settings)
  const characterMicros = SerialPort.characterNanos(settings) / 1000
  console.log(
    `line         ${line}, ${SerialPort.bitsPerCharacter(settings)} bits a character, ` +
      `${characterMicros.toFixed(2)} us each`,
  )

  // The two ends of the cable with nothing plugged in. On a Raspberry Pi the gateway's end is
  // SerialPort.open('/dev/serial0', settings) and nothing after this statement changes.
  const [gateway, node] = SerialPort.pair(settings)

  // A UART carries bytes, and nothing marks where a message ends, so the node frames each
  // reading with COBS: zero becomes the one byte that ends a frame and never appears inside
  // one, which matters here, since the sequence number is full of zeros.
  const texts = ['wind=12.4', 'wind=13.1', 'wind=11.8']
  let sent = 0
  for (const [index, text] of texts.entries()) {
    const frame = cobs.encode(reading(index + 1, text))
    await node.write(frame)
    sent += frame.length
  }
  console.log(
    `node         ${texts.length} readings of ${reading(1, texts[0]).length} bytes, framed as ${sent} bytes`,
  )

  // A read returns whatever has arrived, which is rarely one frame: here it is all three. The
  // decoder splits the stream back into payloads at each delimiter.
  const arrived = await gateway.read(256, 100)
  console.log(`gateway      ${arrived.length} bytes in one read`)
  const decoder = new CobsDecoder()
  const payloads = decoder.feed(arrived)
  for (const payload of payloads) {
    console.log(`reading ${payload.readUInt16BE(0)}    ${payload.subarray(2).toString()}`)
  }

  // What one frame costs on the wire at this speed, start and stop bits included.
  const frameLength = sent / texts.length
  const frameMillis = SerialPort.transferMicros(settings, frameLength) / 1000
  console.log(`on the wire  ${frameMillis.toFixed(2)} ms for a ${frameLength}-byte frame at ${line}`)

  // The node restarts partway through a frame. As it comes back up it sends a lone delimiter,
  // which closes off the half frame, so the gateway drops it rather than gluing it to the next
  // one, and then it sends the reading again.
  const again = cobs.encode(reading(4, 'wind=12.9'))
  await node.write(again.subarray(0, Math.floor(again.length / 2)))
  await node.write(Buffer.from([COBS_DELIMITER_BYTE]))
  await node.write(again)
  const droppedBefore = decoder.discarded
  const resent = decoder.feed(await gateway.read(256, 100))
  const dropped = decoder.discarded - droppedBefore
  console.log(
    `restart      ${dropped} frame cut short and dropped, then ${resent[0].subarray(2).toString()}`,
  )

  // SLIP, the older framing, ends a frame with one reserved byte and escapes that byte and its
  // own escape byte inside one. With no reserved bytes in a reading it costs a byte less than
  // COBS; a payload full of them costs up to twice its length under SLIP, and never more than
  // one byte in 254 over under COBS.
  const first = reading(1, texts[0])
  const slipLength = slip.encode(first).length
  const cobsLength = cobs.encode(first).length
  console.log(
    `framing      ${first.length} payload bytes: ${slipLength} under SLIP, ${cobsLength} under COBS`,
  )

  // The node goes quiet. A read waits for the first byte up to its timeout; on a port with
  // nothing plugged in it resolves at once and counts the wait instead of sleeping through it,
  // so a test of a silent node takes no time.
  const waitedBefore = gateway.waitedMicros
  const quiet = await gateway.read(256, 500)
  const waited = (gateway.waitedMicros - waitedBefore) / 1000
  console.log(`silence      ${quiet.length} bytes in ${waited} ms, counted and not slept`)

  return { settings, payloads, sent, dropped, resent, slipLength, cobsLength, gateway, node, again }
}

main()

from pamoja.hal import SerialPort, SerialSettings
from pamoja.serial import COBS_DELIMITER, CobsDecoder, cobs, slip


def reading(sequence: int, text: str) -> bytes:
    """A reading: a two-byte sequence number, most significant byte first, then its text."""
    return sequence.to_bytes(2, "big") + text.encode()


# The line: 115200 baud, eight data bits, no parity, one stop bit. Ten bits a character.
settings = SerialSettings(115_200)
print(
    f"line         {settings}, {settings.bits_per_character} bits a character, "
    f"{settings.character_nanos / 1_000:.2f} us each"
)

# The two ends of the cable with nothing plugged in. On a Raspberry Pi the gateway's end is
# SerialPort.open("/dev/serial0", settings) and nothing after this statement changes.
gateway, node = SerialPort.pair(settings)

# A UART carries bytes, and nothing marks where a message ends, so the node frames each reading
# with COBS: zero becomes the one byte that ends a frame and never appears inside one, which
# matters here, since the sequence number is full of zeros.
texts = ["wind=12.4", "wind=13.1", "wind=11.8"]
sent = 0
for sequence, text in enumerate(texts, start=1):
    frame = cobs.encode(reading(sequence, text))
    node.write(frame)
    sent += len(frame)
print(
    f"node         {len(texts)} readings of {len(reading(1, texts[0]))} bytes, "
    f"framed as {sent} bytes"
)

# A read returns whatever has arrived, which is rarely one frame: here it is all three. The
# decoder splits the stream back into payloads at each delimiter.
arrived = gateway.read(256, timeout=0.1)
print(f"gateway      {len(arrived)} bytes in one read")
decoder = CobsDecoder()
payloads = decoder.feed(arrived)
for payload in payloads:
    print(f"reading {int.from_bytes(payload[:2], 'big')}    {payload[2:].decode()}")

# What one frame costs on the wire at this speed, start and stop bits included.
frame_length = sent // len(texts)
print(
    f"on the wire  {settings.transfer_micros(frame_length) / 1_000:.2f} ms "
    f"for a {frame_length}-byte frame at {settings}"
)

# The node restarts partway through a frame. As it comes back up it sends a lone delimiter,
# which closes off the half frame, so the gateway drops it rather than gluing it to the next
# one, and then it sends the reading again.
again = cobs.encode(reading(4, "wind=12.9"))
node.write(again[: len(again) // 2])
node.write(bytes([COBS_DELIMITER]))
node.write(again)
dropped_before = decoder.discarded
resent = decoder.feed(gateway.read(256, timeout=0.1))
dropped = decoder.discarded - dropped_before
print(f"restart      {dropped} frame cut short and dropped, then {resent[0][2:].decode()}")

# SLIP, the older framing, ends a frame with one reserved byte and escapes that byte and its
# own escape byte inside one. With no reserved bytes in a reading it costs a byte less than
# COBS; a payload full of them costs up to twice its length under SLIP, and never more than
# one byte in 254 over under COBS.
first = reading(1, texts[0])
slip_length = len(slip.encode(first))
cobs_length = len(cobs.encode(first))
print(
    f"framing      {len(first)} payload bytes: {slip_length} under SLIP, "
    f"{cobs_length} under COBS"
)

# The node goes quiet. A read waits for the first byte up to its timeout; on a port with
# nothing plugged in it returns at once and counts the wait instead of sleeping through it, so
# a test of a silent node takes no time.
waited_before = gateway.waited_micros
quiet = gateway.read(256, timeout=0.5)
waited = (gateway.waited_micros - waited_before) // 1_000
print(f"silence      {len(quiet)} bytes in {waited} ms, counted and not slept")

// A reading is a two-byte sequence number, most significant byte first, then its text.
static byte[] Reading(ushort sequence, string text)
{
    byte[] payload = new byte[2 + Encoding.UTF8.GetByteCount(text)];
    BinaryPrimitives.WriteUInt16BigEndian(payload, sequence);
    Encoding.UTF8.GetBytes(text, payload.AsSpan(2));
    return payload;
}

// The line: 115200 baud, eight data bits, no parity, one stop bit. Ten bits a character.
var settings = new SerialSettings(115_200);
Console.WriteLine(Invariant(
    $"line         {settings}, {settings.BitsPerCharacter} bits a character, {settings.CharacterNanos / 1_000.0:F2} us each"));

// The two ends of the cable with nothing plugged in. On a Raspberry Pi the gateway's end
// is SerialPort.Open("/dev/serial0", settings) and nothing after this statement changes.
var (gateway, node) = SerialPort.Pair(settings);
using (gateway)
using (node)
{
    // A UART carries bytes, and nothing marks where a message ends, so the node frames
    // each reading with COBS: zero becomes the one byte that ends a frame and never
    // appears inside one, which matters here, since the sequence number is full of zeros.
    string[] texts = ["wind=12.4", "wind=13.1", "wind=11.8"];
    int sent = 0;
    for (int index = 0; index < texts.Length; index++)
    {
        byte[] frame = Serial.CobsEncode(Reading((ushort)(index + 1), texts[index]));
        node.Write(frame);
        sent += frame.Length;
    }

    Console.WriteLine(
        $"node         {texts.Length} readings of {Reading(1, texts[0]).Length} bytes, framed as {sent} bytes");

    // A read returns whatever has arrived, which is rarely one frame: here it is all
    // three. The decoder splits the stream back into payloads at each delimiter.
    byte[] arrived = gateway.Read(256, TimeSpan.FromMilliseconds(100));
    Console.WriteLine($"gateway      {arrived.Length} bytes in one read");
    using var decoder = new CobsDecoder();
    byte[][] payloads = decoder.Feed(arrived);
    foreach (byte[] payload in payloads)
    {
        ushort sequence = BinaryPrimitives.ReadUInt16BigEndian(payload);
        Console.WriteLine($"reading {sequence}    {Encoding.UTF8.GetString(payload, 2, payload.Length - 2)}");
    }

    // What one frame costs on the wire at this speed, start and stop bits included.
    int frameLength = sent / texts.Length;
    Console.WriteLine(Invariant(
        $"on the wire  {settings.TransferMicros(frameLength) / 1_000.0:F2} ms for a {frameLength}-byte frame at {settings}"));

    // The node restarts partway through a frame. As it comes back up it sends a lone
    // delimiter, which closes off the half frame, so the gateway drops it rather than
    // gluing it to the next one, and then it sends the reading again.
    byte[] again = Serial.CobsEncode(Reading(4, "wind=12.9"));
    node.Write(again.AsSpan(0, again.Length / 2));
    node.Write([Serial.CobsDelimiter]);
    node.Write(again);
    ulong droppedBefore = decoder.Discarded;
    byte[][] resent = decoder.Feed(gateway.Read(256, TimeSpan.FromMilliseconds(100)));
    ulong dropped = decoder.Discarded - droppedBefore;
    Console.WriteLine(
        $"restart      {dropped} frame cut short and dropped, then {Encoding.UTF8.GetString(resent[0], 2, resent[0].Length - 2)}");

    // SLIP, the older framing, ends a frame with one reserved byte and escapes that byte
    // and its own escape byte inside one. With no reserved bytes in a reading it costs a
    // byte less than COBS; a payload full of them costs up to twice its length under
    // SLIP, and never more than one byte in 254 over under COBS.
    byte[] first = Reading(1, texts[0]);
    int slipLength = Serial.SlipEncode(first).Length;
    int cobsLength = Serial.CobsEncode(first).Length;
    Console.WriteLine(
        $"framing      {first.Length} payload bytes: {slipLength} under SLIP, {cobsLength} under COBS");

    // The node goes quiet. A read waits for the first byte up to its timeout; on a port
    // with nothing plugged in it returns at once and counts the wait instead of sleeping
    // through it, so a test of a silent node takes no time.
    ulong waitedBefore = gateway.WaitedMicros;
    byte[] quiet = gateway.Read(256, TimeSpan.FromMilliseconds(500));
    ulong waited = (gateway.WaitedMicros - waitedBefore) / 1_000;
    Console.WriteLine($"silence      {quiet.Length} bytes in {waited} ms, counted and not slept");

use pamoja_hal::port::{SerialPort, Settings};
use pamoja_serial::cobs::{self, CobsDecoder};

/// The primary UART, on GPIO14 and GPIO15 on every model but the Raspberry Pi 5.
const PORT: &str = "/dev/serial0";

fn main() -> Result<(), Box<dyn Error>> {
    let port = SerialPort::open(PORT, Settings::new(115_200))?;
    let mut decoder: CobsDecoder<64> = CobsDecoder::new();
    let mut buffer = [0u8; 64];
    let mut frame = [0u8; cobs::max_encoded_len(32)];

    for sequence in 1u16..=5 {
        let mut payload = sequence.to_be_bytes().to_vec();
        payload.extend_from_slice(b"ping");
        let framed = cobs::encode(&payload, &mut frame)?;
        let started = Instant::now();
        port.write(&frame[..framed])?;

        // The frame comes back on RX as it goes out on TX. Read until the decoder has it, or
        // until the line has been quiet for 100 ms.
        let mut echoed = None;
        while echoed.is_none() {
            let got = port.read(&mut buffer, Duration::from_millis(100))?;
            if got == 0 {
                break;
            }
            for &byte in &buffer[..got] {
                if let Ok(Some(back)) = decoder.push(byte) {
                    echoed = Some(back.to_vec());
                }
            }
        }
        match echoed {
            Some(back) if back == payload => println!(
                "frame {sequence}  {framed} bytes back in {:.2} ms",
                started.elapsed().as_secs_f64() * 1e3
            ),
            Some(_) => {
                println!("frame {sequence}  came back changed: check the speed and the wiring")
            }
            None => println!(
                "frame {sequence}  nothing came back: check the jumper, and that the console is off"
            ),
        }
        thread::sleep(Duration::from_millis(500));
    }
    Ok(())
}

cd examples/boards/raspberry-pi
cargo run --release --bin uart

import { setTimeout as sleep } from 'node:timers/promises'
import { SerialPort } from '@pamoja/hal'
import { CobsDecoder, cobs } from '@pamoja/serial'

// The primary UART, on GPIO14 and GPIO15 on every model but the Raspberry Pi 5.
const PORT = '/dev/serial0'

async function main(): Promise<void> {
  const port = SerialPort.open(PORT, { baud: 115200 })
  const decoder = new CobsDecoder()

  for (let sequence = 1; sequence <= 5; sequence += 1) {
    const number = Buffer.alloc(2)
    number.writeUInt16BE(sequence)
    const payload = Buffer.concat([number, Buffer.from('ping')])
    const frame = cobs.encode(payload)
    const started = performance.now()
    await port.write(frame)

    // The frame comes back on RX as it goes out on TX. Read until the decoder has it, or
    // until the line has been quiet for 100 ms.
    let echoed: Buffer | undefined
    while (echoed === undefined) {
      const arrived = await port.read(64, 100)
      if (arrived.length === 0) {
        break
      }
      echoed = decoder.feed(arrived)[0]
    }
    if (echoed === undefined) {
      console.log(`frame ${sequence}  nothing came back: check the jumper, and that the console is off`)
    } else if (echoed.equals(payload)) {
      const millis = performance.now() - started
      console.log(`frame ${sequence}  ${frame.length} bytes back in ${millis.toFixed(2)} ms`)
    } else {
      console.log(`frame ${sequence}  came back changed: check the speed and the wiring`)
    }
    await sleep(500)
  }
}

main().catch((error: Error) => {
  console.error(error.message)
  process.exitCode = 1
})

npm --prefix bindings/node run boards
node bindings/node/build/boards/raspberry-pi/uart.js

import time

from pamoja.hal import SerialPort, SerialSettings
from pamoja.serial import CobsDecoder, cobs

# The primary UART, on GPIO14 and GPIO15 on every model but the Raspberry Pi 5.
PORT = "/dev/serial0"


def main() -> None:
    port = SerialPort.open(PORT, SerialSettings(115_200))
    decoder = CobsDecoder()

    for sequence in range(1, 6):
        payload = sequence.to_bytes(2, "big") + b"ping"
        frame = cobs.encode(payload)
        started = time.perf_counter()
        port.write(frame)

        # The frame comes back on RX as it goes out on TX. Read until the decoder has it, or
        # until the line has been quiet for 100 ms.
        echoed = None
        while echoed is None:
            arrived = port.read(64, timeout=0.1)
            if not arrived:
                break
            payloads = decoder.feed(arrived)
            if payloads:
                echoed = payloads[0]
        if echoed is None:
            print(f"frame {sequence}  nothing came back: check the jumper, and that the console is off")
        elif echoed == payload:
            millis = (time.perf_counter() - started) * 1_000
            print(f"frame {sequence}  {len(frame)} bytes back in {millis:.2f} ms")
        else:
            print(f"frame {sequence}  came back changed: check the speed and the wiring")
        time.sleep(0.5)


if __name__ == "__main__":
    main()

python bindings/python/boards/raspberry_pi/uart.py

using System.Buffers.Binary;
using System.Diagnostics;
using System.Globalization;

using Pamoja.Hal;
using Pamoja.Serial;

namespace Boards.RaspberryPi;

/// <summary>
/// A UART self-test: the Pi's own serial port with TX jumpered to RX, sending COBS frames and
/// reading each one straight back. Turn the serial port hardware on and the serial console off
/// (raspi-config, Interface Options, Serial Port), reboot, and put one jumper between GPIO14 and
/// GPIO15.
/// </summary>
public static class Uart
{
    // The primary UART, on GPIO14 and GPIO15 on every model but the Raspberry Pi 5.
    private const string Port = "/dev/serial0";

    /// <summary>Sends five frames and reports each one's return.</summary>
    public static void Run()
    {
        using SerialPort port = SerialPort.Open(Port, new SerialSettings(115_200));
        using var decoder = new CobsDecoder();

        for (ushort sequence = 1; sequence <= 5; sequence++)
        {
            byte[] payload = new byte[2 + "ping"u8.Length];
            BinaryPrimitives.WriteUInt16BigEndian(payload, sequence);
            "ping"u8.CopyTo(payload.AsSpan(2));
            byte[] frame = Serial.CobsEncode(payload);
            var started = Stopwatch.StartNew();
            port.Write(frame);

            // The frame comes back on RX as it goes out on TX. Read until the decoder has it, or
            // until the line has been quiet for 100 ms.
            byte[]? echoed = null;
            while (echoed is null)
            {
                byte[] arrived = port.Read(64, TimeSpan.FromMilliseconds(100));
                if (arrived.Length == 0)
                {
                    break;
                }

                echoed = decoder.Feed(arrived).FirstOrDefault();
            }

            if (echoed is null)
            {
                Console.WriteLine($"frame {sequence}  nothing came back: check the jumper, and that the console is off");
            }
            else if (echoed.SequenceEqual(payload))
            {
                Console.WriteLine(string.Create(
                    CultureInfo.InvariantCulture,
                    $"frame {sequence}  {frame.Length} bytes back in {started.Elapsed.TotalMilliseconds:F2} ms"));
            }
            else
            {
                Console.WriteLine($"frame {sequence}  came back changed: check the speed and the wiring");
            }

            Thread.Sleep(TimeSpan.FromMilliseconds(500));
        }
    }
}

dotnet run --project bindings/dotnet/samples/Pamoja.Boards -- raspberry-pi/uart