1. 车载CAN总线报文采集与解析

任务目标

本案例搭建一个可运行的车联网通信仿真环境，用于模拟车辆与路侧单元（RSU）之间的V2I通信过程。环境基于Python实现，包含车辆节点、RSU节点和通信信道三个核心组件，能够完成消息的发送、接收与传输延迟统计，为后续协议栈实验提供基础仿真平台。

环境依赖

需要Python 3.8+，无需第三方库，仅使用标准库random、time和dataclasses。代码在单文件内完成，可直接运行。

完整实现

import random
import time
from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class Vehicle:
    """车辆节点：周期发送BSM安全消息"""
    id: int
    position: float  # 道路位置（米）
    speed: float     # 速度（m/s）
    sent_count: int = 0
    
    def generate_bsm(self) -> Dict:
        """生成基本安全消息（BSM）"""
        self.sent_count += 1
        return {
            "type": "BSM",
            "vehicle_id": self.id,
            "position": self.position,
            "speed": self.speed,
            "timestamp": time.time()
        }

@dataclass
class RSU:
    """路侧单元：接收消息并记录延迟"""
    id: int
    position: float
    received: List[Dict] = field(default_factory=list)
    
    def receive(self, msg: Dict, delay: float):
        msg["delay_ms"] = round(delay * 1000, 2)
        self.received.append(msg)

class Channel:
    """通信信道：模拟传播延迟与丢包"""
    def __init__(self, loss_rate: float = 0.1):
        self.loss_rate = loss_rate
    
    def transmit(self, sender_pos: float, receiver_pos: float, msg: Dict) -> float | None:
        """返回传播延迟（秒），丢包返回None"""
        if random.random() < self.loss_rate:
            return None
        distance = abs(sender_pos - receiver_pos)
        # 假设传播速度200m/ms（简化模型）
        return distance / 200_000

def run_simulation(duration: float = 5.0, interval: float = 0.5):
    """运行仿真：车辆每interval秒发送一次BSM"""
    # 初始化节点
    vehicle = Vehicle(id=1, position=100.0, speed=20.0)
    rsu = RSU(id=1, position=500.0)
    channel = Channel(loss_rate=0.1)
    
    start = time.time()
    while time.time() - start < duration:
        # 车辆位置随时间更新
        vehicle.position += vehicle.speed * interval
        
        # 生成并发送消息
        bsm = vehicle.generate_bsm()
        delay = channel.transmit(vehicle.position, rsu.position, bsm)
        
        if delay is not None:
            rsu.receive(bsm, delay)
        else:
            print(f"消息 {bsm['vehicle_id']} 在传输中丢失")
        
        time.sleep(interval)
    
    # 输出统计结果
    print(f"\n=== 仿真结束 ===")
    print(f"车辆发送消息总数: {vehicle.sent_count}")
    print(f"RSU成功接收: {len(rsu.received)}")
    print(f"丢包率: {(vehicle.sent_count - len(rsu.received)) / vehicle.sent_count:.1%}")
    
    if rsu.received:
        delays = [m["delay_ms"] for m in rsu.received]
        print(f"平均传输延迟: {sum(delays)/len(delays):.2f} ms")
        print(f"最大延迟: {max(delays):.2f} ms")
        print(f"最小延迟: {min(delays):.2f} ms")
    
    return vehicle, rsu

if __name__ == "__main__":
    # 固定随机种子保证可复现
    random.seed(42)
    run_simulation(duration=3.0, interval=0.5)

运行方式

将代码保存为v2i_sim.py，在终端执行：

python v2i_sim.py

验证方法

程序运行后应输出仿真统计信息。通过以下条件判断仿真环境是否搭建成功：

• 车辆发送消息总数约为duration/interval（3秒/0.5秒=6条），由于循环计时存在微小偏差，实际值可能为5~7条；
• RSU接收消息数小于等于发送总数，丢包率接近设定的10%（因随机种子固定，每次运行结果一致）；
• 延迟值随车辆位置变化：车辆靠近RSU时延迟减小，远离时增大，且所有延迟值在合理范围内（约0.2~2ms）。

若输出显示发送总数大于0、接收数小于等于发送数、延迟数值非负，则说明车辆节点、RSU节点和信道三个组件均正常工作，仿真环境搭建完成。

上图展示了仿真环境的核心通信模型：车辆沿道路行驶并周期性发送BSM消息，RSU固定在500m处接收消息。信道根据两者距离计算传播延迟，并以10%概率模拟丢包。该模型为后续协议栈实验提供了基础的消息传递与延迟统计能力。