CAN and J1939#

use std::time::Duration;

use pamoja_can::bus::{CanBus, Filter};
use pamoja_can::{priority, CanError, CanId, Frame, J1939Id, Signals, NOT_AVAILABLE};

// The nodes by the address each answers to, and the two parameter groups in play.
const ENGINE: u8 = 0;
const GATEWAY: u8 = 1;
const ENGINE_CONTROLLER_1: u32 = 61_444; // carries engine speed
const REQUEST: u32 = 59_904; // asks another node for a parameter group

// Where engine speed sits inside that group, and the scale the standard fixes for it.
const ENGINE_SPEED_AT: usize = 3;
const RPM_PER_BIT: f64 = 0.125;

// J1939 keeps its addressing inside the 29-bit identifier: a priority, the parameter group,
// and the sender's address. A broadcast names no destination.
let speed_id = J1939Id::broadcast(priority::CONTROL, ENGINE_CONTROLLER_1, ENGINE);
println!(
    "engine speed 0x{:08X}: pgn {} at priority {}, from node {ENGINE} to every node",
    speed_id.to_id().raw(),
    speed_id.pgn(),
    speed_id.priority()
);

// A reading starts with every signal marked not available, and the engine writes only its
// speed.
let reading = |rpm: f64| -> Result<Frame, CanError> {
    let mut signals = Signals::new();
    signals.set_u16(ENGINE_SPEED_AT, (rpm / RPM_PER_BIT) as u16);
    Frame::new(speed_id.to_id(), signals.as_bytes())
};
let rpm_of = |frame: &Frame| {
    let raw = frame
        .signals()
        .and_then(|signals| signals.u16(ENGINE_SPEED_AT));
    raw.map(|raw| f64::from(raw) * RPM_PER_BIT)
};
let first = reading(1500.0)?;
let unreported = first
    .data()
    .iter()
    .filter(|&&byte| byte == NOT_AVAILABLE)
    .count();
println!(
    "payload      {:.1} rpm in bytes {} and {}, the other {unreported} not available",
    rpm_of(&first).unwrap_or_default(),
    ENGINE_SPEED_AT + 1,
    ENGINE_SPEED_AT + 2
);

// Four nodes on one bus with nothing plugged in. On a Linux board each is
// CanBus::open("can0"), and nothing after this statement changes.
let engine = CanBus::simulated();
let gateway = engine.join()?;
let laptop = engine.join()?;
let sensor = engine.join()?;

// The gateway keeps engine speed and nothing else; the laptop keeps everything.
gateway.set_filters(&[Filter::pgn(ENGINE_CONTROLLER_1)])?;

// Two engine readings, and between them the coolant sensor, which speaks plain CAN: its
// level in percent on the 11-bit identifier 0x120.
engine.send(&first)?;
sensor.send(&Frame::new(CanId::standard(0x120), &[87])?)?;
engine.send(&reading(1512.5)?)?;

// Every node hears every frame but its own, and keeps what its filters pass.
while let Some(frame) = gateway.receive(Duration::from_millis(10))? {
    let from = J1939Id::from_id(frame.id()).map(|id| id.source());
    println!(
        "gateway      {:.1} rpm from node {}",
        rpm_of(&frame).unwrap_or_default(),
        from.unwrap_or_default()
    );
}
let on_the_bus = engine.sent() + sensor.sent();
println!(
    "gateway      kept {} of the {on_the_bus} frames on the bus",
    gateway.received()
);
let mut heard = Vec::new();
while let Some(frame) = laptop.receive(Duration::from_millis(10))? {
    heard.push(frame);
}
if let Some(plain) = heard
    .iter()
    .find(|frame| J1939Id::from_id(frame.id()).is_none())
{
    println!(
        "laptop       heard {}, among them 0x{:03X}, an 11-bit identifier and no J1939 message",
        heard.len(),
        plain.id().raw()
    );
}

// A request is addressed rather than broadcast: below the PDU1 limit, eight bits of the
// identifier name the node it is for.
let request_id = J1939Id::from_parts(priority::DEFAULT, REQUEST, GATEWAY, ENGINE);
println!(
    "request      pgn {} from node {} to node {}",
    request_id.pgn(),
    request_id.source(),
    request_id.destination().unwrap_or_default()
);

