diff --git a/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java b/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java index f899d0f480..7e4e041600 100644 --- a/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java +++ b/application/src/main/java/org/thingsboard/server/controller/CalculatedFieldController.java @@ -36,6 +36,8 @@ import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfCtx; import org.thingsboard.script.api.tbel.TbelCfSingleValueArg; +import org.thingsboard.script.api.tbel.TbelCfTsDoubleVal; +import org.thingsboard.script.api.tbel.TbelCfTsRollingArg; import org.thingsboard.script.api.tbel.TbelInvokeService; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EventInfo; @@ -242,9 +244,8 @@ public class CalculatedFieldController extends BaseController { ctxAndArgNames.toArray(String[]::new) ); - Object[] args = new Object[ctxAndArgNames.size()]; - args[0] = new TbelCfCtx(arguments); + args[0] = new TbelCfCtx(arguments, getLastUpdateTimestamp(arguments)); for (int i = 1; i < ctxAndArgNames.size(); i++) { var arg = arguments.get(ctxAndArgNames.get(i)); if (arg instanceof TbelCfSingleValueArg svArg) { @@ -267,6 +268,20 @@ public class CalculatedFieldController extends BaseController { return result; } + private long getLastUpdateTimestamp(Map arguments) { + long lastUpdateTimestamp = -1; + for (TbelCfArg entry : arguments.values()) { + if (entry instanceof TbelCfSingleValueArg singleValueArg) { + long ts = singleValueArg.getTs(); + lastUpdateTimestamp = Math.max(lastUpdateTimestamp, ts); + } else if (entry instanceof TbelCfTsRollingArg tsRollingArg) { + long maxTs = tsRollingArg.getValues().stream().mapToLong(TbelCfTsDoubleVal::getTs).max().orElse(-1); + lastUpdateTimestamp = Math.max(lastUpdateTimestamp, maxTs); + } + } + return lastUpdateTimestamp; + } + private & HasTenantId, I extends EntityId> void checkReferencedEntities(CalculatedFieldConfiguration calculatedFieldConfig, SecurityUser user) throws ThingsboardException { List referencedEntityIds = calculatedFieldConfig.getReferencedEntities(); for (EntityId referencedEntityId : referencedEntityIds) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index 80b003b3cc..35879ac9aa 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -35,13 +35,15 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected Map arguments; protected boolean sizeExceedsLimit; + protected long lastUpdateTimestamp = -1; + public BaseCalculatedFieldState(List requiredArguments) { this.requiredArguments = requiredArguments; this.arguments = new HashMap<>(); } public BaseCalculatedFieldState() { - this(new ArrayList<>(), new HashMap<>(), false); + this(new ArrayList<>(), new HashMap<>(), false, -1); } @Override @@ -59,14 +61,21 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { checkArgumentSize(key, newEntry, ctx); ArgumentEntry existingEntry = arguments.get(key); + boolean entryUpdated; if (existingEntry == null || newEntry.isForceResetPrevious()) { validateNewEntry(newEntry); arguments.put(key, newEntry); - stateUpdated = true; + entryUpdated = true; } else { - stateUpdated = existingEntry.updateEntry(newEntry); + entryUpdated = existingEntry.updateEntry(newEntry); + } + + if (entryUpdated) { + stateUpdated = true; + updateLastUpdateTimestamp(newEntry); } + } return stateUpdated; @@ -100,4 +109,15 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected abstract void validateNewEntry(ArgumentEntry newEntry); + private void updateLastUpdateTimestamp(ArgumentEntry entry) { + if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { + this.lastUpdateTimestamp = singleValueArgumentEntry.getTs(); + } else if (entry instanceof TsRollingArgumentEntry tsRollingArgumentEntry) { + Map.Entry lastEntry = tsRollingArgumentEntry.getTsRecords().pollLastEntry(); + if (lastEntry != null) { + this.lastUpdateTimestamp = lastEntry.getKey(); + } + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 674ce3726f..a3fdae319d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; +import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -61,6 +62,7 @@ public class CalculatedFieldCtx { private final List argNames; private Output output; private String expression; + private boolean preserveMsgTs; private TbelInvokeService tbelInvokeService; private CalculatedFieldScriptEngine calculatedFieldScriptEngine; private ThreadLocal customExpression; @@ -94,6 +96,7 @@ public class CalculatedFieldCtx { this.argNames = new ArrayList<>(arguments.keySet()); this.output = configuration.getOutput(); this.expression = configuration.getExpression(); + this.preserveMsgTs = CalculatedFieldType.SIMPLE.equals(calculatedField.getType()) && ((SimpleCalculatedFieldConfiguration) configuration).isPreserveMsgTs(); this.tbelInvokeService = tbelInvokeService; this.maxDataPointsPerRollingArg = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java index fc4ba513d2..6eac3358ba 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java @@ -42,6 +42,8 @@ public interface CalculatedFieldState { Map getArguments(); + long getLastUpdateTimestamp(); + void setRequiredArguments(List requiredArguments); boolean updateState(CalculatedFieldCtx ctx, Map argumentValues); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java index bf00f1b0b1..65ef40330c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java @@ -30,7 +30,6 @@ import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.service.cf.CalculatedFieldResult; import java.util.ArrayList; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -67,7 +66,7 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState { args.add(arg); } } - args.set(0, new TbelCfCtx(arguments)); + args.set(0, new TbelCfCtx(arguments, getLastUpdateTimestamp())); ListenableFuture resultFuture = ctx.getCalculatedFieldScriptEngine().executeJsonAsync(args.toArray()); Output output = ctx.getOutput(); return Futures.transform(resultFuture, diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java index 480b334ac3..d0eba5031c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.service.cf.ctx.state; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; @@ -65,19 +67,35 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState { double expressionResult = expr.evaluate(); Output output = ctx.getOutput(); - Object result; - Integer decimals = output.getDecimalsByDefault(); - if (decimals != null) { - if (decimals.equals(0)) { - result = TbUtils.toInt(expressionResult); - } else { - result = TbUtils.toFixed(expressionResult, decimals); - } - } else { - result = expressionResult; + Object result = formatResult(expressionResult, output.getDecimalsByDefault()); + JsonNode outputResult = createResultJson(ctx.isPreserveMsgTs(), output.getName(), result); + + return Futures.immediateFuture(new CalculatedFieldResult(output.getType(), output.getScope(), outputResult)); + } + + private Object formatResult(double expressionResult, Integer decimals) { + if (decimals == null) { + return expressionResult; } + if (decimals.equals(0)) { + return TbUtils.toInt(expressionResult); + } + return TbUtils.toFixed(expressionResult, decimals); + } + + private JsonNode createResultJson(boolean preserveMsgTs, String outputName, Object result) { + ObjectNode valuesNode = JacksonUtil.newObjectNode(); + valuesNode.set(outputName, JacksonUtil.valueToTree(result)); - return Futures.immediateFuture(new CalculatedFieldResult(output.getType(), output.getScope(), JacksonUtil.valueToTree(Map.of(output.getName(), result)))); + long lastTimestamp = getLastUpdateTimestamp(); + if (preserveMsgTs && lastTimestamp != -1) { + ObjectNode resultNode = JacksonUtil.newObjectNode(); + resultNode.put("ts", lastTimestamp); + resultNode.set("values", valuesNode); + return resultNode; + } else { + return valuesNode; + } } } diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java index 002f668fdc..f65f6bc629 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.cf.configuration.ArgumentType; import org.thingsboard.server.common.data.cf.configuration.Output; import org.thingsboard.server.common.data.cf.configuration.OutputType; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; +import org.thingsboard.server.common.data.cf.configuration.ScriptCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; import org.thingsboard.server.common.data.debug.DebugSettings; import org.thingsboard.server.common.data.id.AssetProfileId; @@ -462,6 +463,87 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes }); } + @Test + public void testSimpleCalculatedFieldWhenPreserveMsgTsIsTrue() throws Exception { + Device testDevice = createDevice("Test device", "1234567890"); + long ts = System.currentTimeMillis() - 300000L; + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts))); + + CalculatedField calculatedField = new CalculatedField(); + calculatedField.setEntityId(testDevice.getId()); + calculatedField.setType(CalculatedFieldType.SIMPLE); + calculatedField.setName("C to F"); + calculatedField.setDebugSettings(DebugSettings.all()); + calculatedField.setConfigurationVersion(1); + + SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration(); + + Argument argument = new Argument(); + ReferencedEntityKey refEntityKey = new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null); + argument.setRefEntityKey(refEntityKey); + config.setArguments(Map.of("T", argument)); + config.setExpression("(T * 9/5) + 32"); + + Output output = new Output(); + output.setName("fahrenheitTemp"); + output.setType(OutputType.TIME_SERIES); + config.setOutput(output); + + config.setPreserveMsgTs(true); + + calculatedField.setConfiguration(config); + + CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class); + + await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode fahrenheitTemp = getLatestTelemetry(testDevice.getId(), "fahrenheitTemp"); + assertThat(fahrenheitTemp).isNotNull(); + assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("ts").asText()).isEqualTo(Long.toString(ts)); + assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("value").asText()).isEqualTo("86.0"); + }); + } + + @Test + public void testScriptCalculatedFieldWhenUsedMsgTsInScript() throws Exception { + Device testDevice = createDevice("Test device", "1234567890"); + long ts = System.currentTimeMillis() - 300000L; + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts))); + + CalculatedField calculatedField = new CalculatedField(); + calculatedField.setEntityId(testDevice.getId()); + calculatedField.setType(CalculatedFieldType.SCRIPT); + calculatedField.setName("C to F"); + calculatedField.setDebugSettings(DebugSettings.all()); + calculatedField.setConfigurationVersion(1); + + ScriptCalculatedFieldConfiguration config = new ScriptCalculatedFieldConfiguration(); + + Argument argument = new Argument(); + ReferencedEntityKey refEntityKey = new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null); + argument.setRefEntityKey(refEntityKey); + config.setArguments(Map.of("T", argument)); + config.setExpression("return {\"ts\": ctx.msgTs, \"values\": {\"fahrenheitTemp\": (T * 1.8) + 32}};"); + + Output output = new Output(); + output.setType(OutputType.TIME_SERIES); + config.setOutput(output); + + calculatedField.setConfiguration(config); + + CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class); + + await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode fahrenheitTemp = getLatestTelemetry(testDevice.getId(), "fahrenheitTemp"); + assertThat(fahrenheitTemp).isNotNull(); + assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("ts").asText()).isEqualTo(Long.toString(ts)); + assertThat(fahrenheitTemp.get("fahrenheitTemp").get(0).get("value").asText()).isEqualTo("86.0"); + }); + } + private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception { return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/SimpleCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/SimpleCalculatedFieldConfiguration.java index 5c0ce71e86..af3cb4d5cd 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/SimpleCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/SimpleCalculatedFieldConfiguration.java @@ -23,6 +23,8 @@ import org.thingsboard.server.common.data.cf.CalculatedFieldType; @EqualsAndHashCode(callSuper = true) public class SimpleCalculatedFieldConfiguration extends BaseCalculatedFieldConfiguration implements CalculatedFieldConfiguration { + private boolean preserveMsgTs; + @Override public CalculatedFieldType getType() { return CalculatedFieldType.SIMPLE; diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java index ce42e2cf3b..7515cb5269 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfCtx.java @@ -24,9 +24,12 @@ public class TbelCfCtx implements TbelCfObject { @Getter private final Map args; + @Getter + private final long msgTs; - public TbelCfCtx(Map args) { + public TbelCfCtx(Map args, long lastUpdateTs) { this.args = Collections.unmodifiableMap(args); + this.msgTs = lastUpdateTs != -1 ? lastUpdateTs : System.currentTimeMillis(); } @Override diff --git a/ui-ngx/src/assets/help/en_US/calculated-field/expression_fn.md b/ui-ngx/src/assets/help/en_US/calculated-field/expression_fn.md index 373c50e2c6..4c54b01499 100644 --- a/ui-ngx/src/assets/help/en_US/calculated-field/expression_fn.md +++ b/ui-ngx/src/assets/help/en_US/calculated-field/expression_fn.md @@ -1,7 +1,7 @@ ## Calculated Field TBEL Script Function The **calculate()** function is a user-defined script that enables custom calculations using [TBEL](${siteBaseUrl}/docs${docPlatformPrefix}/user-guide/tbel/) on telemetry and attribute data. -It receives arguments configured in the calculated field setup, along with an additional `ctx` object that provides access to all arguments. +It receives arguments configured in the calculated field setup, along with an additional `ctx` object that stores `msgTs` and provides access to all arguments. ### Function Signature @@ -44,7 +44,7 @@ Let's modify the function that converts Fahrenheit to Celsius to also return the var temperatureC = (temperatureF - 32) / 1.8; return { "ts": ctx.args.temperatureF.ts, - "values": { "temperatureC": toFixed(temperatureC, 2) } + "values": {"temperatureC": toFixed(temperatureC, 2)} }; ``` @@ -88,7 +88,7 @@ foreach(t: temperature) { } // iterate through all values and calculate the sum using for loop: sum = 0.0; -for(var i = 0; i < temperature.values.size; i++) { +for (var i = 0; i < temperature.values.size; i++) { sum += temperature.values[i].value; } // use built-in function to calculate the sum @@ -146,12 +146,13 @@ function calculate(ctx, altitude, temperature) { Time series rolling arguments can be **merged** to align timestamps across multiple datasets. -| Method | Description | Returns | Example | -|:-----------------------------|:--------------------------------------------------------------------------------------------------------------------------|:----------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `merge(other, settings)` | Merges with another rolling argument. Aligns timestamps and filling missing values with the previous available value. | Merged object with `timeWindow` and aligned values. |

| -| `mergeAll(others, settings)` | Merges multiple rolling arguments. Aligns timestamps and filling missing values with the previous available value. | Merged object with `timeWindow` and aligned values. |

| +| Method | Description | Returns | Example | +|:-----------------------------|:----------------------------------------------------------------------------------------------------------------------|:----------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `merge(other, settings)` | Merges with another rolling argument. Aligns timestamps and filling missing values with the previous available value. | Merged object with `timeWindow` and aligned values. |

| +| `mergeAll(others, settings)` | Merges multiple rolling arguments. Aligns timestamps and filling missing values with the previous available value. | Merged object with `timeWindow` and aligned values. |

| ##### Parameters + | Parameter | Description | |:---------------------|:-------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `other` or `others` | Another rolling argument or array of rolling arguments to merge with. | @@ -187,29 +188,51 @@ function calculate(ctx, temperature, defrost) { The result is a list of issues that may be used to configure alarm rules: ```json -[{ +[ + { "ts": 1741613833843, "values": { - "issue": { - "temperature": -3.12, - "defrostState": false - } + "issue": { + "temperature": -3.12, + "defrostState": false + } } -}, { + }, + { "ts": 1741613923848, "values": { - "issue": { - "temperature": -4.16, - "defrostState": false - } + "issue": { + "temperature": -4.16, + "defrostState": false + } } -}] + } +] ``` ### Function return format The return format depends on the output type configured in the calculated field settings (default: **Time Series**). +### Message timestamp + +The `ctx` object also includes property `msgTs`, which represents the timestamp of the incoming telemetry message that triggered the calculated field execution in milliseconds. + +You can use `ctx.msgTs` to set the timestamp of the resulting output explicitly when returning a time series object. + +```javascript +var temperatureC = (temperatureF - 32) / 1.8; +return { + ts: ctx.msgTs, + values: { + "temperatureC": toFixed(temperatureC, 2) + } +} + +``` + +This ensures that the calculated data point aligns with the timestamp of the triggering telemetry. + ##### Time Series Output The function must return a JSON object or array with or without a timestamp. @@ -246,7 +269,7 @@ With timestamp: "someArray": [1,2,3], "someNestedObject": {"key": "value"} } - } + } } ``` @@ -265,7 +288,7 @@ Array containing multiple timestamps and different values of the `airDensity` : "values": { "airDensity": 1.07 } - } + } ] ```