📚 课程目录第3章 J1939地址声明与动态地址分配

3. J1939地址声明与动态地址分配

任务目标

实现一个最小可运行的 J1939 地址声明状态机：上电后发送 Address Claimed（PGN 60928，优先级 6，目标地址 255），监听总线上的 Address Claimed 报文；若发现同地址且对方 NAME 优先级更高，则从候选地址池中动态选取新地址并重新声明；若地址池耗尽，则发送 Cannot Claim Address（PGN 60928，SA=254）。

环境与依赖

Python 3.8+，仅用标准库（socket、struct、time）。
Linux 下使用 socket.AF_CAN + J1939 协议族；Windows/macOS 无原生 J1939 套接字，可用 python-can 的 can.interface.Bus 替代发送/接收层，本案例核心逻辑与传输层解耦。
需要 can0 接口已 up（sudo ip link set can0 up type can bitrate 250000）。

关键设计

J1939 地址声明依赖 NAME（64 位）比较：数值越小优先级越高。冲突时 NAME 大的一方必须让出地址。状态机包含 IDLE → CLAIMING → CLAIMED，冲突时回到 CLAIMING 并换地址。PGN 60928 数据域为 8 字节 NAME，SA 字段即声明的源地址。

完整实现

import socket, struct, time

PGN_ADDR_CLAIM = 60928
PRIORITY = 6
BROADCAST = 0xFF
CANNOT_CLAIM_SA = 0xFE

def make_name(identity, manufacturer, function, instance):
    # 64-bit NAME: 21b identity, 11b manufacturer, 8b function, 8b instance, 16b reserved
    return ((identity & 0x1FFFFF) << 43) | ((manufacturer & 0x7FF) << 32) \
           | ((function & 0xFF) << 24) | ((instance & 0xFF) << 16)

def pgn_to_canid(pgn, priority, sa):
    dp, pf, ps = (pgn >> 16) & 1, (pgn >> 8) & 0xFF, pgn & 0xFF
    if pf < 240:  # PDU1，目标地址在 PS
        return (priority << 26) | (dp << 24) | (pf << 16) | (sa << 8) | ps
    return (priority << 26) | (dp << 24) | (pf << 16) | (ps << 8) | sa

def canid_to_pgn_sa(canid):
    dp = (canid >> 24) & 1
    pf = (canid >> 16) & 0xFF
    ps = (canid >> 8) & 0xFF
    sa = canid & 0xFF
    pgn = (dp << 16) | (pf << 8) | (ps if pf < 240 else 0)
    return pgn, sa

class AddressClaimer:
    def __init__(self, bus, name, candidates):
        self.bus = bus
        self.name = name
        self.candidates = list(candidates)
        self.sa = None
        self.state = "IDLE"

    def _send_claim(self, sa):
        data = struct.pack("<Q", self.name)
        canid = pgn_to_canid(PGN_ADDR_CLAIM, PRIORITY, sa)
        self.bus.send(canid, data)

    def _send_cannot_claim(self):
        data = struct.pack("<Q", self.name)
        canid = pgn_to_canid(PGN_ADDR_CLAIM, PRIORITY, CANNOT_CLAIM_SA)
        self.bus.send(canid, data)

    def start(self):
        self.state = "CLAIMING"
        while self.candidates:
            self.sa = self.candidates.pop(0)
            self._send_claim(self.sa)
            time.sleep(0.25)  # 等待 250ms 冲突窗口
            if not self._conflict_detected():
                self.state = "CLAIMED"
                return self.sa
        self.state = "FAILED"
        self._send_cannot_claim()
        return None

    def _conflict_detected(self):
        deadline = time.time() + 0.25
        while time.time() < deadline:
            msg = self.bus.recv(timeout=0.05)
            if msg is None:
                continue
            canid, data = msg
            pgn, sa = canid_to_pgn_sa(canid)
            if pgn != PGN_ADDR_CLAIM or sa != self.sa or len(data) < 8:
                continue
            other = struct.unpack("<Q", data[:8])[0]
            if other < self.name:  # 对方 NAME 更小，优先级更高
                return True
        return False

# ---- 最小可运行入口（Linux SocketCAN）----
class SocketCanBus:
    def __init__(self, iface="can0"):
        self.sock = socket.socket(socket.AF_CAN, socket.SOCK_DGRAM, socket.CAN_J1939)
        self.sock.bind((iface,))
    def send(self, canid, data):
        self.sock.send(bytes([canid & 0xFF, (canid >> 8) & 0xFF,
                              (canid >> 16) & 0xFF, (canid >> 24) & 0xFF]) + data)
    def recv(self, timeout=0.1):
        self.sock.settimeout(timeout)
        try:
            frame = self.sock.recv(16)
        except socket.timeout:
            return None
        canid = frame[0] | (frame[1] << 8) | (frame[2] << 16) | (frame[3] << 24)
        return canid, frame[4:]

if __name__ == "__main__":
    bus = SocketCanBus("can0")
    name = make_name(identity=0x12345, manufacturer=0x0AB, function=0x21, instance=0x01)
    claimer = AddressClaimer(bus, name, candidates=[0x80, 0x81, 0x82])
    sa = claimer.start()
    print("state=%s sa=%s" % (claimer.state, sa))

运行与验证

启动 can0，运行脚本：python3 addr_claim.py。
另开终端用 candump can0 观察：应看到 SA=0x80 的 PGN 60928 报文；若总线上有更低 NAME 的同地址节点，脚本会依次尝试 0x81、0x82。
通过条件：脚本打印 state=CLAIMED sa=<候选地址>，且 candump 中该地址的 Address Claimed 报文在 250ms 冲突窗口后无更高优先级节点覆盖。
地址耗尽场景：把 candidates 改为已被占用的地址，脚本应打印 state=FAILED sa=None，并发出 SA=254 的 Cannot Claim 报文。

常见问题

socket.CAN_J1939 在部分内核/发行版不可用，可改用 python-can 的 Bus("socketcan", channel="can0")，只需替换 SocketCanBus 的 send/recv 实现。
NAME 比较必须按无符号 64 位进行，struct.unpack("<Q", ...) 已保证。
冲突窗口 250ms 是 J1939 推荐值，实际总线负载高时可适当放宽。