返回教程正文

配套源码

README.md

app/bricks/generic_can/README.md
generic_can Brick:在 VENTUNO Q 上自定义 CAN 报文app/bricks/generic_can/README.md
Markdown124 行
# Generic CAN Brick API

`generic_can` 是协议无关的 CAN 数据帧 Brick。普通业务代码只需导入:

```python
from generic_can import CanBus, CanFrame, MessageDefinition
```

## CanFrame

```python
frame = CanFrame(
    arbitration_id=0x123,
    data=b"\x01\x02",
    is_extended=False,
    is_fd=False,
    bitrate_switch=False,
    timestamp=0.0,
)
```

- `arbitration_id`:标准帧为 `0..0x7FF`,扩展帧为 `0..0x1FFFFFFF`
- `data`:Classical CAN 为 0~8 字节;CAN FD 只允许 `0~8、12、16、20、24、32、48、64` 字节。
- `bitrate_switch=True` 要求 `is_fd=True`
- `timestamp`:接收时由 python-can 提供;用户构造发送帧时通常保持 `0.0`

`CanFrame` 不导入也不依赖 python-can。

CAN FD 的 DATA 长度必须能够直接对应 DLC。如果协议只有 9 个有意义字节,应由用户按照
协议规定显式补齐到 12 字节;Brick 不会替用户选择填充值,也不会依赖 Backend 隐式补齐。

## MessageDefinition

```python
definition = MessageDefinition(
    arbitration_id=0x301,
    direction="rx",
    is_extended=False,
    is_fd=False,
    bitrate_switch=False,
    fixed_data=None,
    encode=None,
    decode=decode_status,
)
```

- `fixed_data`:固定发送字节。
- `encode(**values)`:把业务参数转换成 bytes-compatible DATA。
- `decode(data)`:把接收 DATA 转换为字典、数字或其他工程值。
- 没有 `decode` 时,`receive(name)` 返回完整 `CanFrame`

## CanBus

```python
bus = CanBus(
    interface="can0",
    messages=MESSAGES,
    raw_queue_size=256,
    message_queue_size=64,
    receiver_poll_s=0.05,
)
```

稳定公开方法:

| 方法 | 返回值 | 行为 |
| --- | --- | --- |
| `open()` | `CanBus` | 打开后端并启动唯一接收线程;可重复调用 |
| `close()` | `None` | 停止线程并释放后端;可重复调用 |
| `send_frame(frame)` | `CanFrame` | 发送一帧原始数据 |
| `receive_frame(timeout_s)` | `CanFrame | None` | 从原始队列取帧,超时返回None |
| `send(name, **values)` | `CanFrame` | 固定或动态编码后发送命名报文 |
| `receive(name, timeout_s)` | 解码结果或`CanFrame` | 从该名称的FIFO队列取下一条待消费报文 |
| `describe()` | `dict` | 返回配置、线程、队列和丢帧统计 |

支持上下文管理器:

```python
with CanBus(interface="can0", messages=MESSAGES) as bus:
    frame = bus.receive_frame(timeout_s=1.0)
```

## Backend边界

普通用户不需要直接使用 backend。测试或扩展底层设备时可以导入:

```python
from generic_can.backends import CanBackend, SocketCANBackend
```

V1 的 `SocketCANBackend` 是唯一导入 python-can 的模块,只打开已有接口:

```python
can.Bus(interface="socketcan", channel=device, fd=True)
```

这里的 `fd=True` 让 socket 能处理 CAN FD 消息字段,不会配置 Linux 链路,也不会执行 `ip link`。所有后端错误统一转为 `CANBackendError`

手动把 `SocketCANBackend` 传给 `CanBus` 时,`backend.device` 必须等于
`CanBus.interface`;否则构造阶段立即抛出 `CANConfigurationError`

## App Lab与普通Python

`compat.py` 在 App Lab 中使用真正的 `arduino.app_utils.brick`;普通 Python 没有该模块时,使用不改变类的兼容装饰器。因此同一份核心源码可由 App Lab 使用,也可由 `FakeBackend` 单元测试直接导入。

## 线程模型

一个 `CanBus` 对应一个 backend 和一个接收线程。业务线程不直接调用 `backend.receive()`。接收线程把帧复制到原始队列以及所有匹配的命名队列,队列满时丢最旧帧并记录计数。

`receive(name)` 不是 request-response API。它只读取该命名 FIFO 队列中的下一条待消费帧,
不会自动判断该帧是否由上一条 `send()` 引起。队列里可能存在发送前已经收到的旧帧。
sequence number、transaction ID、counter 或请求/响应功能码等关联规则必须由具体设备协议
层或业务代码实现。

命名匹配键为:

```python
(arbitration_id, is_extended, is_fd)
```

`describe()["messages"][name]["payload_source"]` 对固定报文为 `"fixed_data"`,对动态发送
报文为 `"encode"`,对纯 RX 报文为 `None`

V1 没有 callback、subscriber、request-response correlation 或协议对象。