20260627:接入深度位移计，单个数据值推送
This commit is contained in:
zhugongping
parent
4e467edb2b
commit
8c12a70df2
@ -51,12 +51,15 @@ import java.util.Map;
|
||||
@Component
|
||||
public class DcpPlatformAdapter {
|
||||
|
||||
private static final DateTimeFormatter ISO_OFFSET_MILLIS =
|
||||
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX");
|
||||
private static final DateTimeFormatter DATE_TIME_MILLIS =
|
||||
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
|
||||
private static final DateTimeFormatter DATE_TIME_SECONDS =
|
||||
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||||
// 组包形势
|
||||
// 成对
|
||||
private static final String ARRAY_MODE_REFERENCE_PAIR = "reference_pair";
|
||||
// 单个
|
||||
private static final String ARRAY_MODE_SINGLE = "single";
|
||||
|
||||
private static final DateTimeFormatter ISO_OFFSET_MILLIS = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX");
|
||||
private static final DateTimeFormatter DATE_TIME_MILLIS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
|
||||
private static final DateTimeFormatter DATE_TIME_SECONDS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||||
|
||||
@Autowired
|
||||
private IPlatformInstanceService instanceService;
|
||||
@ -105,7 +108,8 @@ public class DcpPlatformAdapter {
|
||||
* 推送网关子设备上报数据到 DCP
|
||||
* <p>
|
||||
* deviceId 直接从子映射行获取，不再解析 JSON。
|
||||
* reference 角色仅缓存基准数据，monitor 角色组装 [refItem, monitorItem] 推送。
|
||||
* 数组字段由 requestSchema.arrayMode 控制组包方式：
|
||||
* single 只推当前子设备，reference_pair 组装 [refItem, monitorItem] 推送。
|
||||
*/
|
||||
public void pushGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping,
|
||||
PlatformDeviceSubMapping subMapping,
|
||||
@ -140,15 +144,18 @@ public class DcpPlatformAdapter {
|
||||
// 解析 extra_mapping 中的 itemFieldMappings 和 reportIntervalSec
|
||||
int reportIntervalSec = 300;
|
||||
JSONArray itemFieldMappings = null;
|
||||
String fieldName = "sensorData";
|
||||
String fieldName = resolveArrayFieldName(requestSchema, null);
|
||||
JSONObject extra = null;
|
||||
if (StringUtils.hasText(mapping.getExtraMapping())) {
|
||||
JSONObject extra = JSON.parseObject(mapping.getExtraMapping());
|
||||
extra = JSON.parseObject(mapping.getExtraMapping());
|
||||
reportIntervalSec = extra.getIntValue("reportIntervalSec", 300);
|
||||
itemFieldMappings = extra.getJSONArray("itemFieldMappings");
|
||||
if (StringUtils.hasText(extra.getString("fieldName"))) {
|
||||
fieldName = extra.getString("fieldName");
|
||||
}
|
||||
}
|
||||
JSONObject arrayFieldSchema = findArrayFieldSchema(requestSchema, fieldName);
|
||||
String arrayMode = resolveArrayMode(arrayFieldSchema, extra);
|
||||
|
||||
// 从 onboard_config 取 sensorId
|
||||
String sensorId = reportSerialNumber;
|
||||
@ -170,6 +177,13 @@ public class DcpPlatformAdapter {
|
||||
|
||||
JSONObject currentBodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings);
|
||||
|
||||
// 单点数组推送：sensorData 中只包含当前上报子设备这一条数据。
|
||||
if (ARRAY_MODE_SINGLE.equals(arrayMode)) {
|
||||
pushSingleArrayData(platform, mapping, subMapping, reportSerialNumber, baseUrl,
|
||||
interfaceCode, fieldName, currentItem, currentBodyFields);
|
||||
return;
|
||||
}
|
||||
|
||||
if (subMapping.isReference()) {
|
||||
// 基准点上报：缓存当前基准点，并补推此前已缓存的监测点
|
||||
bufferService.saveReferenceData(
|
||||
@ -216,6 +230,25 @@ public class DcpPlatformAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
private void pushSingleArrayData(PlatformInstance platform, PlatformDeviceMapping mapping,
|
||||
PlatformDeviceSubMapping subMapping, String reportSerialNumber,
|
||||
String baseUrl, String interfaceCode, String fieldName,
|
||||
JSONObject currentItem, JSONObject currentBodyFields) {
|
||||
String deviceId = subMapping.getPlatformDeviceId();
|
||||
if (!StringUtils.hasText(deviceId)) {
|
||||
log.warn("[DCP] 子设备 {} 未分配平台设备ID，跳过单点数组推送", reportSerialNumber);
|
||||
return;
|
||||
}
|
||||
JSONArray sensorData = new JSONArray();
|
||||
sensorData.add(currentItem);
|
||||
JSONArray pushBody = buildArrayPushBody(mapping, currentBodyFields, fieldName, sensorData, deviceId);
|
||||
|
||||
PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber);
|
||||
doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry);
|
||||
pushLogService.asyncSave(logEntry);
|
||||
log.info("[DCP] 子设备 {} 单点数组已推送 deviceId={}", reportSerialNumber, deviceId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 请求 schema 是否包含数组字段
|
||||
*/
|
||||
@ -232,6 +265,33 @@ public class DcpPlatformAdapter {
|
||||
return false;
|
||||
}
|
||||
|
||||
private String resolveArrayFieldName(JSONArray requestSchema, String fallback) {
|
||||
if (StringUtils.hasText(fallback)) {
|
||||
return fallback;
|
||||
}
|
||||
JSONObject firstArrayField = findFirstArrayFieldSchema(requestSchema);
|
||||
if (firstArrayField != null && StringUtils.hasText(firstArrayField.getString("name"))) {
|
||||
return firstArrayField.getString("name");
|
||||
}
|
||||
return "sensorData";
|
||||
}
|
||||
|
||||
private String resolveArrayMode(JSONObject arrayFieldSchema, JSONObject extra) {
|
||||
String mode = null;
|
||||
if (extra != null && StringUtils.hasText(extra.getString("arrayMode"))) {
|
||||
mode = extra.getString("arrayMode");
|
||||
} else if (arrayFieldSchema != null && StringUtils.hasText(arrayFieldSchema.getString("arrayMode"))) {
|
||||
mode = arrayFieldSchema.getString("arrayMode");
|
||||
}
|
||||
if (!StringUtils.hasText(mode)) {
|
||||
return ARRAY_MODE_REFERENCE_PAIR;
|
||||
}
|
||||
if (ARRAY_MODE_SINGLE.equals(mode) || ARRAY_MODE_REFERENCE_PAIR.equals(mode)) {
|
||||
return mode;
|
||||
}
|
||||
throw new RuntimeException("不支持的数组组包模式 arrayMode=" + mode);
|
||||
}
|
||||
|
||||
/** 解析基准点序列号 */
|
||||
private String resolveRefSerialNumber(PlatformDeviceSubMapping subMapping, PlatformDeviceMapping mapping) {
|
||||
try {
|
||||
@ -633,6 +693,19 @@ public class DcpPlatformAdapter {
|
||||
return null;
|
||||
}
|
||||
|
||||
private JSONObject findFirstArrayFieldSchema(JSONArray requestSchema) {
|
||||
if (requestSchema == null || requestSchema.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
for (int i = 0; i < requestSchema.size(); i++) {
|
||||
JSONObject schemaField = requestSchema.getJSONObject(i);
|
||||
if ("array".equalsIgnoreCase(schemaField.getString("type"))) {
|
||||
return schemaField;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private JSONObject copyJsonObject(JSONObject source) {
|
||||
if (source == null || source.isEmpty()) {
|
||||
return new JSONObject();
|
||||
|
||||
@ -295,7 +295,7 @@ public class PlatformFlowExecutor {
|
||||
if (StringUtils.hasText(ctx.getCurrentMonitorSerialNumber())) {
|
||||
return ctx.getCurrentMonitorSerialNumber();
|
||||
}
|
||||
return resolveGatewaySerialNumber(ctx.getDeviceMapping());
|
||||
return normalizeValueByType(fd.getString("type"), resolveGatewaySerialNumber(ctx.getDeviceMapping()));
|
||||
}
|
||||
if ("gateway_serial_number".equals(df)) {
|
||||
return resolveGatewaySerialNumber(ctx.getDeviceMapping());
|
||||
|
||||
@ -127,13 +127,7 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl<PlatformDevice
|
||||
ctx.setReferenceSubMappingOnboardConfig(reference.getOnboardConfig());
|
||||
}
|
||||
|
||||
try {
|
||||
flowExecutor.execute(onboardFlow, ctx);
|
||||
} catch (Exception e) {
|
||||
log.error("[Onboard] 网关子设备 {} 入网失败: {}", monitor.getSubSerialNumber(), e.getMessage(), e);
|
||||
failedMessages.add(monitor.getSubSerialNumber() + ": " + e.getMessage());
|
||||
continue;
|
||||
}
|
||||
flowExecutor.execute(onboardFlow, ctx);
|
||||
|
||||
// 从流程上下文取回平台分配的 ID
|
||||
Object pointId = ctx.getFromContext("step1.id");
|
||||
|
||||
@ -99,7 +99,7 @@
|
||||
<el-divider content-position="left">入参定义（request_schema）</el-divider>
|
||||
<div class="schema-tip">
|
||||
定义接口需要的入参字段。参数来源使用业务化名称：
|
||||
固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、基准点成对传感器列表、设备物模型、上报时间。
|
||||
固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、基准点成对传感器列表、设备物模型、上报时间。数组字段可配置组包模式。
|
||||
</div>
|
||||
<el-table :data="requestSchemaRows" size="mini" border style="width:100%;margin-top:8px">
|
||||
<el-table-column label="字段名" width="130">
|
||||
@ -171,6 +171,14 @@
|
||||
<el-checkbox v-model="scope.row.required" />
|
||||
</template>
|
||||
</el-table-column>
|
||||
<el-table-column label="组包模式" width="130">
|
||||
<template slot-scope="scope">
|
||||
<el-select v-if="scope.row.type === 'array'" v-model="scope.row.arrayMode" size="mini" clearable placeholder="默认配对" style="width:100%">
|
||||
<el-option label="单条" value="single" />
|
||||
<el-option label="基准配对" value="reference_pair" />
|
||||
</el-select>
|
||||
</template>
|
||||
</el-table-column>
|
||||
<el-table-column label="说明" width="140">
|
||||
<template slot-scope="scope">
|
||||
<el-input v-model="scope.row.desc" size="mini" placeholder="字段说明" />
|
||||
@ -424,7 +432,7 @@ export default {
|
||||
}
|
||||
},
|
||||
addRequestField() {
|
||||
this.requestSchemaRows.push({ name: '', type: 'string', source: 'fixed', value: '', field: '', contextKey: '', required: false, desc: '', itemSchema: [] })
|
||||
this.requestSchemaRows.push({ name: '', type: 'string', source: 'fixed', value: '', field: '', contextKey: '', required: false, desc: '', arrayMode: '', itemSchema: [] })
|
||||
},
|
||||
addResponseField() {
|
||||
this.responseSchemaRows.push({ name: '', type: 'string', path: '', desc: '' })
|
||||
@ -457,6 +465,7 @@ export default {
|
||||
// 将表格行序列化为 JSON 存入 form
|
||||
this.form.requestSchema = JSON.stringify(this.requestSchemaRows.filter(r => r.name).map(r => {
|
||||
const field = { name: r.name, type: r.type, source: r.source, required: !!r.required, desc: r.desc }
|
||||
if (r.type === 'array' && r.arrayMode) field.arrayMode = r.arrayMode
|
||||
if (r.type === 'array' && r.itemSchema && r.itemSchema.length > 0) {
|
||||
field.itemSchema = r.itemSchema.filter(s => s.name).map(s => {
|
||||
const item = { name: s.name, type: s.type || 'string', source: s.source || 'thingsModel', desc: s.desc }
|
||||
|
||||
Loading…
Reference in New Issue
Block a user