Raw CAN rámce z Raspberry

Princip ukázky

Ukázka předvádí protokol Raw CAN Frames: hodnoty v pevných bajtech CAN rámce bez jakékoli vyšší vrstvy, tak jak je posílá jednoduchá elektronika nebo jak vypadají při monitorování sběrnice.

Protistranou je Python skript Qenex_RawCan_RandomWalkGenerator.py ze složky Examples vedle QInsight.exe. Běží na Linuxu se SocketCAN, typicky na Raspberry Pi s rozšiřující deskou CAN (CAN HAT), a:

• vysílá pět náhodných procházek rnd1 až rnd5 na identifikátorech 0x100 až 0x104, 50 sad za sekundu; každý rámec má 8 bajtů: float32 hodnota v bajtech 0–3 a float32 čas generátoru v bajtech 4–7,
• přijímá rámce 0x180 až 0x184 (nový offset procházky 1 až 5) a 0x190 až 0x194 (nový krok), float32 v bajtech 0–3. Přesně takový rámec QInsight pošle při zápisu do proměnné se směrem write.

QInsight je na sběrnici přes adaptér PEAK PCAN-USB, 250 kbit/s.

Raspberry + CAN deska ─ 0x100: [float value][float t] ──►  PCAN-USB  ──►  QInsight (graf, watch table)
                     ◄── 0x180: [float offset]        ──                   zápis z watch table

Projekt Qenex_CAN_RawCAN_PythonRandomWalks.qproj je ve složce Examples.

Co ukázka předvádí

• Čtení pěti procházek ze sběrnice v grafu a v signal controlech.
• Dvě proměnné z jednoho rámce: Rnd1 čte bajty 0–3 rámce 0x100, Rnd1Time bajty 4–7 téhož rámce (offset="4").
• Zápis na sběrnici: Rnd1Offset (0x180), Rnd2Offset (0x181) a Rnd1Step (0x190) mají směr write; zápis z watch table odešle rámec a procházka se přesune na nový offset.
• Časové značky z adaptéru: vzorky nesou čas příjmu z PCAN-USB, ne okamžik zpracování na PC.

Co potřebujete

Na straně generátoru postačí i velmi levný hardware. Fotografie ukazuje sestavu, na které byla ukázka ověřena: Raspberry Pi Zero a modul s řadičem MCP2515 a budičem sběrnice TJA1050, propojený s Raspberry přes rozhraní SPI několika vodiči. Modul stojí řádově desítky korun a v Linuxu se chová jako běžné rozhraní SocketCAN.

Modul MCP2515 se v Raspberry Pi OS aktivuje v souboru /boot/firmware/config.txt (u starších systémů /boot/config.txt):

dtparam=spi=on
dtoverlay=mcp2515-can0,oscillator=8000000,interrupt=25

Hodnota oscillator odpovídá krystalu na modulu (nejčastěji 8 MHz, u některých modulů 16 MHz), interrupt číslu GPIO, na který je připojen výstup INT modulu. Po restartu je k dispozici rozhraní can0. Stejně dobře poslouží jakákoli jiná deska CAN pro Raspberry Pi nebo jiný počítač s Linuxem a rozhraním SocketCAN.

• Raspberry Pi s rozhraním SocketCAN can0 a Python 3. Rozhraní aktivujte s rychlostí 250 kbit/s a spusťte generátor:

  sudo ip link set can0 up type can bitrate 250000
  python3 Qenex_RawCan_RandomWalkGenerator.py
  

• PEAK PCAN-USB na PC s QInsight, sběrnice zakončená 120 Ω. V nastavení driveru upravte deviceId na id svého adaptéru (hexadecimálně, viz PEAK CAN), bitrate zůstává 250000.

Projekt

Driver a protokol

Protokol nemá nastavení.

Proměnné

Pracovní plocha

Graf všech pěti procházek, watch table se třemi zapisovatelnými parametry, signal control času generátoru a pět signal controlů hodnot.

Co uvidíte a co vyzkoušet

• Po Startu se rozběhne pět procházek kolem svých offsetů 0, 10, −5, 20 a 5. Rnd1Time roste od spuštění generátoru.
• Zapište Rnd1Offset = 15: procházka 1 se přesune na hodnotu 15 a dále kolísá kolem ní; generátor vypíše rnd1.O = 15.000. Rnd1Step = 5 zvětší její krok, 0.1 ho zmenší.
• Odpojte adaptér PCAN-USB: driver ohlásí ztrátu adaptéru a pokouší se ho otevřít každou sekundu; po opětovném připojení příjem pokračuje a časové značky se znovu synchronizují.
• Do Comm params proměnné zapište canId="zz" a potvrďte tlačítkem Apply: proměnná se v logu ohlásí jako nekomunikovaná s uvedením důvodu, místo aby zůstala bez hlášení.

Skript generátoru

Celý Qenex_RawCan_RandomWalkGenerator.py; rozložení rámců je v docstringu. Pro vlastní zařízení stačí převzít stavbu rámce (struct.pack("<ff", value, t)) a čtení řídicích rámců.