// The engine goes quiet. A receive waits for a frame up to its timeout; on a simulated bus
// it returns at once and counts the wait instead of sleeping through it.
let before = gateway.waited_micros();
let quiet = gateway.receive(Duration::from_millis(500))?;
println!(
    "silent       {} frames in {} ms, counted and not slept",
    usize::from(quiet.is_some()),
    (gateway.waited_micros() - before) / 1_000
);

// Above eight bytes CAN FD encodes a length in steps, and a classic frame refuses a ninth
// byte.
let wide = Frame::fd(speed_id.to_id(), &[0; 32])?;
println!(
    "fd           32 bytes travel at data length code {}",
    wide.dlc()
);
if let Err(error) = Frame::new(speed_id.to_id(), &[0; 9]) {
    println!("classic      refused nine bytes: {error}");
}

import {
  CanBus,
  type CanFrame,
  NOT_AVAILABLE,
  broadcastJ1939,
  composeJ1939,
  decodeJ1939,
  fdFrame,
  filterPgn,
  frame,
  priority,
  signals,
  signalsFrom,
} from '@pamoja/can'

// The nodes by the address each answers to, and the two parameter groups in play.
const ENGINE = 0
const GATEWAY = 1
const ENGINE_CONTROLLER_1 = 61_444 // carries engine speed
const REQUEST = 59_904 // asks another node for a parameter group

// Where engine speed sits inside that group, and the scale the standard fixes for it.
const ENGINE_SPEED_AT = 3
const RPM_PER_BIT = 0.125

// A reading starts with every signal marked not available, and the engine writes only its
// speed.
function reading(speedId: number, rpm: number): CanFrame {
  const reported = signals()
  reported.setU16(ENGINE_SPEED_AT, rpm / RPM_PER_BIT)
  return frame(speedId, reported.bytes, true)
}

function rpmOf(received: CanFrame): number {
  return (signalsFrom(received.data).u16(ENGINE_SPEED_AT) ?? 0) * RPM_PER_BIT
}

function hex(value: number, digits: number): string {
  return `0x${value.toString(16).toUpperCase().padStart(digits, '0')}`
}

async function main() {
  // J1939 keeps its addressing inside the 29-bit identifier: a priority, the parameter group,
  // and the sender's address. A broadcast names no destination.
  const speedId = broadcastJ1939(priority.control, ENGINE_CONTROLLER_1, ENGINE)
  const speed = decodeJ1939(speedId)!
  console.log(
    `engine speed ${hex(speedId, 8)}: pgn ${speed.pgn} at priority ${speed.priority}, ` +
      `from node ${ENGINE} to every node`,
  )

  const first = reading(speedId, 1500)
  const unreported = [...first.data].filter((byte) => byte === NOT_AVAILABLE).length
  console.log(
    `payload      ${rpmOf(first).toFixed(1)} rpm in bytes ${ENGINE_SPEED_AT + 1} and ` +
      `${ENGINE_SPEED_AT + 2}, the other ${unreported} not available`,
  )

  // Four nodes on one bus with nothing plugged in. On a Linux board each is
  // CanBus.open('can0'), and nothing after this statement changes.
  const engine = CanBus.simulated()
  const gateway = engine.join()
  const laptop = engine.join()
  const sensor = engine.join()

  // The gateway keeps engine speed and nothing else; the laptop keeps everything.
  gateway.setFilters([filterPgn(ENGINE_CONTROLLER_1)])

  // Two engine readings, and between them the coolant sensor, which speaks plain CAN: its
  // level in percent on the 11-bit identifier 0x120.
  await engine.send(first)
  await sensor.send(frame(0x120, Buffer.from([87])))
  await engine.send(reading(speedId, 1512.5))

  // Every node hears every frame but its own, and keeps what its filters pass.
  let kept: CanFrame | null
  while ((kept = await gateway.receive(10)) !== null) {
    const from = decodeJ1939(kept.id, kept.extended)?.source ?? 0
    console.log(`gateway      ${rpmOf(kept).toFixed(1)} rpm from node ${from}`)
  }
  const onTheBus = engine.sent + sensor.sent
  console.log(`gateway      kept ${gateway.received} of the ${onTheBus} frames on the bus`)
  const heard: CanFrame[] = []
  let next: CanFrame | null
  while ((next = await laptop.receive(10)) !== null) {
    heard.push(next)
  }
  const plain = heard.find((received) => decodeJ1939(received.id, received.extended) === null)
  if (plain !== undefined) {
    console.log(
      `laptop       heard ${heard.length}, among them ${hex(plain.id, 3)}, ` +
        'an 11-bit identifier and no J1939 message',
    )
  }

  // A request is addressed rather than broadcast: below the PDU1 limit, eight bits of the
  // identifier name the node it is for.
  const request = decodeJ1939(composeJ1939(priority.default, REQUEST, GATEWAY, ENGINE))!
  console.log(
    `request      pgn ${request.pgn} from node ${request.source} to node ${request.destination}`,
  )

  // The engine goes quiet. A receive waits for a frame up to its timeout; on a simulated bus
  // it resolves at once and counts the wait instead of sleeping through it.
  const before = gateway.waitedMicros
  const quiet = await gateway.receive(500)
  const waited = Math.floor((gateway.waitedMicros - before) / 1000)
  console.log(
    `silent       ${quiet === null ? 0 : 1} frames in ${waited} ms, counted and not slept`,
  )

  // Above eight bytes CAN FD encodes a length in steps, and a classic frame refuses a ninth
  // byte.
  const wide = fdFrame(speedId, new Uint8Array(32), true)
  console.log(`fd           32 bytes travel at data length code ${wide.dlc}`)
  try {
    frame(speedId, new Uint8Array(9), true)
  } catch (error) {
    console.log(`classic      refused nine bytes: ${(error as Error).message}`)
  }

  return { speedId, unreported, gateway, heard, request, quiet, wide }
}

