ThbTech/FileManage
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

代码提交点位地图2

Browse Source
This commit is contained in:
lilin 2025-08-01 10:58:32 +08:00
parent 50462faaff
commit 12cd3be1f7
1 changed files with 191 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

388
java/src/main/java/com/yfd/platform/modules/experimentalData/service/impl/TsFilesServiceImpl.java
View File

@ -19,6 +19,7 @@ import java.util.stream.Collectors;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
@ -41,7 +42,6 @@ import com.yfd.platform.component.WebSocketServer;
import com.yfd.platform.config.ResponseResult;
import com.yfd.platform.modules.experimentalData.domain.*;
import com.yfd.platform.modules.experimentalData.mapper.TsFilesMapper;
import com.yfd.platform.modules.experimentalData.mapper.TsNodesMapper;
import com.yfd.platform.modules.experimentalData.mapper.TsTaskMapper;
import com.yfd.platform.modules.experimentalData.service.ITsFilesService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@ -89,8 +89,6 @@ import java.util.zip.*;
import org.apache.commons.codec.binary.Hex;
import static java.lang.Float.parseFloat;
/**
* <p>
@ -5424,125 +5422,6 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
//判断文件后缀是.txt还是.csv
String fileName = tsFiles.getFileName();
String suffix = fileName.substring(fileName.lastIndexOf(".") + 1).toLowerCase();
if ("txt".equals(suffix)) {
if (currentTaskFuture != null && !currentTaskFuture.isDone()) {
currentTaskFuture.cancel(true); // 中断之前的任务
}
currentTaskFuture = executorService.submit(() -> {
StorageSourceConfig config = getStorageSourceConfig("filePath", "local", tsTask.getLocalStorageId());
Path filePath = Paths.get(config.getValue() + tsFiles.getWorkPath() + tsFiles.getFileName());
try (BufferedReader reader = new BufferedReader(new FileReader(filePath.toFile()))) {
reader.readLine(); // 跳过标题行
SimpleNaviData data1 = null;
int result = 0;
// 读取第二行数据
String secondLine = reader.readLine();
if (secondLine != null) {
secondLine = secondLine.trim();
String[] valuesSecond = secondLine.split("\t");
data1 = parseLine(valuesSecond);
}
// 读取第三行数据
String thirdLine = reader.readLine();
if (thirdLine != null) {
thirdLine = thirdLine.trim();
String[] valuesThird = thirdLine.split("\t");
SimpleNaviData data2 = parseLine(valuesThird);
// 假设 UtcTime 是 String 类型，首先转换为 double 类型
double time1 = Double.parseDouble(data1.getUtcTime()); // 如果 getUtcTime() 返回 String 类型
double time2 = Double.parseDouble(data2.getUtcTime()); // 如果 getUtcTime() 返回 String 类型
// 保留小数点后六位
double formattedTime1 = Double.parseDouble(String.format("%.6f", time1));
double formattedTime2 = Double.parseDouble(String.format("%.6f", time2));
// 计算差值
double diff = formattedTime2 - formattedTime1;
// 确保差值不为零，以避免除以零错误
if (diff != 0) {
result = (int) Math.floor(1 / diff);
} else {
result = 1;
}
}
// 不管 samTimes 如何，先读取第一行数据
String firstLine = reader.readLine();
if (firstLine != null) {
firstLine = firstLine.trim();
String[] values = firstLine.split("\t");
if (values.length >= 40) {
// 发送第一行数据
sendData(token, values, 1);
} else {
System.err.println("忽略不完整行: " + firstLine);
}
}
//todo 这个地方该从一下 换成200
int step = samTimes * result; // 计算行号间隔
int lineCount = 1; // 从第一行开始
while (!Thread.currentThread().isInterrupted()) {
lineCount += step; // 直接跳到下一个需要读取的行
String line = null;
// 读取指定行
for (int i = 0; i < step; i++) {
line = reader.readLine();
if (line == null) {
break; // 文件结束
}
}
if (line == null) {
break; // 文件结束
}
line = line.trim();
String[] values = line.split("\t");
if (values.length < 40) {
LOGGER.info("忽略不完整行: " + line);
continue;
}
// 发送数据
sendData(token, values, lineCount);
// 固定休眠 3 秒
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
// 如果休眠期间被中断，则退出循环
Thread.currentThread().interrupt();
break;
}
}
// 发送完成信号（如果需要）
if (!Thread.currentThread().isInterrupted()) {
// ServerSendEventServer.sendMessage(token, "COMPLETED");
}
} catch (Exception e) {
// 捕获异常并记录日志
LOGGER.error("任务执行失败: " + e.getMessage());
e.printStackTrace();
}
});
}else{
TsFiles tsFilesData = tsFilesMapper.selectById(configId);
if (currentTaskFuture != null && !currentTaskFuture.isDone()) {
currentTaskFuture.cancel(true);
@ -5555,8 +5434,8 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
Path basePath = Paths.get(config.getValue() + tsFilesData.getWorkPath());
// 2. 读取配置文件（假设配置文件名为 trj_config.txt）
Path configPath = basePath.resolve("trj_config.txt");
Map<String, String> columnMapping = parseConfigFile(configPath);
Path configPath = basePath.resolve(tsFilesData.getFileName());
Map<String, String> columnMapping = parseConfigFile(configPath, token);
String timeColumn = columnMapping.get("time");
String lonColumn = columnMapping.get("lon");
@ -5567,7 +5446,7 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
timeColumn, lonColumn, latColumn, hgtColumn);
// 3. 获取数据文件路径
Path dataPath = basePath.resolve("VINS.csv");
Path dataPath = basePath.resolve(tsFiles.getFileName());
// 4. 处理数据文件
processDataFile(dataPath, samTimes, token, timeColumn, lonColumn, latColumn, hgtColumn);
@ -5576,9 +5455,6 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
LOGGER.error("任务执行失败: {}", e.getMessage(), e);
}
});
}
} catch (Exception e) {
@ -5590,43 +5466,40 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
}
// 封装发送数据的逻辑
private void sendData(String token, String[] values, int lineCount) {
if (Thread.currentThread().isInterrupted()) {
return;
}
try {
SimpleNaviData data = parseLine(values);
String jsonData = JSONUtil.toJsonStr(data); // 假设 JSONUtil.toJsonStr 不会抛出 IOException
ServerSendEventServer.sendMessageById(token, jsonData); // 假设 sendMessage 不会抛出 IOException
LOGGER.info("Line " + lineCount + " sent at: " + System.currentTimeMillis());
} catch (Exception e) {
// 捕获所有可能的异常
LOGGER.error("发送数据失败: " + e.getMessage());
if (e.getCause() instanceof InterruptedException) {
Thread.currentThread().interrupt(); // 重新设置中断状态
}
}
}
// 辅助方法：解析行数据
private SimpleNaviData parseLine(String[] values) {
SimpleNaviData data = new SimpleNaviData();
data.setUtcTime(getValueSafely(values, 8, "0.0")); // UTC_TIME（索引8）
data.setLat(getValueSafely(values, 17, "0.0")); // LAT（索引17）
data.setLon(getValueSafely(values, 18, "0.0")); // LON（索引18）
data.setAlt(getValueSafely(values, 19, "0.0")); // ALT（索引19）
return data;
}
// 安全获取数组值的方法
private String getValueSafely(String[] values, int index, String defaultValue) {
return (index < values.length) ? values[index] : defaultValue;
}
// // 封装发送数据的逻辑
// private void sendData(String token, String[] values, int lineCount) {
// if (Thread.currentThread().isInterrupted()) {
// return;
// }
//
// try {
// SimpleNaviData data = parseLine(values);
// String jsonData = JSONUtil.toJsonStr(data); // 假设 JSONUtil.toJsonStr 不会抛出 IOException
// ServerSendEventServer.sendMessageById(token, jsonData); // 假设 sendMessage 不会抛出 IOException
// LOGGER.info("Line " + lineCount + " sent at: " + System.currentTimeMillis());
// } catch (Exception e) {
// // 捕获所有可能的异常
// LOGGER.error("发送数据失败: " + e.getMessage());
// if (e.getCause() instanceof InterruptedException) {
// Thread.currentThread().interrupt(); // 重新设置中断状态
// }
// }
// }
//
// // 辅助方法：解析行数据
// private SimpleNaviData parseLine(String[] values) {
// SimpleNaviData data = new SimpleNaviData();
// data.setUtcTime(getValueSafely(values, 8, "0.0")); // UTC_TIME（索引8）
// data.setLat(getValueSafely(values, 17, "0.0")); // LAT（索引17）
// data.setLon(getValueSafely(values, 18, "0.0")); // LON（索引18）
// data.setAlt(getValueSafely(values, 19, "0.0")); // ALT（索引19）
// return data;
// }
//
// // 安全获取数组值的方法
// private String getValueSafely(String[] values, int index, String defaultValue) {
// return (index < values.length) ? values[index] : defaultValue;
// }
/**
@ -5636,7 +5509,7 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
* @return 列名映射Map（time, lon, lat, hgt）
* @throws IOException 文件读取错误
*/
public Map<String, String> parseConfigFile(Path configFilePath) throws IOException {
public Map<String, String> parseConfigFile(Path configFilePath, String token) throws IOException {
Map<String, String> columnMapping = new HashMap<>();
try (BufferedReader reader = Files.newBufferedReader(configFilePath, StandardCharsets.UTF_8)) {
@ -5665,7 +5538,13 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
// 验证是否获取到所有必需的列
if (columnMapping.size() < 4) {
throw new IOException("配置文件缺少必需的列定义");
JSONObject jsonResponse = new JSONObject();
jsonResponse.put("message", "配置文件选择错误，请重新选择！");
// 将 JSON 数据发送到客户端
ServerSendEventServer.sendMessageById(token, jsonResponse.toString());
//throw new IOException("配置文件选择错误，请重新选择！");
}
return columnMapping;
@ -5675,6 +5554,7 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
/**
* 处理数据文件
*/
private void processDataFile(Path dataPath, int samTimes, String token,
String timeColumn, String lonColumn,
String latColumn, String hgtColumn)
@ -5688,48 +5568,58 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
return;
}
// 2. 解析标题行 - 使用逗号分隔
String[] headers = headerLine.split(","); // 改为逗号分隔
// 2. 自动检测分隔符（关键改进）
String delimiter = detectDelimiter(headerLine);
LOGGER.info("检测到分隔符: {}", delimiter.replace("\t", "\\t").replace(" ", "\\s"));
// 添加标题行调试日志
// 3. 解析标题行
String[] headers = headerLine.split(delimiter);
LOGGER.info("数据文件标题行包含 {} 列", headers.length);
LOGGER.debug("完整标题行: {}", String.join("|", headers));
Map<String, Integer> columnIndices = new HashMap<>();
// 3. 查找列索引（修复后的逻辑）
// 4. 查找列索引（改进：大小写不敏感+空格处理）
for (int i = 0; i < headers.length; i++) {
String header = headers[i].trim();
// 使用 equalsIgnoreCase 进行不区分大小写的匹配
if (header.equalsIgnoreCase(timeColumn)) {
// 标准化列名：移除空格和下划线
String normalizedHeader = header.replaceAll("[_\\s]", "")
.toLowerCase();
if (normalizedHeader.equals(timeColumn.replaceAll("[_\\s]", "").toLowerCase())) {
columnIndices.put("time", i);
LOGGER.info("找到 time 列 '{}' 在位置 {}", header, i);
}
if (header.equalsIgnoreCase(lonColumn)) {
if (normalizedHeader.equals(lonColumn.replaceAll("[_\\s]", "").toLowerCase())) {
columnIndices.put("lon", i);
LOGGER.info("找到 lon 列 '{}' 在位置 {}", header, i);
}
if (header.equalsIgnoreCase(latColumn)) {
if (normalizedHeader.equals(latColumn.replaceAll("[_\\s]", "").toLowerCase())) {
columnIndices.put("lat", i);
LOGGER.info("找到 lat 列 '{}' 在位置 {}", header, i);
}
if (header.equalsIgnoreCase(hgtColumn)) {
if (normalizedHeader.equals(hgtColumn.replaceAll("[_\\s]", "").toLowerCase())) {
columnIndices.put("hgt", i);
LOGGER.info("找到 hgt 列 '{}' 在位置 {}", header, i);
}
}
// 4. 验证列是否存在
// 5. 验证列是否存在（改进：提供更友好的错误信息）
List<String> missingColumns = new ArrayList<>();
if (!columnIndices.containsKey("time")) missingColumns.add(timeColumn);
if (!columnIndices.containsKey("lon")) missingColumns.add(lonColumn);
if (!columnIndices.containsKey("lat")) missingColumns.add(latColumn);
if (!columnIndices.containsKey("hgt")) missingColumns.add(hgtColumn);
if (!columnIndices.containsKey("time")) {
missingColumns.add("时间列(" + timeColumn + ")");
LOGGER.warn("未找到时间列，尝试的列名: {}", timeColumn);
}
if (!columnIndices.containsKey("lon")) {
missingColumns.add("经度列(" + lonColumn + ")");
}
if (!columnIndices.containsKey("lat")) {
missingColumns.add("纬度列(" + latColumn + ")");
}
if (!columnIndices.containsKey("hgt")) {
missingColumns.add("高度列(" + hgtColumn + ")");
}
if (!missingColumns.isEmpty()) {
LOGGER.error("以下列在数据文件中不存在: {}", String.join(", ", missingColumns));
LOGGER.error("数据文件标题行: {}", headerLine);
LOGGER.error("缺少必需列: {}", String.join(", ", missingColumns));
LOGGER.error("可用列: {}", String.join(", ", headers));
return;
}
@ -5741,12 +5631,16 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
LOGGER.info("列索引: time={}, lon={}, lat={}, hgt={}",
timeIndex, lonIndex, latIndex, hgtIndex);
// 5. 读取并处理数据行
// 6. 计算实际步长（关键修复）
int step = samTimes * 200; // 与之前逻辑保持一致
if (step <= 0) step = 1; // 避免除零错误
// 7. 读取并处理数据行
String line;
int lineCount = 0;
int sendCount = 0;
int lineCountData = 0;
float timeValue0 = 0;
float timeValue1 = 0;
int result = 0; //
@ -5776,32 +5670,36 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
}
while (!Thread.currentThread().isInterrupted() && (line = reader.readLine()) != null) {
lineCount++;
// 使用逗号分隔数据行
String[] values = line.split(","); // 改为逗号分隔
// 跳过空行
if (line.trim().isEmpty()) continue;
// 使用检测到的分隔符
String[] values = line.split(delimiter);
// 跳过不符合步长的行（第一行总是发送）
if (lineCount > 1 && (lineCount - 1) % (samTimes * result) != 0) {
continue;
}
// 检查数据完整性
int maxIndex = getMaxIndex(timeIndex, lonIndex, latIndex, hgtIndex);
// 检查数据完整性（动态计算最大索引）
int maxIndex = Math.max(Math.max(timeIndex, lonIndex),
Math.max(latIndex, hgtIndex));
if (values.length <= maxIndex) {
LOGGER.warn("行 {} 数据不完整: {}", lineCount, line);
LOGGER.warn("行 {} 数据不完整 (需要 {} 列, 实际 {} 列): {}",
lineCount, maxIndex + 1, values.length, line);
continue;
}
// 发送数据
sendDataCsv(token, values, lineCount, timeIndex, lonIndex, latIndex, hgtIndex);
sendCount++;
// 每次发送后都休眠3秒
try {
Thread.sleep(3000);
Thread.sleep(500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
@ -5809,11 +5707,71 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
}
if (!Thread.currentThread().isInterrupted()) {
LOGGER.info("文件处理完成，共处理 {} 行", lineCount);
LOGGER.info("文件处理完成: 总行数={}, 发送行数={}", lineCount, sendCount);
}
}
}
// 自动检测分隔符（支持两种格式）
private String detectDelimiter(String headerLine) {
// 优先检查制表符
if (headerLine.contains("\t")) {
return "\t";
}
// 检查逗号
if (headerLine.contains(",")) {
return ",";
}
// 检查空格（多个空格）
if (headerLine.contains(" ")) {
return "\\s+";
}
// 默认使用制表符
return "\t";
}
// 安全获取数组值的方法
private String getValueSafely(String[] values, int index, String defaultValue) {
return (index < values.length) ? values[index] : defaultValue;
}
// 改进的发送方法（处理各种数据格式）
private void sendDataCsv(String token, String[] values, int lineCount,
int timeIndex, int lonIndex,
int latIndex, int hgtIndex) {
try {
SimpleNaviData data = new SimpleNaviData();
// 处理可能带空格的数值
data.setUtcTime(values[timeIndex].trim());
data.setLon(parseDoubleSafe(values[lonIndex].trim(), 0.0));
data.setLat(parseDoubleSafe(values[latIndex].trim(), 0.0));
data.setAlt(parseDoubleSafe(values[hgtIndex].trim(), 0.0));
String jsonData = JSONUtil.toJsonStr(data);
ServerSendEventServer.sendMessageById(token, jsonData);
LOGGER.info("发送行 {}: UTC={}, Lon={}, Lat={}, Alt={}",
lineCount, data.getUtcTime(), data.getLon(),
data.getLat(), data.getAlt());
} catch (Exception e) {
LOGGER.error("发送行 {} 失败: {}", lineCount, e.getMessage());
}
}
// 安全解析double值
private String parseDoubleSafe(String value, double defaultValue) {
try {
return String.valueOf(Double.parseDouble(value));
} catch (NumberFormatException e) {
LOGGER.warn("数值解析失败: '{}', 使用默认值 {}", value, defaultValue);
return String.valueOf(defaultValue);
}
}
// 辅助方法：获取最大索引
private int getMaxIndex(int... indices) {
int max = -1;
@ -5824,34 +5782,6 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
}
private void sendDataCsv(String token, String[] values, int lineCount,
int timeIndex, int lonIndex, int latIndex, int hgtIndex) {
if (Thread.currentThread().isInterrupted()) {
return;
}
try {
SimpleNaviData data = new SimpleNaviData();
data.setUtcTime(getValueSafely(values, timeIndex, "0.0"));
data.setLon(getValueSafely(values, lonIndex, "0.0"));
data.setLat(getValueSafely(values, latIndex, "0.0"));
data.setAlt(getValueSafely(values, hgtIndex, "0.0"));
String jsonData = JSONUtil.toJsonStr(data);
ServerSendEventServer.sendMessageById(token, jsonData);
LOGGER.info("Line {} sent at: {}", lineCount, System.currentTimeMillis());
} catch (Exception e) {
LOGGER.error("发送数据失败: {}", e.getMessage());
if (e.getCause() instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
}
}
/**
* 查询文件内容接口
*
@ -6346,8 +6276,6 @@ public class TsFilesServiceImpl extends ServiceImpl<TsFilesMapper, TsFiles> impl
}
}