From fbd7e5e7ad0e185f2b0fca086f024bf620f3d468 Mon Sep 17 00:00:00 2001 From: Tian jianyong <11429339@qq.com> Date: Thu, 16 Oct 2025 18:33:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=97=A5=E5=BF=97=E7=AE=A1?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- IFLOW.md | 142 ++ qaup-admin/src/main/resources/logback.xml | 72 +- .../logging_config.cpython-313.pyc | Bin 0 -> 1835 bytes tools/logging_config.py | 55 + tools/mock_adxp.py | 18 +- tools/mock_airport.py | 71 +- tools/mock_traffic_light.py | 42 +- tools/mock_unmanned_vehicle.py | 1909 ++++++++--------- 8 files changed, 1281 insertions(+), 1028 deletions(-) create mode 100644 IFLOW.md create mode 100644 tools/__pycache__/logging_config.cpython-313.pyc create mode 100644 tools/logging_config.py diff --git a/IFLOW.md b/IFLOW.md new file mode 100644 index 00000000..31904de5 --- /dev/null +++ b/IFLOW.md @@ -0,0 +1,142 @@ +# QAUP-Management 项目概览 + +## 项目简介 + +QAUP-Management 是一个机场无人车冲突管理平台，集成了机场冲突避免系统。该项目基于若依框架（RuoYi）开发，采用 Maven 多模块架构，提供了完整的车辆管理、空间分析、实时监控等功能。 + +核心模块 `qaup-collision` 实现了无人车与飞机、其他车辆之间的冲突检测算法，利用 PostGIS 进行空间数据分析，并通过 WebSocket 实现实时数据推送。 + +## 技术栈 + +- **后端框架**: Spring Boot 3.5.3, Spring MVC, Spring Data JPA, MyBatis +- **数据库**: PostgreSQL 与 PostGIS 扩展（空间数据处理） +- **缓存**: Redis +- **实时通信**: WebSocket + STOMP +- **空间计算**: JTS (Java Topology Suite), GeoTools +- **构建工具**: Maven 3.6+ +- **Java版本**: JDK 21 +- **API文档**: SpringDoc OpenAPI (Swagger UI) +- **数据库迁移**: Flyway + +## 项目结构 + +``` +QAUP-Management/ +├── qaup-admin/ # Web服务入口，集成所有模块 +├── qaup-framework/ # 核心框架，提供通用功能 +├── qaup-system/ # 系统管理模块（用户、角色、权限等） +├── qaup-collision/ # 冲突避免系统模块（核心业务逻辑） +├── qaup-common/ # 公共工具类和基础组件 +├── qaup-quartz/ # 定时任务调度 +├── qaup-generator/ # 代码生成器 +├── qaup-ui/ # 前端Vue项目 +├── sql/ # 数据库初始化脚本 +├── deploy/ # 部署相关脚本和配置 +├── tools/ # 开发和测试工具 +└── doc/ # 项目文档 +``` + +## 核心模块详解 + +### qaup-collision (冲突避免系统) + +这是项目的核心模块，负责实现无人车冲突检测与管理功能。 + +**主要特性**: +- 基于 PostGIS 的空间数据分析与几何计算 +- WebSocket 实时车辆位置监控与数据推送 +- 机场区域（跑道、滑行道、停机坪）配置与监控 +- 车辆与飞机、车辆与车辆间的实时冲突检测算法 +- 统一数据访问适配器 `QuapDataAdapter`，避免重复 DAO 开发 +- Redis 缓存支持高性能数据存取 + +**关键组件**: +- `QuapDataAdapter`: 统一数据访问接口，连接若依系统数据 +- `WebSocketConfig`: WebSocket 配置，支持实时数据推送 +- `VehicleLocationService`: 车辆位置管理服务 +- 空间实体类：继承空间基类并配置 PostGIS 映射 + +## 构建与运行 + +### 环境准备 + +- JDK 21 +- Maven 3.6+ +- PostgreSQL (启用 PostGIS 扩展) +- Redis + +### 快速启动 + +1. **数据库配置**: + - 创建 PostgreSQL 数据库并启用 PostGIS 扩展 + - 执行 SQL 初始化脚本 (位于 `sql/` 目录) + - 配置 `qaup-admin/src/main/resources/application.yml` 中的数据库连接信息 + +2. **启动依赖服务**: + ```bash + # 启动 Redis + redis-server + + # 启动 PostgreSQL (确保已创建数据库并启用 PostGIS) + ``` + +3. **编译和启动**: + ```bash + # 清理并编译整个项目 + mvn clean install + + # 启动应用 + cd qaup-admin + mvn spring-boot:run + + # 或者运行打包后的jar + java -jar target/qaup-admin.jar + ``` + +4. **使用启动脚本**: + ```bash + # 复制环境配置模板 + cp .env.example .env + # 修改 .env 中的配置值 + # 启动应用 + ./qaup.sh start + # 停止应用 + ./qaup.sh stop + # 重启应用 + ./qaup.sh restart + # 查看状态 + ./qaup.sh status + ``` + +### 访问系统 + +- 管理后台: `http://localhost:8080` +- WebSocket 端点: `ws://localhost:8080/collision` +- API 文档: `http://localhost:8080/swagger-ui.html` + +## 开发指南 + +### 项目合并说明 + +本项目已将独立的 CollisionAvoidanceSystem 项目合并到若依框架中，通过以下方式实现： + +1. **模块化集成**: 创建 `qaup-collision` 模块，保持功能独立性 +2. **数据适配器**: `QuapDataAdapter` 提供统一数据访问，连接若依 Service 层 +3. **配置整合**: 将 PostGIS、JPA、WebSocket 等配置整合到若依配置体系 +4. **版本兼容**: 升级 Spring Boot 版本至 3.5.3，使用 JDK 21 + +### collision 模块开发要点 + +- 新增空间实体时，需继承空间基类并正确配置 PostGIS 映射 +- WebSocket 消息通过 `/topic` 前缀向客户端广播 +- 使用 `QuapDataAdapter` 获取车辆和司机数据，避免直接访问 DAO +- 所有空间计算应使用 JTS 和 GeoTools 库 + +### 数据采集与处理 + +系统支持从多个数据源采集数据： +- 机场数据 API +- ADXP 适配器服务 +- 无人车厂商 API + +数据采集配置可在 `application.yml` 中调整，包括采集间隔、超时时间等参数。 \ No newline at end of file diff --git a/qaup-admin/src/main/resources/logback.xml b/qaup-admin/src/main/resources/logback.xml index fdacf927..61d0ad5a 100644 --- a/qaup-admin/src/main/resources/logback.xml +++ b/qaup-admin/src/main/resources/logback.xml @@ -15,12 +15,16 @@ ${log.path}/sys-info.log - - + + - ${log.path}/sys-info.%d{yyyy-MM-dd}.log - - 60 + ${log.path}/sys-info.%d{yyyy-MM-dd}.%i.log + + 30 + + 100MB + + 10GB ${log.pattern} @@ -37,12 +41,16 @@ ${log.path}/sys-error.log - - + + - ${log.path}/sys-error.%d{yyyy-MM-dd}.log - - 60 + ${log.path}/sys-error.%d{yyyy-MM-dd}.%i.log + + 30 + + 100MB + + 10GB ${log.pattern} @@ -60,21 +68,51 @@ ${log.path}/sys-user.log - + - ${log.path}/sys-user.%d{yyyy-MM-dd}.log - - 60 + ${log.path}/sys-user.%d{yyyy-MM-dd}.%i.log + + 30 + + 100MB + + 10GB ${log.pattern} + + + 0 + 512 + false + + + + + 0 + 512 + false + + + + + 0 + 512 + false + + + + + + + @@ -82,12 +120,12 @@ - - + + - + \ No newline at end of file diff --git a/tools/__pycache__/logging_config.cpython-313.pyc b/tools/__pycache__/logging_config.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..cfa6f90d2db15b965329316917cfbc50f4a9b3b7 GIT binary patch literal 1835 zcmah~TTB~A6rJ^Ry!b&9Tmp%ZK@e~R-b6~I)~XtXv?OgxXl?CBOC_|im&J*_YwfO^ zkfMlGsenkxM5+s^Qi)1L^r2F|+NcT5SN{uGvNe4uQZbLOP?`RBW*@;gNV$?{=FUBH z?wLEY&15psAE&^HaDVf5)<)Y;st85K`k3a;P z%Xa&yHEKI$joKT*6-u@XRM4?Sf|r66IuwyqPECkHPa?@*;KX>6mlFWYf-O5n#7JYB z6Ay@#a;JsJ=r`v<1pCz%sa$CUS?V;U4@Em`?=WV$FnAdk$y{Zu-Q+TP;;-!C*{2LG zV0Tpj^lJqtv=Mg|Id(A?m>EO|>xlSX#IP0<5daub`y%4GwokCOLuQ<%SSxFbQc=rm zHqlw3qs~O^_ATJF;fSyTiT7G_%Y!LC<>Rs^x?ZXz1h=;_aq>iYjI3=bgQhZPebOqWvNd1gql2oF? zPw|Sv2{;<`B8#LTTbfgckTaf;2?#m(>j_0wNY1M~o>Y+25RF^{uj;}PRPc}xBRHf# z6^cw=1muaTGS8)q_XD|lBss;YDla29Ub5@aTs&@of;6&esUc3Johq_xqLJ(QX(4xI z-8D%ASR8T?ZG@a$h54q*atisC*qb4}<3jPgCqQG$J zw8ET|Qfcfn@1JGg>+0o1T(cA}s*Ea0Nrf?Lg71Mim=HqgVQqi-V(5b8W}JpRJPN|i zDA;KNz5`idLdZ%J(#+7Rw|(^d-%ju6^7OU4*QRgYy*Vos>O;%*p+6y=t+`V(bv_S| z?l^$gU+^@oc$#L^RZr_^Ptit?_1)^5dMj^h+;#x>!R(prnYH@kMR#4%-&8!@TJ#_N z*JVFUkGpqj!I73iBeT-TtTvt;_hB;s(L!C@N?lvg-Ced?{PegBbGZBk*t`szXM=gz zS%mck*t!f`XFte8hD3b@*su&6W;*k*y$HRA$K>Jhr!@ygd&_utAn1_2xRYJDLA6F@ zLVNFp?d?lFeOgt%1#$~P2M8Lzo|593B>#qd47=n%K_Qjfwo(*ThJf<@0bGB96HnVi V^CMr6JRJG`!~XvOLmkv8{s)<<>b3v? literal 0 HcmV?d00001 diff --git a/tools/logging_config.py b/tools/logging_config.py new file mode 100644 index 00000000..9140a1d7 --- /dev/null +++ b/tools/logging_config.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import logging +import os +from logging.handlers import RotatingFileHandler + +def setup_logger(name, log_file, level=logging.INFO, max_bytes=10*1024*1024, backup_count=5): + """ + 配置日志记录器 + + Args: + name: 日志记录器名称 + log_file: 日志文件路径 + level: 日志级别 + max_bytes: 单个日志文件最大字节数（默认10MB） + backup_count: 保留的备份日志文件数量 + + Returns: + logging.Logger: 配置好的日志记录器 + """ + # 创建logs目录（如果不存在） + log_dir = os.path.dirname(log_file) + if log_dir and not os.path.exists(log_dir): + os.makedirs(log_dir) + + # 创建日志记录器 + logger = logging.getLogger(name) + logger.setLevel(level) + + # 避免重复添加处理器 + if not logger.handlers: + # 创建文件处理器（带滚动） + file_handler = RotatingFileHandler( + log_file, + maxBytes=max_bytes, + backupCount=backup_count, + encoding='utf-8' + ) + + # 创建控制台处理器 + console_handler = logging.StreamHandler() + + # 设置格式 + formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + file_handler.setFormatter(formatter) + console_handler.setFormatter(formatter) + + # 添加处理器到记录器 + logger.addHandler(file_handler) + logger.addHandler(console_handler) + + return logger \ No newline at end of file diff --git a/tools/mock_adxp.py b/tools/mock_adxp.py index 484b3f64..5537ddbf 100644 --- a/tools/mock_adxp.py +++ b/tools/mock_adxp.py @@ -28,13 +28,19 @@ from datetime import datetime from http.server import BaseHTTPRequestHandler from typing import List, Tuple, Optional -# 日志配置 - 只显示关键信息 -logging.basicConfig( - level=logging.WARNING, # 只显示 WARNING 和 ERROR，减少海量日志 - format="%(asctime)s [%(levelname)s] %(message)s" +# 导入统一的日志配置 +import sys +import os +sys.path.append(os.path.dirname(os.path.abspath(__file__))) +from logging_config import setup_logger + +# 设置当前模块的日志记录器 +logger = setup_logger( + name='mock_adxp', + log_file='logs/mock_adxp.log', + max_bytes=10*1024*1024, # 10MB + backup_count=5 ) -logger = logging.getLogger("mock-adxp") -logger.setLevel(logging.INFO) # 自己的日志保持 INFO # 服务代码 SERVICE_CODES = { diff --git a/tools/mock_airport.py b/tools/mock_airport.py index f2938253..7a6e816a 100644 --- a/tools/mock_airport.py +++ b/tools/mock_airport.py @@ -7,18 +7,17 @@ import threading import atexit from typing import Any, Literal, final, TypedDict -# 创建 logs 目录（如果不存在） -if not os.path.exists('logs'): - os.makedirs('logs') +# 导入统一的日志配置 +import sys +sys.path.append(os.path.dirname(os.path.abspath(__file__))) +from logging_config import setup_logger -# 配置日志 -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('logs/mock_server.log'), - logging.StreamHandler() - ] +# 设置当前模块的日志记录器 +logger = setup_logger( + name='mock_airport', + log_file='logs/mock_airport.log', + max_bytes=10*1024*1024, # 10MB + backup_count=5 ) app = Flask(__name__) @@ -1584,14 +1583,14 @@ class AircraftRouteManager: """ self.flight_configs[flight_no] = config self.followers[flight_no] = AircraftRouteFollower(flight_no) - logging.info(f"已添加飞机 {flight_no} 到路径跟随系统，行为模式: {config.get('primary_behavior', 'cycle')}") + logger.info(f"已添加飞机 {flight_no} 到路径跟随系统，行为模式: {config.get('primary_behavior', 'cycle')}") def update_aircraft_position(self, aircraft: dict[str, Any], elapsed_time: float): """统一的航空器位置更新入口""" flight_no = aircraft.get("flightNo") if flight_no not in self.followers: - logging.warning(f"飞机 {flight_no} 未注册到路径跟随系统") + logger.warning(f"飞机 {flight_no} 未注册到路径跟随系统") return follower = self.followers[flight_no] @@ -1955,7 +1954,7 @@ class BackgroundUpdateManager: self.running = False if self.update_thread: self.update_thread.join(timeout=2.0) - logging.info("后台更新管理器已停止") + logger.info("后台更新管理器已停止") def _update_loop(self): """主更新循环""" @@ -1978,7 +1977,7 @@ class BackgroundUpdateManager: time.sleep(UPDATE_INTERVAL) except Exception as e: - logging.error(f"后台更新出错: {str(e)}") + logger.error(f"后台更新出错: {str(e)}") time.sleep(1.0) # 出错时短暂休眠后继续 # 全局后台更新管理器实例 @@ -1988,15 +1987,15 @@ def check_auth() -> bool: auth_header = request.headers.get('Authorization') if not auth_header: - print("认证失败: 缺少Authorization头") + logger.warning("认证失败: 缺少Authorization头") return False # 直接比较完整的Authorization头 result = auth_header == AUTH_TOKEN if not result: - print(f"认证失败: token不匹配") - print(f"收到的token: {auth_header}") - print(f"期望的token: {AUTH_TOKEN}") + logger.warning(f"认证失败: token不匹配") + logger.debug(f"收到的token: {auth_header}") + logger.debug(f"期望的token: {AUTH_TOKEN}") return result def to_float_safe(x: object) -> float | None: @@ -2137,7 +2136,7 @@ def get_vehicle_positions(): "data": response_data }) except Exception as e: - print(f"Error in get_vehicle_positions: {str(e)}") + logger.error(f"Error in get_vehicle_positions: {str(e)}") return jsonify({ "status": 500, "msg": str(e), @@ -2203,12 +2202,12 @@ def login(): except: pass - print(f"收到登录请求: username={username}, password={password}") - print(f"请求 URL: {request.url}") - print(f"请求方法: {request.method}") - print(f"请求参数: {request.args}") - print(f"请求表单: {request.form}") - print(f"请求JSON: {request.get_json(silent=True)}") + logger.info(f"收到登录请求: username={username}, password={password}") + logger.debug(f"请求 URL: {request.url}") + logger.debug(f"请求方法: {request.method}") + logger.debug(f"请求参数: {request.args}") + logger.debug(f"请求表单: {request.form}") + logger.debug(f"请求JSON: {request.get_json(silent=True)}") if not username or not password: return jsonify({ @@ -2281,7 +2280,7 @@ def get_arrival_taxiway_route(): contact_cross = request.args.get('contactCross', 'F1') seat = request.args.get('seat', '138') - logging.info(f"进港路线查询: inRunway={in_runway}, outRunway={out_runway}, contactCross={contact_cross}, seat={seat}") + logger.info(f"进港路线查询: inRunway={in_runway}, outRunway={out_runway}, contactCross={contact_cross}, seat={seat}") # 根据参数匹配对应的航班路由 matching_flight = None @@ -2296,10 +2295,10 @@ def get_arrival_taxiway_route(): # 如果找到匹配的航班，返回对应路由，否则使用MU5123作为默认 if matching_flight and matching_flight in aircraft_routes: route_data = aircraft_routes[matching_flight]["arrival"] - logging.info(f"匹配航班 {matching_flight} 的进港路由") + logger.info(f"匹配航班 {matching_flight} 的进港路由") else: route_data = aircraft_routes["MU5123"]["arrival"] - logging.info(f"未找到匹配航班，使用MU5123的进港路由作为默认") + logger.info(f"未找到匹配航班，使用MU5123的进港路由作为默认") # 将路由数据转换为CGCS2000坐标系格式 converted_route_data = convert_route_to_cgcs2000(route_data) @@ -2328,7 +2327,7 @@ def get_departure_taxiway_route(): out_runway = request.args.get('outRunway', '34') start_seat = request.args.get('startSeat', '138') - logging.info(f"出港路线查询: inRunway={in_runway}, outRunway={out_runway}, startSeat={start_seat}") + logger.info(f"出港路线查询: inRunway={in_runway}, outRunway={out_runway}, startSeat={start_seat}") # 根据参数匹配对应的航班路由 matching_flight = None @@ -2342,10 +2341,10 @@ def get_departure_taxiway_route(): # 如果找到匹配的航班，返回对应路由，否则使用CA1234作为默认 if matching_flight and matching_flight in aircraft_routes: route_data = aircraft_routes[matching_flight]["departure"] - logging.info(f"匹配航班 {matching_flight} 的出港路由") + logger.info(f"匹配航班 {matching_flight} 的出港路由") else: route_data = aircraft_routes["CA1234"]["departure"] - logging.info(f"未找到匹配航班，使用CA1234的出港路由作为默认") + logger.info(f"未找到匹配航班，使用CA1234的出港路由作为默认") # 将路由数据转换为CGCS2000坐标系格式 converted_route_data = convert_route_to_cgcs2000(route_data) @@ -2384,9 +2383,9 @@ def get_active_flights() -> list[dict[str, Any]]: } active_flights.append(flight_notification) - logging.info(f"航班 {flight_no} {event_type}通知中 (路径状态: {follower.flight_status}, 进度: {follower.route_progress:.2f})") + logger.info(f"航班 {flight_no} {event_type}通知中 (路径状态: {follower.flight_status}, 进度: {follower.route_progress:.2f})") else: - logging.warning(f"航班 {flight_no} 在路径跟随器中但不在flight_data中") + logger.warning(f"航班 {flight_no} 在路径跟随器中但不在flight_data中") return active_flights @@ -2407,7 +2406,7 @@ def get_flights(): # 获取当前活跃航班 active_flights = get_active_flights() - logging.info(f"进出港航班查询: 返回 {len(active_flights)} 个活跃航班") + logger.info(f"进出港航班查询: 返回 {len(active_flights)} 个活跃航班") return jsonify({ "status": 200, @@ -2416,7 +2415,7 @@ def get_flights(): }) except Exception as e: - logging.error(f"进出港航班查询失败: {str(e)}") + logger.error(f"进出港航班查询失败: {str(e)}") return jsonify({ "status": 500, "msg": f"查询失败: {str(e)}", diff --git a/tools/mock_traffic_light.py b/tools/mock_traffic_light.py index d838b36f..5bd32b1e 100644 --- a/tools/mock_traffic_light.py +++ b/tools/mock_traffic_light.py @@ -17,6 +17,20 @@ from datetime import datetime import signal from typing import TypedDict +# 导入统一的日志配置 +import sys +import os +sys.path.append(os.path.dirname(os.path.abspath(__file__))) +from logging_config import setup_logger + +# 设置当前模块的日志记录器 +logger = setup_logger( + name='mock_traffic_light', + log_file='logs/mock_traffic_light.log', + max_bytes=10*1024*1024, # 10MB + backup_count=5 +) + class TLState(TypedDict): name: str duration: int @@ -82,10 +96,10 @@ class TrafficLightDevice: self.socket.connect((self.host, self.port)) # 连接成功后取消超时设置 self.socket.settimeout(None) - print(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ 已连接到服务器 {self.host}:{self.port}") + logger.info(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ 已连接到服务器 {self.host}:{self.port}") return True except Exception as e: - print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 连接失败: {e}") + logger.error(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 连接失败: {e}") if self.socket: try: self.socket.close() @@ -99,7 +113,7 @@ class TrafficLightDevice: if self.socket: try: self.socket.close() - print(f"[{datetime.now().strftime('%H:%M:%S')}] 已断开连接") + logger.info(f"[{datetime.now().strftime('%H:%M:%S')}] 已断开连接") except: pass self.socket = None @@ -107,15 +121,15 @@ class TrafficLightDevice: def send_signal(self, di_signal): """发送DI信号到服务器""" if not self.socket: - print("未连接到服务器") + logger.warning("未连接到服务器") return False try: self.socket.sendall(di_signal.encode('utf-8')) - print(f"[{datetime.now().strftime('%H:%M:%S')}] 📡 发送信号: {di_signal}") + logger.info(f"[{datetime.now().strftime('%H:%M:%S')}] 📡 发送信号: {di_signal}") return True except Exception as e: - print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 发送失败: {e}") + logger.error(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 发送失败: {e}") # 发送失败时关闭socket，触发重连 try: self.socket.close() @@ -160,9 +174,9 @@ class TrafficLightDevice: def run_continuous_simulation(self): """运行连续的红绿灯信号模拟""" - print(f"开始连续发送红绿灯信号，间隔: {self.interval}秒") - print("按 Ctrl+C 停止模拟") - print("服务端断开时会自动重连...") + logger.info(f"开始连续发送红绿灯信号，间隔: {self.interval}秒") + logger.info("按 Ctrl+C 停止模拟") + logger.info("服务端断开时会自动重连...") self.running = True start_time = time.time() @@ -174,7 +188,7 @@ class TrafficLightDevice: if not connected: connected = self.connect() if not connected: - print("连接失败，5秒后重试...") + logger.warning("连接失败，5秒后重试...") time.sleep(5) continue else: @@ -196,22 +210,22 @@ class TrafficLightDevice: # 发送信号 success = self.send_signal(di_signal) if not success: - print("发送失败，连接已断开") + logger.warning("发送失败，连接已断开") self.disconnect() connected = False continue - print(f"当前状态: {state_name}") + logger.info(f"当前状态: {state_name}") # 等待下一次发送 time.sleep(self.interval) except KeyboardInterrupt: - print("\n收到停止信号...") + logger.info("\n收到停止信号...") finally: self.running = False self.disconnect() - print("红绿灯模拟已停止") + logger.info("红绿灯模拟已停止") def test_signal_states(self): """测试各种信号状态""" diff --git a/tools/mock_unmanned_vehicle.py b/tools/mock_unmanned_vehicle.py index 60edb68a..f74e86d0 100644 --- a/tools/mock_unmanned_vehicle.py +++ b/tools/mock_unmanned_vehicle.py @@ -1,958 +1,957 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -无人车厂商平台模拟服务 -Unmanned Vehicle Manufacturer Platform Mock Server - -提供无人车位置上报、状态查询、控制指令等API接口 -模拟真实的无人车厂商平台与机场管理系统的交互 -""" - -from flask import Flask, jsonify, request -import time -import math -import logging -import os -from typing import Any, Literal, final, TypedDict - -# 创建 logs 目录（如果不存在） -if not os.path.exists('logs'): - os.makedirs('logs') - -# 配置日志 -logging.basicConfig( - level=logging.DEBUG, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('logs/unmanned_vehicle_mock_server.log'), - logging.StreamHandler() - ] -) - -app = Flask(__name__) - -# 为位置与移动对象定义类型，避免使用 Any -class Point(TypedDict): - latitude: float - longitude: float - -# 必填字段 -class MovingObjectRequired(TypedDict): - latitude: float - longitude: float - time: int - speed: float - moving_to_end: bool - start_point: Point - end_point: Point - -# 可选字段（total=False） -class MovingObject(MovingObjectRequired, total=False): - vehicleNo: str - wait_until: float - direction: float - -# 地球半径（米） -EARTH_RADIUS = 6378137.0 - -# 指令优先级配置 -COMMAND_PRIORITIES = { - "ALERT": 5, - "WARNING": 3, - "PARKING": 2, - "RESUME": 1 -} - -# 距离配置（米） -DIST_300M = 300 -DIST_200M = 200 -DIST_150M = 150 -DIST_100M = 100 -DIST_50M = 50 - -# 时间配置（秒） -WAIT_TIME_AFTER_RETURN = 0.1 # 返回起点后的等待时间 -UPDATE_INTERVAL = 1.0 # 位置更新间隔, 默认 1 秒 - -# 无人车路径配置 -# 无人车A 鲁B567 路径 -UNMANNED_A_START = {"longitude": 120.083084, "latitude": 36.369696} # 无人车A起点 -UNMANNED_A_END = {"longitude": 120.084637, "latitude": 36.365617} # 无人车A终点 - -# 无人车B 鲁B579 路径 -UNMANNED_B_START = {"longitude": 120.086965, "latitude": 36.368599} # 无人车B起点 -UNMANNED_B_END = {"longitude": 120.086263, "latitude": 36.370484} # 无人车B终点 - -# 车辆配置 -DEFAULT_VEHICLE_SPEED = 30.0 # km/h，默认速度 -EMERGENCY_BRAKE_DECELERATION = 0.8 # 紧急制动减速度 (每次更新减速 80%) -NORMAL_BRAKE_DECELERATION = 0.2 # 正常制动减速度 (每次更新减速 20%) - -# 认证相关配置 -AUTH_USERNAME = "dianxin" -AUTH_PASSWORD = "dianxin@123" -AUTH_TOKEN = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE3MzI3ODMwOTAsInVzZXJuYW1lIjoiYWRtaW4ifQ.y9feEL_9NT8UzED9NNkb0Ln6C-PBoufiSHWobWe5vWY" - -def meters_to_degrees(meters: float, latitude: float) -> tuple[float, float]: - """ - 将米转换为度，考虑纬度的影响 - """ - # 纬度方向：1度 = 111,319.9米 - meters_per_deg_lat: float = 111319.9 - # 经度方向：1度 = 111,319.9 * cos(latitude)米 - meters_per_deg_lon: float = meters_per_deg_lat * math.cos(math.radians(latitude)) - return meters / meters_per_deg_lat, meters / meters_per_deg_lon - -def to_float_safe(x: object) -> float | None: - """ - 安全将值转换为 float: - - 允许 int/float - - 允许非布尔的数字字符串 - - 允许 dict，优先取 'value'、'longitude'、'latitude' 等常见键 - - 无法解析则返回 None - """ - # 排除布尔值（bool 是 int 的子类，但不应当被视为数值） - if isinstance(x, bool): - return None - if isinstance(x, (int, float)): - return float(x) - if isinstance(x, str): - try: - return float(x.strip()) - except Exception: - return None - if isinstance(x, dict): - for key in ("value", "longitude", "latitude"): - if key in x: - return to_float_safe(x[key]) - return None - return None - -def calculate_distance_to_target(vehicle: dict[str, float], target_lat: float, target_lon: float) -> float: - """计算车辆到目标位的距离（米）""" - # 转换为弧度 - lat1_rad = math.radians(vehicle["latitude"]) - lon1_rad = math.radians(vehicle["longitude"]) - lat2_rad = math.radians(target_lat) - lon2_rad = math.radians(target_lon) - - dlat = lat2_rad - lat1_rad - dlon = lon2_rad - lon1_rad - - # 使用 Haversine 公式计算距离 - a = math.sin(dlat/2) * math.sin(dlat/2) + \ - math.cos(lat1_rad) * math.cos(lat2_rad) * \ - math.sin(dlon/2) * math.sin(dlon/2) - c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a)) - return EARTH_RADIUS * c - -def update_position_with_vector( - obj: dict[str, Any], - target_point: "Point", - start_point: "Point", - speed: float, - elapsed_time: float, - return_to_start: bool = False -) -> bool: - """ - 基于向量的位置更新逻辑 - - Args: - obj: 需要更新位置的对象（车辆） - target_point: 目标点坐标 {"latitude": float, "longitude": float} - start_point: 起点坐标 {"latitude": float, "longitude": float} - speed: 移动速度（km/h） - elapsed_time: 经过的时间（秒） - return_to_start: 是否在到达目标点时返回起点 - """ - # 检查是否在等待状态 - if "wait_until" not in obj: - obj["wait_until"] = 0.0 - - current_time = time.time() - # 基于类型安全读取 wait_until，避免 float 与 object 比较 - _wu = to_float_safe(obj.get("wait_until")) - wait_until = 0.0 if _wu is None else float(_wu) - if current_time < wait_until: - return False # 还在等待中，不更新位置 - - # 计算这一步要移动的距离（米）并转换为经纬度 - speed_mps = float(speed) * 1000.0 / 3600.0 - distance = speed_mps * float(elapsed_time) - - # 累计里程到总里程中 - if "total_mileage" not in obj: - obj["total_mileage"] = 0.0 - obj["total_mileage"] = to_float_safe(obj.get("total_mileage", 0.0)) or 0.0 - obj["total_mileage"] += distance - - # 安全读取并转换当前位置与目标点/起点经纬度为 float - cur_lat = to_float_safe(obj.get("latitude")) - cur_lon = to_float_safe(obj.get("longitude")) - tgt_lat = to_float_safe(target_point.get("latitude")) - tgt_lon = to_float_safe(target_point.get("longitude")) - if cur_lat is None or cur_lon is None or tgt_lat is None or tgt_lon is None: - # 无法解析经纬度，跳过更新 - return False - - # 将移动距离转换为经纬度变化量 - move_dlat, move_dlon = meters_to_degrees(distance, cur_lat) - - # 将5米的判断距离转换为经纬度（优化：从15米减少到5米，减少颤抖） - check_dlat, check_dlon = meters_to_degrees(5.0, cur_lat) - - # 计算当前位置到终点的向量（经纬度空间） - vector_lat = float(tgt_lat) - float(cur_lat) - vector_lon = float(tgt_lon) - float(cur_lon) - - # 计算向量长度（经纬度空间） - vector_length = math.sqrt(vector_lat * vector_lat + vector_lon * vector_lon) - - # 检查是否到达终点 - if vector_length <= math.sqrt(check_dlat * check_dlat + check_dlon * check_dlon): - # 精确设置到目标点位置，避免位置偏差 - obj["latitude"] = float(tgt_lat) - obj["longitude"] = float(tgt_lon) - - if return_to_start: - # 返回起点并设置等待时间 - obj["wait_until"] = current_time + WAIT_TIME_AFTER_RETURN - print(f"无人车 {obj.get('vehicleNo')} 返回起点，等待{WAIT_TIME_AFTER_RETURN}秒后继续") - else: - # 设置短暂等待时间，避免立即切换方向导致的颤抖 - obj["wait_until"] = current_time + 0.1 # 0.1秒的短暂等待 - print(f"无人车 {obj.get('vehicleNo')} 到达目标点，等待0.1秒后切换方向") - return True # 表示已到达终点 - else: - # 归一化方向向量 - unit_lat = vector_lat / vector_length - unit_lon = vector_lon / vector_length - - # 计算这一步的位置变化 - dlat = move_dlat * unit_lat - dlon = move_dlon * unit_lon - - # 更新位置（写回为 float） - obj["latitude"] = float(cur_lat + dlat) - obj["longitude"] = float(cur_lon + dlon) - return False # 表示正在移动中 - -def update_object_position(obj: dict[str, Any], elapsed_time: float) -> None: - """统一的位置更新函数 - 用于无人车的往复运动""" - # 获取对象类型和标识 - obj_id = obj.get("vehicleNo") - - # 根据移动方向确定目标点 - if obj["moving_to_end"]: - target_raw = obj["end_point"] - start_raw = obj["start_point"] - else: - target_raw = obj["start_point"] - start_raw = obj["end_point"] - - # 构造类型安全的 Point - t_lat = to_float_safe(target_raw.get("latitude")) if isinstance(target_raw, dict) else None - t_lon = to_float_safe(target_raw.get("longitude")) if isinstance(target_raw, dict) else None - s_lat = to_float_safe(start_raw.get("latitude")) if isinstance(start_raw, dict) else None - s_lon = to_float_safe(start_raw.get("longitude")) if isinstance(start_raw, dict) else None - if t_lat is None or t_lon is None or s_lat is None or s_lon is None: - # 无法解析坐标，直接更新时间戳后返回 - obj["time"] = int(time.time() * 1000) - return - - target_point: Point = {"latitude": float(t_lat), "longitude": float(t_lon)} - start_point: Point = {"latitude": float(s_lat), "longitude": float(s_lon)} - - # 读取速度为 float - spd = to_float_safe(obj.get("speed")) - if spd is None: - spd = 0.0 - - # 更新位置 - reached_target = update_position_with_vector( - obj, - target_point, - start_point, - float(spd), - float(elapsed_time), - return_to_start=False - ) - - if reached_target: - # 切换移动方向 - obj["moving_to_end"] = not obj["moving_to_end"] - target_name = "终点" if not obj["moving_to_end"] else "起点" - print(f"无人车 {obj_id} 到达{'终点' if obj['moving_to_end'] else '起点'}，开始前往{target_name}") - - # 更新时间戳 - obj["time"] = int(time.time() * 1000) - -def update_vehicle_position(vehicle: dict[str, Any], elapsed_time: float) -> None: - """更新无人车位置 - 使用统一的位置更新函数""" - update_object_position(vehicle, elapsed_time) - -# 无人车数据配置 -current_timestamp = time.time() -unmanned_vehicle_data = [ - { - "vehicleNo": "鲁B567", # 无人车A - "longitude": UNMANNED_A_START["longitude"], - "latitude": UNMANNED_A_START["latitude"], - "time": int(current_timestamp * 1000), - "direction": 90.0, # 90度 - "speed": 25.0, # 25km/h - "start_point": UNMANNED_A_START, # 起点 - "end_point": UNMANNED_A_END, # 终点 - "moving_to_end": True, # 当前是否向终点移动 - # 任务和里程相关字段 - "total_mileage": 0.0, # 总里程（米） - "mission_start_time": int(current_timestamp * 1000), # 任务开始时间 - "mission_id": "MISSION_A_001", # 任务ID - "mission_type": "PATROL_TRANSPORT" # 任务类型：巡逻运输 - }, - { - "vehicleNo": "鲁B579", # 无人车B - "longitude": UNMANNED_B_START["longitude"], - "latitude": UNMANNED_B_START["latitude"], - "time": int(current_timestamp * 1000), - "direction": 270.0, # 270度 - "speed": 25.0, # 25km/h - "start_point": UNMANNED_B_START, # 起点 - "end_point": UNMANNED_B_END, # 终点 - "moving_to_end": True, # 当前是否向终点移动 - # 任务和里程相关字段 - "total_mileage": 0.0, # 总里程（米） - "mission_start_time": int(current_timestamp * 1000), # 任务开始时间 - "mission_id": "MISSION_B_001", # 任务ID - "mission_type": "CARGO_TRANSPORT" # 任务类型：货物运输 - } -] - -# 添加车辆状态类 -@final -class VehicleState: - vehicle_no: str - is_running: bool - current_command: str | None - command_priority: int - target_speed: float - brake_mode: Literal['emergency', 'normal'] | None - command_reason: str | None - last_command_time: float - traffic_light_state: str | None - target_lat: float | None - target_lon: float | None - is_traffic_light_stop: bool - - def __init__(self, vehicle_no: str) -> None: - self.vehicle_no = vehicle_no - self.is_running = True # 运行状态 - self.current_command = None # 当前执行的指令 - self.command_priority = 0 # 当前指令优先级 - self.target_speed = float(DEFAULT_VEHICLE_SPEED) # 目标速度 - self.brake_mode = None # 制动模式：'emergency' 或 'normal' - self.command_reason = None # 指令原因 - self.last_command_time = float(time.time()) # 最后一次指令时间 - self.traffic_light_state = None # 当前红绿灯状态 - self.target_lat = None # 目标纬度 - self.target_lon = None # 目标经度 - self.is_traffic_light_stop = False # 是否因红灯停车 - self.remoteControlActive = False # 远程控制是否激活 - - def can_be_overridden_by(self, command_type: str) -> bool: - """判断当前指令是否可以被新指令覆盖""" - # 红绿灯信号不参与指令优先级判断 - if command_type in ["RED", "GREEN", "YELLOW"]: - return True - - # 类型安全获取优先级，避免 Unknown | None 作为 dict key - new_priority = COMMAND_PRIORITIES.get(str(command_type), 0) - current_key = self.current_command if isinstance(self.current_command, str) else None - current_priority = COMMAND_PRIORITIES.get(current_key or "", 0) - - # 相同类型的指令可以覆盖（因为需要持续发送） - if isinstance(self.current_command, str) and command_type == self.current_command: - return True - - # ALERT 指令可以被 RESUME 解除 - if self.current_command == "ALERT": - return command_type == "RESUME" - - # WARNING 指令可以被 RESUME 或相同及更高优先级指令覆盖 - if self.current_command == "WARNING": - return command_type == "RESUME" or new_priority >= current_priority - - # 其他情况下，相同或更高优先级可以覆盖 - return new_priority >= current_priority - - def update_command(self, command_type: str, target_lat: float | None = None, target_lon: float | None = None) -> None: - """更新指令状态""" - # 更新目标位置 - if target_lat is not None and target_lon is not None: - self.target_lat = target_lat - self.target_lon = target_lon - - # 如果是红绿灯状态，只更新状态，不影响其他指令 - if command_type in ["RED", "GREEN", "YELLOW"]: - old_state = self.traffic_light_state - self.traffic_light_state = command_type - - # 更新红灯停车状态 - if command_type == "RED": - self.is_traffic_light_stop = True - elif command_type in ["GREEN", "YELLOW"]: - self.is_traffic_light_stop = False - - print(f"无人车 {self.vehicle_no} 收到红绿灯信号: {command_type} (原状态: {old_state})") - else: - # 其他指令正常更新 - self.current_command = command_type - self.command_priority = COMMAND_PRIORITIES.get(command_type, 0) - - # RESUME 指令不清除红绿灯状态，红绿灯状态由信号灯独立控制 - if command_type == "RESUME": - # 清除其他状态 - self.brake_mode = None - self.command_reason = None - - # 更新运行状态 - self.is_running = self.can_move() - print(f"更新无人车 {self.vehicle_no} 状态: command={self.current_command}, traffic_light={self.traffic_light_state}, priority={self.command_priority}, is_running={self.is_running}") - - def can_move(self) -> bool: - """检查车辆是否可以移动""" - # 如果有告警指令，不能移动 - if self.current_command == "ALERT": - return False - - # 如果有预警指令，不能移动 - if self.current_command == "WARNING": - return False - - # 如果有安全停靠指令，不能移动 - if self.current_command == "PARKING": - return False - - # 如果是红灯停车状态，不能移动 - if self.is_traffic_light_stop: - return False - - # 其他情况可以移动 - return True - - def log_state(self) -> None: - """记录当前车辆状态""" - print(f""" -无人车状态: -- 车辆编号: {self.vehicle_no} -- 运行状态: {'运行中' if self.is_running else '已停止'} -- 当前指令: {self.current_command} -- 红绿灯状态: {self.traffic_light_state} -- 红灯停车: {'是' if self.is_traffic_light_stop else '否'} -- 指令优先级: {self.command_priority} -- 目标速度: {self.target_speed} -- 制动模式: {self.brake_mode} -- 指令原因: {self.command_reason} -- 目标位置: ({self.target_lat}, {self.target_lon}) -""") - -# 车辆状态管理 -vehicle_states = {} -for vehicle in unmanned_vehicle_data: - vehicle_states[str(vehicle["vehicleNo"])] = VehicleState(str(vehicle["vehicleNo"])) - -# 记录上次更新时间 -last_unmanned_vehicle_update_time = time.time() - -def check_auth() -> bool: - """检查认证""" - auth_header = request.headers.get('Authorization') - - if not auth_header: - print("认证失败: 缺少Authorization头") - return False - - # 直接比较完整的Authorization头 - result = auth_header == AUTH_TOKEN - if not result: - print(f"认证失败: token不匹配") - print(f"收到的token: {auth_header}") - print(f"期望的token: {AUTH_TOKEN}") - return result - -class RequestVehicleCommandPayload(TypedDict, total=False): - vehicleID: str - commandType: str - commandReason: str - signalState: str - latitude: float - longitude: float - transId: str - - -# 生成符合API规范的完整车辆状态数据 -def generate_comprehensive_vehicle_status(vehicle_id: str) -> dict[str, Any]: - """生成符合universal_autonomous_vehicle_api规范的完整车辆状态数据""" - current_time = time.time() - timestamp_ms = int(current_time * 1000) - - # 获取车辆基础数据 - vehicle_data_item = None - for v in unmanned_vehicle_data: - if v["vehicleNo"] == vehicle_id: - vehicle_data_item = v - break - - if not vehicle_data_item: - return {} - - # 获取车辆状态 - vehicle_state = vehicle_states.get(vehicle_id) - - # 基础位置和运动数据 - lat_val = to_float_safe(vehicle_data_item.get("latitude")) - lon_val = to_float_safe(vehicle_data_item.get("longitude")) - speed_val = to_float_safe(vehicle_data_item.get("speed", 0)) - direction_val = to_float_safe(vehicle_data_item.get("direction", 0)) - - # 生成完整状态数据 - status_data = { - "vehicleInfo": { - "vehicleId": vehicle_id - }, - "operationalStatus": { - "powerStatus": "ON", - "systemHealth": "HEALTHY" if vehicle_state and vehicle_state.is_running else "DEGRADED", - "operationalMode": "AUTONOMOUS", - "emergencyStatus": "CRITICAL" if vehicle_state and vehicle_state.current_command == "ALERT" - else "WARNING" if vehicle_state and vehicle_state.current_command == "WARNING" - else "NORMAL", - "lastHeartbeat": timestamp_ms - }, - "controlStatus": { - "controlMode": "AUTONOMOUS", - "controlAuthority": "SYSTEM", - "remoteControlActive": vehicle_state.remoteControlActive if vehicle_state else False - }, - "motionStatus": { - "position": { - "latitude": round(lat_val, 6) if lat_val is not None else 0.0, - "longitude": round(lon_val, 6) if lon_val is not None else 0.0 - }, - "velocity": { - "speed": round(speed_val * 1000 / 3600, 2) if speed_val is not None else 0.0, # 转换为 m/s - "direction": round(math.radians(direction_val), 4) if direction_val is not None else 0.0 # 转换为弧度 - } - }, - "safetyStatus": { - "collisionAvoidanceActive": vehicle_state.is_running if vehicle_state else True, - "emergencyBrakingReady": True, - "pathPlanningStatus": "ACTIVE" if vehicle_state and vehicle_state.is_running else "INACTIVE", - "obstacleDetectionStatus": "ACTIVE" if vehicle_state and vehicle_state.is_running else "INACTIVE", - "minimumRiskManeuverTriggered": vehicle_state.current_command == "ALERT" if vehicle_state else False - }, - "sensorStatus": { - "gps": { - "status": "ACTIVE", - "accuracy": 0.5, # 0.5米精度 - "lastUpdate": timestamp_ms - } - }, - "batteryStatus": { - "mainBattery": { - "chargeLevel": 85.5, # 85.5% - "voltage": 48.2, # 48.2V - "current": -15.3, # -15.3A (放电) - "temperature": 35.2, # 35.2°C - "chargingStatus": "DISCHARGING" - } - }, - "communicationStatus": { - "v2xStatus": "CONNECTED", - "cellularSignalStrength": -65, # -65dBm - "wifiStatus": "CONNECTED", - "cloudConnectivity": "ONLINE" - }, - "missionContext": { - "currentMission": { - "missionId": vehicle_data_item.get("mission_id", "UNKNOWN"), - "missionType": vehicle_data_item.get("mission_type", "UNKNOWN"), - "startTime": vehicle_data_item.get("mission_start_time", timestamp_ms), - "estimatedEndTime": vehicle_data_item.get("mission_start_time", timestamp_ms) + 3600000, # 1小时后 - "progress": round(calculate_mission_progress(vehicle_data_item), 1), - "totalMileage": round(to_float_safe(vehicle_data_item.get("total_mileage", 0.0)) or 0.0, 1) - }, - "waypoints": [ - { - "waypointId": f"{vehicle_id}_START", - "latitude": round(to_float_safe(vehicle_data_item.get("start_point", {}).get("latitude", 0.0)) or 0.0, 6), - "longitude": round(to_float_safe(vehicle_data_item.get("start_point", {}).get("longitude", 0.0)) or 0.0, 6), - "status": "COMPLETED" if not vehicle_data_item.get("moving_to_end", True) else "PENDING" - }, - { - "waypointId": f"{vehicle_id}_END", - "latitude": round(to_float_safe(vehicle_data_item.get("end_point", {}).get("latitude", 0.0)) or 0.0, 6), - "longitude": round(to_float_safe(vehicle_data_item.get("end_point", {}).get("longitude", 0.0)) or 0.0, 6), - "status": "COMPLETED" if vehicle_data_item.get("moving_to_end", True) else "PENDING" - } - ] - } - } - - return status_data - -def calculate_mission_progress(vehicle_data: dict[str, Any]) -> float: - """计算任务进度百分比""" - try: - # 获取车辆当前位置 - cur_lat = to_float_safe(vehicle_data.get("latitude")) - cur_lon = to_float_safe(vehicle_data.get("longitude")) - - # 获取起点和终点坐标 - start_point = vehicle_data.get("start_point", {}) - end_point = vehicle_data.get("end_point", {}) - - start_lat = to_float_safe(start_point.get("latitude")) - start_lon = to_float_safe(start_point.get("longitude")) - end_lat = to_float_safe(end_point.get("latitude")) - end_lon = to_float_safe(end_point.get("longitude")) - - if any(v is None for v in [cur_lat, cur_lon, start_lat, start_lon, end_lat, end_lon]): - return 0.0 - - # 确保所有值都不为None，使用断言来帮助类型检查器理解 - assert cur_lat is not None and cur_lon is not None - assert start_lat is not None and start_lon is not None - assert end_lat is not None and end_lon is not None - - # 现在可以安全地转换为float类型 - start_lat_safe = float(start_lat) - start_lon_safe = float(start_lon) - end_lat_safe = float(end_lat) - end_lon_safe = float(end_lon) - cur_lat_safe = float(cur_lat) - cur_lon_safe = float(cur_lon) - - # 计算起点到终点的总距离 - total_distance = calculate_distance_to_target( - {"latitude": start_lat_safe, "longitude": start_lon_safe}, - end_lat_safe, end_lon_safe - ) - - if total_distance == 0: - return 100.0 - - # 计算当前位置到起点的距离 - distance_from_start = calculate_distance_to_target( - {"latitude": cur_lat_safe, "longitude": cur_lon_safe}, - start_lat_safe, start_lon_safe - ) - - # 计算当前位置到终点的距离 - distance_to_end = calculate_distance_to_target( - {"latitude": cur_lat_safe, "longitude": cur_lon_safe}, - end_lat_safe, end_lon_safe - ) - - # 根据移动方向计算进度 - if vehicle_data.get("moving_to_end", True): - # 向终点移动：进度 = (已走距离 / 总距离) * 100 - progress = (distance_from_start / total_distance) * 100.0 - else: - # 向起点移动：进度 = (1 - 距离起点的距离 / 总距离) * 100 + 100 - progress = 100.0 + ((total_distance - distance_from_start) / total_distance) * 100.0 - - # 确保进度在合理范围内 - return max(0.0, min(200.0, progress)) - - except Exception: - return 0.0 - -def filter_vehicle_status_fields(status_data: dict[str, Any], fields: str | None) -> dict[str, Any]: - """根据fields参数过滤车辆状态字段""" - if not fields: - return status_data - - # 解析fields参数 - requested_fields = [f.strip() for f in fields.split(',')] - filtered_data = {} - - # 可用的字段组 - available_fields = { - "vehicleInfo", "operationalStatus", "controlStatus", "motionStatus", - "safetyStatus", "sensorStatus", "batteryStatus", "communicationStatus" - } - - for field in requested_fields: - if field in available_fields and field in status_data: - filtered_data[field] = status_data[field] - - return filtered_data - - -# API 端点实现 - -@app.route('/api/VehicleCommandInfo', methods=['POST']) -def handle_vehicle_command(): - """车辆控制指令接口""" - # 提前定义并注解，避免未绑定与类型未知 - data: RequestVehicleCommandPayload = {} - try: - data = request.get_json(silent=True) or {} - vehicle_id = data.get("vehicleID") - command_type = data.get("commandType", "").upper() - reason = data.get("commandReason", "").upper() - signal_state = data.get("signalState", "").upper() - target_lat = data.get("latitude", None) - target_lon = data.get("longitude", None) - - logging.info(f"🎛️ 收到无人车控制指令: vehicle_id={vehicle_id}, type={command_type}, reason={reason}, signal_state={signal_state}") - logging.info(f"完整请求数据: {data}") - - # 检查 SIGNAL 类型命令必须包含 signalState - if command_type == "SIGNAL" and not signal_state: - logging.warning(f"SIGNAL 类型命令缺少 signalState") - return jsonify({ - "code": 400, - "msg": "SIGNAL command must include signalState", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }), 400 - - # 检查车辆是否存在 - vehicle_state = vehicle_states.get(vehicle_id) - if not vehicle_state: - logging.warning(f"未找到无人车: {vehicle_id}") - return jsonify({ - "code": 404, - "msg": f"Vehicle {vehicle_id} not found", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }), 404 - - # 打印当前车辆状态 - logging.info(f"当前无人车状态: vehicle={vehicle_id}") - vehicle_state.log_state() - - # 检查指令优先级 - check_command = signal_state if command_type == "SIGNAL" else command_type - can_override = vehicle_state.can_be_overridden_by(check_command) - logging.info(f"指令优先级检查: vehicle={vehicle_id}, current_command={str(vehicle_state.current_command)}, new_command={check_command}, can_override={can_override}") - - if not can_override: - logging.warning(f"指令优先级过低: vehicle={vehicle_id}, current_priority={vehicle_state.command_priority}, command={check_command}") - return jsonify({ - "code": 400, - "msg": "Command priority too low", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }), 400 - - # 处理不同类型的指令 - if command_type == "SIGNAL": - # 更新红绿灯状态和指令状态 - vehicle_state.update_command(signal_state, target_lat, target_lon) - logging.info(f"无人车 {vehicle_id} 更新状态: command={command_type}, traffic_light={vehicle_state.traffic_light_state}") - return jsonify({ - "code": 200, - "msg": "Traffic light state updated", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }) - elif command_type in ["ALERT", "WARNING"]: - # 查找当前车辆 - current_vehicle = None - for v in unmanned_vehicle_data: - if v["vehicleNo"] == vehicle_id: - current_vehicle = v - break - - # 执行告警指令直接停车 - logging.info(f"执行{'紧急' if command_type == 'ALERT' else '正常'}制动: vehicle={vehicle_id}") - vehicle_state.current_command = command_type - vehicle_state.command_priority = COMMAND_PRIORITIES.get(command_type, 0) - vehicle_state.is_running = False - vehicle_state.target_speed = 0 - vehicle_state.brake_mode = "emergency" if command_type == "ALERT" else "normal" - vehicle_state.target_lat = target_lat - vehicle_state.target_lon = target_lon - - return jsonify({ - "code": 200, - "msg": "Command executed successfully", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }) - - elif command_type == "RESUME": - logging.info(f"执行恢复指令: vehicle_id={vehicle_id}") - # 检查当前车辆 - current_vehicle = None - for v in unmanned_vehicle_data: - if v["vehicleNo"] == vehicle_id: - current_vehicle = v - break - - # 清除限制性指令 - if vehicle_state.current_command in ["ALERT", "WARNING"]: - logging.info(f"清除限制性指令: vehicle={vehicle_id}") - vehicle_state.current_command = None - vehicle_state.command_priority = 0 - vehicle_state.is_running = True - vehicle_state.target_speed = DEFAULT_VEHICLE_SPEED - vehicle_state.brake_mode = None - vehicle_state.target_lat = None - vehicle_state.target_lon = None - - # 更新车辆运行状态 - vehicle_state.is_running = vehicle_state.can_move() - if vehicle_state.is_running and current_vehicle: - logging.info(f"无人车 {vehicle_id} 恢复运行") - - # 记录状态变化但不更新指令 - vehicle_state.command_reason = reason - vehicle_state.last_command_time = time.time() - - logging.info(f"Vehicle {vehicle_id} state updated: running={vehicle_state.is_running}, " - f"command={vehicle_state.current_command}, traffic_light={vehicle_state.traffic_light_state}, " - f"reason={reason}, priority={vehicle_state.command_priority}, " - f"target_speed={vehicle_state.target_speed}, brake_mode={vehicle_state.brake_mode}") - - return jsonify({ - "code": 200, - "msg": "Command executed successfully", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }) - - return jsonify({ - "code": 200, - "msg": "Command executed successfully", - "transId": data.get("transId"), - "timestamp": int(time.time() * 1000) - }) - - except Exception as e: - logging.error(f"Error handling vehicle command: {str(e)}") - return jsonify({ - "code": 500, - "msg": str(e), - "transId": data.get("transId", ""), - "timestamp": int(time.time() * 1000) - }), 500 - - - -@app.route('/api/v1/vehicles//status', methods=['GET', 'OPTIONS']) -def get_vehicle_status_universal(vehicle_id): - """通用无人车状态上报接口 - 符合universal_autonomous_vehicle_api规范""" - if request.method == 'OPTIONS': - return '', 204 - - logging.info(f"🔍 收到无人车状态查询请求: vehicleId={vehicle_id}") - - # 检查vehicleId格式（字母数字 3~20 位） - if not vehicle_id or len(vehicle_id) < 3 or len(vehicle_id) > 20: - return jsonify({ - "code": 400, - "message": "Bad Request", - "timestamp": int(time.time() * 1000), - "error": { - "type": "VALIDATION_ERROR", - "details": "vehicleId must be 3-20 alphanumeric characters", - "field": "vehicleId" - } - }), 400 - - # 检查车辆是否存在 - vehicle_exists = False - for v in unmanned_vehicle_data: - if v["vehicleNo"] == vehicle_id: - vehicle_exists = True - break - - if not vehicle_exists: - return jsonify({ - "code": 404, - "message": "Not Found", - "timestamp": int(time.time() * 1000), - "error": { - "type": "RESOURCE_NOT_FOUND", - "details": f"Vehicle {vehicle_id} not found", - "field": "vehicleId" - } - }), 404 - - try: - # 更新车辆位置数据 - current_time = time.time() - global last_unmanned_vehicle_update_time - elapsed_time = current_time - last_unmanned_vehicle_update_time - if elapsed_time >= UPDATE_INTERVAL: - for vehicle in unmanned_vehicle_data: - update_vehicle_position(vehicle, UPDATE_INTERVAL) - vehicle["time"] = int(current_time * 1000) - last_unmanned_vehicle_update_time = current_time - - # 生成完整的车辆状态数据 - status_data = generate_comprehensive_vehicle_status(vehicle_id) - if not status_data: - return jsonify({ - "code": 500, - "message": "Internal Server Error", - "timestamp": int(time.time() * 1000), - "error": { - "type": "DATA_ERROR", - "details": "Failed to generate vehicle status data", - "field": None - } - }), 500 - - # 处理fields查询参数 - fields = request.args.get('fields') - if fields: - status_data = filter_vehicle_status_fields(status_data, fields) - - # 返回成功响应 - battery_level = status_data.get('batteryStatus', {}).get('mainBattery', {}).get('chargeLevel', '未知') - logging.info(f"🚗 返回无人车状态数据: vehicleId={vehicle_id}, 电量={battery_level}%") - return jsonify({ - "code": 200, - "message": "success", - "timestamp": int(time.time() * 1000), - "data": status_data - }) - - except Exception as e: - logging.error(f"Error in get_vehicle_status_universal: {str(e)}") - return jsonify({ - "code": 500, - "message": "Internal Server Error", - "timestamp": int(time.time() * 1000), - "error": { - "type": "SERVER_ERROR", - "details": str(e), - "field": None - } - }), 500 - - -# 设置CORS -@app.after_request -def add_cors_headers(response): - response.headers['Access-Control-Allow-Origin'] = '*' - response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization' - response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS' - response.headers['Access-Control-Max-Age'] = '3600' - return response - +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +无人车厂商平台模拟服务 +Unmanned Vehicle Manufacturer Platform Mock Server + +提供无人车位置上报、状态查询、控制指令等API接口 +模拟真实的无人车厂商平台与机场管理系统的交互 +""" + +from flask import Flask, jsonify, request +import time +import math +import logging +import os +from typing import Any, Literal, final, TypedDict + +# 导入统一的日志配置 +import sys +sys.path.append(os.path.dirname(os.path.abspath(__file__))) +from logging_config import setup_logger + +# 设置当前模块的日志记录器 +logger = setup_logger( + name='mock_unmanned_vehicle', + log_file='logs/mock_unmanned_vehicle.log', + max_bytes=10*1024*1024, # 10MB + backup_count=5 +) + +app = Flask(__name__) + +# 为位置与移动对象定义类型，避免使用 Any +class Point(TypedDict): + latitude: float + longitude: float + +# 必填字段 +class MovingObjectRequired(TypedDict): + latitude: float + longitude: float + time: int + speed: float + moving_to_end: bool + start_point: Point + end_point: Point + +# 可选字段（total=False） +class MovingObject(MovingObjectRequired, total=False): + vehicleNo: str + wait_until: float + direction: float + +# 地球半径（米） +EARTH_RADIUS = 6378137.0 + +# 指令优先级配置 +COMMAND_PRIORITIES = { + "ALERT": 5, + "WARNING": 3, + "PARKING": 2, + "RESUME": 1 +} + +# 距离配置（米） +DIST_300M = 300 +DIST_200M = 200 +DIST_150M = 150 +DIST_100M = 100 +DIST_50M = 50 + +# 时间配置（秒） +WAIT_TIME_AFTER_RETURN = 0.1 # 返回起点后的等待时间 +UPDATE_INTERVAL = 1.0 # 位置更新间隔, 默认 1 秒 + +# 无人车路径配置 +# 无人车A 鲁B567 路径 +UNMANNED_A_START = {"longitude": 120.083084, "latitude": 36.369696} # 无人车A起点 +UNMANNED_A_END = {"longitude": 120.084637, "latitude": 36.365617} # 无人车A终点 + +# 无人车B 鲁B579 路径 +UNMANNED_B_START = {"longitude": 120.086965, "latitude": 36.368599} # 无人车B起点 +UNMANNED_B_END = {"longitude": 120.086263, "latitude": 36.370484} # 无人车B终点 + +# 车辆配置 +DEFAULT_VEHICLE_SPEED = 30.0 # km/h，默认速度 +EMERGENCY_BRAKE_DECELERATION = 0.8 # 紧急制动减速度 (每次更新减速 80%) +NORMAL_BRAKE_DECELERATION = 0.2 # 正常制动减速度 (每次更新减速 20%) + +# 认证相关配置 +AUTH_USERNAME = "dianxin" +AUTH_PASSWORD = "dianxin@123" +AUTH_TOKEN = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE3MzI3ODMwOTAsInVzZXJuYW1lIjoiYWRtaW4ifQ.y9feEL_9NT8UzED9NNkb0Ln6C-PBoufiSHWobWe5vWY" + +def meters_to_degrees(meters: float, latitude: float) -> tuple[float, float]: + """ + 将米转换为度，考虑纬度的影响 + """ + # 纬度方向：1度 = 111,319.9米 + meters_per_deg_lat: float = 111319.9 + # 经度方向：1度 = 111,319.9 * cos(latitude)米 + meters_per_deg_lon: float = meters_per_deg_lat * math.cos(math.radians(latitude)) + return meters / meters_per_deg_lat, meters / meters_per_deg_lon + +def to_float_safe(x: object) -> float | None: + """ + 安全将值转换为 float: + - 允许 int/float + - 允许非布尔的数字字符串 + - 允许 dict，优先取 'value'、'longitude'、'latitude' 等常见键 + - 无法解析则返回 None + """ + # 排除布尔值（bool 是 int 的子类，但不应当被视为数值） + if isinstance(x, bool): + return None + if isinstance(x, (int, float)): + return float(x) + if isinstance(x, str): + try: + return float(x.strip()) + except Exception: + return None + if isinstance(x, dict): + for key in ("value", "longitude", "latitude"): + if key in x: + return to_float_safe(x[key]) + return None + return None + +def calculate_distance_to_target(vehicle: dict[str, float], target_lat: float, target_lon: float) -> float: + """计算车辆到目标位的距离（米）""" + # 转换为弧度 + lat1_rad = math.radians(vehicle["latitude"]) + lon1_rad = math.radians(vehicle["longitude"]) + lat2_rad = math.radians(target_lat) + lon2_rad = math.radians(target_lon) + + dlat = lat2_rad - lat1_rad + dlon = lon2_rad - lon1_rad + + # 使用 Haversine 公式计算距离 + a = math.sin(dlat/2) * math.sin(dlat/2) + \ + math.cos(lat1_rad) * math.cos(lat2_rad) * \ + math.sin(dlon/2) * math.sin(dlon/2) + c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a)) + return EARTH_RADIUS * c + +def update_position_with_vector( + obj: dict[str, Any], + target_point: "Point", + start_point: "Point", + speed: float, + elapsed_time: float, + return_to_start: bool = False +) -> bool: + """ + 基于向量的位置更新逻辑 + + Args: + obj: 需要更新位置的对象（车辆） + target_point: 目标点坐标 {"latitude": float, "longitude": float} + start_point: 起点坐标 {"latitude": float, "longitude": float} + speed: 移动速度（km/h） + elapsed_time: 经过的时间（秒） + return_to_start: 是否在到达目标点时返回起点 + """ + # 检查是否在等待状态 + if "wait_until" not in obj: + obj["wait_until"] = 0.0 + + current_time = time.time() + # 基于类型安全读取 wait_until，避免 float 与 object 比较 + _wu = to_float_safe(obj.get("wait_until")) + wait_until = 0.0 if _wu is None else float(_wu) + if current_time < wait_until: + return False # 还在等待中，不更新位置 + + # 计算这一步要移动的距离（米）并转换为经纬度 + speed_mps = float(speed) * 1000.0 / 3600.0 + distance = speed_mps * float(elapsed_time) + + # 累计里程到总里程中 + if "total_mileage" not in obj: + obj["total_mileage"] = 0.0 + obj["total_mileage"] = to_float_safe(obj.get("total_mileage", 0.0)) or 0.0 + obj["total_mileage"] += distance + + # 安全读取并转换当前位置与目标点/起点经纬度为 float + cur_lat = to_float_safe(obj.get("latitude")) + cur_lon = to_float_safe(obj.get("longitude")) + tgt_lat = to_float_safe(target_point.get("latitude")) + tgt_lon = to_float_safe(target_point.get("longitude")) + if cur_lat is None or cur_lon is None or tgt_lat is None or tgt_lon is None: + # 无法解析经纬度，跳过更新 + return False + + # 将移动距离转换为经纬度变化量 + move_dlat, move_dlon = meters_to_degrees(distance, cur_lat) + + # 将5米的判断距离转换为经纬度（优化：从15米减少到5米，减少颤抖） + check_dlat, check_dlon = meters_to_degrees(5.0, cur_lat) + + # 计算当前位置到终点的向量（经纬度空间） + vector_lat = float(tgt_lat) - float(cur_lat) + vector_lon = float(tgt_lon) - float(cur_lon) + + # 计算向量长度（经纬度空间） + vector_length = math.sqrt(vector_lat * vector_lat + vector_lon * vector_lon) + + # 检查是否到达终点 + if vector_length <= math.sqrt(check_dlat * check_dlat + check_dlon * check_dlon): + # 精确设置到目标点位置，避免位置偏差 + obj["latitude"] = float(tgt_lat) + obj["longitude"] = float(tgt_lon) + + if return_to_start: + # 返回起点并设置等待时间 + obj["wait_until"] = current_time + WAIT_TIME_AFTER_RETURN + print(f"无人车 {obj.get('vehicleNo')} 返回起点，等待{WAIT_TIME_AFTER_RETURN}秒后继续") + else: + # 设置短暂等待时间，避免立即切换方向导致的颤抖 + obj["wait_until"] = current_time + 0.1 # 0.1秒的短暂等待 + print(f"无人车 {obj.get('vehicleNo')} 到达目标点，等待0.1秒后切换方向") + return True # 表示已到达终点 + else: + # 归一化方向向量 + unit_lat = vector_lat / vector_length + unit_lon = vector_lon / vector_length + + # 计算这一步的位置变化 + dlat = move_dlat * unit_lat + dlon = move_dlon * unit_lon + + # 更新位置（写回为 float） + obj["latitude"] = float(cur_lat + dlat) + obj["longitude"] = float(cur_lon + dlon) + return False # 表示正在移动中 + +def update_object_position(obj: dict[str, Any], elapsed_time: float) -> None: + """统一的位置更新函数 - 用于无人车的往复运动""" + # 获取对象类型和标识 + obj_id = obj.get("vehicleNo") + + # 根据移动方向确定目标点 + if obj["moving_to_end"]: + target_raw = obj["end_point"] + start_raw = obj["start_point"] + else: + target_raw = obj["start_point"] + start_raw = obj["end_point"] + + # 构造类型安全的 Point + t_lat = to_float_safe(target_raw.get("latitude")) if isinstance(target_raw, dict) else None + t_lon = to_float_safe(target_raw.get("longitude")) if isinstance(target_raw, dict) else None + s_lat = to_float_safe(start_raw.get("latitude")) if isinstance(start_raw, dict) else None + s_lon = to_float_safe(start_raw.get("longitude")) if isinstance(start_raw, dict) else None + if t_lat is None or t_lon is None or s_lat is None or s_lon is None: + # 无法解析坐标，直接更新时间戳后返回 + obj["time"] = int(time.time() * 1000) + return + + target_point: Point = {"latitude": float(t_lat), "longitude": float(t_lon)} + start_point: Point = {"latitude": float(s_lat), "longitude": float(s_lon)} + + # 读取速度为 float + spd = to_float_safe(obj.get("speed")) + if spd is None: + spd = 0.0 + + # 更新位置 + reached_target = update_position_with_vector( + obj, + target_point, + start_point, + float(spd), + float(elapsed_time), + return_to_start=False + ) + + if reached_target: + # 切换移动方向 + obj["moving_to_end"] = not obj["moving_to_end"] + target_name = "终点" if not obj["moving_to_end"] else "起点" + print(f"无人车 {obj_id} 到达{'终点' if obj['moving_to_end'] else '起点'}，开始前往{target_name}") + + # 更新时间戳 + obj["time"] = int(time.time() * 1000) + +def update_vehicle_position(vehicle: dict[str, Any], elapsed_time: float) -> None: + """更新无人车位置 - 使用统一的位置更新函数""" + update_object_position(vehicle, elapsed_time) + +# 无人车数据配置 +current_timestamp = time.time() +unmanned_vehicle_data = [ + { + "vehicleNo": "鲁B567", # 无人车A + "longitude": UNMANNED_A_START["longitude"], + "latitude": UNMANNED_A_START["latitude"], + "time": int(current_timestamp * 1000), + "direction": 90.0, # 90度 + "speed": 25.0, # 25km/h + "start_point": UNMANNED_A_START, # 起点 + "end_point": UNMANNED_A_END, # 终点 + "moving_to_end": True, # 当前是否向终点移动 + # 任务和里程相关字段 + "total_mileage": 0.0, # 总里程（米） + "mission_start_time": int(current_timestamp * 1000), # 任务开始时间 + "mission_id": "MISSION_A_001", # 任务ID + "mission_type": "PATROL_TRANSPORT" # 任务类型：巡逻运输 + }, + { + "vehicleNo": "鲁B579", # 无人车B + "longitude": UNMANNED_B_START["longitude"], + "latitude": UNMANNED_B_START["latitude"], + "time": int(current_timestamp * 1000), + "direction": 270.0, # 270度 + "speed": 25.0, # 25km/h + "start_point": UNMANNED_B_START, # 起点 + "end_point": UNMANNED_B_END, # 终点 + "moving_to_end": True, # 当前是否向终点移动 + # 任务和里程相关字段 + "total_mileage": 0.0, # 总里程（米） + "mission_start_time": int(current_timestamp * 1000), # 任务开始时间 + "mission_id": "MISSION_B_001", # 任务ID + "mission_type": "CARGO_TRANSPORT" # 任务类型：货物运输 + } +] + +# 添加车辆状态类 +@final +class VehicleState: + vehicle_no: str + is_running: bool + current_command: str | None + command_priority: int + target_speed: float + brake_mode: Literal['emergency', 'normal'] | None + command_reason: str | None + last_command_time: float + traffic_light_state: str | None + target_lat: float | None + target_lon: float | None + is_traffic_light_stop: bool + + def __init__(self, vehicle_no: str) -> None: + self.vehicle_no = vehicle_no + self.is_running = True # 运行状态 + self.current_command = None # 当前执行的指令 + self.command_priority = 0 # 当前指令优先级 + self.target_speed = float(DEFAULT_VEHICLE_SPEED) # 目标速度 + self.brake_mode = None # 制动模式：'emergency' 或 'normal' + self.command_reason = None # 指令原因 + self.last_command_time = float(time.time()) # 最后一次指令时间 + self.traffic_light_state = None # 当前红绿灯状态 + self.target_lat = None # 目标纬度 + self.target_lon = None # 目标经度 + self.is_traffic_light_stop = False # 是否因红灯停车 + self.remoteControlActive = False # 远程控制是否激活 + + def can_be_overridden_by(self, command_type: str) -> bool: + """判断当前指令是否可以被新指令覆盖""" + # 红绿灯信号不参与指令优先级判断 + if command_type in ["RED", "GREEN", "YELLOW"]: + return True + + # 类型安全获取优先级，避免 Unknown | None 作为 dict key + new_priority = COMMAND_PRIORITIES.get(str(command_type), 0) + current_key = self.current_command if isinstance(self.current_command, str) else None + current_priority = COMMAND_PRIORITIES.get(current_key or "", 0) + + # 相同类型的指令可以覆盖（因为需要持续发送） + if isinstance(self.current_command, str) and command_type == self.current_command: + return True + + # ALERT 指令可以被 RESUME 解除 + if self.current_command == "ALERT": + return command_type == "RESUME" + + # WARNING 指令可以被 RESUME 或相同及更高优先级指令覆盖 + if self.current_command == "WARNING": + return command_type == "RESUME" or new_priority >= current_priority + + # 其他情况下，相同或更高优先级可以覆盖 + return new_priority >= current_priority + + def update_command(self, command_type: str, target_lat: float | None = None, target_lon: float | None = None) -> None: + """更新指令状态""" + # 更新目标位置 + if target_lat is not None and target_lon is not None: + self.target_lat = target_lat + self.target_lon = target_lon + + # 如果是红绿灯状态，只更新状态，不影响其他指令 + if command_type in ["RED", "GREEN", "YELLOW"]: + old_state = self.traffic_light_state + self.traffic_light_state = command_type + + # 更新红灯停车状态 + if command_type == "RED": + self.is_traffic_light_stop = True + elif command_type in ["GREEN", "YELLOW"]: + self.is_traffic_light_stop = False + + print(f"无人车 {self.vehicle_no} 收到红绿灯信号: {command_type} (原状态: {old_state})") + else: + # 其他指令正常更新 + self.current_command = command_type + self.command_priority = COMMAND_PRIORITIES.get(command_type, 0) + + # RESUME 指令不清除红绿灯状态，红绿灯状态由信号灯独立控制 + if command_type == "RESUME": + # 清除其他状态 + self.brake_mode = None + self.command_reason = None + + # 更新运行状态 + self.is_running = self.can_move() + print(f"更新无人车 {self.vehicle_no} 状态: command={self.current_command}, traffic_light={self.traffic_light_state}, priority={self.command_priority}, is_running={self.is_running}") + + def can_move(self) -> bool: + """检查车辆是否可以移动""" + # 如果有告警指令，不能移动 + if self.current_command == "ALERT": + return False + + # 如果有预警指令，不能移动 + if self.current_command == "WARNING": + return False + + # 如果有安全停靠指令，不能移动 + if self.current_command == "PARKING": + return False + + # 如果是红灯停车状态，不能移动 + if self.is_traffic_light_stop: + return False + + # 其他情况可以移动 + return True + + def log_state(self) -> None: + """记录当前车辆状态""" + print(f""" +无人车状态: +- 车辆编号: {self.vehicle_no} +- 运行状态: {'运行中' if self.is_running else '已停止'} +- 当前指令: {self.current_command} +- 红绿灯状态: {self.traffic_light_state} +- 红灯停车: {'是' if self.is_traffic_light_stop else '否'} +- 指令优先级: {self.command_priority} +- 目标速度: {self.target_speed} +- 制动模式: {self.brake_mode} +- 指令原因: {self.command_reason} +- 目标位置: ({self.target_lat}, {self.target_lon}) +""") + +# 车辆状态管理 +vehicle_states = {} +for vehicle in unmanned_vehicle_data: + vehicle_states[str(vehicle["vehicleNo"])] = VehicleState(str(vehicle["vehicleNo"])) + +# 记录上次更新时间 +last_unmanned_vehicle_update_time = time.time() + +def check_auth() -> bool: + """检查认证""" + auth_header = request.headers.get('Authorization') + + if not auth_header: + print("认证失败: 缺少Authorization头") + return False + + # 直接比较完整的Authorization头 + result = auth_header == AUTH_TOKEN + if not result: + print(f"认证失败: token不匹配") + print(f"收到的token: {auth_header}") + print(f"期望的token: {AUTH_TOKEN}") + return result + +class RequestVehicleCommandPayload(TypedDict, total=False): + vehicleID: str + commandType: str + commandReason: str + signalState: str + latitude: float + longitude: float + transId: str + + +# 生成符合API规范的完整车辆状态数据 +def generate_comprehensive_vehicle_status(vehicle_id: str) -> dict[str, Any]: + """生成符合universal_autonomous_vehicle_api规范的完整车辆状态数据""" + current_time = time.time() + timestamp_ms = int(current_time * 1000) + + # 获取车辆基础数据 + vehicle_data_item = None + for v in unmanned_vehicle_data: + if v["vehicleNo"] == vehicle_id: + vehicle_data_item = v + break + + if not vehicle_data_item: + return {} + + # 获取车辆状态 + vehicle_state = vehicle_states.get(vehicle_id) + + # 基础位置和运动数据 + lat_val = to_float_safe(vehicle_data_item.get("latitude")) + lon_val = to_float_safe(vehicle_data_item.get("longitude")) + speed_val = to_float_safe(vehicle_data_item.get("speed", 0)) + direction_val = to_float_safe(vehicle_data_item.get("direction", 0)) + + # 生成完整状态数据 + status_data = { + "vehicleInfo": { + "vehicleId": vehicle_id + }, + "operationalStatus": { + "powerStatus": "ON", + "systemHealth": "HEALTHY" if vehicle_state and vehicle_state.is_running else "DEGRADED", + "operationalMode": "AUTONOMOUS", + "emergencyStatus": "CRITICAL" if vehicle_state and vehicle_state.current_command == "ALERT" + else "WARNING" if vehicle_state and vehicle_state.current_command == "WARNING" + else "NORMAL", + "lastHeartbeat": timestamp_ms + }, + "controlStatus": { + "controlMode": "AUTONOMOUS", + "controlAuthority": "SYSTEM", + "remoteControlActive": vehicle_state.remoteControlActive if vehicle_state else False + }, + "motionStatus": { + "position": { + "latitude": round(lat_val, 6) if lat_val is not None else 0.0, + "longitude": round(lon_val, 6) if lon_val is not None else 0.0 + }, + "velocity": { + "speed": round(speed_val * 1000 / 3600, 2) if speed_val is not None else 0.0, # 转换为 m/s + "direction": round(math.radians(direction_val), 4) if direction_val is not None else 0.0 # 转换为弧度 + } + }, + "safetyStatus": { + "collisionAvoidanceActive": vehicle_state.is_running if vehicle_state else True, + "emergencyBrakingReady": True, + "pathPlanningStatus": "ACTIVE" if vehicle_state and vehicle_state.is_running else "INACTIVE", + "obstacleDetectionStatus": "ACTIVE" if vehicle_state and vehicle_state.is_running else "INACTIVE", + "minimumRiskManeuverTriggered": vehicle_state.current_command == "ALERT" if vehicle_state else False + }, + "sensorStatus": { + "gps": { + "status": "ACTIVE", + "accuracy": 0.5, # 0.5米精度 + "lastUpdate": timestamp_ms + } + }, + "batteryStatus": { + "mainBattery": { + "chargeLevel": 85.5, # 85.5% + "voltage": 48.2, # 48.2V + "current": -15.3, # -15.3A (放电) + "temperature": 35.2, # 35.2°C + "chargingStatus": "DISCHARGING" + } + }, + "communicationStatus": { + "v2xStatus": "CONNECTED", + "cellularSignalStrength": -65, # -65dBm + "wifiStatus": "CONNECTED", + "cloudConnectivity": "ONLINE" + }, + "missionContext": { + "currentMission": { + "missionId": vehicle_data_item.get("mission_id", "UNKNOWN"), + "missionType": vehicle_data_item.get("mission_type", "UNKNOWN"), + "startTime": vehicle_data_item.get("mission_start_time", timestamp_ms), + "estimatedEndTime": vehicle_data_item.get("mission_start_time", timestamp_ms) + 3600000, # 1小时后 + "progress": round(calculate_mission_progress(vehicle_data_item), 1), + "totalMileage": round(to_float_safe(vehicle_data_item.get("total_mileage", 0.0)) or 0.0, 1) + }, + "waypoints": [ + { + "waypointId": f"{vehicle_id}_START", + "latitude": round(to_float_safe(vehicle_data_item.get("start_point", {}).get("latitude", 0.0)) or 0.0, 6), + "longitude": round(to_float_safe(vehicle_data_item.get("start_point", {}).get("longitude", 0.0)) or 0.0, 6), + "status": "COMPLETED" if not vehicle_data_item.get("moving_to_end", True) else "PENDING" + }, + { + "waypointId": f"{vehicle_id}_END", + "latitude": round(to_float_safe(vehicle_data_item.get("end_point", {}).get("latitude", 0.0)) or 0.0, 6), + "longitude": round(to_float_safe(vehicle_data_item.get("end_point", {}).get("longitude", 0.0)) or 0.0, 6), + "status": "COMPLETED" if vehicle_data_item.get("moving_to_end", True) else "PENDING" + } + ] + } + } + + return status_data + +def calculate_mission_progress(vehicle_data: dict[str, Any]) -> float: + """计算任务进度百分比""" + try: + # 获取车辆当前位置 + cur_lat = to_float_safe(vehicle_data.get("latitude")) + cur_lon = to_float_safe(vehicle_data.get("longitude")) + + # 获取起点和终点坐标 + start_point = vehicle_data.get("start_point", {}) + end_point = vehicle_data.get("end_point", {}) + + start_lat = to_float_safe(start_point.get("latitude")) + start_lon = to_float_safe(start_point.get("longitude")) + end_lat = to_float_safe(end_point.get("latitude")) + end_lon = to_float_safe(end_point.get("longitude")) + + if any(v is None for v in [cur_lat, cur_lon, start_lat, start_lon, end_lat, end_lon]): + return 0.0 + + # 确保所有值都不为None，使用断言来帮助类型检查器理解 + assert cur_lat is not None and cur_lon is not None + assert start_lat is not None and start_lon is not None + assert end_lat is not None and end_lon is not None + + # 现在可以安全地转换为float类型 + start_lat_safe = float(start_lat) + start_lon_safe = float(start_lon) + end_lat_safe = float(end_lat) + end_lon_safe = float(end_lon) + cur_lat_safe = float(cur_lat) + cur_lon_safe = float(cur_lon) + + # 计算起点到终点的总距离 + total_distance = calculate_distance_to_target( + {"latitude": start_lat_safe, "longitude": start_lon_safe}, + end_lat_safe, end_lon_safe + ) + + if total_distance == 0: + return 100.0 + + # 计算当前位置到起点的距离 + distance_from_start = calculate_distance_to_target( + {"latitude": cur_lat_safe, "longitude": cur_lon_safe}, + start_lat_safe, start_lon_safe + ) + + # 计算当前位置到终点的距离 + distance_to_end = calculate_distance_to_target( + {"latitude": cur_lat_safe, "longitude": cur_lon_safe}, + end_lat_safe, end_lon_safe + ) + + # 根据移动方向计算进度 + if vehicle_data.get("moving_to_end", True): + # 向终点移动：进度 = (已走距离 / 总距离) * 100 + progress = (distance_from_start / total_distance) * 100.0 + else: + # 向起点移动：进度 = (1 - 距离起点的距离 / 总距离) * 100 + 100 + progress = 100.0 + ((total_distance - distance_from_start) / total_distance) * 100.0 + + # 确保进度在合理范围内 + return max(0.0, min(200.0, progress)) + + except Exception: + return 0.0 + +def filter_vehicle_status_fields(status_data: dict[str, Any], fields: str | None) -> dict[str, Any]: + """根据fields参数过滤车辆状态字段""" + if not fields: + return status_data + + # 解析fields参数 + requested_fields = [f.strip() for f in fields.split(',')] + filtered_data = {} + + # 可用的字段组 + available_fields = { + "vehicleInfo", "operationalStatus", "controlStatus", "motionStatus", + "safetyStatus", "sensorStatus", "batteryStatus", "communicationStatus" + } + + for field in requested_fields: + if field in available_fields and field in status_data: + filtered_data[field] = status_data[field] + + return filtered_data + + +# API 端点实现 + +@app.route('/api/VehicleCommandInfo', methods=['POST']) +def handle_vehicle_command(): + """车辆控制指令接口""" + # 提前定义并注解，避免未绑定与类型未知 + data: RequestVehicleCommandPayload = {} + try: + data = request.get_json(silent=True) or {} + vehicle_id = data.get("vehicleID") + command_type = data.get("commandType", "").upper() + reason = data.get("commandReason", "").upper() + signal_state = data.get("signalState", "").upper() + target_lat = data.get("latitude", None) + target_lon = data.get("longitude", None) + + logger.info(f"🎛️ 收到无人车控制指令: vehicle_id={vehicle_id}, type={command_type}, reason={reason}, signal_state={signal_state}") + logger.info(f"完整请求数据: {data}") + + # 检查 SIGNAL 类型命令必须包含 signalState + if command_type == "SIGNAL" and not signal_state: + logger.warning(f"SIGNAL 类型命令缺少 signalState") + return jsonify({ + "code": 400, + "msg": "SIGNAL command must include signalState", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }), 400 + + # 检查车辆是否存在 + vehicle_state = vehicle_states.get(vehicle_id) + if not vehicle_state: + logger.warning(f"未找到无人车: {vehicle_id}") + return jsonify({ + "code": 404, + "msg": f"Vehicle {vehicle_id} not found", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }), 404 + + # 打印当前车辆状态 + logger.info(f"当前无人车状态: vehicle={vehicle_id}") + vehicle_state.log_state() + + # 检查指令优先级 + check_command = signal_state if command_type == "SIGNAL" else command_type + can_override = vehicle_state.can_be_overridden_by(check_command) + logger.info(f"指令优先级检查: vehicle={vehicle_id}, current_command={str(vehicle_state.current_command)}, new_command={check_command}, can_override={can_override}") + + if not can_override: + logger.warning(f"指令优先级过低: vehicle={vehicle_id}, current_priority={vehicle_state.command_priority}, command={check_command}") + return jsonify({ + "code": 400, + "msg": "Command priority too low", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }), 400 + + # 处理不同类型的指令 + if command_type == "SIGNAL": + # 更新红绿灯状态和指令状态 + vehicle_state.update_command(signal_state, target_lat, target_lon) + logger.info(f"无人车 {vehicle_id} 更新状态: command={command_type}, traffic_light={vehicle_state.traffic_light_state}") + return jsonify({ + "code": 200, + "msg": "Traffic light state updated", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }) + elif command_type in ["ALERT", "WARNING"]: + # 查找当前车辆 + current_vehicle = None + for v in unmanned_vehicle_data: + if v["vehicleNo"] == vehicle_id: + current_vehicle = v + break + + # 执行告警指令直接停车 + logger.info(f"执行{'紧急' if command_type == 'ALERT' else '正常'}制动: vehicle={vehicle_id}") + vehicle_state.current_command = command_type + vehicle_state.command_priority = COMMAND_PRIORITIES.get(command_type, 0) + vehicle_state.is_running = False + vehicle_state.target_speed = 0 + vehicle_state.brake_mode = "emergency" if command_type == "ALERT" else "normal" + vehicle_state.target_lat = target_lat + vehicle_state.target_lon = target_lon + + return jsonify({ + "code": 200, + "msg": "Command executed successfully", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }) + + elif command_type == "RESUME": + logger.info(f"执行恢复指令: vehicle_id={vehicle_id}") + # 检查当前车辆 + current_vehicle = None + for v in unmanned_vehicle_data: + if v["vehicleNo"] == vehicle_id: + current_vehicle = v + break + + # 清除限制性指令 + if vehicle_state.current_command in ["ALERT", "WARNING"]: + logger.info(f"清除限制性指令: vehicle={vehicle_id}") + vehicle_state.current_command = None + vehicle_state.command_priority = 0 + vehicle_state.is_running = True + vehicle_state.target_speed = DEFAULT_VEHICLE_SPEED + vehicle_state.brake_mode = None + vehicle_state.target_lat = None + vehicle_state.target_lon = None + + # 更新车辆运行状态 + vehicle_state.is_running = vehicle_state.can_move() + if vehicle_state.is_running and current_vehicle: + logger.info(f"无人车 {vehicle_id} 恢复运行") + + # 记录状态变化但不更新指令 + vehicle_state.command_reason = reason + vehicle_state.last_command_time = time.time() + + logger.info(f"Vehicle {vehicle_id} state updated: running={vehicle_state.is_running}, " + f"command={vehicle_state.current_command}, traffic_light={vehicle_state.traffic_light_state}, " + f"reason={reason}, priority={vehicle_state.command_priority}, " + f"target_speed={vehicle_state.target_speed}, brake_mode={vehicle_state.brake_mode}") + + return jsonify({ + "code": 200, + "msg": "Command executed successfully", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }) + + return jsonify({ + "code": 200, + "msg": "Command executed successfully", + "transId": data.get("transId"), + "timestamp": int(time.time() * 1000) + }) + + except Exception as e: + logger.error(f"Error handling vehicle command: {str(e)}") + return jsonify({ + "code": 500, + "msg": str(e), + "transId": data.get("transId", ""), + "timestamp": int(time.time() * 1000) + }), 500 + + + +@app.route('/api/v1/vehicles//status', methods=['GET', 'OPTIONS']) +def get_vehicle_status_universal(vehicle_id): + """通用无人车状态上报接口 - 符合universal_autonomous_vehicle_api规范""" + if request.method == 'OPTIONS': + return '', 204 + + logger.info(f"🔍 收到无人车状态查询请求: vehicleId={vehicle_id}") + + # 检查vehicleId格式（字母数字 3~20 位） + if not vehicle_id or len(vehicle_id) < 3 or len(vehicle_id) > 20: + return jsonify({ + "code": 400, + "message": "Bad Request", + "timestamp": int(time.time() * 1000), + "error": { + "type": "VALIDATION_ERROR", + "details": "vehicleId must be 3-20 alphanumeric characters", + "field": "vehicleId" + } + }), 400 + + # 检查车辆是否存在 + vehicle_exists = False + for v in unmanned_vehicle_data: + if v["vehicleNo"] == vehicle_id: + vehicle_exists = True + break + + if not vehicle_exists: + return jsonify({ + "code": 404, + "message": "Not Found", + "timestamp": int(time.time() * 1000), + "error": { + "type": "RESOURCE_NOT_FOUND", + "details": f"Vehicle {vehicle_id} not found", + "field": "vehicleId" + } + }), 404 + + try: + # 更新车辆位置数据 + current_time = time.time() + global last_unmanned_vehicle_update_time + elapsed_time = current_time - last_unmanned_vehicle_update_time + if elapsed_time >= UPDATE_INTERVAL: + for vehicle in unmanned_vehicle_data: + update_vehicle_position(vehicle, UPDATE_INTERVAL) + vehicle["time"] = int(current_time * 1000) + last_unmanned_vehicle_update_time = current_time + + # 生成完整的车辆状态数据 + status_data = generate_comprehensive_vehicle_status(vehicle_id) + if not status_data: + return jsonify({ + "code": 500, + "message": "Internal Server Error", + "timestamp": int(time.time() * 1000), + "error": { + "type": "DATA_ERROR", + "details": "Failed to generate vehicle status data", + "field": None + } + }), 500 + + # 处理fields查询参数 + fields = request.args.get('fields') + if fields: + status_data = filter_vehicle_status_fields(status_data, fields) + + # 返回成功响应 + battery_level = status_data.get('batteryStatus', {}).get('mainBattery', {}).get('chargeLevel', '未知') + logger.info(f"🚗 返回无人车状态数据: vehicleId={vehicle_id}, 电量={battery_level}%") + return jsonify({ + "code": 200, + "message": "success", + "timestamp": int(time.time() * 1000), + "data": status_data + }) + + except Exception as e: + logger.error(f"Error in get_vehicle_status_universal: {str(e)}") + return jsonify({ + "code": 500, + "message": "Internal Server Error", + "timestamp": int(time.time() * 1000), + "error": { + "type": "SERVER_ERROR", + "details": str(e), + "field": None + } + }), 500 + + +# 设置CORS +@app.after_request +def add_cors_headers(response): + response.headers['Access-Control-Allow-Origin'] = '*' + response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization' + response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS' + response.headers['Access-Control-Max-Age'] = '3600' + return response + if __name__ == '__main__': - print("🚗 无人车厂商平台模拟服务启动中...") - print("📍 服务端点:") - print(" - 车辆控制指令: POST /api/VehicleCommandInfo") - print(" - 通用状态接口: GET /api/v1/vehicles//status") - print(" - 车辆状态获取: GET /openApi/getVehicleStatus") - print(f"🌐 服务地址: http://localhost:8091") - print("🔐 认证Token: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...") + logger.info("🚗 无人车厂商平台模拟服务启动中...") + logger.info("📍 服务端点:") + logger.info(" - 车辆控制指令: POST /api/VehicleCommandInfo") + logger.info(" - 通用状态接口: GET /api/v1/vehicles//status") + logger.info(" - 车辆状态获取: GET /openApi/getVehicleStatus") + logger.info(f"🌐 服务地址: http://localhost:8091") + logger.info("🔐 认证Token: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9...") app.run(host='localhost', port=8091, debug=True) \ No newline at end of file