main()

from pamoja.can import (
    NOT_AVAILABLE,
    CanBus,
    CanFilter,
    CanFrame,
    Priority,
    broadcast_j1939,
    compose_j1939,
    decode_j1939,
    fd_frame,
    frame,
    signals,
    signals_from,
)
from pamoja.core import PamojaError

# The nodes by the address each answers to, and the two parameter groups in play.
ENGINE = 0
GATEWAY = 1
ENGINE_CONTROLLER_1 = 61_444  # carries engine speed
REQUEST = 59_904  # asks another node for a parameter group

# Where engine speed sits inside that group, and the scale the standard fixes for it.
ENGINE_SPEED_AT = 3
RPM_PER_BIT = 0.125


def reading(speed_id: int, rpm: float) -> CanFrame:
    """A reading: every signal marked not available but the engine's speed."""
    reported = signals()
    reported.set_u16(ENGINE_SPEED_AT, int(rpm / RPM_PER_BIT))
    return frame(speed_id, reported.bytes, extended=True)


def rpm_of(received: CanFrame) -> float:
    """The engine speed a reading carries."""
    return (signals_from(received.data).u16(ENGINE_SPEED_AT) or 0) * RPM_PER_BIT


# J1939 keeps its addressing inside the 29-bit identifier: a priority, the parameter group, and
# the sender's address. A broadcast names no destination.
speed_id = broadcast_j1939(Priority.CONTROL, ENGINE_CONTROLLER_1, ENGINE)
speed = decode_j1939(speed_id)
print(
    f"engine speed 0x{speed_id:08X}: pgn {speed.pgn} at priority {speed.priority}, "
    f"from node {ENGINE} to every node"
)

first = reading(speed_id, 1500)
unreported = sum(1 for byte in first.data if byte == NOT_AVAILABLE)
print(
    f"payload      {rpm_of(first):.1f} rpm in bytes {ENGINE_SPEED_AT + 1} and "
    f"{ENGINE_SPEED_AT + 2}, the other {unreported} not available"
)

# Four nodes on one bus with nothing plugged in. On a Linux board each is
# CanBus.open("can0"), and nothing after this statement changes.
engine = CanBus.simulated()
gateway = engine.join()
laptop = engine.join()
sensor = engine.join()

# The gateway keeps engine speed and nothing else; the laptop keeps everything.
gateway.set_filters([CanFilter.pgn(ENGINE_CONTROLLER_1)])

# Two engine readings, and between them the coolant sensor, which speaks plain CAN: its level
# in percent on the 11-bit identifier 0x120.
engine.send(first)
sensor.send(frame(0x120, bytes([87])))
engine.send(reading(speed_id, 1512.5))

# Every node hears every frame but its own, and keeps what its filters pass.
while (kept := gateway.receive(timeout=0.01)) is not None:
    source = decode_j1939(kept.id, kept.extended).source
    print(f"gateway      {rpm_of(kept):.1f} rpm from node {source}")
on_the_bus = engine.sent + sensor.sent
print(f"gateway      kept {gateway.received} of the {on_the_bus} frames on the bus")
heard = []
while (received := laptop.receive(timeout=0.01)) is not None:
    heard.append(received)
