2. 基于PC5接口的V2V直接通信实现

任务说明

本案例用PC5直连（Sidelink）在车与车之间建立V2V通信：发送端周期广播BSM（Basic Safety Message），内容为车辆ID、经纬度、速度；接收端解析并打印。使用Python + UDP组播模拟PC5广播信道，聚焦通信逻辑本身，便于在单机或局域网复现。

环境与依赖

• Python 3.8+，仅用标准库（socket、struct、json、time、threading）。
• 两台设备需在同一局域网，或单机用回环地址测试。
• PC5真实部署时替换为Sidelink socket/DSRC网卡，本案例用UDP组播承载相同报文格式。

报文格式

固定二进制头 + JSON负载，便于解析与扩展：

• magic(4B)="V2X1"，msg_type(1B)=1 表示BSM，seq(4B) 序号，payload_len(2B)，payload(JSON)。
• payload字段：id、lat、lon、speed、ts。

通信流程

完整实现

import socket, struct, json, time, threading

GROUP, PORT = "239.0.0.1", 5000
MAGIC = b"V2X1"
MSG_BSM = 1
HEADER = struct.Struct("!4sBIH")  # magic, type, seq, payload_len

def pack_bsm(seq, vid, lat, lon, speed):
    payload = json.dumps({"id": vid, "lat": lat, "lon": lon,
                          "speed": speed, "ts": time.time()}).encode()
    return HEADER.pack(MAGIC, MSG_BSM, seq, len(payload)) + payload

def parse_packet(data):
    if len(data) < HEADER.size:
        return None
    magic, mtype, seq, plen = HEADER.unpack(data[:HEADER.size])
    if magic != MAGIC or mtype != MSG_BSM:
        return None
    body = data[HEADER.size:HEADER.size + plen]
    if len(body) != plen:
        return None
    return seq, json.loads(body.decode())

def sender(vid, lat, lon, speed, interval=1.0, count=5):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1)
    for seq in range(count):
        pkt = pack_bsm(seq, vid, lat, lon, speed)
        s.sendto(pkt, (GROUP, PORT))
        print(f"[TX] seq={seq} id={vid} speed={speed}")
        time.sleep(interval)
    s.close()

def receiver(stop_after=5):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(("", PORT))
    mreq = struct.pack("4s4s", socket.inet_aton(GROUP),
                       socket.inet_aton("0.0.0.0"))
    s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
    s.settimeout(3.0)
    got = 0
    while got < stop_after:
        try:
            data, _ = s.recvfrom(2048)
        except socket.timeout:
            break
        r = parse_packet(data)
        if r:
            seq, b = r
            print(f"[RX] seq={seq} id={b['id']} lat={b['lat']:.5f} "
                  f"lon={b['lon']:.5f} speed={b['speed']} m/s")
            got += 1
    s.close()

if __name__ == "__main__":
    t = threading.Thread(target=receiver, args=(5,), daemon=True)
    t.start()
    time.sleep(0.3)
    sender("CAR-A", 31.23040, 121.47370, 12.5, interval=0.5, count=5)
    t.join()

运行方式

python v2v_pc5.py

单机即可运行；跨机测试时把两台设备接入同一局域网，接收端先启动，发送端后启动。

验证方法与通过条件

• 接收端打印5条 [RX] 记录，字段id、lat、lon、speed与发送端一致。
• 发送端打印5条 [TX]，seq从0到4递增。
• 若接收端无输出，检查防火墙是否放行UDP 5000，或组播是否被交换机过滤。

常见问题

• 组播不通：确认 IP_MULTICAST_TTL 已设置，跨网段需路由器支持IGMP。
• 真实PC5部署：将UDP组播替换为Sidelink socket或DSRC网卡发送，报文格式与解析逻辑不变。