#!/usr/bin/env python3
"""Random-walk CAN generator - the counterpart of QInsight's Raw CAN Frames protocol.

Runs on a Linux machine with a SocketCAN interface, typically a Raspberry Pi with a CAN HAT
(QInsight sits on the other end of the bus with a PEAK PCAN-USB adapter):

    sudo ip link set can0 up type can bitrate 250000
    python3 Qenex_RawCan_RandomWalkGenerator.py              # can0, 50 sets of frames per second
    python3 Qenex_RawCan_RandomWalkGenerator.py --channel can1 --hz 20

Shipped with QInsight in the Examples folder; used by the example
Qenex_CAN_RawCAN_PythonRandomWalks.qproj.

Frames sent (standard 11-bit ids, DLC 8, little-endian):
    id 0x100 .. 0x104   rnd1 .. rnd5   bytes 0-3 float32 value, bytes 4-7 float32 time t [s]
Each signal is a random walk: every sample moves by a random step in +-step and a weak
mean reversion pulls it back towards its offset O, so it never drifts off the chart.

Frames accepted (what QInsight sends for a Raw CAN variable with direction="write"):
    id 0x180 .. 0x184   bytes 0-3 float32   new offset O of rnd1 .. rnd5
    id 0x190 .. 0x194   bytes 0-3 float32   new step size of rnd1 .. rnd5
The walk jumps to the new offset at once, so a write is visible immediately on the graph.
"""
import argparse
import random
import socket
import struct
import threading
import time

# id -> random-walk parameters
#   O          = offset (the value the walk hovers around)
#   step       = max size of one random step; each sample moves by +-step
#   reversion  = mean-reversion strength (0..1); pulls the value back towards O
SIGNALS = {
    0x100: {"O":  0.0, "step": 0.5, "reversion": 0.01},
    0x101: {"O": 10.0, "step": 1.0, "reversion": 0.02},
    0x102: {"O": -5.0, "step": 0.3, "reversion": 0.015},
    0x103: {"O": 20.0, "step": 2.0, "reversion": 0.01},
    0x104: {"O":  5.0, "step": 0.8, "reversion": 0.03},
}
OFFSET_ID_BASE = 0x180   # 0x180 + n sets O of the n-th signal
STEP_ID_BASE = 0x190     # 0x190 + n sets step of the n-th signal

FRAME_FORMAT = "=IB3x8s"  # SocketCAN frame: id (4), dlc (1), padding (3), data (8)
CAN_EFF_FLAG = 0x80000000

lock = threading.Lock()
STATE = {cid: p["O"] for cid, p in SIGNALS.items()}   # current value of every walk


def step_walk(can_id):
    """Advance one random walk by one step and return the new value."""
    with lock:
        p = SIGNALS[can_id]
        x = STATE[can_id]
        x = x + random.uniform(-p["step"], p["step"]) + p["reversion"] * (p["O"] - x)
        STATE[can_id] = x
        return x


def build_frame(can_id, value, t):
    data = struct.pack("<ff", value, t)        # 8 B: float32 value + float32 t
    return struct.pack(FRAME_FORMAT, can_id, len(data), data)


def apply_frame(raw):
    """A received frame: 0x18n / 0x19n with a float32 sets offset / step of signal n."""
    can_id, dlc, data = struct.unpack(FRAME_FORMAT, raw)
    if can_id & CAN_EFF_FLAG or dlc < 4:
        return
    can_id &= 0x7FF
    for base, key in ((OFFSET_ID_BASE, "O"), (STEP_ID_BASE, "step")):
        n = can_id - base
        if 0 <= n < len(SIGNALS):
            signal_id = 0x100 + n
            value = struct.unpack("<f", data[:4])[0]
            with lock:
                SIGNALS[signal_id][key] = value
                if key == "O":
                    STATE[signal_id] = value      # jump there, so the write shows at once
            print(f"rnd{n + 1}.{key} = {value:.3f}", flush=True)
            return


def reader_thread(sock, stop):
    while not stop.is_set():
        try:
            raw = sock.recv(16)
        except OSError:
            break
        if len(raw) == 16:
            apply_frame(raw)


def main():
    parser = argparse.ArgumentParser(description="Raw CAN random-walk generator for QInsight (SocketCAN).")
    parser.add_argument("--channel", default="can0", help="SocketCAN interface (default can0)")
    parser.add_argument("--hz", type=float, default=50.0, help="sets of 5 frames per second (default 50)")
    args = parser.parse_args()

    sock = socket.socket(socket.AF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
    sock.bind((args.channel,))
    stop = threading.Event()
    threading.Thread(target=reader_thread, args=(sock, stop), daemon=True).start()

    t0 = time.monotonic()
    period = 1.0 / args.hz
    next_t = time.monotonic()
    print(f"Raw CAN generator on {args.channel}: ids 0x100-0x104 @ {args.hz:g} Hz, "
          f"writes on 0x180-0x184 (offset) and 0x190-0x194 (step). Ctrl+C to stop.", flush=True)
    try:
        while True:
            t = time.monotonic() - t0
            for can_id in SIGNALS:                 # 5 frames, one signal each
                sock.send(build_frame(can_id, step_walk(can_id), t))
            next_t += period
            time.sleep(max(0.0, next_t - time.monotonic()))
    except KeyboardInterrupt:
        pass
    finally:
        stop.set()
        sock.close()


if __name__ == "__main__":
    main()

Související kapitoly

• Raw CAN Frames v referenci protokolů
• PEAK CAN v referenci driverů
• XCP on CAN na NUCLEO-H723ZG — tentýž adaptér s protokolem XCP