992 lines
38 KiB
Python
992 lines
38 KiB
Python
from flask import Flask, jsonify, request
|
||
import time
|
||
import math
|
||
import random
|
||
import logging
|
||
import os
|
||
import uuid
|
||
|
||
# 创建 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/mock_server.log'),
|
||
logging.StreamHandler()
|
||
]
|
||
)
|
||
|
||
app = Flask(__name__)
|
||
|
||
# 地球半径（米）
|
||
EARTH_RADIUS = 6378137.0
|
||
|
||
COMMAND_PRIORITIES = {
|
||
"ALERT": 5,
|
||
"WARNING": 3,
|
||
"PARKING": 2,
|
||
"RESUME": 1
|
||
}
|
||
|
||
def meters_to_degrees(meters, latitude):
|
||
"""
|
||
将米转换为度，考虑纬度的影响
|
||
"""
|
||
# 纬度方向：1度 = 111,319.9米
|
||
meters_per_deg_lat = 111319.9
|
||
# 经度方向：1度 = 111,319.9 * cos(latitude)米
|
||
meters_per_deg_lon = meters_per_deg_lat * math.cos(math.radians(latitude))
|
||
return meters / meters_per_deg_lat, meters / meters_per_deg_lon
|
||
|
||
# 距离配置（米）
|
||
DIST_300M = 300
|
||
DIST_200M = 200
|
||
DIST_150M = 150
|
||
DIST_100M = 100
|
||
DIST_50M = 50
|
||
|
||
# 时间配置（秒）
|
||
WAIT_TIME_AFTER_RETURN = 0.1 # 返回起点后的等待时间（进一步优化：减少到0.1秒）
|
||
UPDATE_INTERVAL = 1.0 # 位置更新间隔, 默认 1 秒
|
||
TRAFFIC_LIGHT_SWITCH_INTERVAL = 10.0 # 红绿灯切换间隔
|
||
|
||
# 根据 route.md 定义的坐标点
|
||
|
||
# 飞机 CA1234 路径
|
||
CA_START = {"longitude": 120.086263, "latitude": 36.370484} # 飞机起点
|
||
CA_END = {"longitude": 120.080996, "latitude": 36.369105} # 飞机终点
|
||
|
||
# 飞机 MU5123 路径
|
||
MU_START = {"longitude": 120.088076, "latitude": 36.374179} # 飞机起点
|
||
MU_END = {"longitude": 120.077971, "latitude": 36.371503} # 飞机终点
|
||
|
||
# 特勤车 鲁B123 路径
|
||
SPECIAL_VEHICLE_START = {"longitude": 120.080801, "latitude": 36.366626} # 特勤车起点
|
||
SPECIAL_VEHICLE_END = {"longitude": 120.083899, "latitude": 36.367403} # 特勤车终点
|
||
|
||
# 普通车 鲁B234 路径
|
||
NORMAL_VEHICLE_START = {"longitude": 120.087259, "latitude": 36.368299} # 普通车起点
|
||
NORMAL_VEHICLE_END = {"longitude": 120.083899, "latitude": 36.367403} # 普通车终点
|
||
|
||
# 无人车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终点
|
||
|
||
# 飞机和车辆尺寸（半径 米）
|
||
AIRCRAFT_SIZE_M = 30.0
|
||
VEHICLE_SIZE_M = 10.0
|
||
|
||
# 飞机数据 - 根据 route.md 配置
|
||
aircraft_data = [
|
||
{
|
||
"flightNo": "CA1234", # 根据route.md更新航班号
|
||
"longitude": CA_START["longitude"],
|
||
"latitude": CA_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"altitude": 0.0,
|
||
"trackNumber": 1001,
|
||
"speed": 50.0, # 根据route.md设置为50km/h
|
||
"start_point": CA_START, # 起点
|
||
"end_point": CA_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
},
|
||
{
|
||
"flightNo": "MU5123", # 根据route.md更新航班号
|
||
"longitude": MU_START["longitude"],
|
||
"latitude": MU_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"altitude": 0.0,
|
||
"trackNumber": 1002,
|
||
"speed": 50.0, # 根据route.md设置为50km/h
|
||
"start_point": MU_START, # 起点
|
||
"end_point": MU_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
}
|
||
]
|
||
|
||
# 车辆数据，根据 route.md 配置
|
||
DEFAULT_VEHICLE_SPEED = 30.0 # km/h，默认速度
|
||
EMERGENCY_BRAKE_DECELERATION = 0.8 # 紧急制动减速度 (每次更新减速 80%)
|
||
NORMAL_BRAKE_DECELERATION = 0.2 # 正常制动减速度 (每次更新减速 20%)
|
||
|
||
vehicle_data = [
|
||
{
|
||
"vehicleNo": "鲁B123", # 特勤车，根据route.md更新
|
||
"longitude": SPECIAL_VEHICLE_START["longitude"],
|
||
"latitude": SPECIAL_VEHICLE_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"direction": 0.0, # 方向角
|
||
"speed": 30.0, # 根据route.md设置为30km/h
|
||
"start_point": SPECIAL_VEHICLE_START, # 起点
|
||
"end_point": SPECIAL_VEHICLE_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
},
|
||
{
|
||
"vehicleNo": "鲁B234", # 普通车，根据route.md更新
|
||
"longitude": NORMAL_VEHICLE_START["longitude"],
|
||
"latitude": NORMAL_VEHICLE_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"direction": 180.0, # 根据route.md设置为180度
|
||
"speed": 30.0, # 根据route.md设置为30km/h
|
||
"start_point": NORMAL_VEHICLE_START, # 起点
|
||
"end_point": NORMAL_VEHICLE_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
},
|
||
{
|
||
"vehicleNo": "鲁B567", # 无人车A，根据route.md更新
|
||
"longitude": UNMANNED_A_START["longitude"],
|
||
"latitude": UNMANNED_A_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"direction": 90.0, # 根据route.md设置为90度
|
||
"speed": 25.0, # 根据route.md设置为25km/h
|
||
"start_point": UNMANNED_A_START, # 起点
|
||
"end_point": UNMANNED_A_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
},
|
||
{
|
||
"vehicleNo": "鲁B579", # 无人车B，根据route.md更新
|
||
"longitude": UNMANNED_B_START["longitude"],
|
||
"latitude": UNMANNED_B_START["latitude"],
|
||
"time": int(time.time() * 1000),
|
||
"direction": 270.0, # 根据route.md设置为270度
|
||
"speed": 25.0, # 根据route.md设置为25km/h
|
||
"start_point": UNMANNED_B_START, # 起点
|
||
"end_point": UNMANNED_B_END, # 终点
|
||
"moving_to_end": True # 当前是否向终点移动
|
||
}
|
||
]
|
||
|
||
# 车辆分类
|
||
airport_vehicle_data = [v for v in vehicle_data if v["vehicleNo"] in ["鲁B123", "鲁B234"]] # 特勤车和普通车
|
||
unmanned_vehicle_data = [v for v in vehicle_data if v["vehicleNo"] in ["鲁B567", "鲁B579"]] # 无人车
|
||
|
||
# 添加车辆状态类
|
||
class VehicleState:
|
||
def __init__(self, vehicle_no):
|
||
self.vehicle_no = vehicle_no
|
||
self.is_running = True # 运行状态
|
||
self.current_command = None # 当前执行的指令
|
||
self.command_priority = 0 # 当前指令优先级
|
||
self.target_speed = DEFAULT_VEHICLE_SPEED # 目标速度
|
||
self.brake_mode = None # 制动模式：'emergency' 或 'normal'
|
||
self.command_reason = None # 指令原因
|
||
self.last_command_time = time.time() # 最后一次指令时间
|
||
self.traffic_light_state = None # 当前红绿灯状态
|
||
self.target_lat = None # 目标纬度
|
||
self.target_lon = None # 目标经度
|
||
self.is_traffic_light_stop = False # 是否因红灯停车
|
||
|
||
def can_be_overridden_by(self, command_type):
|
||
"""判断当前指令是否可以被新指令覆盖"""
|
||
# 红绿灯信号不参与指令优先级判断
|
||
if command_type in ["RED", "GREEN", "YELLOW"]:
|
||
return True
|
||
|
||
new_priority = COMMAND_PRIORITIES.get(command_type, 0)
|
||
current_priority = COMMAND_PRIORITIES.get(self.current_command, 0)
|
||
|
||
# 相同类型的指令可以覆盖（因为需要持续发送）
|
||
if 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, target_lat=None, target_lon=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):
|
||
"""检查车辆是否可以移动"""
|
||
# 如果有告警指令，不能移动
|
||
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):
|
||
"""记录当前车辆状态"""
|
||
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 vehicle_data:
|
||
vehicle_states[vehicle["vehicleNo"]] = VehicleState(vehicle["vehicleNo"])
|
||
|
||
def calculate_distance_to_target(vehicle, target_lat, target_lon):
|
||
"""计算车辆到目标位的距离（米）"""
|
||
dlat = target_lat - vehicle["latitude"]
|
||
dlon = target_lon - vehicle["longitude"]
|
||
|
||
# 使用 Haversine 公式计算距离
|
||
R = 6371000 # 地球半径（米）
|
||
a = math.sin(dlat/2) * math.sin(dlat/2) + \
|
||
math.cos(math.radians(vehicle["latitude"])) * math.cos(math.radians(target_lat)) * \
|
||
math.sin(dlon/2) * math.sin(dlon/2)
|
||
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
|
||
return R * c
|
||
|
||
@app.route('/api/VehicleCommandInfo', methods=['POST'])
|
||
def handle_vehicle_command():
|
||
try:
|
||
data = request.json
|
||
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)
|
||
|
||
print(f"收到车辆控制指令: vehicle_id={vehicle_id}, type={command_type}, reason={reason}, signal_state={signal_state}")
|
||
print(f"完整请求数据: {data}")
|
||
|
||
# 检查 SIGNAL 类型命令必须包含 signalState
|
||
if command_type == "SIGNAL" and not signal_state:
|
||
print(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:
|
||
print(f"未找到车辆: {vehicle_id}")
|
||
return jsonify({
|
||
"code": 404,
|
||
"msg": f"Vehicle {vehicle_id} not found",
|
||
"transId": data.get("transId"),
|
||
"timestamp": int(time.time() * 1000)
|
||
}), 404
|
||
|
||
# 打印当前车辆状态
|
||
print(f"当前车辆状态: vehicle={vehicle_id}")
|
||
vehicle_state.log_state()
|
||
|
||
# 特勤车（鲁B123）只响应红绿灯信号
|
||
if vehicle_id == "鲁B123":
|
||
if command_type != "SIGNAL":
|
||
print(f"特勤车辆忽略非红绿灯指令: {vehicle_id}")
|
||
return jsonify({
|
||
"code": 200,
|
||
"msg": "Special vehicle only responds to traffic light signals",
|
||
"transId": data.get("transId"),
|
||
"timestamp": int(time.time() * 1000)
|
||
})
|
||
# 更新红绿灯状态和指令状态
|
||
vehicle_state.update_command(signal_state, target_lat, target_lon)
|
||
print(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)
|
||
})
|
||
|
||
# 检查指令优先级并添加详细日志
|
||
check_command = signal_state if command_type == "SIGNAL" else command_type
|
||
can_override = vehicle_state.can_be_overridden_by(check_command)
|
||
print(f"指令优先级检查: vehicle={vehicle_id}, current_command={vehicle_state.current_command}, "
|
||
f"new_command={check_command}, can_override={can_override}")
|
||
|
||
if not can_override:
|
||
print(f"指令优先级过低: vehicle={vehicle_id}, current_priority={vehicle_state.command_priority}, "
|
||
f"command={check_command}")
|
||
return jsonify({
|
||
"code": 400,
|
||
"msg": "Command priority too low",
|
||
"transId": data.get("transId"),
|
||
"timestamp": int(time.time() * 1000)
|
||
})
|
||
|
||
# 处理不同类型的指令
|
||
if command_type == "SIGNAL":
|
||
# 更新红绿灯状态和指令状态
|
||
vehicle_state.update_command(signal_state, target_lat, target_lon)
|
||
print(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 vehicle_data:
|
||
if v["vehicleNo"] == vehicle_id:
|
||
current_vehicle = v
|
||
break
|
||
|
||
# 执行告警指令直接停车
|
||
print(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
|
||
# 注释掉：不再直接设置车辆速度为0，让位置更新逻辑统一处理
|
||
# if current_vehicle:
|
||
# current_vehicle["speed"] = 0
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"msg": "Command executed successfully",
|
||
"transId": data.get("transId"),
|
||
"timestamp": int(time.time() * 1000)
|
||
})
|
||
|
||
elif command_type == "RESUME":
|
||
print(f"执行恢复指令: vehicle_id={vehicle_id}")
|
||
# 检查当前车辆
|
||
current_vehicle = None
|
||
for v in vehicle_data:
|
||
if v["vehicleNo"] == vehicle_id:
|
||
current_vehicle = v
|
||
break
|
||
|
||
# 清除限制性指令
|
||
if vehicle_state.current_command in ["ALERT", "WARNING"]:
|
||
print(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:
|
||
print(f"车辆 {vehicle_id} 恢复运行")
|
||
# 注释掉：不再直接设置车辆速度，让位置更新逻辑统一处理
|
||
# current_vehicle["speed"] = DEFAULT_VEHICLE_SPEED
|
||
|
||
# 记录状态变化但不更新指令
|
||
vehicle_state.command_reason = reason
|
||
vehicle_state.last_command_time = time.time()
|
||
|
||
print(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:
|
||
print(f"Error handling vehicle command: {str(e)}")
|
||
return jsonify({
|
||
"code": 500,
|
||
"msg": str(e),
|
||
"transId": data.get("transId", ""),
|
||
"timestamp": int(time.time() * 1000)
|
||
}), 500
|
||
|
||
# 删除了复杂的红绿灯和路口计算函数，因为已简化为往复运动逻辑
|
||
|
||
def update_position_with_vector(obj, target_point, start_point, speed, elapsed_time, return_to_start=False):
|
||
"""
|
||
基于向量的位置更新逻辑
|
||
|
||
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
|
||
|
||
current_time = time.time()
|
||
if current_time < obj["wait_until"]:
|
||
return False # 还在等待中，不更新位置
|
||
|
||
# 计算这一步要移动的距离（米）并转换为经纬度
|
||
speed_mps = speed * 1000 / 3600
|
||
distance = speed_mps * elapsed_time
|
||
|
||
# 将移动距离转换为经纬度变化量
|
||
move_dlat, move_dlon = meters_to_degrees(distance, obj["latitude"])
|
||
|
||
# 将5米的判断距离转换为经纬度（优化：从15米减少到5米，减少颤抖）
|
||
check_dlat, check_dlon = meters_to_degrees(5, obj["latitude"])
|
||
|
||
# 计算当前位置到终点的向量
|
||
vector_lat = target_point["latitude"] - obj["latitude"]
|
||
vector_lon = target_point["longitude"] - obj["longitude"]
|
||
|
||
# 计算向量长度（经纬度空间）
|
||
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"] = target_point["latitude"]
|
||
obj["longitude"] = target_point["longitude"]
|
||
|
||
if return_to_start:
|
||
# 返回起点并设置等待时间
|
||
obj["wait_until"] = current_time + WAIT_TIME_AFTER_RETURN
|
||
print(f"对象 {obj.get('flightNo', obj.get('vehicleNo'))} 返回起点，等待{WAIT_TIME_AFTER_RETURN}秒后继续")
|
||
else:
|
||
# 设置短暂等待时间，避免立即切换方向导致的颤抖
|
||
obj["wait_until"] = current_time + 0.1 # 0.1秒的短暂等待
|
||
print(f"对象 {obj.get('flightNo', 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
|
||
|
||
# 更新位置
|
||
obj["latitude"] += dlat
|
||
obj["longitude"] += dlon
|
||
return False # 表示正在移动中
|
||
|
||
def update_object_position(obj, elapsed_time):
|
||
"""统一的位置更新函数 - 用于飞机和车辆的往复运动"""
|
||
# 获取对象类型和标识
|
||
obj_type = "航空器" if "flightNo" in obj else "车辆"
|
||
obj_id = obj.get("flightNo", obj.get("vehicleNo"))
|
||
|
||
# 根据移动方向确定目标点
|
||
if obj["moving_to_end"]:
|
||
target_point = obj["end_point"]
|
||
else:
|
||
target_point = obj["start_point"]
|
||
|
||
# 更新位置
|
||
reached_target = update_position_with_vector(
|
||
obj,
|
||
target_point,
|
||
obj["start_point"] if obj["moving_to_end"] else obj["end_point"],
|
||
obj["speed"],
|
||
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_type} {obj_id} 到达{'终点' if obj['moving_to_end'] else '起点'}，开始前往{target_name}")
|
||
# 注意：等待时间已经在update_position_with_vector中设置了，不需要重复设置
|
||
|
||
# 更新时间戳
|
||
obj["time"] = int(time.time() * 1000)
|
||
|
||
def update_aircraft_position(aircraft, elapsed_time):
|
||
"""更新航空器位置 - 使用统一的位置更新函数"""
|
||
update_object_position(aircraft, elapsed_time)
|
||
|
||
def update_vehicle_position(vehicle, elapsed_time):
|
||
"""更新车辆位置 - 使用统一的位置更新函数"""
|
||
update_object_position(vehicle, elapsed_time)
|
||
|
||
# 简化的路口信息（保留API兼容性）
|
||
INTERSECTIONS = {
|
||
"TL001": {
|
||
"id": "INT001",
|
||
"name": "路口1",
|
||
"latitude": 36.369696,
|
||
"longitude": 120.083084
|
||
},
|
||
"TL002": {
|
||
"id": "INT002",
|
||
"name": "路口2",
|
||
"latitude": 36.365617,
|
||
"longitude": 120.084637
|
||
}
|
||
}
|
||
|
||
def generate_traffic_light_data():
|
||
"""生成红绿灯数据"""
|
||
traffic_light_data = []
|
||
|
||
# 生成两个固定路口的红绿灯数据
|
||
for tl_id, intersection in INTERSECTIONS.items():
|
||
signal = {
|
||
"id": tl_id,
|
||
"state": 1, # 0: RED, 1: GREEN，测试时保持绿灯
|
||
"timestamp": int(time.time() * 1000),
|
||
"intersection": intersection["id"],
|
||
"position": {
|
||
"latitude": intersection["latitude"],
|
||
"longitude": intersection["longitude"]
|
||
}
|
||
}
|
||
traffic_light_data.append(signal)
|
||
|
||
return traffic_light_data
|
||
|
||
# 红绿灯数据
|
||
traffic_light_data = generate_traffic_light_data()
|
||
|
||
# 分别记录航空器、机场车辆和无人车的上次更新时间
|
||
last_aircraft_update_time = time.time()
|
||
last_vehicle_update_time = time.time() # 机场车辆更新时间
|
||
last_unmanned_vehicle_update_time = time.time() # 无人车更新时间
|
||
last_light_switch_time = time.time()
|
||
|
||
def check_auth():
|
||
auth_header = request.headers.get('Authorization')
|
||
if not auth_header or auth_header != AUTH_TOKEN:
|
||
return False
|
||
return True
|
||
|
||
@app.route('/openApi/getCurrentFlightPositions', methods=['GET', 'OPTIONS'])
|
||
def get_flight_positions():
|
||
"""获取当前航空器位置信息"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
|
||
if not check_auth():
|
||
return jsonify({
|
||
"status": 401,
|
||
"msg": "认证失败",
|
||
"data": None
|
||
}), 401
|
||
global last_aircraft_update_time
|
||
current_time = time.time()
|
||
elapsed_time = current_time - last_aircraft_update_time
|
||
|
||
# 只在达到更新间隔时更新位置
|
||
if elapsed_time >= UPDATE_INTERVAL:
|
||
for aircraft in aircraft_data:
|
||
update_aircraft_position(aircraft, UPDATE_INTERVAL)
|
||
aircraft["time"] = int(current_time * 1000)
|
||
last_aircraft_update_time = current_time
|
||
|
||
# 创建符合 API 格式的响应数据
|
||
response_data = []
|
||
for aircraft in aircraft_data:
|
||
api_aircraft = {
|
||
"flightNo": aircraft["flightNo"],
|
||
"longitude": round(aircraft["longitude"], 6), # 经纬度保留6位小数
|
||
"latitude": round(aircraft["latitude"], 6), # 经纬度保留6位小数
|
||
"time": aircraft["time"]
|
||
}
|
||
response_data.append(api_aircraft)
|
||
|
||
return jsonify({
|
||
"status": 200,
|
||
"msg": "当前航空器实时位置数据",
|
||
"data": response_data
|
||
})
|
||
|
||
def switch_traffic_light_state():
|
||
"""统一处理红绿灯状态切换"""
|
||
global last_light_switch_time
|
||
current_time = time.time()
|
||
|
||
# 西路口红绿灯每15秒切换一次
|
||
elapsed_since_switch = current_time - last_light_switch_time
|
||
if elapsed_since_switch >= TRAFFIC_LIGHT_SWITCH_INTERVAL: # 使用常量
|
||
# 获取当前状态
|
||
current_state = traffic_light_data[0]["state"]
|
||
|
||
# 根据当前状态决定下一个状态
|
||
if current_state == 0: # 当前是红灯
|
||
# 切换到黄灯（表示红绿灯故障）
|
||
traffic_light_data[0]["state"] = 2 # 2 表示黄灯
|
||
print(f"西路口红绿灯状态切换为: 黄灯（红绿灯故障）")
|
||
elif current_state == 2: # 当前是黄灯
|
||
# 从黄灯切换到绿灯
|
||
traffic_light_data[0]["state"] = 1 # 1 表示绿灯
|
||
print(f"西路口红绿灯状态切换为: 绿灯")
|
||
else: # 当前是绿灯
|
||
# 从绿灯切换到红灯
|
||
traffic_light_data[0]["state"] = 0 # 0 表示红灯
|
||
print(f"西路口红绿灯状态切换为: 红灯")
|
||
|
||
last_light_switch_time = current_time
|
||
|
||
# 更新东路口红绿灯（根据航空器位置）
|
||
if aircraft_data:
|
||
aircraft = aircraft_data[0]
|
||
lat_diff = abs(aircraft["latitude"] - T6_INTERSECTION["latitude"]) * 111319.9
|
||
|
||
old_state = traffic_light_data[1]["state"]
|
||
traffic_light_data[1]["state"] = 1 if lat_diff > DIST_50M else 1
|
||
|
||
if old_state != traffic_light_data[1]["state"]:
|
||
print(f"东路口红绿灯状态切换为: {'绿灯' if traffic_light_data[1]['state'] == 1 else '红灯'}")
|
||
|
||
@app.route('/openApi/getCurrentVehiclePositions', methods=['GET', 'OPTIONS'])
|
||
def get_vehicle_positions():
|
||
"""获取当前车辆位置信息（仅机场车辆）"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
if not check_auth():
|
||
return jsonify({
|
||
"status": 401,
|
||
"msg": "认证失败",
|
||
"data": None
|
||
}), 401
|
||
global last_vehicle_update_time
|
||
current_time = time.time()
|
||
elapsed_time = current_time - last_vehicle_update_time
|
||
try:
|
||
# 只在达到更新间隔时更新位置
|
||
if elapsed_time >= UPDATE_INTERVAL:
|
||
for vehicle in airport_vehicle_data:
|
||
update_vehicle_position(vehicle, UPDATE_INTERVAL)
|
||
vehicle["time"] = int(current_time * 1000)
|
||
last_vehicle_update_time = current_time
|
||
response_data = []
|
||
for v in airport_vehicle_data:
|
||
v_out = {
|
||
"vehicleNo": v.get("vehicleNo"),
|
||
"longitude": round(v.get("longitude"), 6), # 经纬度保留6位小数
|
||
"latitude": round(v.get("latitude"), 6), # 经纬度保留6位小数
|
||
"time": v.get("time")
|
||
}
|
||
response_data.append(v_out)
|
||
return jsonify({
|
||
"status": 200,
|
||
"msg": "当前车辆实时位置数据",
|
||
"data": response_data
|
||
})
|
||
except Exception as e:
|
||
print(f"Error in get_vehicle_positions: {str(e)}")
|
||
return jsonify({
|
||
"status": 500,
|
||
"msg": str(e),
|
||
"data": None
|
||
}), 500
|
||
|
||
@app.route('/openApi/getTrafficLightSignals', methods=['GET', 'OPTIONS'])
|
||
def get_traffic_light_signals():
|
||
"""获取红绿灯信号状态 (旧的轮询接口)"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
|
||
# --- 注释掉原有逻辑 (2025-05-06 by AI Assistant) ---
|
||
# 原因: 为了测试新的 TrafficLightHttpServer 推送机制，暂时禁用此轮询接口返回数据。
|
||
# 模拟和发送红绿灯状态的功能现在由 tools/mock_traffic_light_client.py 负责。
|
||
# 如需恢复轮询测试，请取消下面的注释并注释掉新的 return 语句。
|
||
# # 更新红绿灯状态
|
||
# # switch_traffic_light_state() # 采用模拟的红绿灯脚本，注释掉此处的调用
|
||
#
|
||
# # 更新时间戳
|
||
# for signal in traffic_light_data:
|
||
# signal[\"timestamp\"] = int(time.time() * 1000)
|
||
#
|
||
# return jsonify({
|
||
# \"status\": 200,
|
||
# \"msg\": \"获取红绿灯信号成功\",
|
||
# \"data\": traffic_light_data
|
||
# })
|
||
# --- 注释结束 ---
|
||
|
||
# 新增: 返回空数据，以禁用轮询数据源
|
||
return jsonify({
|
||
"status": 200,
|
||
"msg": "Traffic light polling endpoint disabled to test push mechanism.",
|
||
"data": []
|
||
})
|
||
|
||
@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
|
||
|
||
def check_vehicle_states():
|
||
"""定期检查所有车辆状态"""
|
||
for vehicle_no, state in vehicle_states.items():
|
||
state.log_state()
|
||
|
||
# 认证相关配置
|
||
AUTH_USERNAME = "dianxin"
|
||
AUTH_PASSWORD = "dianxin@123"
|
||
AUTH_TOKEN = "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE3MzI3ODMwOTAsInVzZXJuYW1lIjoiYWRtaW4ifQ.y9feEL_9NT8UzED9NNkb0Ln6C-PBoufiSHWobWe5vWY"
|
||
|
||
@app.route('/login', methods=['POST', 'OPTIONS'])
|
||
def login():
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
|
||
# 从 URL query parameters 中获取用户名和密码
|
||
username = request.args.get('username')
|
||
password = request.args.get('password')
|
||
|
||
print(f"收到登录请求: username={username}, password={password}")
|
||
print(f"请求 URL: {request.url}")
|
||
print(f"请求方法: {request.method}")
|
||
print(f"请求参数: {request.args}")
|
||
|
||
if not username or not password:
|
||
return jsonify({
|
||
"status": 400,
|
||
"msg": "缺少用户名或密码",
|
||
"data": None
|
||
}), 400
|
||
|
||
if username == AUTH_USERNAME and password == AUTH_PASSWORD:
|
||
return jsonify({
|
||
"status": 200,
|
||
"msg": "登入成功",
|
||
"data": AUTH_TOKEN
|
||
})
|
||
else:
|
||
return jsonify({
|
||
"status": 401,
|
||
"msg": "用户名或密码错误",
|
||
"data": None
|
||
}), 401
|
||
|
||
@app.route('/openApi/getVehicleStatus', methods=['GET', 'OPTIONS'])
|
||
def get_vehicle_status():
|
||
"""获取无人车状态信息"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
|
||
vehicle_id = request.args.get('vehicleId')
|
||
if not vehicle_id:
|
||
return jsonify({
|
||
"status": 400,
|
||
"msg": "缺少 vehicleId 参数",
|
||
"data": None
|
||
}), 400
|
||
|
||
# 查找对应的车辆
|
||
vehicle_state = vehicle_states.get(vehicle_id)
|
||
if not vehicle_state:
|
||
return jsonify({
|
||
"status": 404,
|
||
"msg": f"未找到车辆 {vehicle_id}",
|
||
"data": None
|
||
}), 404
|
||
|
||
# 返回车辆状态
|
||
return jsonify({
|
||
"status": 200,
|
||
"msg": "获取车辆状态成功",
|
||
"data": {
|
||
"vehicleId": vehicle_id,
|
||
"status": "NORMAL" if vehicle_state.is_running else "STOPPED",
|
||
"command": vehicle_state.current_command,
|
||
"commandPriority": vehicle_state.command_priority,
|
||
"trafficLightState": vehicle_state.traffic_light_state,
|
||
"timestamp": int(time.time() * 1000)
|
||
}
|
||
})
|
||
|
||
# 无人车位置数据生成函数
|
||
def generate_unmanned_vehicle_location_data():
|
||
"""生成无人车位置数据，符合官方API格式"""
|
||
unmanned_vehicles = []
|
||
|
||
# 从现有vehicle_data中筛选无人车（QN开头的车辆）
|
||
for vehicle in unmanned_vehicle_data:
|
||
location_info = {
|
||
"transId": str(uuid.uuid4()),
|
||
"timestamp": int(time.time() * 1000),
|
||
"vehicleID": vehicle["vehicleNo"],
|
||
"longitude": round(vehicle.get("longitude"), 6), # 经纬度保留6位小数
|
||
"latitude": round(vehicle.get("latitude"), 6) # 经纬度保留6位小数
|
||
}
|
||
unmanned_vehicles.append(location_info)
|
||
|
||
return unmanned_vehicles
|
||
|
||
# 无人车状态数据生成函数
|
||
def generate_unmanned_vehicle_state_data(vehicle_id=None, is_single=True):
|
||
"""生成无人车状态数据，符合官方API格式"""
|
||
vehicle_states_data = []
|
||
|
||
if is_single and vehicle_id:
|
||
# 单个车辆状态查询
|
||
vehicle_state = vehicle_states.get(vehicle_id)
|
||
if vehicle_state:
|
||
state_info = {
|
||
"transId": str(uuid.uuid4()),
|
||
"timestamp": int(time.time() * 1000),
|
||
"vehicleID": vehicle_id,
|
||
"loginState": True,
|
||
"faultInfo": [],
|
||
"activeSafety": not vehicle_state.is_running,
|
||
"RC": False,
|
||
"Command": 1 if vehicle_state.current_command == "ALERT" else 0,
|
||
"airportInfo": [],
|
||
"vehicleMode": 2, # 自动模式
|
||
"gearState": 2, # D档
|
||
"chassisReady": vehicle_state.is_running,
|
||
"collisionStatus": False,
|
||
"clearance": 1 if vehicle_state.is_running else 0,
|
||
"turnSignalStstus": 0,
|
||
"pointCloud": []
|
||
}
|
||
vehicle_states_data.append(state_info)
|
||
else:
|
||
# 所有车辆状态查询
|
||
for vehicle in unmanned_vehicle_data:
|
||
vehicle_state = vehicle_states.get(vehicle["vehicleNo"])
|
||
state_info = {
|
||
"transId": str(uuid.uuid4()),
|
||
"timestamp": int(time.time() * 1000),
|
||
"vehicleID": vehicle["vehicleNo"],
|
||
"loginState": True,
|
||
"faultInfo": [],
|
||
"activeSafety": not vehicle_state.is_running if vehicle_state else False,
|
||
"RC": False,
|
||
"Command": 1 if vehicle_state and vehicle_state.current_command == "ALERT" else 0,
|
||
"airportInfo": [],
|
||
"vehicleMode": 2,
|
||
"gearState": 2,
|
||
"chassisReady": vehicle_state.is_running if vehicle_state else True,
|
||
"collisionStatus": False,
|
||
"clearance": 1 if vehicle_state and vehicle_state.is_running else 0,
|
||
"turnSignalStstus": 0,
|
||
"pointCloud": []
|
||
}
|
||
vehicle_states_data.append(state_info)
|
||
|
||
return vehicle_states_data
|
||
|
||
# 无人车API接口
|
||
@app.route('/api/VehicleLocationInfo', methods=['GET', 'OPTIONS'])
|
||
def get_unmanned_vehicle_location():
|
||
"""无人车位置上报接口"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
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
|
||
# 生成无人车位置数据
|
||
location_data = generate_unmanned_vehicle_location_data()
|
||
|
||
return jsonify(location_data)
|
||
except Exception as e:
|
||
print(f"Error in get_unmanned_vehicle_location: {str(e)}")
|
||
return jsonify([]), 500
|
||
|
||
@app.route('/api/VehicleStateInfo', methods=['POST', 'OPTIONS'])
|
||
def get_unmanned_vehicle_state():
|
||
"""无人车状态查询接口"""
|
||
if request.method == 'OPTIONS':
|
||
return '', 204
|
||
|
||
try:
|
||
data = request.json
|
||
vehicle_id = data.get("vehicleID") if data else None
|
||
is_single = data.get("isSingle", True) if data else False
|
||
|
||
print(f"收到无人车状态查询请求: vehicle_id={vehicle_id}, is_single={is_single}")
|
||
|
||
# 生成无人车状态数据
|
||
state_data = generate_unmanned_vehicle_state_data(vehicle_id, is_single)
|
||
|
||
print(f"返回无人车状态数据，数量: {len(state_data)}")
|
||
return jsonify(state_data)
|
||
|
||
except Exception as e:
|
||
print(f"Error in get_unmanned_vehicle_state: {str(e)}")
|
||
return jsonify([]), 500
|
||
|
||
if __name__ == '__main__':
|
||
app.run(host='localhost', port=8090, debug=True) |