From 77f5cdc88003b9b0dfbc706d65f2d4dfdb78a306 Mon Sep 17 00:00:00 2001 From: Tian jianyong <11429339@qq.com> Date: Thu, 7 Aug 2025 10:54:15 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E6=B6=88=E6=81=AF=E6=A0=BC?= =?UTF-8?q?=E5=BC=8F=E5=A4=84=E7=90=86=E5=92=8Cwebsocket=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E7=BB=9F=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- qaup-admin/src/main/resources/application.yml | 10 - .../server/TrafficLightSignalHandler.java | 4 +- .../server/TrafficLightTcpServer.java | 146 +++++- .../service/TrafficLightDataCollector.java | 16 +- .../parser/TrafficLightSignalParser.java | 143 +++--- ...DataProcessingServiceTrafficLightTest.java | 2 - tools/test_traffic_light.py | 436 ++++++++++-------- tools/test_websocket.html | 117 ++++- 8 files changed, 566 insertions(+), 308 deletions(-) diff --git a/qaup-admin/src/main/resources/application.yml b/qaup-admin/src/main/resources/application.yml index 5aaabfb4..b00be434 100644 --- a/qaup-admin/src/main/resources/application.yml +++ b/qaup-admin/src/main/resources/application.yml @@ -36,12 +36,6 @@ logging: level: com.qaup: debug org.springframework: warn - # 红绿灯系统日志级别 - com.qaup.collision.datacollector.server: INFO - com.qaup.collision.datacollector.service.TrafficLightDataCollector: INFO - com.qaup.collision.dataprocessing.parser.TrafficLightSignalParser: INFO - com.qaup.collision.dataprocessing.service.TrafficLightService: INFO - com.qaup.collision.dataprocessing.service.IntersectionService: INFO # 用户配置 user: @@ -251,10 +245,6 @@ traffic: intersection: # 默认路口ID（当信号中没有指定时使用） default-id: "DEFAULT_INTERSECTION" - # 默认纬度坐标 - default-latitude: 0.0 - # 默认经度坐标 - default-longitude: 0.0 processing: # 是否启用统计功能 diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightSignalHandler.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightSignalHandler.java index b0ecfa70..4ca5362d 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightSignalHandler.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightSignalHandler.java @@ -12,6 +12,8 @@ public interface TrafficLightSignalHandler { * * @param clientId 客户端ID * @param signal 信号内容 + * @param ipAddress 客户端IP地址 + * @param port 客户端端口 */ - void handleSignal(String clientId, String signal); + void handleSignal(String clientId, String signal, String ipAddress, Integer port); } \ No newline at end of file diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightTcpServer.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightTcpServer.java index ded04c9d..64554b7f 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightTcpServer.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/server/TrafficLightTcpServer.java @@ -174,24 +174,26 @@ public class TrafficLightTcpServer { // 设置连接超时 clientSocket.setSoTimeout(connectionTimeout); - // 创建输入流读取器 - BufferedReader reader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream(), "UTF-8")); + // 使用字节流读取数据（兼容真实红绿灯设备） + java.io.InputStream inputStream = clientSocket.getInputStream(); + byte[] buffer = new byte[1024]; + StringBuilder messageBuffer = new StringBuilder(); - String message; - while (serverRunning.get() && !clientSocket.isClosed() && (message = reader.readLine()) != null) { - if (!message.trim().isEmpty()) { - // 更新连接活动时间 - connection.updateLastActivity(); - - // 处理接收到的消息 - processReceivedMessage(clientId, message.trim()); - - totalMessages.incrementAndGet(); - log.debug("🚦 接收到红绿灯信号: {} - {}", clientId, message); + while (serverRunning.get() && !clientSocket.isClosed()) { + int bytesRead = inputStream.read(buffer); + if (bytesRead == -1) { + break; // 连接关闭 } + + // 将读取的字节转换为字符串 + String data = new String(buffer, 0, bytesRead, "UTF-8"); + messageBuffer.append(data); + + // 处理缓冲区中的完整JSON消息 + processBufferedMessages(clientId, messageBuffer, connection); } - } catch (SocketTimeoutException e) { + } catch (java.net.SocketTimeoutException e) { log.warn("客户端连接超时: {}", clientId); } catch (IOException e) { if (serverRunning.get()) { @@ -204,16 +206,126 @@ public class TrafficLightTcpServer { } } + /** + * 处理缓冲区中的消息，提取完整的JSON对象 + * + * @param clientId 客户端ID + * @param messageBuffer 消息缓冲区 + * @param connection 客户端连接 + */ + private void processBufferedMessages(String clientId, StringBuilder messageBuffer, ClientConnection connection) { + String bufferContent = messageBuffer.toString(); + int startIndex = 0; + + while (startIndex < bufferContent.length()) { + // 查找JSON对象的开始位置 + int jsonStart = bufferContent.indexOf('{', startIndex); + if (jsonStart == -1) { + break; // 没有找到JSON开始标记 + } + + // 查找匹配的JSON对象结束位置 + int jsonEnd = findJsonEnd(bufferContent, jsonStart); + if (jsonEnd == -1) { + // 没有找到完整的JSON对象，保留剩余数据 + messageBuffer.delete(0, jsonStart); + return; + } + + // 提取完整的JSON消息 + String jsonMessage = bufferContent.substring(jsonStart, jsonEnd + 1); + + // 更新连接活动时间 + connection.updateLastActivity(); + + // 处理接收到的消息 + processReceivedMessage(clientId, jsonMessage.trim(), connection.getSocket()); + + totalMessages.incrementAndGet(); + log.debug("🚦 接收到红绿灯信号: {} - {}", clientId, jsonMessage); + + // 移动到下一个可能的JSON对象 + startIndex = jsonEnd + 1; + } + + // 清理已处理的数据 + if (startIndex > 0) { + messageBuffer.delete(0, startIndex); + } + } + + /** + * 查找JSON对象的结束位置（通过大括号匹配） + * + * @param content 内容字符串 + * @param startIndex JSON开始位置 + * @return JSON结束位置，如果没有找到返回-1 + */ + private int findJsonEnd(String content, int startIndex) { + int braceCount = 0; + boolean inString = false; + boolean escaped = false; + + for (int i = startIndex; i < content.length(); i++) { + char c = content.charAt(i); + + if (escaped) { + escaped = false; + continue; + } + + if (c == '\\') { + escaped = true; + continue; + } + + if (c == '"') { + inString = !inString; + continue; + } + + if (!inString) { + if (c == '{') { + braceCount++; + } else if (c == '}') { + braceCount--; + if (braceCount == 0) { + return i; // 找到匹配的结束大括号 + } + } + } + } + + return -1; // 没有找到完整的JSON对象 + } + /** * 处理接收到的消息 * * @param clientId 客户端ID * @param message 消息内容 + * @param clientSocket 客户端socket连接 */ - private void processReceivedMessage(String clientId, String message) { + private void processReceivedMessage(String clientId, String message, Socket clientSocket) { try { - // 委托给信号处理器处理 - signalHandler.handleSignal(clientId, message); + // 忽略协议消息（rw_prot），只处理DI信号消息 + if (message.contains("\"rw_prot\"")) { + log.debug("🚦 忽略协议消息: clientId={}, message={}", clientId, message); + return; + } + + // 只处理包含DI信号的消息 + if (message.contains("\"DI-")) { + // 提取客户端IP地址和端口 + String ipAddress = clientSocket.getInetAddress().getHostAddress(); + int port = clientSocket.getPort(); + + // 委托给信号处理器处理，传递网络地址信息 + signalHandler.handleSignal(clientId, message, ipAddress, port); + log.debug("🚦 处理DI信号: clientId={}, ip={}, port={}", clientId, ipAddress, port); + } else { + log.debug("🚦 忽略非DI信号消息: clientId={}, message={}", clientId, message); + } } catch (Exception e) { log.error("🚦 处理红绿灯信号异常: clientId={}, message={}", clientId, message, e); errorCount.incrementAndGet(); diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/TrafficLightDataCollector.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/TrafficLightDataCollector.java index db68bfc9..290ed659 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/TrafficLightDataCollector.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/TrafficLightDataCollector.java @@ -42,9 +42,11 @@ public class TrafficLightDataCollector implements TrafficLightSignalHandler { * * @param clientId 客户端ID * @param signal 信号内容 + * @param ipAddress 客户端IP地址 + * @param port 客户端端口 */ @Override - public void handleSignal(String clientId, String signal) { + public void handleSignal(String clientId, String signal, String ipAddress, Integer port) { totalSignalsReceived.incrementAndGet(); if (signal == null || signal.trim().isEmpty()) { @@ -59,18 +61,20 @@ public class TrafficLightDataCollector implements TrafficLightSignalHandler { // 记录调试日志 if (trafficLightProperties.getProcessing().isEnableDebugLog()) { - log.debug("处理红绿灯信号: clientId={}, signal={}", clientId, signal); + log.debug("处理红绿灯信号: clientId={}, signal={}, ip={}, port={}", + clientId, signal, ipAddress, port); } - // 委托给数据处理服务处理 - dataProcessingService.processTrafficLightSignal(signal.trim()); + // 委托给数据处理服务处理，传递网络地址信息 + dataProcessingService.processTrafficLightSignalWithAddress(signal.trim(), ipAddress, port); validSignalsProcessed.incrementAndGet(); - log.debug("🚦 成功处理红绿灯信号: clientId={}", clientId); + log.debug("🚦 成功处理红绿灯信号: clientId={}, ip={}, port={}", clientId, ipAddress, port); } catch (Exception e) { - log.error("🚦 处理红绿灯信号异常: clientId={}, signal={}", clientId, signal, e); + log.error("🚦 处理红绿灯信号异常: clientId={}, signal={}, ip={}, port={}", + clientId, signal, ipAddress, port, e); processingErrors.incrementAndGet(); } } diff --git a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/parser/TrafficLightSignalParser.java b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/parser/TrafficLightSignalParser.java index b7a4dd28..de7d90ae 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/parser/TrafficLightSignalParser.java +++ b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/parser/TrafficLightSignalParser.java @@ -24,68 +24,68 @@ import java.util.concurrent.atomic.AtomicLong; @Slf4j @Component public class TrafficLightSignalParser { - + private final ObjectMapper objectMapper; - + // 解析统计信息 private final AtomicLong totalParsed = new AtomicLong(0); private final AtomicLong successfulParsed = new AtomicLong(0); private final AtomicLong failedParsed = new AtomicLong(0); private final AtomicLong invalidSignals = new AtomicLong(0); - + public TrafficLightSignalParser() { this.objectMapper = new ObjectMapper(); } - + /** * 解析红绿灯JSON信号并结合连接信息创建TrafficLightStatus对象 * * @param jsonSignal 纯JSON信号字符串，如: {"DI-01":0,"DI-11":1,"DI-12":0,...} - * @param ipAddress 从socket连接中获取的IP地址 - * @param port 从socket连接中获取的端口号 + * @param ipAddress 从socket连接中获取的IP地址 + * @param port 从socket连接中获取的端口号 * @return 解析后的红绿灯状态，解析失败时返回安全默认状态 */ public TrafficLightStatus parseSignalWithAddress(String jsonSignal, String ipAddress, Integer port) { totalParsed.incrementAndGet(); - + if (!isValidSignal(jsonSignal)) { log.warn("🚦 无效的红绿灯JSON信号格式: {}", jsonSignal); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault(generateDeviceIdentifier(ipAddress, port)); } - + if (ipAddress == null || ipAddress.trim().isEmpty()) { log.warn("🚦 IP地址不能为空"); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault("UNKNOWN_DEVICE"); } - + // 验证IP地址格式 if (!isValidIpAddressFormat(ipAddress.trim())) { log.warn("🚦 IP地址格式无效: {}", ipAddress); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault(generateDeviceIdentifier(ipAddress, port)); } - + if (port == null || port <= 0 || port > 65535) { log.warn("🚦 端口号无效: {}", port); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault(generateDeviceIdentifier(ipAddress, port)); } - + try { // 解析JSON数据 JsonNode rootNode = objectMapper.readTree(jsonSignal); - + // 解析南北方向信号状态 (DI-11: 北红, DI-12: 北黄, DI-13: 北绿) SignalState nsStatus = parseDirectionSignal(rootNode, "DI-11", "DI-12", "DI-13"); - + // 解析东西方向信号状态 (DI-14: 东红, DI-15: 东黄, DI-16: 东绿) SignalState ewStatus = parseDirectionSignal(rootNode, "DI-14", "DI-15", "DI-16"); - + // 生成设备标识符（使用IP:端口组合） String deviceIdentifier = generateDeviceIdentifier(ipAddress, port); - + // 创建状态对象 TrafficLightStatus status = TrafficLightStatus.builder() .deviceId(null) // 设备ID可能为空，使用IP:端口作为标识 @@ -96,37 +96,37 @@ public class TrafficLightSignalParser { .timestamp(System.currentTimeMillis() * 1000) // 微秒级时间戳 .rawSignal(jsonSignal) .build(); - + // 验证解析结果 if (!status.isValid()) { - log.warn("🚦 解析的红绿灯状态无效: ipAddress={}, port={}, nsStatus={}, ewStatus={}", + log.warn("🚦 解析的红绿灯状态无效: ipAddress={}, port={}, nsStatus={}, ewStatus={}", ipAddress, port, nsStatus, ewStatus); invalidSignals.incrementAndGet(); return TrafficLightStatus.createSafeDefault(deviceIdentifier); } - + // 检查冲突状态 if (status.hasConflict()) { - log.error("🚦 检测到红绿灯冲突状态（两个方向都是绿灯）: ipAddress={}, port={}, 状态={}", + log.error("🚦 检测到红绿灯冲突状态（两个方向都是绿灯）: ipAddress={}, port={}, 状态={}", ipAddress, port, status.getStatusDescription()); // 冲突时返回安全状态 return TrafficLightStatus.createSafeDefault(deviceIdentifier); } - + successfulParsed.incrementAndGet(); - log.debug("🚦 成功解析红绿灯信号: ipAddress={}, port={}, 状态={}", - ipAddress, port, status.getStatusDescription()); - + log.debug("🚦 成功解析红绿灯信号: ipAddress={}, port={}, 状态={}", + ipAddress, port, status.getStatusDescription()); + return status; - + } catch (Exception e) { - log.error("🚦 解析红绿灯信号异常: jsonSignal={}, ipAddress={}, port={}", - jsonSignal, ipAddress, port, e); + log.error("🚦 解析红绿灯信号异常: jsonSignal={}, ipAddress={}, port={}", + jsonSignal, ipAddress, port, e); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault(generateDeviceIdentifier(ipAddress, port)); } } - + /** * 解析原始JSON信号为TrafficLightStatus对象（兼容旧格式） * @@ -135,16 +135,16 @@ public class TrafficLightSignalParser { */ public TrafficLightStatus parseSignal(String rawJsonSignal) { totalParsed.incrementAndGet(); - + if (!isValidSignal(rawJsonSignal)) { log.warn("🚦 无效的红绿灯信号格式: {}", rawJsonSignal); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault("UNKNOWN_DEVICE"); } - + try { JsonNode rootNode = objectMapper.readTree(rawJsonSignal); - + // 提取设备ID String deviceId = extractDeviceId(rootNode); if (deviceId == null) { @@ -152,13 +152,13 @@ public class TrafficLightSignalParser { failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault("UNKNOWN_DEVICE"); } - + // 解析南北方向信号状态 (DI-11: 北红, DI-12: 北黄, DI-13: 北绿) SignalState nsStatus = parseDirectionSignal(rootNode, "DI-11", "DI-12", "DI-13"); - + // 解析东西方向信号状态 (DI-14: 东红, DI-15: 东黄, DI-16: 东绿) SignalState ewStatus = parseDirectionSignal(rootNode, "DI-14", "DI-15", "DI-16"); - + // 创建状态对象 TrafficLightStatus status = TrafficLightStatus.builder() .deviceId(deviceId) @@ -167,35 +167,35 @@ public class TrafficLightSignalParser { .timestamp(System.currentTimeMillis() * 1000) // 微秒级时间戳 .rawSignal(rawJsonSignal) .build(); - + // 验证解析结果 if (!status.isValid()) { - log.warn("🚦 解析的红绿灯状态无效: deviceId={}, nsStatus={}, ewStatus={}", + log.warn("🚦 解析的红绿灯状态无效: deviceId={}, nsStatus={}, ewStatus={}", deviceId, nsStatus, ewStatus); invalidSignals.incrementAndGet(); return TrafficLightStatus.createSafeDefault(deviceId); } - + // 检查冲突状态 if (status.hasConflict()) { - log.error("🚦 检测到红绿灯冲突状态（两个方向都是绿灯）: deviceId={}, 状态={}", + log.error("🚦 检测到红绿灯冲突状态（两个方向都是绿灯）: deviceId={}, 状态={}", deviceId, status.getStatusDescription()); // 冲突时返回安全状态 return TrafficLightStatus.createSafeDefault(deviceId); } - + successfulParsed.incrementAndGet(); log.debug("🚦 成功解析红绿灯信号: deviceId={}, 状态={}", deviceId, status.getStatusDescription()); - + return status; - + } catch (Exception e) { log.error("🚦 解析红绿灯信号异常: {}", rawJsonSignal, e); failedParsed.incrementAndGet(); return TrafficLightStatus.createSafeDefault("UNKNOWN_DEVICE"); } } - + /** * 验证IP地址格式是否有效 * @@ -206,13 +206,13 @@ public class TrafficLightSignalParser { if (ipAddress == null || ipAddress.trim().isEmpty()) { return false; } - + try { String[] parts = ipAddress.split("\\."); if (parts.length != 4) { return false; } - + for (String part : parts) { int num = Integer.parseInt(part); if (num < 0 || num > 255) { @@ -224,12 +224,12 @@ public class TrafficLightSignalParser { return false; } } - + /** * 生成设备标识符 * * @param ipAddress IP地址 - * @param port 端口号 + * @param port 端口号 * @return 设备标识符，格式为 "IP:端口" */ private String generateDeviceIdentifier(String ipAddress, Integer port) { @@ -241,7 +241,7 @@ public class TrafficLightSignalParser { } return ipAddress.trim() + ":" + port; } - + /** * 验证信号格式是否有效（兼容旧格式） * @@ -252,27 +252,27 @@ public class TrafficLightSignalParser { if (rawJsonSignal == null || rawJsonSignal.trim().isEmpty()) { return false; } - + try { JsonNode rootNode = objectMapper.readTree(rawJsonSignal); - + // 检查是否为JSON对象 if (!rootNode.isObject()) { return false; } - + // 检查是否包含必要的DI字段 boolean hasRequiredFields = rootNode.has("DI-11") || rootNode.has("DI-12") || rootNode.has("DI-13") || - rootNode.has("DI-14") || rootNode.has("DI-15") || rootNode.has("DI-16"); - + rootNode.has("DI-14") || rootNode.has("DI-15") || rootNode.has("DI-16"); + return hasRequiredFields; - + } catch (Exception e) { log.debug("信号格式验证失败: {}", rawJsonSignal, e); return false; } } - + /** * 提取设备ID * 优先从device_id字段获取，如果没有则尝试其他可能的字段 @@ -285,35 +285,36 @@ public class TrafficLightSignalParser { if (rootNode.has("device_id")) { return rootNode.get("device_id").asText(); } - + // 尝试其他可能的字段名 - String[] possibleFields = {"deviceId", "id", "device", "equipment_id"}; + String[] possibleFields = { "deviceId", "id", "device", "equipment_id" }; for (String field : possibleFields) { if (rootNode.has(field)) { return rootNode.get(field).asText(); } } - + // 如果都没有，返回null，调用方会使用默认值 return null; } - + /** * 解析单个方向的信号状态 * - * @param rootNode JSON根节点 - * @param redField 红灯字段名 + * @param rootNode JSON根节点 + * @param redField 红灯字段名 * @param yellowField 黄灯字段名 - * @param greenField 绿灯字段名 + * @param greenField 绿灯字段名 * @return 该方向的信号状态 */ - private SignalState parseDirectionSignal(JsonNode rootNode, String redField, String yellowField, String greenField) { + private SignalState parseDirectionSignal(JsonNode rootNode, String redField, String yellowField, + String greenField) { try { // 获取各个信号的值（1表示激活，0表示未激活） int redValue = rootNode.has(redField) ? rootNode.get(redField).asInt(0) : 0; int yellowValue = rootNode.has(yellowField) ? rootNode.get(yellowField).asInt(0) : 0; int greenValue = rootNode.has(greenField) ? rootNode.get(greenField).asInt(0) : 0; - + // 根据优先级确定状态：红灯 > 黄灯 > 绿灯 if (redValue == 1) { return SignalState.RED; @@ -323,18 +324,18 @@ public class TrafficLightSignalParser { return SignalState.GREEN; } else { // 如果没有任何信号激活，默认为红灯（安全状态） - log.debug("方向信号无激活状态，默认为红灯: red={}, yellow={}, green={}", - redValue, yellowValue, greenValue); + log.debug("方向信号无激活状态，默认为红灯: red={}, yellow={}, green={}", + redValue, yellowValue, greenValue); return SignalState.RED; } - + } catch (Exception e) { - log.warn("解析方向信号异常，默认为红灯: redField={}, yellowField={}, greenField={}", + log.warn("解析方向信号异常，默认为红灯: redField={}, yellowField={}, greenField={}", redField, yellowField, greenField, e); return SignalState.RED; } } - + /** * 获取解析统计信息 * @@ -349,7 +350,7 @@ public class TrafficLightSignalParser { .successRate(calculateSuccessRate()) .build(); } - + /** * 计算成功率 * @@ -362,7 +363,7 @@ public class TrafficLightSignalParser { } return (double) successfulParsed.get() / total * 100.0; } - + /** * 重置统计信息 */ @@ -372,7 +373,7 @@ public class TrafficLightSignalParser { failedParsed.set(0); invalidSignals.set(0); } - + /** * 解析统计信息数据类 */ @@ -386,7 +387,7 @@ public class TrafficLightSignalParser { private long failedParsed; private long invalidSignals; private double successRate; - + @Override public String toString() { return String.format("解析统计 - 总计:%d, 成功:%d, 失败:%d, 无效:%d, 成功率:%.2f%%", diff --git a/qaup-collision/src/test/java/com/qaup/collision/dataprocessing/service/DataProcessingServiceTrafficLightTest.java b/qaup-collision/src/test/java/com/qaup/collision/dataprocessing/service/DataProcessingServiceTrafficLightTest.java index aad66020..da77c588 100644 --- a/qaup-collision/src/test/java/com/qaup/collision/dataprocessing/service/DataProcessingServiceTrafficLightTest.java +++ b/qaup-collision/src/test/java/com/qaup/collision/dataprocessing/service/DataProcessingServiceTrafficLightTest.java @@ -5,7 +5,6 @@ import com.qaup.collision.common.model.spatial.Intersection; import com.qaup.collision.dataprocessing.model.TrafficLightStatus; import com.qaup.collision.dataprocessing.parser.TrafficLightSignalParser; import com.qaup.collision.common.model.enums.SignalState; -import com.qaup.collision.websocket.message.TrafficLightStatusPayload; import com.qaup.collision.websocket.event.TrafficLightStatusEvent; import org.junit.jupiter.api.Test; @@ -18,7 +17,6 @@ import org.springframework.context.ApplicationEventPublisher; import java.util.Optional; -import static org.junit.jupiter.api.Assertions.*; import static org.mockito.Mockito.*; /** diff --git a/tools/test_traffic_light.py b/tools/test_traffic_light.py index e84971cd..dc5789b7 100644 --- a/tools/test_traffic_light.py +++ b/tools/test_traffic_light.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 """ -红绿灯IP地址增强功能端到端测试脚本 +红绿灯设备模拟器 -此脚本模拟红绿灯硬件发送包含IP地址和端口信息的消息， -用于测试整个系统的端到端功能。 +此脚本模拟真实的红绿灯硬件设备，通过TCP连接发送纯JSON格式的DI信号数据。 +服务器会从TCP连接中获取设备的IP地址和端口信息。 使用方法: - python test_traffic_light_ip_enhancement.py --host localhost --port 8082 + python test_traffic_light.py --host localhost --port 8082 --interval 1 """ import socket @@ -15,24 +15,24 @@ import json import argparse import sys from datetime import datetime -import threading +import signal -class TrafficLightSimulator: - """红绿灯模拟器""" +class TrafficLightDevice: + """红绿灯设备模拟器""" - def __init__(self, host='localhost', port=8082): + def __init__(self, host='localhost', port=8082, interval=1): self.host = host self.port = port + self.interval = interval # 发送间隔（秒） self.running = False + self.socket = None - def create_message(self, ip_address, device_port, ns_red=0, ns_yellow=0, ns_green=0, - ew_red=0, ew_yellow=0, ew_green=0): + def create_di_signal(self, ns_red=0, ns_yellow=0, ns_green=0, + ew_red=0, ew_yellow=0, ew_green=0): """ - 创建红绿灯消息 + 创建红绿灯DI信号数据（纯JSON格式） Args: - ip_address: 设备IP地址 - device_port: 设备端口 ns_red: 南北红灯状态 (0或1) ns_yellow: 南北黄灯状态 (0或1) ns_green: 南北绿灯状态 (0或1) @@ -41,9 +41,9 @@ class TrafficLightSimulator: ew_green: 东西绿灯状态 (0或1) Returns: - 格式化的消息字符串 + JSON格式的DI信号字符串 """ - # DI信号数据 + # DI信号数据（真实红绿灯设备发送的格式） di_data = { "DI-01": 0, "DI-02": 0, @@ -57,228 +57,264 @@ class TrafficLightSimulator: "DI-18": 0 } - # 创建完整消息格式: ('IP地址', 端口) - {DI数据} - message = f"('{ip_address}', {device_port}) - {json.dumps(di_data, separators=(',', ':'))}" - return message + return json.dumps(di_data, separators=(',', ':')) - def send_message(self, message): - """发送消息到服务器""" + def connect(self): + """连接到服务器""" try: - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.settimeout(5) # 5秒超时 - sock.connect((self.host, self.port)) - sock.sendall(message.encode('utf-8')) - print(f"[{datetime.now().strftime('%H:%M:%S')}] 发送消息: {message[:100]}...") - return True + # 如果已有连接，先关闭 + if self.socket: + try: + self.socket.close() + except: + pass + self.socket = None + + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + # 设置连接超时 + self.socket.settimeout(10) + self.socket.connect((self.host, self.port)) + # 连接成功后取消超时设置 + self.socket.settimeout(None) + print(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}") + print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 连接失败: {e}") + if self.socket: + try: + self.socket.close() + except: + pass + self.socket = None return False - def simulate_traffic_light_cycle(self, ip_address, device_port, cycles=5): + def disconnect(self): + """断开连接""" + if self.socket: + try: + self.socket.close() + print(f"[{datetime.now().strftime('%H:%M:%S')}] 已断开连接") + except: + pass + self.socket = None + + def send_signal(self, di_signal): + """发送DI信号到服务器""" + if not self.socket: + print("未连接到服务器") + return False + + try: + self.socket.sendall(di_signal.encode('utf-8')) + print(f"[{datetime.now().strftime('%H:%M:%S')}] 📡 发送信号: {di_signal}") + return True + except Exception as e: + print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ 发送失败: {e}") + # 发送失败时关闭socket，触发重连 + try: + self.socket.close() + except: + pass + self.socket = None + return False + + def get_current_state(self, cycle_time): """ - 模拟红绿灯周期变化 + 根据时间计算当前红绿灯状态 Args: - ip_address: 设备IP地址 - device_port: 设备端口 - cycles: 循环次数 + cycle_time: 当前周期时间（秒） + + Returns: + tuple: (状态名称, (ns_red, ns_yellow, ns_green), (ew_red, ew_yellow, ew_green)) """ - print(f"开始模拟红绿灯 {ip_address}:{device_port} 的信号周期...") - - # 红绿灯状态序列：南北红+东西绿 -> 南北红+东西黄 -> 南北绿+东西红 -> 南北黄+东西红 + # 红绿灯状态序列，每个状态持续时间（秒） states = [ + {"name": "南北红+东西绿", "duration": 30, "ns": (1,0,0), "ew": (0,0,1)}, + {"name": "南北红+东西黄", "duration": 3, "ns": (1,0,0), "ew": (0,1,0)}, + {"name": "南北绿+东西红", "duration": 25, "ns": (0,0,1), "ew": (1,0,0)}, + {"name": "南北黄+东西红", "duration": 3, "ns": (0,1,0), "ew": (1,0,0)}, + ] + + # 计算总周期时间 + total_cycle_time = sum(state["duration"] for state in states) + + # 计算当前在周期中的位置 + current_time = cycle_time % total_cycle_time + + # 找到当前状态 + elapsed = 0 + for state in states: + if current_time < elapsed + state["duration"]: + return state["name"], state["ns"], state["ew"] + elapsed += state["duration"] + + # 默认返回第一个状态 + return states[0]["name"], states[0]["ns"], states[0]["ew"] + + def run_continuous_simulation(self): + """运行连续的红绿灯信号模拟""" + print(f"开始连续发送红绿灯信号，间隔: {self.interval}秒") + print("按 Ctrl+C 停止模拟") + print("服务端断开时会自动重连...") + + self.running = True + start_time = time.time() + connected = False + + try: + while self.running: + # 如果未连接，尝试连接 + if not connected: + connected = self.connect() + if not connected: + print("连接失败，5秒后重试...") + time.sleep(5) + continue + else: + # 重新连接成功，重置开始时间以保持状态同步 + start_time = time.time() + + # 计算当前周期时间 + current_time = time.time() - start_time + + # 获取当前状态 + state_name, ns_state, ew_state = self.get_current_state(current_time) + + # 创建DI信号 + di_signal = self.create_di_signal( + ns_state[0], ns_state[1], ns_state[2], # 南北红黄绿 + ew_state[0], ew_state[1], ew_state[2] # 东西红黄绿 + ) + + # 发送信号 + success = self.send_signal(di_signal) + if not success: + print("发送失败，连接已断开") + self.disconnect() + connected = False + continue + + print(f"当前状态: {state_name}") + + # 等待下一次发送 + time.sleep(self.interval) + + except KeyboardInterrupt: + print("\n收到停止信号...") + finally: + self.running = False + self.disconnect() + print("红绿灯模拟已停止") + + def test_signal_states(self): + """测试各种信号状态""" + if not self.connect(): + return + + print("开始测试各种红绿灯信号状态...") + + test_cases = [ {"name": "南北红+东西绿", "ns": (1,0,0), "ew": (0,0,1)}, {"name": "南北红+东西黄", "ns": (1,0,0), "ew": (0,1,0)}, {"name": "南北绿+东西红", "ns": (0,0,1), "ew": (1,0,0)}, {"name": "南北黄+东西红", "ns": (0,1,0), "ew": (1,0,0)}, + {"name": "全红状态", "ns": (1,0,0), "ew": (1,0,0)}, + {"name": "无信号状态", "ns": (0,0,0), "ew": (0,0,0)}, + {"name": "冲突状态（两绿）", "ns": (0,0,1), "ew": (0,0,1)}, ] - for cycle in range(cycles): - print(f"\n=== 第 {cycle + 1} 个周期 ===") - - for state in states: - if not self.running: - break - - ns_red, ns_yellow, ns_green = state["ns"] - ew_red, ew_yellow, ew_green = state["ew"] + try: + for i, case in enumerate(test_cases, 1): + print(f"\n测试 {i}/{len(test_cases)}: {case['name']}") - message = self.create_message( - ip_address, device_port, - ns_red, ns_yellow, ns_green, - ew_red, ew_yellow, ew_green + di_signal = self.create_di_signal( + case["ns"][0], case["ns"][1], case["ns"][2], + case["ew"][0], case["ew"][1], case["ew"][2] ) - print(f"状态: {state['name']}") - success = self.send_message(message) - + success = self.send_signal(di_signal) if not success: - print(f"发送失败，跳过此状态") - continue + print("发送失败") + break - # 每个状态持续3秒 - time.sleep(3) - - if not self.running: - break - - print(f"\n红绿灯 {ip_address}:{device_port} 模拟完成") + time.sleep(2) # 每个状态持续2秒 + + except KeyboardInterrupt: + print("\n测试被中断") + finally: + self.disconnect() - def test_multiple_devices(self): - """测试多个设备同时发送信号""" - devices = [ - {"ip": "192.168.1.100", "port": 8082}, - {"ip": "192.168.1.101", "port": 8082}, - {"ip": "10.0.0.1", "port": 9090}, - {"ip": "172.16.0.1", "port": 8083}, + def test_invalid_json(self): + """测试无效JSON格式""" + if not self.connect(): + return + + print("开始测试无效JSON格式...") + + invalid_signals = [ + "invalid json", + "{\"DI-11\":\"invalid_value\"}", # 无效值类型 + "{\"invalid_field\":1}", # 无效字段 + "not json at all", + "{\"DI-11\":1,}", # 语法错误 + "", # 空字符串 ] - print("开始测试多设备并发信号发送...") - self.running = True - - threads = [] - for device in devices: - thread = threading.Thread( - target=self.simulate_traffic_light_cycle, - args=(device["ip"], device["port"], 3) - ) - threads.append(thread) - thread.start() - time.sleep(0.5) # 错开启动时间 - try: - for thread in threads: - thread.join() + for i, signal in enumerate(invalid_signals, 1): + print(f"\n测试无效JSON {i}/{len(invalid_signals)}: {signal[:30]}...") + success = self.send_signal(signal) + if not success: + print("发送失败") + time.sleep(1) + except KeyboardInterrupt: - print("\n收到中断信号，停止测试...") + print("\n测试被中断") + finally: + self.disconnect() + + def setup_signal_handler(self): + """设置信号处理器""" + def signal_handler(signum, frame): + print(f"\n收到信号 {signum}，正在停止...") self.running = False - for thread in threads: - thread.join(timeout=1) - - def test_edge_cases(self): - """测试边缘情况""" - print("开始测试边缘情况...") - test_cases = [ - { - "name": "冲突状态（两个方向都是绿灯）", - "ip": "192.168.1.200", - "port": 8084, - "ns": (0,0,1), # 南北绿 - "ew": (0,0,1), # 东西绿 - }, - { - "name": "全红状态", - "ip": "192.168.1.201", - "port": 8085, - "ns": (1,0,0), # 南北红 - "ew": (1,0,0), # 东西红 - }, - { - "name": "无信号状态", - "ip": "192.168.1.202", - "port": 8086, - "ns": (0,0,0), # 南北无信号 - "ew": (0,0,0), # 东西无信号 - }, - ] - - for case in test_cases: - print(f"\n测试: {case['name']}") - message = self.create_message( - case["ip"], case["port"], - case["ns"][0], case["ns"][1], case["ns"][2], - case["ew"][0], case["ew"][1], case["ew"][2] - ) - self.send_message(message) - time.sleep(1) - - def test_invalid_formats(self): - """测试无效格式消息""" - print("开始测试无效格式消息...") - - invalid_messages = [ - "invalid message", - "('192.168.1.1') - {\"DI-11\":1}", # 缺少端口 - "192.168.1.1, 8082 - {\"DI-11\":1}", # 格式错误 - "('192.168.1.1', 8082)", # 缺少DI数据 - "('192.168.1.1', 8082) - invalid_json", # 无效JSON - "('256.1.1.1', 8082) - {\"DI-11\":1}", # 无效IP地址 - "('192.168.1.1', 70000) - {\"DI-11\":1}", # 无效端口 - ] - - for i, message in enumerate(invalid_messages, 1): - print(f"\n测试无效消息 {i}: {message[:50]}...") - self.send_message(message) - time.sleep(0.5) - - def run_comprehensive_test(self): - """运行综合测试""" - print("=" * 60) - print("红绿灯IP地址增强功能综合测试") - print("=" * 60) - - try: - # 1. 测试单个设备 - print("\n1. 测试单个设备信号周期...") - self.running = True - self.simulate_traffic_light_cycle("192.168.1.100", 8082, 2) - - # 2. 测试多设备并发 - print("\n2. 测试多设备并发...") - self.test_multiple_devices() - - # 3. 测试边缘情况 - print("\n3. 测试边缘情况...") - self.test_edge_cases() - - # 4. 测试无效格式 - print("\n4. 测试无效格式...") - self.test_invalid_formats() - - print("\n" + "=" * 60) - print("综合测试完成！") - print("请检查服务器日志以验证消息处理结果。") - print("=" * 60) - - except KeyboardInterrupt: - print("\n测试被用户中断") - self.running = False - except Exception as e: - print(f"\n测试过程中发生错误: {e}") + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) def main(): - parser = argparse.ArgumentParser(description='红绿灯IP地址增强功能测试工具') + parser = argparse.ArgumentParser(description='红绿灯设备模拟器') parser.add_argument('--host', default='localhost', help='服务器主机地址 (默认: localhost)') parser.add_argument('--port', type=int, default=8082, help='服务器端口 (默认: 8082)') - parser.add_argument('--mode', choices=['single', 'multi', 'edge', 'invalid', 'comprehensive'], - default='comprehensive', help='测试模式') - parser.add_argument('--ip', default='192.168.1.100', help='设备IP地址 (单设备模式)') - parser.add_argument('--device-port', type=int, default=8082, help='设备端口 (单设备模式)') - parser.add_argument('--cycles', type=int, default=3, help='信号周期数 (单设备模式)') + parser.add_argument('--interval', type=float, default=1.0, help='信号发送间隔秒数 (默认: 1.0)') + parser.add_argument('--mode', choices=['continuous', 'test-states', 'test-invalid'], + default='continuous', help='运行模式') args = parser.parse_args() - simulator = TrafficLightSimulator(args.host, args.port) + device = TrafficLightDevice(args.host, args.port, args.interval) + device.setup_signal_handler() - print(f"连接到服务器: {args.host}:{args.port}") + print("=" * 60) + print("红绿灯设备模拟器") + print("=" * 60) + print(f"服务器地址: {args.host}:{args.port}") + print(f"发送间隔: {args.interval}秒") + print(f"运行模式: {args.mode}") + print("=" * 60) - if args.mode == 'single': - print(f"单设备测试模式: {args.ip}:{args.device_port}") - simulator.running = True - simulator.simulate_traffic_light_cycle(args.ip, args.device_port, args.cycles) - elif args.mode == 'multi': - print("多设备并发测试模式") - simulator.test_multiple_devices() - elif args.mode == 'edge': - print("边缘情况测试模式") - simulator.test_edge_cases() - elif args.mode == 'invalid': - print("无效格式测试模式") - simulator.test_invalid_formats() - else: - print("综合测试模式") - simulator.run_comprehensive_test() + if args.mode == 'continuous': + print("\n连续模拟模式 - 模拟真实红绿灯设备持续发送信号") + print("设备会通过TCP连接发送纯JSON格式的DI信号") + print("服务器将从TCP连接中获取设备的IP地址和端口") + device.run_continuous_simulation() + elif args.mode == 'test-states': + print("\n状态测试模式 - 测试各种红绿灯状态") + device.test_signal_states() + elif args.mode == 'test-invalid': + print("\n无效数据测试模式 - 测试无效JSON格式") + device.test_invalid_json() if __name__ == '__main__': main() \ No newline at end of file diff --git a/tools/test_websocket.html b/tools/test_websocket.html index 08f3047b..710f217a 100644 --- a/tools/test_websocket.html +++ b/tools/test_websocket.html @@ -91,11 +91,46 @@ .log-success { background: #d4edda; color: #155724; } .log-error { background: #f8d7da; color: #721c24; } .log-warning { background: #fff3cd; color: #856404; } + + .message-stats { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); + gap: 8px; + width: 100%; + font-size: 12px !important; + } + + .message-stats div { + font-size: 12px !important; + }
+ +
+

