Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260401:支持修改轮询任务；修改微信推送模板；支持网关设备跨计算模型数据推送

Browse Source
This commit is contained in:
George 2026-04-01 22:17:35 +08:00
parent 6dbfa2f09c
commit a3a5b6a719
5 changed files with 320 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
fastbee-iot-data/pom.xml
View File

@ -15,6 +15,12 @@
<artifactId>mqtt-broker</artifactId>
</dependency>
<!-- IoT 业务能力（物模型、网关从机关系等），用于推送时计算网关跨设备计算量 -->
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-iot-service</artifactId>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-mq</artifactId>

169
fastbee-iot-data/src/main/java/com/fastbee/iot/data/push/ThingsModelPushHandler.java
View File

@ -8,14 +8,20 @@ import com.fastbee.common.core.notify.config.WeChatConfigParams;
import com.fastbee.common.core.notify.msg.WechatMsgParams;
import com.fastbee.common.core.redis.RedisKeyBuilder;
import com.fastbee.common.enums.DataLevelEnum;
import com.fastbee.common.utils.CaculateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.wechat.WechatUtils;
import com.fastbee.common.wechat.WeChatAppResult;
import com.fastbee.iot.cache.ITSLCache;
import com.fastbee.iot.domain.SubGateway;
import com.fastbee.iot.domain.PushModelItem;
import com.fastbee.iot.domain.SocialUser;
import com.fastbee.iot.domain.ThingsModel;
import com.fastbee.iot.domain.ThingsModelPushConfig;
import com.fastbee.iot.model.ThingsModels.ValueItem;
import com.fastbee.iot.model.gateWay.SubDeviceListVO;
import com.fastbee.iot.service.ISubGatewayService;
import com.fastbee.iot.service.IThingsModelService;
import com.fastbee.iot.service.IThingsModelPushConfigService;
import com.fastbee.iot.service.ISocialUserService;
import com.fastbee.notify.core.wechat.vo.TemplateDataVo;
@ -52,6 +58,8 @@ public class ThingsModelPushHandler {
private static final RestTemplate REST_TEMPLATE = new RestTemplate();
private static final Pattern VAR_PATTERN = Pattern.compile("\\{\\{([^}]+)}}");
/** 公式中跨设备引用：{slaveId:N.identifier} */
private static final Pattern CROSS_DEVICE_REF = Pattern.compile("\\{slaveId:(\\d+)\\.([^}]+)}");
private static final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@ -65,6 +73,10 @@ public class ThingsModelPushHandler {
private ISocialUserService socialUserService;
@Resource
private ITSLCache itslCache;
@Resource
private ISubGatewayService subGatewayService;
@Resource
private IThingsModelService thingsModelService;
/**
* 按推送配置执行推送
@ -148,13 +160,24 @@ public class ThingsModelPushHandler {
levelTextMap.put(lvl, new ArrayList<>());
}
Map<String, ValueItem> cacheMap = itslCache.getSimpleThingsModelValueItemByCache(redisKey);
Map<Integer, SubDeviceListVO> subDeviceBySlaveId = null;
Map<String, ThingsModel> calculatedModelCache = new HashMap<>();
for (PushModelItem item : config.getItems()) {
ValueItem valueItem = cacheMap.get(item.getIdentifier());
if (valueItem == null) {
continue;
String value;
Date ts;
if (valueItem != null) {
value = valueItem.getValue();
ts = valueItem.getTs();
} else {
// 如果缓存里没有，尝试按网关计算物模型公式计算（跨设备引用）
if (subDeviceBySlaveId == null) {
subDeviceBySlaveId = loadSubDeviceBySlaveId(config.getDeviceId());
}
ComputedValue computed = tryComputeGatewayCalculatedValue(config, serialNumber, item.getIdentifier(), subDeviceBySlaveId, calculatedModelCache);
value = computed != null ? computed.value : null;
ts = computed != null ? computed.ts : null;
}
String value = valueItem.getValue();
Date ts = valueItem.getTs();
if (config.getLastPushTime() != null && ts != null && ts.before(config.getLastPushTime())) {
log.warn("[Push] configId={} sn={} item={} value={} 时间{}早于上次推送时间{}，跳过",
config.getId(), serialNumber, item.getIdentifier(), value, ts, config.getLastPushTime());
@ -195,6 +218,133 @@ public class ThingsModelPushHandler {
return sendTemplateMsg(config, msgParams, serialNumber, chosen, dataText, pushUrl, openIdMap);
}
private static final class ComputedValue {
private final String value;
private final Date ts;
private ComputedValue(String value, Date ts) {
this.value = value;
this.ts = ts;
}
}
private Map<Integer, SubDeviceListVO> loadSubDeviceBySlaveId(Long gwDeviceId) {
if (gwDeviceId == null) {
return Collections.emptyMap();
}
try {
SubGateway subQ = new SubGateway();
subQ.setGwDeviceId(gwDeviceId);
List<SubDeviceListVO> list = subGatewayService.selectGatewayList(subQ);
if (CollectionUtils.isEmpty(list)) {
return Collections.emptyMap();
}
Map<Integer, SubDeviceListVO> map = new HashMap<>();
for (SubDeviceListVO vo : list) {
if (vo != null && vo.getSlaveId() != null) {
map.put(vo.getSlaveId(), vo);
}
}
return map;
} catch (Exception e) {
log.warn("[Push] loadSubDeviceBySlaveId gwDeviceId={} 异常: {}", gwDeviceId, e.getMessage());
return Collections.emptyMap();
}
}
/**
* 计算网关跨设备计算物模型的当前值（只基于各从机当前缓存值）
*/
private ComputedValue tryComputeGatewayCalculatedValue(ThingsModelPushConfig config,
String gwSerialNumber,
String identifier,
Map<Integer, SubDeviceListVO> subDeviceBySlaveId,
Map<String, ThingsModel> calculatedModelCache) {
if (config == null || config.getProductId() == null || StringUtils.isEmpty(identifier)) {
return null;
}
ThingsModel model;
if (calculatedModelCache != null && calculatedModelCache.containsKey(identifier)) {
model = calculatedModelCache.get(identifier);
} else {
ThingsModel query = new ThingsModel();
query.setProductId(config.getProductId());
query.setIdentifier(identifier);
query.setIsCalculated(1);
query.setSerialNumber(gwSerialNumber);
try {
model = thingsModelService.selectSingleThingsModel(query);
} catch (Exception e) {
return null;
}
if (calculatedModelCache != null) {
calculatedModelCache.put(identifier, model);
}
}
if (model == null || model.getIsCalculated() == null ||
model.getIsCalculated() != 1 || StringUtils.isEmpty(model.getFormula())) {
return null;
}
Map<String, String> refs = parseCrossDeviceRefs(model.getFormula());
if (refs.isEmpty()) {
return null;
}
Map<String, String> replaceMap = new HashMap<>();
Date maxTs = null;
for (Map.Entry<String, String> ref : refs.entrySet()) {
String refKey = ref.getKey(); // slaveId:N.identifier
int slaveId;
try {
slaveId = Integer.parseInt(refKey.split("\\.")[0].replace("slaveId:", ""));
} catch (Exception e) {
continue;
}
SubDeviceListVO sub = subDeviceBySlaveId != null ? subDeviceBySlaveId.get(slaveId) : null;
String subSn = sub != null ? sub.getSubDeviceNo() : null;
Long subProductId = sub != null ? sub.getSubProductId() : null;
String subIdentifier = ref.getValue();
String val = "0";
if (subProductId != null && StringUtils.isNotEmpty(subSn) && StringUtils.isNotEmpty(subIdentifier)) {
try {
ValueItem v = thingsModelService.getCacheIdentifierValue(subProductId, subSn, subIdentifier);
if (v != null && StringUtils.isNotEmpty(v.getValue())) {
val = v.getValue();
}
if (v != null && v.getTs() != null) {
if (maxTs == null || v.getTs().after(maxTs)) {
maxTs = v.getTs();
}
}
} catch (Exception ignored) {
}
}
replaceMap.put("{" + refKey + "}", val);
}
try {
Object computed = CaculateUtils.execute(model.getFormula(), replaceMap);
String computedStr = computed != null ? computed.toString() : null;
return new ComputedValue(computedStr, maxTs);
} catch (Exception e) {
log.warn("[Push] configId={} sn={} identifier={} 公式计算失败 formula={} err={}",
config.getId(), gwSerialNumber, identifier, model.getFormula(), e.getMessage());
return null;
}
}
private Map<String, String> parseCrossDeviceRefs(String formula) {
Map<String, String> refs = new LinkedHashMap<>();
if (StringUtils.isEmpty(formula)) {
return refs;
}
Matcher m = CROSS_DEVICE_REF.matcher(formula);
while (m.find()) {
String key = "slaveId:" + m.group(1) + "." + m.group(2);
refs.put(key, m.group(2));
}
return refs;
}
/**
* 发送单条微信模板消息。
* 从通知模板的 content 字段提取 {{变量.DATA}} 变量名顺序，
@ -206,12 +356,12 @@ public class ThingsModelPushHandler {
List<String> varNames = extractVarNames(msgParams.getContent());
Map<String, String> map = new LinkedHashMap<>();
// String deviceDisplay = StringUtils.isNotEmpty(config.getDeviceName())
// ? config.getDeviceName() : serialNumber;
String deviceDisplay = StringUtils.isNotEmpty(config.getDeviceName())
? config.getDeviceName() : serialNumber;
for (int i = 0; i < varNames.size(); i++) {
String varName = varNames.get(i);
if (i == 0) {
map.put(varName, serialNumber);
map.put(varName, deviceDisplay);
} else if (i == 1) {
map.put(varName, level.getLabel());
} else if (i == 2) {
@ -284,9 +434,4 @@ public class ThingsModelPushHandler {
return resp;
}
public static void main(String[] args) {
String value = "WaterT=1.76;WaterVol=-6.34;T=20";
String value1 = "[修正]参术儿康糖浆10ml*10支1盒装儿童小...";
System.out.println("value.length() = " + value.length());
}
}

93
fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/ModbusJobServiceImpl.java
View File

@ -3,7 +3,9 @@ package com.fastbee.iot.service.impl;
import com.fastbee.common.core.device.DeviceAndProtocol;
import com.fastbee.common.core.mq.SubDeviceBo;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.exception.job.TaskException;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.iot.domain.DeviceJob;
import com.fastbee.iot.domain.ModbusJob;
import com.fastbee.iot.enums.DeviceType;
@ -172,16 +174,103 @@ public class ModbusJobServiceImpl implements IModbusJobService {
*/
@Override
public int updateModbusJob(ModbusJob modbusJob) {
if (modbusJob == null || modbusJob.getTaskId() == null) {
throw new ServiceException("任务ID不能为空");
}
ModbusJob existingJob = modbusJobMapper.selectModbusJobByTaskId(modbusJob.getTaskId());
if (existingJob == null) {
throw new ServiceException("轮询任务不存在");
}
// cycleType 由前端传递，不再根据 remark 判断
// 如果状态从1（暂停）改为0（正常），需要清空执行时间
if (modbusJob.getTaskId() != null && modbusJob.getStatus() != null && "0".equals(modbusJob.getStatus())) {
ModbusJob existingJob = modbusJobMapper.selectModbusJobByTaskId(modbusJob.getTaskId());
if (existingJob != null && "1".equals(existingJob.getStatus())) {
// 状态从1改为0，清空执行时间
modbusJobMapper.clearExecuteTime(modbusJob.getTaskId());
}
}
return modbusJobMapper.updateModbusJob(modbusJob);
// 是否修改了时间周期（remark/cycleType），如修改则需要同步更新/重绑 iot_device_job 并刷新 quartz
boolean scheduleChanged = (modbusJob.getRemark() != null) || (modbusJob.getCycleType() != null);
Long oldDeviceJobId = existingJob.getJobId();
if (scheduleChanged) {
// 计算新的 remark / cycleType（允许只传 cycleType 或只传 remark）
String newRemark = modbusJob.getRemark() != null ? modbusJob.getRemark() : existingJob.getRemark();
Integer newCycleType = modbusJob.getCycleType() != null ? modbusJob.getCycleType() : existingJob.getCycleType();
if (newCycleType == null) {
newCycleType = 1;
}
if (oldDeviceJobId == null) {
throw new ServiceException("轮询任务缺少关联的设备定时任务(jobId)");
}
DeviceJob oldDeviceJob = deviceJobService.selectJobById(oldDeviceJobId);
if (oldDeviceJob == null) {
throw new ServiceException("关联的设备定时任务不存在(jobId=" + oldDeviceJobId + ")");
}
// 优先复用同设备、同remark的 DeviceJob（避免重复创建）
DeviceJob query = new DeviceJob();
query.setDeviceId(oldDeviceJob.getDeviceId());
query.setJobType(5);
query.setRemark(newRemark);
List<DeviceJob> candidates = deviceJobService.selectJobList(query);
DeviceJob targetDeviceJob = CollectionUtils.isEmpty(candidates) ? null : candidates.get(0);
if (targetDeviceJob == null) {
DeviceJob create = new DeviceJob();
create.setDeviceId(oldDeviceJob.getDeviceId());
create.setSerialNumber(oldDeviceJob.getSerialNumber());
create.setProductId(oldDeviceJob.getProductId());
create.setJobType(5);
create.setJobName(StringUtils.isNotEmpty(oldDeviceJob.getJobName()) ? oldDeviceJob.getJobName() : "modbus poll");
create.setJobGroup(StringUtils.isNotEmpty(oldDeviceJob.getJobGroup()) ? oldDeviceJob.getJobGroup() : "MODBUS");
create.setMisfirePolicy(StringUtils.isNotEmpty(oldDeviceJob.getMisfirePolicy()) ? oldDeviceJob.getMisfirePolicy() : "2");
create.setConcurrent(StringUtils.isNotEmpty(oldDeviceJob.getConcurrent()) ? oldDeviceJob.getConcurrent() : "1");
create.setIsAdvance(oldDeviceJob.getIsAdvance() != null ? oldDeviceJob.getIsAdvance() : 0);
create.setStatus("0");
create.setRemark(newRemark);
if (isPowerOnTrigger(newCycleType)) {
create.setCronExpression(null);
} else {
SceneModelTagCycleVO vo = JobCronUtils.handleCronCycle(newCycleType, newRemark);
create.setCronExpression(vo.getCron());
}
try {
deviceJobService.insertJob(create);
} catch (SchedulerException | TaskException e) {
throw new ServiceException(e.getMessage());
}
targetDeviceJob = create;
}
// 绑定到新的/复用的 DeviceJob
modbusJob.setJobId(targetDeviceJob.getJobId());
}
int rows = modbusJobMapper.updateModbusJob(modbusJob);
// 如发生重绑且旧 jobId 无引用，则删除旧 DeviceJob（同时会清理 quartz trigger/job）
if (scheduleChanged && oldDeviceJobId != null && modbusJob.getJobId() != null && !oldDeviceJobId.equals(modbusJob.getJobId())) {
ModbusJob q = new ModbusJob();
q.setJobId(oldDeviceJobId);
List<ModbusJob> left = modbusJobMapper.selectModbusJobList(q);
if (CollectionUtils.isEmpty(left)) {
try {
deviceJobService.deleteJobByIds(new Long[]{oldDeviceJobId});
} catch (SchedulerException e) {
throw new ServiceException(e.getMessage());
}
}
}
return rows;
}
/**

8
yack-web/src/api/iot/modbusJob.js
View File

@ -27,15 +27,11 @@ export function addJob(data) {
}
// 修改轮训任务列
export function updateJob(taskId, status) {
const data = {
taskId: taskId,
status: status,
};
export function updateJob(data) {
return request({
url: '/modbus/job',
method: 'put',
data: data,
data,
});
}

65
yack-web/src/views/iot/device/device-modbus-task.vue
View File

@ -86,6 +86,7 @@
<el-table-column :label="$t('device.device-modbus-task.384302-59')" align="center" prop="remarkStr"></el-table-column>
<el-table-column :label="$t('device.device-modbus-task.384302-60')" align="center" class-name="small-padding fixed-width">
<template slot-scope="scope">
<el-button size="mini" type="text" icon="el-icon-edit" @click="handleEdit(scope.row)" v-hasPermi="['modbus:job:edit']">编辑</el-button>
<el-button size="mini" type="text" icon="el-icon-delete" @click="handleDelete(scope.row)" v-hasPermi="['modbus:job:remove']">{{ $t('device.device-modbus-task.384302-61') }}</el-button>
</template>
</el-table-column>
@ -517,6 +518,50 @@ export default {
};
},
methods: {
// 回填时间周期（remark -> cycles1/cycles2）
parseRemarkToCycles(cycleType, remark) {
// 默认值
this.cycles1 = [{ interval: '300', time: '', week: '', day: '' }];
this.cycles2 = [{ type: 'day', time: '00', week: '', day: '', toType: '1', toTime: '02', toWeek: '', toDay: '' }];
if (!remark) return;
let list = [];
try {
list = JSON.parse(remark);
} catch (e) {
return;
}
if (!Array.isArray(list) || list.length === 0) return;
const first = list[0] || {};
if (cycleType === 2) {
// 自定义时间段（当前页面若未展示 cycles2，则仅做数据回填保底）
this.cycles2 = list;
return;
}
// 周期循环
if (first.interval != null) {
this.cycles1 = [{ interval: String(first.interval), time: '', week: '', day: '' }];
return;
}
if (first.type === 'hour') {
this.cycles1 = [{ interval: 'hour', time: '', week: '', day: '' }];
return;
}
if (first.type === 'day') {
this.cycles1 = [{ interval: 'day', time: String(first.time || '01'), week: '', day: '' }];
return;
}
if (first.type === 'week') {
this.cycles1 = [{ interval: 'week', week: String(first.week || '1'), time: String(first.time || '01'), day: '' }];
return;
}
if (first.type === 'month') {
this.cycles1 = [{ interval: 'month', day: String(first.day || '1'), time: String(first.time || '01'), week: '' }];
return;
}
},
/** 查询轮训任务列列表 */
getList() {
this.loading = true;
@ -871,7 +916,7 @@ export default {
this.$modal
.confirm(this.$t('device.device-modbus-task.384302-44', [text + '""' + row.jobName]))
.then(function () {
return updateJob(row.taskId, row.status);
return updateJob({ taskId: row.taskId, status: row.status });
})
.then(() => {
this.$modal.msgSuccess(text + this.$t('device.device-modbus-task.384302-45'));
@ -880,6 +925,24 @@ export default {
row.status = row.status === '0' ? '1' : '0';
});
},
// 编辑轮询任务（修改时间周期/指令等）
handleEdit(row) {
this.resetCreateForm();
this.editName = true;
getJob(row.taskId).then((res) => {
const data = (res && res.data) || {};
this.createForm = Object.assign({}, this.createForm, data);
// 回填指令
this.createCode = data.command || '';
// 回填周期
const cycleType = data.cycleType != null ? data.cycleType : 1;
this.createForm.cycleType = cycleType;
this.parseRemarkToCycles(cycleType, data.remark);
// 打开弹窗
this.editDialog = true;
});
},
// 格式化显示CRON描述
formatCronDisplay(item) {
let result = '';