Refactor WebSocketMessageBroadcaster to enhance position update handling and introduce payload sanitization，增加了飞机平滑
This commit is contained in:
parent
dc9967e10d
commit
bbc3a6f5d6
@ -1,34 +1,35 @@
|
||||
package com.qaup.collision.websocket.broadcaster;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.qaup.collision.websocket.cache.MessageCacheService;
|
||||
import com.qaup.collision.websocket.event.CollisionWarningEvent;
|
||||
import com.qaup.collision.websocket.event.GeofenceAlertWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.PathConflictAlertWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.PositionUpdateEvent;
|
||||
import com.qaup.collision.websocket.event.RuleExecutionStatusWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.RuleStateChangeWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.RuleViolationWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.GeofenceAlertWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.PathConflictAlertWebSocketEvent;
|
||||
import com.qaup.collision.websocket.event.VehicleCommandEvent;
|
||||
import com.qaup.collision.websocket.event.VehicleStatusUpdateEvent;
|
||||
import com.qaup.collision.websocket.handler.CollisionWebSocketHandler;
|
||||
import com.qaup.collision.websocket.message.CollisionWarningPayload;
|
||||
import com.qaup.collision.websocket.message.MessageTypeConstants;
|
||||
import com.qaup.collision.websocket.message.GeofenceAlertPayload;
|
||||
import com.qaup.collision.websocket.message.MessageTypeConstants;
|
||||
import com.qaup.collision.websocket.message.PathConflictAlertMessage;
|
||||
import com.qaup.collision.websocket.message.PositionUpdatePayload;
|
||||
import com.qaup.collision.websocket.message.RuleExecutionStatusPayload;
|
||||
import com.qaup.collision.websocket.message.RuleStateChangePayload;
|
||||
import com.qaup.collision.websocket.message.RuleViolationPayload;
|
||||
import com.qaup.collision.websocket.message.UniversalMessage;
|
||||
import com.qaup.collision.websocket.message.VehicleCommandPayload;
|
||||
import com.qaup.collision.websocket.message.VehicleStatusUpdatePayload;
|
||||
import com.qaup.collision.websocket.message.UniversalMessage;
|
||||
import com.qaup.collision.websocket.message.PathConflictAlertMessage;
|
||||
import com.qaup.collision.websocket.handler.CollisionWebSocketHandler;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
@ -36,65 +37,92 @@ import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* WebSocket统一消息广播器
|
||||
* 监听所有WebSocket事件，使用原生WebSocket统一推送到控制台前端
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class WebSocketMessageBroadcaster {
|
||||
|
||||
private final MessageCacheService messageCacheService;
|
||||
private final CollisionWebSocketHandler collisionWebSocketHandler; // 注入实例
|
||||
private final ObjectMapper objectMapper; // JSON序列化器
|
||||
|
||||
/**
|
||||
* 高频位置更新做“覆盖式合并”，只保留每个objectId的最新一条，避免消息堆积。
|
||||
* flush线程如果跑不过来，下一轮会直接跳过（丢旧保实时）。
|
||||
*/
|
||||
private static final int REACQUIRE_AFTER_REJECTS = 3;
|
||||
private static final double EARTH_RADIUS_METERS = 6371000.0;
|
||||
|
||||
private final MessageCacheService messageCacheService;
|
||||
private final CollisionWebSocketHandler collisionWebSocketHandler;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
private final Map<String, CoalescedPositionUpdate> latestPositionUpdates = new ConcurrentHashMap<>();
|
||||
private final Map<String, PositionTrackState> positionTrackStates = new ConcurrentHashMap<>();
|
||||
private final AtomicBoolean positionFlushInProgress = new AtomicBoolean(false);
|
||||
|
||||
@Value("${websocket.position.filter.ema-alpha:0.45}")
|
||||
private double positionEmaAlpha;
|
||||
|
||||
@Value("${websocket.position.filter.jitter-meter:2.5}")
|
||||
private double jitterMeterThreshold;
|
||||
|
||||
@Value("${websocket.position.filter.max-speed-mps-aircraft:150.0}")
|
||||
private double maxSpeedMpsAircraft;
|
||||
|
||||
@Value("${websocket.position.filter.max-speed-mps-vehicle:35.0}")
|
||||
private double maxSpeedMpsVehicle;
|
||||
|
||||
@Value("${websocket.position.filter.max-speed-mps-unmanned:15.0}")
|
||||
private double maxSpeedMpsUnmanned;
|
||||
|
||||
@Value("${websocket.position.filter.max-jump-margin-meter:20.0}")
|
||||
private double maxJumpMarginMeter;
|
||||
|
||||
@Value("${websocket.position.filter.state-ttl-ms:120000}")
|
||||
private long positionTrackStateTtlMs;
|
||||
|
||||
private record CoalescedPositionUpdate(PositionUpdatePayload payload, long timestamp) {}
|
||||
|
||||
public WebSocketMessageBroadcaster(MessageCacheService messageCacheService, CollisionWebSocketHandler collisionWebSocketHandler, @Qualifier("websocketObjectMapper") ObjectMapper objectMapper) {
|
||||
private static final class PositionTrackState {
|
||||
private boolean initialized;
|
||||
private long lastPayloadTimestampMs;
|
||||
private long lastAcceptedAtMs;
|
||||
private double lastRawLatitude;
|
||||
private double lastRawLongitude;
|
||||
private double lastFilteredLatitude;
|
||||
private double lastFilteredLongitude;
|
||||
private int rejectedCount;
|
||||
}
|
||||
|
||||
public WebSocketMessageBroadcaster(
|
||||
MessageCacheService messageCacheService,
|
||||
CollisionWebSocketHandler collisionWebSocketHandler,
|
||||
@Qualifier("websocketObjectMapper") ObjectMapper objectMapper) {
|
||||
this.messageCacheService = messageCacheService;
|
||||
this.collisionWebSocketHandler = collisionWebSocketHandler;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理位置更新事件
|
||||
* @param event 位置更新事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handlePositionUpdate(PositionUpdateEvent event) {
|
||||
try {
|
||||
if (event == null || event.getPayload() == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!(event.getPayload() instanceof PositionUpdatePayload payload)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (payload.getObjectId() == null || payload.getObjectId().isBlank()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 覆盖式合并：同一对象只保留最新的
|
||||
latestPositionUpdates.put(payload.getObjectId(), new CoalescedPositionUpdate(payload, event.getTimestamp()));
|
||||
PositionUpdatePayload sanitizedPayload = sanitizePositionPayload(payload);
|
||||
if (sanitizedPayload == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
latestPositionUpdates.put(
|
||||
sanitizedPayload.getObjectId(),
|
||||
new CoalescedPositionUpdate(sanitizedPayload, event.getTimestamp())
|
||||
);
|
||||
} catch (Exception e) {
|
||||
log.error("合并位置更新事件失败", e);
|
||||
log.error("Failed to merge position update", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 定期刷新合并后的高频位置消息。
|
||||
* 丢旧保实时：如果上一轮flush未完成，本轮直接跳过。
|
||||
*/
|
||||
@Scheduled(fixedDelayString = "${websocket.position.flush-interval-ms:250}")
|
||||
public void flushCoalescedPositionUpdates() {
|
||||
if (!positionFlushInProgress.compareAndSet(false, true)) {
|
||||
@ -106,7 +134,6 @@ public class WebSocketMessageBroadcaster {
|
||||
return;
|
||||
}
|
||||
|
||||
// 没有客户端时直接清空，避免无意义积压
|
||||
if (collisionWebSocketHandler.getConnectionCount() <= 0) {
|
||||
latestPositionUpdates.clear();
|
||||
return;
|
||||
@ -125,48 +152,43 @@ public class WebSocketMessageBroadcaster {
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.error("刷新合并位置更新失败", e);
|
||||
log.error("Failed to flush coalesced position updates", e);
|
||||
} finally {
|
||||
positionFlushInProgress.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理车辆状态更新事件
|
||||
* 符合universal_autonomous_vehicle_api规范
|
||||
*/
|
||||
@Scheduled(fixedDelayString = "${websocket.position.filter.cleanup-interval-ms:60000}")
|
||||
public void cleanupPositionTrackStates() {
|
||||
long now = System.currentTimeMillis();
|
||||
positionTrackStates.entrySet().removeIf(entry -> now - entry.getValue().lastAcceptedAtMs > positionTrackStateTtlMs);
|
||||
}
|
||||
|
||||
@EventListener
|
||||
@Async
|
||||
public void handleVehicleStatusUpdate(VehicleStatusUpdateEvent event) {
|
||||
try {
|
||||
if (event.getPayload() == null) {
|
||||
log.warn("车辆状态更新事件负载为空，跳过处理");
|
||||
log.warn("Vehicle status update payload is null");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
VehicleStatusUpdatePayload payload = event.getPayload();
|
||||
|
||||
UniversalMessage<VehicleStatusUpdatePayload> message = UniversalMessage.<VehicleStatusUpdatePayload>builder()
|
||||
.type(MessageTypeConstants.VEHICLE_STATUS_UPDATE)
|
||||
.payload(payload)
|
||||
.timestamp(event.getTimestamp())
|
||||
.messageId(generateMessageId())
|
||||
.build();
|
||||
|
||||
.type(MessageTypeConstants.VEHICLE_STATUS_UPDATE)
|
||||
.payload(payload)
|
||||
.timestamp(event.getTimestamp())
|
||||
.messageId(generateMessageId())
|
||||
.build();
|
||||
|
||||
broadcastMessage(message);
|
||||
|
||||
log.debug("发送车辆状态更新WebSocket消息: vehicleId={}, 消息ID={}",
|
||||
payload.getVehicleId(), message.getMessageId());
|
||||
|
||||
log.debug("Vehicle status message published: vehicleId={}, messageId={}",
|
||||
payload.getVehicleId(), message.getMessageId());
|
||||
} catch (Exception e) {
|
||||
log.error("处理车辆状态更新事件异常", e);
|
||||
log.error("Failed to handle vehicle status update", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理车辆指令事件
|
||||
* @param event 车辆指令事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleVehicleCommand(VehicleCommandEvent event) {
|
||||
UniversalMessage<VehicleCommandPayload> message = UniversalMessage.<VehicleCommandPayload>builder()
|
||||
@ -175,15 +197,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((VehicleCommandPayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 处理碰撞预警事件
|
||||
* @param event 碰撞预警事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleCollisionWarning(CollisionWarningEvent event) {
|
||||
UniversalMessage<CollisionWarningPayload> message = UniversalMessage.<CollisionWarningPayload>builder()
|
||||
@ -192,16 +209,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((CollisionWarningPayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
// === 规则事件处理方法 ===
|
||||
|
||||
/**
|
||||
* 处理规则违规事件
|
||||
* @param event 规则违规事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleRuleViolation(RuleViolationWebSocketEvent event) {
|
||||
UniversalMessage<RuleViolationPayload> message = UniversalMessage.<RuleViolationPayload>builder()
|
||||
@ -210,14 +221,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((RuleViolationPayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理规则执行状态事件
|
||||
* @param event 规则执行状态事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleRuleExecutionStatus(RuleExecutionStatusWebSocketEvent event) {
|
||||
UniversalMessage<RuleExecutionStatusPayload> message = UniversalMessage.<RuleExecutionStatusPayload>builder()
|
||||
@ -226,14 +233,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((RuleExecutionStatusPayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理规则状态变更事件
|
||||
* @param event 规则状态变更事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleRuleStateChange(RuleStateChangeWebSocketEvent event) {
|
||||
UniversalMessage<RuleStateChangePayload> message = UniversalMessage.<RuleStateChangePayload>builder()
|
||||
@ -242,14 +245,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((RuleStateChangePayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理路径冲突告警事件
|
||||
* @param event 路径冲突告警事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handlePathConflictAlert(PathConflictAlertWebSocketEvent event) {
|
||||
UniversalMessage<PathConflictAlertMessage> message = UniversalMessage.<PathConflictAlertMessage>builder()
|
||||
@ -258,14 +257,10 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((PathConflictAlertMessage) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理电子围栏告警事件
|
||||
* @param event 电子围栏告警事件
|
||||
*/
|
||||
|
||||
@EventListener
|
||||
public void handleGeofenceAlert(GeofenceAlertWebSocketEvent event) {
|
||||
UniversalMessage<GeofenceAlertPayload> message = UniversalMessage.<GeofenceAlertPayload>builder()
|
||||
@ -274,97 +269,193 @@ public class WebSocketMessageBroadcaster {
|
||||
.messageId(generateMessageId())
|
||||
.payload((GeofenceAlertPayload) event.getPayload())
|
||||
.build();
|
||||
|
||||
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 统一消息广播方法 - 使用原生WebSocket发送JSON消息（公共方法）
|
||||
* @param message 要广播的消息
|
||||
*/
|
||||
|
||||
public void broadcastMessage(UniversalMessage<?> message) {
|
||||
broadcastMessageInternal(message);
|
||||
}
|
||||
|
||||
/**
|
||||
* 统一消息广播方法 - 使用原生WebSocket发送JSON消息（内部方法）
|
||||
* @param message 要广播的消息
|
||||
*/
|
||||
|
||||
private void broadcastMessageInternal(UniversalMessage<?> message) {
|
||||
try {
|
||||
// 使用Jackson ObjectMapper将UniversalMessage序列化为JSON字符串
|
||||
// 这样前端可以获得消息类型、时间戳、消息ID等完整信息
|
||||
String jsonMessage = objectMapper.writeValueAsString(message);
|
||||
this.collisionWebSocketHandler.broadcastMessage(jsonMessage);
|
||||
|
||||
// 根据消息类型决定是否缓存（避免高频消息阻塞Redis）
|
||||
// 高频实时消息（如位置更新）不缓存，追求实时性
|
||||
// 低频重要消息（如碰撞预警）缓存，保证不丢失
|
||||
if (shouldCacheMessage(message.getType())) {
|
||||
messageCacheService.cacheMessage(message);
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
System.err.println("Failed to broadcast message via native WebSocket: " + e.getMessage());
|
||||
e.printStackTrace();
|
||||
log.error("Failed to broadcast message", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 判断消息是否应该缓存
|
||||
*
|
||||
* @param messageType 消息类型
|
||||
* @return true=需要缓存，false=不需要缓存
|
||||
*/
|
||||
private boolean shouldCacheMessage(String messageType) {
|
||||
// 高频实时消息 - 不缓存（追求实时性，避免Redis性能瓶颈）
|
||||
// 这些消息更新频繁，用户关心的是最新状态，历史数据价值不高
|
||||
if (MessageTypeConstants.POSITION_UPDATE.equals(messageType) ||
|
||||
MessageTypeConstants.TRAFFIC_LIGHT_STATUS.equals(messageType) ||
|
||||
MessageTypeConstants.HEARTBEAT.equals(messageType) ||
|
||||
MessageTypeConstants.VEHICLE_STATUS_UPDATE.equals(messageType)) {
|
||||
if (MessageTypeConstants.POSITION_UPDATE.equals(messageType)
|
||||
|| MessageTypeConstants.TRAFFIC_LIGHT_STATUS.equals(messageType)
|
||||
|| MessageTypeConstants.HEARTBEAT.equals(messageType)
|
||||
|| MessageTypeConstants.VEHICLE_STATUS_UPDATE.equals(messageType)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 低频重要消息 - 缓存（保证不丢失）
|
||||
// 包括：碰撞预警、规则违规、路径冲突、电子围栏、航班通知、车辆指令等
|
||||
// 这些事件频率低但重要性高，用户需要完整的历史记录
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成唯一消息ID
|
||||
* @return 消息ID
|
||||
*/
|
||||
|
||||
private String generateMessageId() {
|
||||
return UUID.randomUUID().toString();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取最近消息（用于客户端重连恢复）
|
||||
* @param messageType 消息类型，为空则获取所有类型
|
||||
* @param count 消息数量
|
||||
* @return 最近消息列表
|
||||
*/
|
||||
|
||||
public void sendRecentMessages(String messageType, int count) {
|
||||
try {
|
||||
List<UniversalMessage<?>> recentMessages = messageType == null || messageType.trim().isEmpty()
|
||||
List<UniversalMessage<?>> recentMessages = messageType == null || messageType.trim().isEmpty()
|
||||
? messageCacheService.getAllRecentMessages(count)
|
||||
: messageCacheService.getRecentMessages(messageType, count);
|
||||
|
||||
// 逐个发送最近消息
|
||||
|
||||
recentMessages.forEach(this::broadcastMessageInternal);
|
||||
|
||||
} catch (Exception e) {
|
||||
System.err.println("Failed to send recent messages: " + e.getMessage());
|
||||
log.error("Failed to send recent messages", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取当前WebSocket连接数
|
||||
* @return 连接数量
|
||||
*/
|
||||
|
||||
public int getOnlineCount() {
|
||||
return this.collisionWebSocketHandler.getConnectionCount();
|
||||
}
|
||||
}
|
||||
|
||||
private PositionUpdatePayload sanitizePositionPayload(PositionUpdatePayload payload) {
|
||||
if (payload.getPosition() == null
|
||||
|| payload.getPosition().getLatitude() == null
|
||||
|| payload.getPosition().getLongitude() == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
double latitude = payload.getPosition().getLatitude();
|
||||
double longitude = payload.getPosition().getLongitude();
|
||||
if (!isValidCoordinate(latitude, longitude)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
long payloadTimestamp = payload.getTimestamp() != null ? payload.getTimestamp() : System.currentTimeMillis();
|
||||
PositionTrackState state = positionTrackStates.computeIfAbsent(payload.getObjectId(), k -> new PositionTrackState());
|
||||
|
||||
double filteredLatitude = latitude;
|
||||
double filteredLongitude = longitude;
|
||||
|
||||
synchronized (state) {
|
||||
if (!state.initialized) {
|
||||
initializeTrackState(state, latitude, longitude, payloadTimestamp);
|
||||
} else {
|
||||
if (payloadTimestamp <= state.lastPayloadTimestampMs) {
|
||||
state.rejectedCount++;
|
||||
return null;
|
||||
}
|
||||
|
||||
long dtMs = payloadTimestamp - state.lastPayloadTimestampMs;
|
||||
double dtSec = dtMs / 1000.0;
|
||||
double rawDistance = calculateDistanceMeters(
|
||||
state.lastRawLatitude,
|
||||
state.lastRawLongitude,
|
||||
latitude,
|
||||
longitude
|
||||
);
|
||||
|
||||
double maxAllowedJump = resolveMaxAllowedJumpMeters(payload.getObjectType(), dtSec);
|
||||
boolean isOutlierJump = rawDistance > maxAllowedJump;
|
||||
if (isOutlierJump && state.rejectedCount < REACQUIRE_AFTER_REJECTS - 1) {
|
||||
state.rejectedCount++;
|
||||
log.debug("Drop outlier position update: objectId={}, jumpMeter={}, allowedMeter={}",
|
||||
payload.getObjectId(), rawDistance, maxAllowedJump);
|
||||
return null;
|
||||
}
|
||||
|
||||
if (isOutlierJump) {
|
||||
state.rejectedCount = 0;
|
||||
filteredLatitude = latitude;
|
||||
filteredLongitude = longitude;
|
||||
} else {
|
||||
state.rejectedCount = 0;
|
||||
if (rawDistance <= jitterMeterThreshold) {
|
||||
filteredLatitude = state.lastFilteredLatitude;
|
||||
filteredLongitude = state.lastFilteredLongitude;
|
||||
} else {
|
||||
double alpha = clamp(positionEmaAlpha, 0.05, 1.0);
|
||||
filteredLatitude = state.lastFilteredLatitude + alpha * (latitude - state.lastFilteredLatitude);
|
||||
filteredLongitude = state.lastFilteredLongitude + alpha * (longitude - state.lastFilteredLongitude);
|
||||
}
|
||||
}
|
||||
|
||||
state.lastRawLatitude = latitude;
|
||||
state.lastRawLongitude = longitude;
|
||||
state.lastPayloadTimestampMs = payloadTimestamp;
|
||||
state.lastFilteredLatitude = filteredLatitude;
|
||||
state.lastFilteredLongitude = filteredLongitude;
|
||||
state.lastAcceptedAtMs = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
|
||||
PositionUpdatePayload.Position filteredPosition = PositionUpdatePayload.Position.builder()
|
||||
.latitude(filteredLatitude)
|
||||
.longitude(filteredLongitude)
|
||||
.build();
|
||||
|
||||
return PositionUpdatePayload.builder()
|
||||
.objectId(payload.getObjectId())
|
||||
.objectType(payload.getObjectType())
|
||||
.position(filteredPosition)
|
||||
.heading(payload.getHeading())
|
||||
.speed(payload.getSpeed())
|
||||
.timestamp(payloadTimestamp)
|
||||
.build();
|
||||
}
|
||||
|
||||
private void initializeTrackState(PositionTrackState state, double latitude, double longitude, long payloadTimestamp) {
|
||||
state.initialized = true;
|
||||
state.lastPayloadTimestampMs = payloadTimestamp;
|
||||
state.lastAcceptedAtMs = System.currentTimeMillis();
|
||||
state.lastRawLatitude = latitude;
|
||||
state.lastRawLongitude = longitude;
|
||||
state.lastFilteredLatitude = latitude;
|
||||
state.lastFilteredLongitude = longitude;
|
||||
state.rejectedCount = 0;
|
||||
}
|
||||
|
||||
private boolean isValidCoordinate(double latitude, double longitude) {
|
||||
return latitude >= -90.0 && latitude <= 90.0 && longitude >= -180.0 && longitude <= 180.0;
|
||||
}
|
||||
|
||||
private double resolveMaxAllowedJumpMeters(String objectType, double dtSec) {
|
||||
double safeDtSec = Math.max(dtSec, 0.05);
|
||||
double speedMps = resolveMaxSpeedMps(objectType);
|
||||
return speedMps * safeDtSec + Math.max(maxJumpMarginMeter, 0.0);
|
||||
}
|
||||
|
||||
private double resolveMaxSpeedMps(String objectType) {
|
||||
if (objectType == null) {
|
||||
return maxSpeedMpsVehicle;
|
||||
}
|
||||
return switch (objectType) {
|
||||
case "AIRCRAFT" -> maxSpeedMpsAircraft;
|
||||
case "UNMANNED_VEHICLE" -> maxSpeedMpsUnmanned;
|
||||
case "SPECIAL_VEHICLE", "NORMAL_VEHICLE" -> maxSpeedMpsVehicle;
|
||||
default -> maxSpeedMpsVehicle;
|
||||
};
|
||||
}
|
||||
|
||||
private double calculateDistanceMeters(double lat1, double lon1, double lat2, double lon2) {
|
||||
double lat1Rad = Math.toRadians(lat1);
|
||||
double lat2Rad = Math.toRadians(lat2);
|
||||
double deltaLat = lat2Rad - lat1Rad;
|
||||
double deltaLon = Math.toRadians(lon2 - lon1);
|
||||
|
||||
double a = Math.sin(deltaLat / 2.0) * Math.sin(deltaLat / 2.0)
|
||||
+ Math.cos(lat1Rad) * Math.cos(lat2Rad)
|
||||
* Math.sin(deltaLon / 2.0) * Math.sin(deltaLon / 2.0);
|
||||
double c = 2.0 * Math.atan2(Math.sqrt(a), Math.sqrt(1.0 - a));
|
||||
return EARTH_RADIUS_METERS * c;
|
||||
}
|
||||
|
||||
private double clamp(double value, double min, double max) {
|
||||
if (value < min) {
|
||||
return min;
|
||||
}
|
||||
return Math.min(value, max);
|
||||
}
|
||||
}
|
||||
|
||||
12
命令.md
12
命令.md
@ -293,3 +293,15 @@ root@root:/home/project_20250804/qaup# curl -s -X POST http://10.64.58.228:8086/
|
||||
root@root:/home/project_20250804/qaup#
|
||||
root@root:/home/project_20250804/qaup#
|
||||
```
|
||||
|
||||
docker exec -i qaup-redis redis-cli HMGET "flightBiz:" inRunway outRunway inRunwayTs outRunwayTs
|
||||
|
||||
for F in MU9692; do
|
||||
BIZ=$(docker exec -i "qaup-redis" redis-cli HGET "flight:$F" activeBizKey | tr -d '\r' | sed 's/^"//;s/"$//')
|
||||
FR=$(docker exec -i "qaup-redis" redis-cli HMGET "flight:$F" startSeat seat | tr '\n' ' ')
|
||||
BR=""
|
||||
if [ -n "$BIZ" ]; then
|
||||
BR=$(docker exec -i "qaup-redis" redis-cli HMGET "flightBiz:$BIZ" startSeat seat | tr '\n' ' ')
|
||||
fi
|
||||
echo "$F | biz=$BIZ | flight=[$FR] | biz=[$BR]"
|
||||
done
|
||||
Loading…
Reference in New Issue
Block a user