车联网一体化测试：平台与智能网联汽车的端到端协同验证分享

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""云控平台API接口自动化测试脚本测试内容：感知融合 / 路径规划 / 协同决策 三大模块依赖：pip install requests pytest pytest-html"""import requestsimport timeimport jsonimport pytestfrom datetime import datetime# ============ 配置区（根据实际项目修改） ============BASE_URL = "https://cloudctrl.example.com/api/v2"TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.test_token_replace_me"HEADERS = {    "Authorization": f"Bearer {TOKEN}",    "Content-Type": "application/json",    "X-Request-ID": "",}TIMEOUT = 5.0  # 接口超时时间（秒）# ============ 辅助函数 ============def make_request_id():    "# 生成唯一请求ID，方便日志追踪"    return f"test-{datetime.now().strftime("%Y%m%d%H%M%S%f")}"def api_request(method, endpoint, payload=None):    "# 统一请求封装：自动注入请求ID，自动记录耗时"    url = f"{BASE_URL}/{endpoint}"    HEADERS["X-Request-ID"] = make_request_id()    start = time.perf_counter()    try:        resp = requests.request(            method=method,            url=url,            headers=HEADERS,            json=payload,            timeout=TIMEOUT        )        elapsed = (time.perf_counter() - start) * 1000  # ms        return resp, elapsed    except requests.exceptions.Timeout:        pytest.fail(f"请求超时（{TIMEOUT}s）：{endpoint}")    except Exception as e:\n        pytest.fail(f"请求异常：{e}")# ============ 测试用例 ============class TestFusionAPI:    """感知融合模块测试"""    def test_fusion_report(self):        """测试：MEC上报感知数据，云控平台融合成功"""        payload = {            "mec_id": "MEC-001",            "intersection_id": "INT-A001",            "timestamp": datetime.now().isoformat() + "Z",            "objects": [                {                    "id": "obj-001",                    "type": "vehicle",                    "position": {"lat": 31.2304, "lon": 121.4737},                    "velocity": 15.5,                    "heading": 90.0                },                {                    "id": "obj-002",                    "type": "pedestrian",                    "position": {"lat": 31.2305, "lon": 121.4738},                    "velocity": 1.2,                    "heading": 45.0                }            ]        }        resp, elapsed = api_request("POST", "fusion/report", payload=payload)        assert resp.status_code == 200, f"融合上报失败：{resp.text}"        data = resp.json()        assert data.get("code") == 0, "业务返回码非0"        assert "fusion_id" in data, "缺少融合ID"        assert elapsed < 200, f"融合接口响应时间{elapsed:.0f}ms，超过200ms阈值"        print(f"✓ 融合上报成功，响应{elapsed:.0f}ms，fusion_id={data['fusion_id']}")class TestRouteAPI:    """路径规划模块测试"""    def test_route_plan(self):        """测试：单车路径规划请求与响应"""        payload = {            "vehicle_id": "VH-20260816001",            "start": {"lat": 31.2304, "lon": 121.4737},            "end": {"lat": 31.2400, "lon": 121.4800},            "preference": "fastest"        }        resp, elapsed = api_request("POST", "route/plan", payload=payload)        assert resp.status_code == 200        data = resp.json()        assert "route" in data and len(data["route"]["waypoints"]) > 0        assert elapsed < 300, f"路径规划{elapsed:.0f}ms，超过300ms阈值"        print(f"✓ 路径规划成功，{len(data['route']['waypoints'])}个航点，耗时{elapsed:.0f}ms")class TestControlAPI:    """协同决策模块测试"""    def test_signal_control(self):        """测试：信号灯配时调整指令下发"""        payload = {            "intersection_id": "INT-A001",            "phase": 2,            "green_duration": 35,            "reason": "高峰期流量调控"        }        resp, elapsed = api_request("POST", "control/signal", payload=payload)        assert resp.status_code == 200        data = resp.json()        assert data.get("control_id") is not None        assert elapsed < 500, f"控制指令{elapsed:.0f}ms，超过500ms阈值"        print(f"✓ 信号灯控制指令下发成功，control_id={data['control_id']}")

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""感知融合与目标跟踪测试脚本功能：模拟Camera+Radar多传感器融合 + 卡尔曼滤波 + 匈牙利匹配依赖：pip install numpy scipy"""import mathimport randomfrom dataclasses import dataclassfrom typing import List, Tuple, Optional# ============ 目标数据模型 ============@dataclassclass DetectedObject:    """检测到的目标对象"""    sensor: str    obj_id: str    x: float    y: float    vx: float = 0.0    vy: float = 0.0    confidence: float = 0.0    timestamp: float = 0.0@dataclassclass TrackedObject:    """跟踪目标：融合后的稳定目标"""    track_id: int    x: float    y: float    vx: float    vy: float    age: int = 0    confidence: float = 0.0    missed_frames: int = 0# ============ 卡尔曼滤波器（恒速模型） ============class KalmanFilter:    """卡尔曼滤波器 - 状态预测与更新，状态向量 [x, y, vx, vy]"""    def __init__(self, dt: float = 0.1):        self.dt = dt        self.x = [0.0, 0.0, 0.0, 0.0]        self.P = [[1,0,0,0],[0,1,0,0],[0,0,0.5,0],[0,0,0,0.5]]        self.F = [[1,0,dt,0],[0,1,0,dt],[0,0,1,0],[0,0,0,1]]        q = 0.05        self.Q = [[q*dt**4/4,0,q*dt**3/2,0],[0,q*dt**4/4,0,q*dt**3/2],                  [q*dt**3/2,0,q*dt**2,0],[0,q*dt**3/2,0,q*dt**2]]    def predict(self):        """状态预测"""        new_x = [sum(self.F[i][j] * self.x[j] for j in range(4)) for i in range(4)]        self.x = new_x        return self.x    def update(self, z):        """状态更新（z为观测值[x, y]）"""        H = [[1,0,0,0],[0,1,0,0]]        y = [z[i] - sum(H[i][j]*self.x[j] for j in range(4)) for i in range(2)]        # 简化更新（省略协方差更新完整实现）        self.x[0] += y[0] * 0.8        self.x[1] += y[1] * 0.8        return self.x# ============ 融合测试主流程 ============def test_sensor_fusion():    """多传感器融合测试主函数"""    print("="*50)    print("🔍 多传感器融合测试")    print("="*50)    # 模拟摄像头检测结果    cam_objects = [        DetectedObject(sensor="camera", obj_id="cam_001",                       x=15.3, y=2.1, vx=12.0, vy=0.5, confidence=0.92),        DetectedObject(sensor="camera", obj_id="cam_002",                       x=28.7, y=-1.5, vx=8.0, vy=-0.3, confidence=0.85),    ]    # 模拟雷达检测结果    radar_objects = [        DetectedObject(sensor="radar", obj_id="rad_001",                       x=15.5, y=2.0, vx=12.2, vy=0.3, confidence=0.95),        DetectedObject(sensor="radar", obj_id="rad_002",                       x=35.0, y=3.0, vx=5.0, vy=0.0, confidence=0.70),    ]    # 融合：基于距离匹配    fused = []    used_radar = set()    for cam in cam_objects:        best_match = None        best_dist = 3.0  # 最大匹配距离3米        for rad in radar_objects:            if rad.obj_id in used_radar:                continue            dist = math.sqrt((cam.x - rad.x)**2 + (cam.y - rad.y)**2)            if dist < best_dist:                best_dist = dist                best_match = rad        if best_match:            used_radar.add(best_match.obj_id)            # 加权融合            w_cam = cam.confidence / (cam.confidence + best_match.confidence)            w_rad = 1 - w_cam            fused_x = cam.x * w_cam + best_match.x * w_rad            fused_y = cam.y * w_cam + best_match.y * w_rad            fused_vx = cam.vx * w_cam + best_match.vx * w_rad            fused_vy = cam.vy * w_cam + best_match.vy * w_rad            fused_conf = max(cam.confidence, best_match.confidence)            fused.append({                "id": f"fused_{len(fused)+1}",                "x": round(fused_x, 2),                "y": round(fused_y, 2),                "vx": round(fused_vx, 2),                "vy": round(fused_vy, 2),                "confidence": round(fused_conf, 3),                "sources": ["camera", "radar"]            })            print(f"  ✅ 融合目标: cam={cam.obj_id} + rad={best_match.obj_id}")            print(f"     位置({fused_x:.1f}, {fused_y:.1f}) 速度({fused_vx:.1f}, {fused_vy:.1f}) 置信度={fused_conf:.3f}")        else:            fused.append({"id": f"fused_{len(fused)+1}", "x": cam.x, "y": cam.y,                          "vx": cam.vx, "vy": cam.vy, "confidence": cam.confidence,                          "sources": ["camera"]})    # 未匹配的雷达目标    for rad in radar_objects:        if rad.obj_id not in used_radar:            fused.append({"id": f"fused_{len(fused)+1}", "x": rad.x, "y": rad.y,                          "vx": rad.vx, "vy": rad.vy, "confidence": rad.confidence,                          "sources": ["radar"]})            print(f"  ⚠ 仅雷达检测: {rad.obj_id}")    print(f"\n📊 融合结果：共{len(fused)}个目标")    for obj in fused:        print(f"  {obj['id']}: pos({obj['x']}, {obj['y']}) vel({obj['vx']}, {obj['vy']}) conf={obj['confidence']} src={obj['sources']}")    # 断言    assert len(fused) >= 2, "融合目标数应≥2"    assert all(o["confidence"] > 0.5 for o in fused), "置信度应>0.5"    print("\n✅ 测试通过：融合结果符合预期")if __name__ == "__main__":    test_sensor_fusion()