3. V2I通信

任务目标

实现一个最小可运行的 V2I 交互：车辆端周期性发送请求，信号灯端返回当前相位（红/绿/黄）与剩余倒计时；车辆端解析后打印相位和秒数。用本地 UDP 回环模拟空口链路，便于单机验证。

环境与依赖

• Python 3.8+，仅用标准库 socket、json、time、threading。
• 无需第三方库，无需真实 RSU/OBU 硬件。

协议约定

车辆端发送 JSON 请求 {"type":"spat_req","veh_id":"V001"}；信号灯端返回 {"type":"spat","phase":"red|green|yellow","remain":N,"ts":...}。相位按固定周期循环，倒计时由当前时刻在周期内的位置计算，保证返回值与时间一致。

完整代码

import socket, json, time, threading

HOST, PORT = "127.0.0.1", 9100
# 相位周期：绿 20s -> 黄 3s -> 红 15s，循环
CYCLE = [("green", 20), ("yellow", 3), ("red", 15)]
TOTAL = sum(d for _, d in CYCLE)

def current_spat(now=None):
    t = (now if now is not None else time.time()) % TOTAL
    acc = 0
    for phase, dur in CYCLE:
        if t < acc + dur:
            return phase, int(acc + dur - t) + 1
        acc += dur
    return CYCLE[0][0], CYCLE[0][1]

def rsu_server(stop):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind((HOST, PORT))
    s.settimeout(0.5)
    while not stop.is_set():
        try:
            data, addr = s.recvfrom(1024)
        except socket.timeout:
            continue
        try:
            req = json.loads(data.decode())
        except json.JSONDecodeError:
            continue
        if req.get("type") != "spat_req":
            continue
        phase, remain = current_spat()
        resp = {"type": "spat", "phase": phase, "remain": remain, "ts": time.time()}
        s.sendto(json.dumps(resp).encode(), addr)
    s.close()

def obu_client(rounds=5, interval=1.0):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(2.0)
    for i in range(rounds):
        req = {"type": "spat_req", "veh_id": "V001"}
        s.sendto(json.dumps(req).encode(), (HOST, PORT))
        try:
            data, _ = s.recvfrom(1024)
            msg = json.loads(data.decode())
            print(f"[{i+1}] phase={msg['phase']} remain={msg['remain']}s")
        except socket.timeout:
            print(f"[{i+1}] timeout")
        time.sleep(interval)
    s.close()

if __name__ == "__main__":
    stop = threading.Event()
    t = threading.Thread(target=rsu_server, args=(stop,), daemon=True)
    t.start()
    time.sleep(0.2)
    obu_client()
    stop.set()
    t.join(timeout=1.0)

运行方式

python v2i_spat.py

验证方法与通过条件

• 程序启动后应连续打印 5 行形如 [1] phase=green remain=18s 的结果。
• 通过条件：每行 phase 属于 green/yellow/red，remain 为正整数且不超过对应相位时长（20/3/15）。
• 可重复运行观察：由于相位由 time.time() % TOTAL 决定，不同时刻启动会得到不同相位，属预期行为。
• 若出现 timeout，检查 9100 端口是否被占用或防火墙是否拦截本地 UDP。