plain = next((f for f in heard if decode_j1939(f.id, f.extended) is None), None)
if plain is not None:
    print(
        f"laptop       heard {len(heard)}, among them 0x{plain.id:03X}, "
        "an 11-bit identifier and no J1939 message"
    )

# A request is addressed rather than broadcast: below the PDU1 limit, eight bits of the
# identifier name the node it is for.
request = decode_j1939(compose_j1939(Priority.DEFAULT, REQUEST, GATEWAY, ENGINE))
print(f"request      pgn {request.pgn} from node {request.source} to node {request.destination}")

# The engine goes quiet. A receive waits for a frame up to its timeout; on a simulated bus it
# returns at once and counts the wait instead of sleeping through it.
before = gateway.waited_micros
quiet = gateway.receive(timeout=0.5)
waited = (gateway.waited_micros - before) // 1_000
print(f"silent       {0 if quiet is None else 1} frames in {waited} ms, counted and not slept")

# Above eight bytes CAN FD encodes a length in steps, and a classic frame refuses a ninth byte.
wide = fd_frame(speed_id, bytes(32), extended=True)
print(f"fd           32 bytes travel at data length code {wide.dlc}")
try:
    frame(speed_id, bytes(9), extended=True)
except PamojaError as error:
    print(f"classic      refused nine bytes: {error}")

// The nodes by the address each answers to, and the two parameter groups in play.
const byte Engine = 0;
const byte Gateway = 1;
const uint EngineController1 = 61_444; // carries engine speed
const uint Request = 59_904; // asks another node for a parameter group

// Where engine speed sits inside that group, and the scale the standard fixes for it.
const int EngineSpeedAt = 3;
const double RpmPerBit = 0.125;

// J1939 keeps its addressing inside the 29-bit identifier: a priority, the parameter
// group, and the sender's address. A broadcast names no destination.
uint speedId = Can.BroadcastJ1939(J1939Priority.Control, EngineController1, Engine);
J1939Message speed = Can.DecodeJ1939(speedId)!;
Console.WriteLine(
    $"engine speed 0x{speedId:X8}: pgn {speed.Pgn} at priority {speed.Priority}, from node {Engine} to every node");

// A reading starts with every signal marked not available, and the engine writes only
// its speed.
CanFrame Reading(double rpm)
{
    Signals reported = Signals.New();
    reported.SetU16(EngineSpeedAt, (ushort)(rpm / RpmPerBit));
    return Can.Frame(speedId, reported.ToArray(), extended: true);
}

static double RpmOf(CanFrame received) =>
    (Signals.From(received.Data).U16(EngineSpeedAt) ?? 0) * RpmPerBit;

CanFrame first = Reading(1500);
int unreported = first.Data.Count(value => value == Signals.NotAvailable);
Console.WriteLine(Invariant(
    $"payload      {RpmOf(first):F1} rpm in bytes {EngineSpeedAt + 1} and {EngineSpeedAt + 2}, the other {unreported} not available"));

// Four nodes on one bus with nothing plugged in. On a Linux board each is
// CanBus.Open("can0"), and nothing after this statement changes.
using CanBus engine = CanBus.Simulated();
using CanBus gateway = engine.Join();
using CanBus laptop = engine.Join();
using CanBus sensor = engine.Join();

// The gateway keeps engine speed and nothing else; the laptop keeps everything.
gateway.SetFilters(CanFilter.Pgn(EngineController1));

// Two engine readings, and between them the coolant sensor, which speaks plain CAN: its
// level in percent on the 11-bit identifier 0x120.
engine.Send(first);
sensor.Send(Can.Frame(0x120, [87]));
engine.Send(Reading(1512.5));

// Every node hears every frame but its own, and keeps what its filters pass.
while (gateway.Receive(TimeSpan.FromMilliseconds(10)) is { } kept)
{
    byte from = Can.DecodeJ1939(kept.Id, kept.Extended)?.Source ?? 0;
    Console.WriteLine(Invariant($"gateway      {RpmOf(kept):F1} rpm from node {from}"));
}

long onTheBus = engine.Sent + sensor.Sent;
Console.WriteLine($"gateway      kept {gateway.Received} of the {onTheBus} frames on the bus");
var heard = new List<CanFrame>();
while (laptop.Receive(TimeSpan.FromMilliseconds(10)) is { } received)
{
    heard.Add(received);
}