📊 消息统计

+
+
+
总消息数: 0
+
位置更新: 0
+
碰撞预警: 0
+
规则违规: 0
+
红绿灯状态: 0
+
路口红绿灯: 0
+
航空器路由: 0
+
路径冲突: 0
+
车辆指令: 0
+
规则执行状态: 0
+
规则状态变更: 0
+
心跳响应: 0
+
连接确认: 0
+
未知消息: 0
+
+
+
+

🚗 冲突检测WebSocket

@@ -129,6 +164,7 @@ +
@@ -138,6 +174,24 @@ let collisionWebSocket = null; let reconnectAttempts = 0; const maxReconnectAttempts = 5; + + // 消息统计 + let messageStats = { + total: 0, + position_update: 0, + collision_warning: 0, + rule_violation: 0, + traffic_light_status: 0, + intersection_traffic_light_status: 0, + aircraftRouteUpdate: 0, + path_conflict_alert: 0, + vehicle_command: 0, + rule_execution_status: 0, + rule_state_change: 0, + pong: 0, + connection: 0, + unknown: 0 + }; // 通用日志函数 function log(containerId, message, type = 'info') { @@ -156,6 +210,32 @@ statusDiv.className = `status ${connected ? 'connected' : 'disconnected'}`; statusDiv.textContent = `状态：${connected ? '已连接' : '未连接'}${message ? ' - ' + message : ''}`; } + + // 更新消息统计 + function updateMessageStats(messageType) { + messageStats.total++; + if (messageStats.hasOwnProperty(messageType)) { + messageStats[messageType]++; + } else { + messageStats.unknown++; + } + + // 更新页面显示 + document.getElementById('totalCount').textContent = messageStats.total; + document.getElementById('positionUpdateCount').textContent = messageStats.position_update; + document.getElementById('collisionWarningCount').textContent = messageStats.collision_warning; + document.getElementById('ruleViolationCount').textContent = messageStats.rule_violation; + document.getElementById('trafficLightStatusCount').textContent = messageStats.traffic_light_status; + document.getElementById('intersectionTrafficLightCount').textContent = messageStats.intersection_traffic_light_status; + document.getElementById('aircraftRouteCount').textContent = messageStats.aircraftRouteUpdate; + document.getElementById('pathConflictCount').textContent = messageStats.path_conflict_alert; + document.getElementById('vehicleCommandCount').textContent = messageStats.vehicle_command; + document.getElementById('ruleExecutionCount').textContent = messageStats.rule_execution_status; + document.getElementById('ruleStateChangeCount').textContent = messageStats.rule_state_change; + document.getElementById('pongCount').textContent = messageStats.pong; + document.getElementById('connectionCount').textContent = messageStats.connection; + document.getElementById('unknownCount').textContent = messageStats.unknown; + } // =================== 冲突检测WebSocket =================== function connectCollisionWebSocket() { @@ -246,6 +326,9 @@ } function handleCollisionMessage(message) { + // 更新消息统计 + updateMessageStats(message.type); + switch (message.type) { case 'connection': log('collisionLog', `连接确认: ${message.message}`, 'success'); @@ -282,6 +365,13 @@ case 'traffic_light_status': log('collisionLog', `红绿灯状态: ${message.payload?.intersection_id || '未知路口'}`, 'info'); break; + case 'intersection_traffic_light_status': + const intersectionId = message.payload?.intersection_id || '未知路口'; + const deviceId = message.payload?.device_id || '未知设备'; + const nsStatus = message.payload?.ns_status || '未知'; + const ewStatus = message.payload?.ew_status || '未知'; + log('collisionLog', `🚦 路口红绿灯状态: ${intersectionId} - 设备${deviceId} - 南北:${nsStatus} 东西:${ewStatus}`, 'info'); + break; case 'aircraftRouteUpdate': const flightNo = message.payload?.flightNo || '未知航班'; const routeType = message.payload?.routeType || '未知路由'; @@ -324,6 +414,31 @@ }, i * 100); } } + + function resetMessageStats() { + // 重置所有统计数据 + for (let key in messageStats) { + messageStats[key] = 0; + } + + // 更新页面显示 + document.getElementById('totalCount').textContent = '0'; + document.getElementById('positionUpdateCount').textContent = '0'; + document.getElementById('collisionWarningCount').textContent = '0'; + document.getElementById('ruleViolationCount').textContent = '0'; + document.getElementById('trafficLightStatusCount').textContent = '0'; + document.getElementById('intersectionTrafficLightCount').textContent = '0'; + document.getElementById('aircraftRouteCount').textContent = '0'; + document.getElementById('pathConflictCount').textContent = '0'; + document.getElementById('vehicleCommandCount').textContent = '0'; + document.getElementById('ruleExecutionCount').textContent = '0'; + document.getElementById('ruleStateChangeCount').textContent = '0'; + document.getElementById('pongCount').textContent = '0'; + document.getElementById('connectionCount').textContent = '0'; + document.getElementById('unknownCount').textContent = '0'; + + log('collisionLog', '消息统计已重置', 'info'); + } // 页面加载完成 window.onload = function() { @@ -336,4 +451,4 @@ }; - \ No newline at end of file + \ No newline at end of file