if (heard.Find(received => Can.DecodeJ1939(received.Id, received.Extended) is null) is { } plain)
{
    Console.WriteLine(
        $"laptop       heard {heard.Count}, among them 0x{plain.Id:X3}, an 11-bit identifier and no J1939 message");
}

// A request is addressed rather than broadcast: below the PDU1 limit, eight bits of the
// identifier name the node it is for.
J1939Message request = Can.DecodeJ1939(
    Can.ComposeJ1939((byte)J1939Priority.Normal, Request, Gateway, Engine))!;
Console.WriteLine($"request      pgn {request.Pgn} from node {request.Source} to node {request.Destination}");

// The engine goes quiet. A receive waits for a frame up to its timeout; on a simulated
// bus it returns at once and counts the wait instead of sleeping through it.
ulong before = gateway.WaitedMicros;
CanFrame? quiet = gateway.Receive(TimeSpan.FromMilliseconds(500));
Console.WriteLine(
    $"silent       {(quiet is null ? 0 : 1)} frames in {(gateway.WaitedMicros - before) / 1_000} ms, counted and not slept");

// Above eight bytes CAN FD encodes a length in steps, and a classic frame refuses a
// ninth byte.
CanFrame wide = Can.FdFrame(speedId, new byte[32], extended: true);
Console.WriteLine($"fd           32 bytes travel at data length code {wide.Dlc}");
try
{
    Can.Frame(speedId, new byte[9], extended: true);
}
catch (PamojaException error)
{
    Console.WriteLine($"classic      refused nine bytes: {error.Message}");
}

use pamoja_can::bus::CanBus;
use pamoja_can::J1939Id;

/// The interface the MCP2515 overlay makes.
const INTERFACE: &str = "can0";

/// Engine speed's parameter group, where the speed sits in it, and its scale.
const ENGINE_CONTROLLER_1: u32 = 61_444;
const ENGINE_SPEED_AT: usize = 3;
const RPM_PER_BIT: f64 = 0.125;

fn main() -> Result<(), Box<dyn Error>> {
    // The monitor only listens and sends nothing, so it is safe on a running machine's bus.
    let bus = CanBus::open(INTERFACE)?;
    let started = Instant::now();
    while started.elapsed() < Duration::from_secs(10) {
        let Some(frame) = bus.receive(Duration::from_secs(1))? else {
            println!("quiet for a second: check the bit rate, the wiring, and the termination");
            continue;
        };
        let at = started.elapsed().as_secs_f64();
        match J1939Id::from_id(frame.id()) {
            Some(id) if id.pgn() == ENGINE_CONTROLLER_1 => {
                let raw = frame
                    .signals()
                    .and_then(|signals| signals.u16(ENGINE_SPEED_AT));
                let rpm = raw.map_or(0.0, |raw| f64::from(raw) * RPM_PER_BIT);
                println!(
                    "{at:7.3} s  pgn {} from {}: {rpm:.1} rpm",
                    id.pgn(),
                    id.source()
                );
            }
            Some(id) => println!(
                "{at:7.3} s  pgn {} from {}, {} bytes",
                id.pgn(),
                id.source(),
                frame.len()
            ),
            None => println!(
                "{at:7.3} s  0x{:03X}, {} bytes, an 11-bit identifier",
                frame.id().raw(),
                frame.len()
            ),
        }
    }
    println!("{} frames in ten seconds on {INTERFACE}", bus.received());
    Ok(())
}

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

import { CanBus, decodeJ1939, signalsFrom } from '@pamoja/can'

// The interface the MCP2515 overlay makes.
const INTERFACE = 'can0'

// Engine speed's parameter group, where the speed sits in it, and its scale.
const ENGINE_CONTROLLER_1 = 61_444
const ENGINE_SPEED_AT = 3
const RPM_PER_BIT = 0.125

async function main(): Promise<void> {
  // The monitor only listens and sends nothing, so it is safe on a running machine's bus.
  const bus = CanBus.open(INTERFACE)
  const started = performance.now()
  while (performance.now() - started < 10_000) {
    const frame = await bus.receive(1000)
    if (frame === null) {
      console.log('quiet for a second: check the bit rate, the wiring, and the termination')
      continue
    }
    const at = ((performance.now() - started) / 1000).toFixed(3).padStart(7)
    const id = decodeJ1939(frame.id, frame.extended)
    if (id !== null && id.pgn === ENGINE_CONTROLLER_1) {
      const rpm = (signalsFrom(frame.data).u16(ENGINE_SPEED_AT) ?? 0) * RPM_PER_BIT
      console.log(`${at} s  pgn ${id.pgn} from ${id.source}: ${rpm.toFixed(1)} rpm`)
    } else if (id !== null) {
      console.log(`${at} s  pgn ${id.pgn} from ${id.source}, ${frame.len} bytes`)
    } else {
      const hex = frame.id.toString(16).toUpperCase().padStart(3, '0')
      console.log(`${at} s  0x${hex}, ${frame.len} bytes, an 11-bit identifier`)
    }
  }
  console.log(`${bus.received} frames in ten seconds on ${INTERFACE}`)
}

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

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

import time

from pamoja.can import CanBus, decode_j1939, signals_from

# The interface the MCP2515 overlay makes.
INTERFACE = "can0"

# Engine speed's parameter group, where the speed sits in it, and its scale.
ENGINE_CONTROLLER_1 = 61_444
ENGINE_SPEED_AT = 3
RPM_PER_BIT = 0.125


def main() -> None:
    # The monitor only listens and sends nothing, so it is safe on a running machine's bus.
    bus = CanBus.open(INTERFACE)
    started = time.perf_counter()
    while time.perf_counter() - started < 10:
        frame = bus.receive(timeout=1)
        if frame is None:
            print("quiet for a second: check the bit rate, the wiring, and the termination")
            continue
        at = time.perf_counter() - started
        message = decode_j1939(frame.id, frame.extended)
        if message is not None and message.pgn == ENGINE_CONTROLLER_1:
            rpm = (signals_from(frame.data).u16(ENGINE_SPEED_AT) or 0) * RPM_PER_BIT
            print(f"{at:7.3f} s  pgn {message.pgn} from {message.source}: {rpm:.1f} rpm")
        elif message is not None:
            print(f"{at:7.3f} s  pgn {message.pgn} from {message.source}, {frame.len} bytes")
        else:
            print(f"{at:7.3f} s  0x{frame.id:03X}, {frame.len} bytes, an 11-bit identifier")
    print(f"{bus.received} frames in ten seconds on {INTERFACE}")


if __name__ == "__main__":
    main()

python bindings/python/boards/raspberry_pi/can.py

using System.Diagnostics;
using System.Globalization;

using Pamoja.Can;

namespace Boards.RaspberryPi;

/// <summary>
/// A CAN bus monitor: listens on the Pi's CAN interface, through an MCP2515, and names each J1939
/// message it hears. Load the controller's overlay and bring the interface up at the bus's bit
/// rate first.
/// </summary>
public static class CanMonitor
{
    // The interface the MCP2515 overlay makes.
    private const string Interface = "can0";

    // Engine speed's parameter group, where the speed sits in it, and its scale.
    private const uint EngineController1 = 61_444;
    private const int EngineSpeedAt = 3;
    private const double RpmPerBit = 0.125;

    /// <summary>Listens for ten seconds and names each frame.</summary>
    public static void Run()
    {
        // The monitor only listens and sends nothing, so it is safe on a running machine's bus.
        using CanBus bus = CanBus.Open(Interface);
        var started = Stopwatch.StartNew();
        while (started.Elapsed < TimeSpan.FromSeconds(10))
        {
            CanFrame? frame = bus.Receive(TimeSpan.FromSeconds(1));
            if (frame is null)
            {
                Console.WriteLine("quiet for a second: check the bit rate, the wiring, and the termination");
                continue;
            }

            string at = started.Elapsed.TotalSeconds.ToString("F3", CultureInfo.InvariantCulture).PadLeft(7);
            J1939Message? message = Can.DecodeJ1939(frame.Id, frame.Extended);
            if (message is { Pgn: EngineController1 })
            {
                double rpm = (Signals.From(frame.Data).U16(EngineSpeedAt) ?? 0) * RpmPerBit;
                Console.WriteLine(string.Create(
                    CultureInfo.InvariantCulture,
                    $"{at} s  pgn {message.Pgn} from {message.Source}: {rpm:F1} rpm"));
            }
            else if (message is not null)
            {
                Console.WriteLine($"{at} s  pgn {message.Pgn} from {message.Source}, {frame.Length} bytes");
            }
            else
            {
                Console.WriteLine($"{at} s  0x{frame.Id:X3}, {frame.Length} bytes, an 11-bit identifier");
            }
        }

        Console.WriteLine($"{bus.Received} frames in ten seconds on {Interface}");
    }
}

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