IrynaMatveieva
1 year ago
59 changed files with 2527 additions and 70 deletions
@ -0,0 +1,49 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.actors.calculatedField; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
|
|||
@Data |
|||
public class CalculatedFieldRelatedEntityMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final CalculatedFieldCtx calculatedField; |
|||
private final TbCallback callback; |
|||
|
|||
public CalculatedFieldRelatedEntityMsg(TenantId tenantId, |
|||
EntityId entityId, |
|||
CalculatedFieldCtx calculatedField, |
|||
TbCallback callback) { |
|||
this.tenantId = tenantId; |
|||
this.entityId = entityId; |
|||
this.calculatedField = calculatedField; |
|||
this.callback = callback; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_RELATED_ENTITY_MSG; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,72 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.script.api.tbel.TbelCfArg; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntryType; |
|||
|
|||
import java.util.Map; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
public class AggArgumentEntry implements ArgumentEntry { |
|||
|
|||
private final Map<EntityId, ArgumentEntry> aggInputs; |
|||
|
|||
private boolean forceResetPrevious; |
|||
|
|||
@Override |
|||
public ArgumentEntryType getType() { |
|||
return ArgumentEntryType.AGGREGATE_LATEST; |
|||
} |
|||
|
|||
@Override |
|||
public Object getValue() { |
|||
return aggInputs; |
|||
} |
|||
|
|||
@Override |
|||
public boolean updateEntry(ArgumentEntry entry) { |
|||
if (entry instanceof AggArgumentEntry aggArgumentEntry) { |
|||
aggInputs.putAll(aggArgumentEntry.aggInputs); |
|||
return true; |
|||
} else if (entry instanceof AggSingleArgumentEntry aggSingleArgumentEntry) { |
|||
if (aggSingleArgumentEntry.isDeleted()) { |
|||
aggInputs.remove(aggSingleArgumentEntry.getEntityId()); |
|||
} else { |
|||
aggInputs.put(aggSingleArgumentEntry.getEntityId(), aggSingleArgumentEntry); |
|||
} |
|||
return true; |
|||
} else { |
|||
throw new IllegalArgumentException("Unsupported argument entry type for aggregation argument entry: " + entry.getType()); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public boolean isEmpty() { |
|||
return aggInputs.isEmpty(); |
|||
} |
|||
|
|||
@Override |
|||
public TbelCfArg toTbelCfArg() { |
|||
return null; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,83 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.BasicKvEntry; |
|||
import org.thingsboard.server.common.data.kv.KvEntry; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntryType; |
|||
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class AggSingleArgumentEntry extends SingleValueArgumentEntry { |
|||
|
|||
private EntityId entityId; |
|||
private boolean deleted; |
|||
|
|||
public AggSingleArgumentEntry(EntityId entityId, TsKvProto entry) { |
|||
super(entry); |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
public AggSingleArgumentEntry(EntityId entityId, AttributeValueProto entry) { |
|||
super(entry); |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
public AggSingleArgumentEntry(EntityId entityId, KvEntry entry) { |
|||
super(entry); |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
public AggSingleArgumentEntry(EntityId entityId, long ts, BasicKvEntry kvEntryValue, Long version) { |
|||
super(ts, kvEntryValue, version); |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
@Override |
|||
public boolean updateEntry(ArgumentEntry entry) { |
|||
if (entry instanceof AggSingleArgumentEntry singleValueEntry) { |
|||
if (singleValueEntry.getTs() <= ts) { |
|||
return false; |
|||
} |
|||
|
|||
Long newVersion = singleValueEntry.getVersion(); |
|||
if (newVersion == null || this.version == null || newVersion > this.version) { |
|||
this.ts = singleValueEntry.getTs(); |
|||
this.version = newVersion; |
|||
this.kvEntryValue = singleValueEntry.getKvEntryValue(); |
|||
this.entityId = singleValueEntry.getEntityId(); |
|||
return true; |
|||
} |
|||
} else { |
|||
throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType()); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public ArgumentEntryType getType() { |
|||
return ArgumentEntryType.AGGREGATE_LATEST_SINGLE; |
|||
} |
|||
} |
|||
@ -0,0 +1,161 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.actors.TbActorRef; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
import org.thingsboard.server.common.data.cf.configuration.Output; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunctionInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggKeyInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldResult; |
|||
import org.thingsboard.server.service.cf.TelemetryCalculatedFieldResult; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; |
|||
import org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
import org.thingsboard.server.service.cf.ctx.state.aggregation.function.AggEntry; |
|||
import org.thingsboard.server.service.cf.ctx.state.aggregation.function.AggFunctionFactory; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.Map.Entry; |
|||
|
|||
@Slf4j |
|||
@Data |
|||
public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedFieldState { |
|||
|
|||
private long lastArgsRefreshTs = -1; |
|||
private long lastMetricsEvalTs = -1; |
|||
private long deduplicationInterval = -1; |
|||
private Map<String, AggMetric> metrics; |
|||
|
|||
private final Map<EntityId, Map<String, ArgumentEntry>> inputs = new HashMap<>(); |
|||
|
|||
public LatestValuesAggregationCalculatedFieldState(EntityId entityId) { |
|||
super(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) { |
|||
super.setCtx(ctx, actorCtx); |
|||
var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); |
|||
metrics = configuration.getMetrics(); |
|||
deduplicationInterval = configuration.getDeduplicationIntervalMillis(); |
|||
} |
|||
|
|||
@Override |
|||
public void reset() { // must reset everything dependent on arguments
|
|||
super.reset(); |
|||
lastArgsRefreshTs = -1; |
|||
lastMetricsEvalTs = -1; |
|||
metrics = null; |
|||
} |
|||
|
|||
@Override |
|||
public void init() { |
|||
super.init(); |
|||
// long scheduledUpdateIntervalMillis = ctx.getScheduledUpdateIntervalMillis();
|
|||
// ctx.scheduleReevaluation(scheduledUpdateIntervalMillis, actorCtx);
|
|||
} |
|||
|
|||
@Override |
|||
public CalculatedFieldType getType() { |
|||
return CalculatedFieldType.LATEST_VALUES_AGGREGATION; |
|||
} |
|||
|
|||
@Override |
|||
public Map<String, ArgumentEntry> update(Map<String, ArgumentEntry> argumentValues, CalculatedFieldCtx ctx) { |
|||
Map<String, ArgumentEntry> updatedArguments = super.update(argumentValues, ctx); |
|||
lastArgsRefreshTs = System.currentTimeMillis(); |
|||
for (Map.Entry<String, ArgumentEntry> argEntry : arguments.entrySet()) { |
|||
String key = argEntry.getKey(); |
|||
AggArgumentEntry aggArgumentEntry = (AggArgumentEntry) argEntry.getValue(); |
|||
Map<EntityId, ArgumentEntry> aggInputs = aggArgumentEntry.getAggInputs(); |
|||
aggInputs.forEach((entityId, argumentEntry) -> { |
|||
inputs.computeIfAbsent(entityId, k -> new HashMap<>()).put(key, argumentEntry); |
|||
}); |
|||
} |
|||
return updatedArguments; |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx) throws Exception { |
|||
boolean intervalPassed = lastMetricsEvalTs <= System.currentTimeMillis() - deduplicationInterval; |
|||
boolean argsUpdatedDuringInterval = lastArgsRefreshTs > lastMetricsEvalTs; |
|||
if (intervalPassed && argsUpdatedDuringInterval) { |
|||
ObjectNode aggResult = JacksonUtil.newObjectNode(); |
|||
for (Entry<String, AggMetric> entry : metrics.entrySet()) { |
|||
String metricKey = entry.getKey(); |
|||
AggMetric metric = entry.getValue(); |
|||
|
|||
AggEntry aggMetric = AggFunctionFactory.createAggFunction(metric.getFunction()); |
|||
|
|||
for (Map<String, ArgumentEntry> entityInputs : inputs.values()) { |
|||
if (applyAggregation(metric.getFilter(), entityInputs)) { |
|||
Object arg = resolveAggregationInput(metric.getInput(), entityInputs); |
|||
if (arg != null) { |
|||
aggMetric.update(arg); |
|||
} |
|||
} |
|||
} |
|||
|
|||
aggMetric.result().ifPresent(result -> { |
|||
aggResult.set(metricKey, JacksonUtil.valueToTree(result)); |
|||
}); |
|||
} |
|||
Output output = ctx.getOutput(); |
|||
lastMetricsEvalTs = System.currentTimeMillis(); |
|||
ctx.scheduleReevaluation(deduplicationInterval, actorCtx); |
|||
return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() |
|||
.type(output.getType()) |
|||
.scope(output.getScope()) |
|||
.result(aggResult) |
|||
.build()); |
|||
} else { |
|||
return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() |
|||
.result(null) |
|||
.build()); |
|||
} |
|||
} |
|||
|
|||
private boolean applyAggregation(String filter, Map<String, ArgumentEntry> entityInputs) throws Exception { |
|||
if (filter == null || filter.isEmpty()) { |
|||
return true; |
|||
} else { |
|||
Object filterResult = ctx.evaluateTbelExpression(filter, entityInputs, getLatestTimestamp()).get(); |
|||
return filterResult instanceof Boolean booleanResult && booleanResult; |
|||
} |
|||
} |
|||
|
|||
private Object resolveAggregationInput(AggInput aggInput, Map<String, ArgumentEntry> entityInputs) throws Exception { |
|||
if (aggInput instanceof AggFunctionInput functionInput) { |
|||
return ctx.evaluateTbelExpression(functionInput.getFunction(), entityInputs, getLatestTimestamp()).get(); |
|||
} else { |
|||
String inputKey = ((AggKeyInput) aggInput).getKey(); |
|||
return entityInputs.get(inputKey).getValue(); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,65 @@ |
|||
{ |
|||
"type": "LATEST_VALUES_AGGREGATION", |
|||
"name": "Occupied spaces", |
|||
"debugSettings": { |
|||
"failuresEnabled": true, |
|||
"allEnabled": true, |
|||
"allEnabledUntil": 1769907492297 |
|||
}, |
|||
"entityId": { |
|||
"entityType": "ASSET", |
|||
"id": "cc830710-a4cf-11f0-87cb-2d6683c4fccf" |
|||
}, |
|||
"configuration": { |
|||
"type": "LATEST_VALUES_AGGREGATION", |
|||
"source": { |
|||
"relation": { |
|||
"direction": "FROM", |
|||
"relationType": "Contains" |
|||
}, |
|||
"entityProfiles": [ |
|||
{ |
|||
"entityType": "DEVICE_PROFILE", |
|||
"id": "d7a05580-a4cf-11f0-87cb-2d6683c4fccf" |
|||
} |
|||
] |
|||
}, |
|||
"inputs": { |
|||
"oc": { |
|||
"key": "occupied", |
|||
"type": "TS_LATEST" |
|||
} |
|||
}, |
|||
"deduplicationIntervalMillis": 10000, |
|||
"metrics": { |
|||
"totalSpaces": { |
|||
"function": "COUNT", |
|||
"input": { |
|||
"type": "function", |
|||
"function" : "return 1;" |
|||
} |
|||
}, |
|||
"occupiedSpaces": { |
|||
"function": "COUNT", |
|||
"filter": "return oc == true", |
|||
"input": { |
|||
"type": "key", |
|||
"key" : "oc" |
|||
} |
|||
}, |
|||
"freeSpaces": { |
|||
"function": "COUNT", |
|||
"filter": "return oc == false", |
|||
"input": { |
|||
"type": "key", |
|||
"key" : "oc" |
|||
} |
|||
} |
|||
}, |
|||
"output": { |
|||
"type": "TIME_SERIES", |
|||
"decimals": 2 |
|||
}, |
|||
"useLatestTsFromInputs": "true" |
|||
} |
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonSubTypes; |
|||
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
@JsonTypeInfo( |
|||
use = JsonTypeInfo.Id.NAME, |
|||
include = JsonTypeInfo.As.PROPERTY, |
|||
property = "type" |
|||
) |
|||
@JsonSubTypes({ |
|||
@JsonSubTypes.Type(value = AvgAggEntry.class, name = "AVG"), |
|||
@JsonSubTypes.Type(value = CountAggEntry.class, name = "COUNT"), |
|||
@JsonSubTypes.Type(value = CountUniqueAggEntry.class, name = "COUNT_UNIQUE"), |
|||
@JsonSubTypes.Type(value = MaxAggEntry.class, name = "MAX"), |
|||
@JsonSubTypes.Type(value = MinAggEntry.class, name = "MIN"), |
|||
@JsonSubTypes.Type(value = SumAggEntry.class, name = "SUM") |
|||
}) |
|||
public interface AggEntry { |
|||
|
|||
AggFunction getType(); |
|||
|
|||
void update(Object value); |
|||
|
|||
Optional<Object> result(); |
|||
|
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
public class AggFunctionFactory { |
|||
|
|||
public static AggEntry createAggFunction(AggFunction function) { |
|||
return switch (function) { |
|||
case MIN -> new MinAggEntry(); |
|||
case MAX -> new MaxAggEntry(); |
|||
case SUM -> new SumAggEntry(); |
|||
case AVG -> new AvgAggEntry(); |
|||
case COUNT -> new CountAggEntry(); |
|||
case COUNT_UNIQUE -> new CountUniqueAggEntry(); |
|||
}; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
import java.math.BigDecimal; |
|||
import java.math.RoundingMode; |
|||
|
|||
public class AvgAggEntry extends BaseAggEntry { |
|||
|
|||
private BigDecimal sum = BigDecimal.ZERO; |
|||
private long count = 0L; |
|||
|
|||
@Override |
|||
protected void doUpdate(double value) { |
|||
if (value != 0.0) { |
|||
sum = sum.add(BigDecimal.valueOf(value)); |
|||
} |
|||
this.count++; |
|||
} |
|||
|
|||
@Override |
|||
protected double prepareResult() { |
|||
return sum.divide(BigDecimal.valueOf(count), 2, RoundingMode.HALF_UP).doubleValue(); |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.AVG; |
|||
} |
|||
} |
|||
@ -0,0 +1,52 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
public abstract class BaseAggEntry implements AggEntry { |
|||
|
|||
private boolean hasResult = false; |
|||
|
|||
@Override |
|||
public void update(Object value) { |
|||
doUpdate(extractDoubleValue(value)); |
|||
hasResult = true; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Object> result() { |
|||
if (hasResult) { |
|||
hasResult = false; |
|||
return Optional.of(prepareResult()); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
} |
|||
|
|||
protected abstract void doUpdate(double value); |
|||
|
|||
protected abstract double prepareResult(); |
|||
|
|||
protected double extractDoubleValue(Object value) { |
|||
try { |
|||
return Double.parseDouble(value.toString()); |
|||
} catch (Exception e) { |
|||
throw new NumberFormatException("Cannot parse value " + value.toString()); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
public class CountAggEntry implements AggEntry { |
|||
|
|||
private long count = 0L; |
|||
|
|||
@Override |
|||
public void update(Object value) { |
|||
count++; |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Object> result() { |
|||
return Optional.of(count); |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.COUNT; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
import java.util.HashSet; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
|
|||
public class CountUniqueAggEntry implements AggEntry { |
|||
|
|||
private Set<String> items; |
|||
|
|||
@Override |
|||
public void update(Object value) { |
|||
if (value != null) { |
|||
items.add(JacksonUtil.toString(value)); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public Optional<Object> result() { |
|||
return Optional.of(items.size()); |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.COUNT_UNIQUE; |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
public class MaxAggEntry extends BaseAggEntry { |
|||
|
|||
private double max = -Double.MAX_VALUE; |
|||
|
|||
@Override |
|||
protected void doUpdate(double value) { |
|||
if (value > max) { |
|||
max = value; |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected double prepareResult() { |
|||
return max; |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.MAX; |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
public class MinAggEntry extends BaseAggEntry { |
|||
|
|||
private double min = Double.MAX_VALUE; |
|||
|
|||
@Override |
|||
protected void doUpdate(double value) { |
|||
if (value < min) { |
|||
min = value; |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected double prepareResult() { |
|||
return min; |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.MIN; |
|||
} |
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.cf.ctx.state.aggregation.function; |
|||
|
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunction; |
|||
|
|||
import java.math.BigDecimal; |
|||
|
|||
public class SumAggEntry extends BaseAggEntry { |
|||
|
|||
private BigDecimal sum = BigDecimal.ZERO; |
|||
|
|||
@Override |
|||
protected void doUpdate(double value) { |
|||
if (value != 0.0) { |
|||
sum = sum.add(BigDecimal.valueOf(value)); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected double prepareResult() { |
|||
return sum.doubleValue(); |
|||
} |
|||
|
|||
@Override |
|||
public AggFunction getType() { |
|||
return AggFunction.SUM; |
|||
} |
|||
} |
|||
@ -0,0 +1,314 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.cf; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import org.junit.After; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.DeviceProfile; |
|||
import org.thingsboard.server.common.data.Tenant; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.asset.Asset; |
|||
import org.thingsboard.server.common.data.asset.AssetProfile; |
|||
import org.thingsboard.server.common.data.cf.CalculatedField; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
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.aggregation.AggFunction; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunctionInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggKeyInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggSource; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration; |
|||
import org.thingsboard.server.common.data.debug.DebugSettings; |
|||
import org.thingsboard.server.common.data.device.data.DefaultDeviceConfiguration; |
|||
import org.thingsboard.server.common.data.device.data.DefaultDeviceTransportConfiguration; |
|||
import org.thingsboard.server.common.data.device.data.DeviceData; |
|||
import org.thingsboard.server.common.data.id.AssetProfileId; |
|||
import org.thingsboard.server.common.data.id.DeviceProfileId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.common.data.relation.EntitySearchDirection; |
|||
import org.thingsboard.server.common.data.relation.RelationPathLevel; |
|||
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
|||
import org.thingsboard.server.common.data.security.Authority; |
|||
import org.thingsboard.server.controller.AbstractControllerTest; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.awaitility.Awaitility.await; |
|||
import static org.thingsboard.server.cf.CalculatedFieldIntegrationTest.POLL_INTERVAL; |
|||
|
|||
@DaoSqlTest |
|||
public class LatestValuesAggregationCalculatedFieldTest extends AbstractControllerTest { |
|||
|
|||
private Tenant savedTenant; |
|||
|
|||
private DeviceProfile deviceProfile; |
|||
private Device device1; |
|||
private String accessToken1 = "1234567890111"; |
|||
private Device device2; |
|||
private String accessToken2 = "1234567890222"; |
|||
|
|||
private AssetProfile assetProfile; |
|||
private Asset asset; |
|||
|
|||
private CalculatedField calculatedField; |
|||
|
|||
private long deduplicationInterval = 10000; |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
loginSysAdmin(); |
|||
|
|||
Tenant tenant = new Tenant(); |
|||
tenant.setTitle("My tenant"); |
|||
savedTenant = saveTenant(tenant); |
|||
assertThat(savedTenant).isNotNull(); |
|||
|
|||
User tenantAdmin = new User(); |
|||
tenantAdmin.setAuthority(Authority.TENANT_ADMIN); |
|||
tenantAdmin.setTenantId(savedTenant.getId()); |
|||
tenantAdmin.setEmail("tenant@thingsboard.org"); |
|||
tenantAdmin.setFirstName("John"); |
|||
tenantAdmin.setLastName("Doe"); |
|||
|
|||
createUserAndLogin(tenantAdmin, "testPassword"); |
|||
|
|||
deviceProfile = doPost("/api/deviceProfile", createDeviceProfile("Device Profile"), DeviceProfile.class); |
|||
device1 = createDevice("Device 1", deviceProfile.getId(), accessToken1); |
|||
device2 = createDevice("Device 2", deviceProfile.getId(), accessToken2); |
|||
|
|||
postTelemetry(device1.getId(), "{\"occupied\":true}"); |
|||
postTelemetry(device2.getId(), "{\"occupied\":false}"); |
|||
|
|||
assetProfile = doPost("/api/assetProfile", createAssetProfile("Asset Profile"), AssetProfile.class); |
|||
asset = createAsset("Asset", assetProfile.getId()); |
|||
|
|||
createEntityRelation(asset.getId(), device1.getId(), "Contains"); |
|||
createEntityRelation(asset.getId(), device2.getId(), "Contains"); |
|||
|
|||
calculatedField = createOccupancyCF(asset.getId(), List.of(deviceProfile.getId())); |
|||
|
|||
checkInitialCalculation(); |
|||
} |
|||
|
|||
@After |
|||
public void afterTest() throws Exception { |
|||
loginSysAdmin(); |
|||
|
|||
deleteTenant(savedTenant.getId()); |
|||
} |
|||
|
|||
@Test |
|||
public void testUpdateTelemetry_checkMetricsCalculation() throws Exception { |
|||
postTelemetry(device1.getId(), "{\"occupied\":false}"); |
|||
|
|||
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("0"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testUpdateTelemetry_checkMetricsCalculationNotExecutedUntilDeduplicationInterval() throws Exception { |
|||
postTelemetry(device1.getId(), "{\"occupied\":false}"); |
|||
|
|||
await().alias("update telemetry -> no changes").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(this::checkInitialCalculationValues); |
|||
|
|||
postTelemetry(device2.getId(), "{\"occupied\":false}"); |
|||
|
|||
await().alias("create CF and perform initial calculation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("0"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testChangeProfile_checkMetricsCalculation() throws Exception { |
|||
DeviceProfile deviceProfile2 = doPost("/api/deviceProfile", createDeviceProfile("Device Profile 2"), DeviceProfile.class); |
|||
device1.setDeviceProfileId(deviceProfile2.getId()); |
|||
device1 = doPost("/api/device?accessToken=" + accessToken1, device1, Device.class); |
|||
|
|||
postTelemetry(device1.getId(), "{\"occupied\":false}"); |
|||
|
|||
await().alias("change profile and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("0"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testAddEntityToProfile_checkMetricsCalculation() throws Exception { |
|||
Device device3 = createDevice("Device 3", deviceProfile.getId(), "1234567890333"); |
|||
|
|||
postTelemetry(device3.getId(), "{\"occupied\":true}"); |
|||
|
|||
await().alias("add entity to profile and no calculation (there is no relation between device and asset)").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(this::checkInitialCalculationValues); |
|||
|
|||
createEntityRelation(asset.getId(), device3.getId(), "Contains"); |
|||
|
|||
await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("3"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testDeleteRelation_checkMetricsCalculation() throws Exception { |
|||
deleteEntityRelation(new EntityRelation(asset.getId(), device1.getId(), "Contains", RelationTypeGroup.COMMON)); |
|||
|
|||
await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("0"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
}); |
|||
} |
|||
|
|||
private void checkInitialCalculation() { |
|||
await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(this::checkInitialCalculationValues); |
|||
} |
|||
|
|||
private void checkInitialCalculationValues() throws Exception { |
|||
ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); |
|||
assertThat(occupancy).isNotNull(); |
|||
assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("1"); |
|||
assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2"); |
|||
} |
|||
|
|||
private CalculatedField createOccupancyCF(EntityId entityId, List<EntityId> profiles) { |
|||
Map<String, AggMetric> aggMetrics = new HashMap<>(); |
|||
|
|||
AggMetric freeSpaces = new AggMetric(); |
|||
freeSpaces.setFunction(AggFunction.COUNT); |
|||
freeSpaces.setFilter("return oc == false"); |
|||
freeSpaces.setInput(new AggKeyInput("oc")); |
|||
aggMetrics.put("freeSpaces", freeSpaces); |
|||
|
|||
AggMetric occupiedSpaces = new AggMetric(); |
|||
occupiedSpaces.setFunction(AggFunction.COUNT); |
|||
occupiedSpaces.setFilter("return oc == true"); |
|||
occupiedSpaces.setInput(new AggKeyInput("oc")); |
|||
aggMetrics.put("occupiedSpaces", occupiedSpaces); |
|||
|
|||
AggMetric totalSpaces = new AggMetric(); |
|||
totalSpaces.setFunction(AggFunction.COUNT); |
|||
totalSpaces.setInput(new AggFunctionInput("return 1;")); |
|||
aggMetrics.put("totalSpaces", totalSpaces); |
|||
|
|||
Output output = new Output(); |
|||
output.setType(OutputType.TIME_SERIES); |
|||
|
|||
return createAggCf("Occupied spaces", entityId, |
|||
buildSource(EntitySearchDirection.FROM, "Contains", profiles), |
|||
Map.of("oc", new ReferencedEntityKey("occupied", ArgumentType.TS_LATEST, null)), |
|||
aggMetrics, |
|||
output); |
|||
} |
|||
|
|||
private AggSource buildSource(EntitySearchDirection direction, String relationType, List<EntityId> profiles) { |
|||
AggSource source = new AggSource(); |
|||
source.setRelation(new RelationPathLevel(direction, relationType)); |
|||
source.setEntityProfiles(profiles); |
|||
return source; |
|||
} |
|||
|
|||
private CalculatedField createAggCf(String name, |
|||
EntityId entityId, |
|||
AggSource aggSource, |
|||
Map<String, ReferencedEntityKey> inputs, |
|||
Map<String, AggMetric> metrics, |
|||
Output output) { |
|||
CalculatedField calculatedField = new CalculatedField(); |
|||
calculatedField.setName(name); |
|||
calculatedField.setEntityId(entityId); |
|||
calculatedField.setType(CalculatedFieldType.LATEST_VALUES_AGGREGATION); |
|||
|
|||
LatestValuesAggregationCalculatedFieldConfiguration configuration = new LatestValuesAggregationCalculatedFieldConfiguration(); |
|||
configuration.setSource(aggSource); |
|||
configuration.setInputs(inputs); |
|||
configuration.setDeduplicationIntervalMillis(deduplicationInterval); |
|||
configuration.setMetrics(metrics); |
|||
configuration.setOutput(output); |
|||
|
|||
calculatedField.setConfiguration(configuration); |
|||
calculatedField.setDebugSettings(DebugSettings.all()); |
|||
return saveCalculatedField(calculatedField); |
|||
} |
|||
|
|||
private Device createDevice(String name, DeviceProfileId deviceProfileId, String accessToken) { |
|||
Device device = new Device(); |
|||
device.setName(name); |
|||
device.setDeviceProfileId(deviceProfileId); |
|||
DeviceData deviceData = new DeviceData(); |
|||
deviceData.setTransportConfiguration(new DefaultDeviceTransportConfiguration()); |
|||
deviceData.setConfiguration(new DefaultDeviceConfiguration()); |
|||
device.setDeviceData(deviceData); |
|||
return doPost("/api/device?accessToken=" + accessToken, device, Device.class); |
|||
} |
|||
|
|||
private Asset createAsset(String name, AssetProfileId assetProfileId) { |
|||
Asset asset = new Asset(); |
|||
asset.setName(name); |
|||
asset.setAssetProfileId(assetProfileId); |
|||
return doPost("/api/asset", asset, Asset.class); |
|||
} |
|||
|
|||
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); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
public enum AggFunction { |
|||
MIN, MAX, SUM, AVG, COUNT, COUNT_UNIQUE |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public class AggFunctionInput implements AggInput { |
|||
|
|||
private String function; |
|||
|
|||
@Override |
|||
public String getType() { |
|||
return "function"; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import com.fasterxml.jackson.annotation.JsonSubTypes; |
|||
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
|||
|
|||
@JsonTypeInfo( |
|||
use = JsonTypeInfo.Id.NAME, |
|||
include = JsonTypeInfo.As.PROPERTY, |
|||
property = "type" |
|||
) |
|||
@JsonSubTypes({ |
|||
@JsonSubTypes.Type(value = AggKeyInput.class, name = "key"), |
|||
@JsonSubTypes.Type(value = AggFunctionInput.class, name = "function") |
|||
}) |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public interface AggInput { |
|||
|
|||
String getType(); |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public class AggKeyInput implements AggInput { |
|||
|
|||
private String key; |
|||
|
|||
@Override |
|||
public String getType() { |
|||
return "key"; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import com.fasterxml.jackson.annotation.JsonInclude; |
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
@JsonInclude(JsonInclude.Include.NON_NULL) |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public class AggMetric { |
|||
|
|||
private AggFunction function; |
|||
private String filter; |
|||
private AggInput input; |
|||
|
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.relation.RelationPathLevel; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class AggSource { |
|||
|
|||
private RelationPathLevel relation; |
|||
private List<EntityId> entityProfiles; |
|||
|
|||
} |
|||
@ -0,0 +1,104 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.cf.configuration.ArgumentType; |
|||
import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
|
|||
import java.util.List; |
|||
import java.util.function.Predicate; |
|||
|
|||
@Data |
|||
@Builder |
|||
public class CfAggTrigger { |
|||
|
|||
private List<EntityId> entityProfiles; |
|||
private List<ReferencedEntityKey> inputs; |
|||
|
|||
public boolean matches(EntityId profileId, Predicate<CfAggTrigger> cfAggTrigger) { |
|||
if (matchesProfile(profileId)) { |
|||
return cfAggTrigger.test(this); |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
public boolean matchesProfile(EntityId profileId) { |
|||
return entityProfiles.contains(profileId); |
|||
} |
|||
|
|||
public boolean matchesTimeSeries(List<TsKvEntry> telemetry) { |
|||
if (telemetry == null || telemetry.isEmpty()) { |
|||
return false; |
|||
} |
|||
for (TsKvEntry tsKvEntry : telemetry) { |
|||
ReferencedEntityKey latestKey = new ReferencedEntityKey(tsKvEntry.getKey(), ArgumentType.TS_LATEST, null); |
|||
if (inputs.contains(latestKey)) { |
|||
return true; |
|||
} |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
public boolean matchesAttributes(List<AttributeKvEntry> attributes, AttributeScope scope) { |
|||
if (attributes == null || attributes.isEmpty()) { |
|||
return false; |
|||
} |
|||
for (AttributeKvEntry attributeKvEntry : attributes) { |
|||
ReferencedEntityKey latestKey = new ReferencedEntityKey(attributeKvEntry.getKey(), ArgumentType.ATTRIBUTE, scope); |
|||
if (inputs.contains(latestKey)) { |
|||
return true; |
|||
} |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
public boolean matchesTimeSeriesKeys(List<String> telemetry) { |
|||
if (telemetry == null || telemetry.isEmpty()) { |
|||
return false; |
|||
} |
|||
|
|||
for (String key : telemetry) { |
|||
ReferencedEntityKey latestKey = new ReferencedEntityKey(key, ArgumentType.TS_LATEST, null); |
|||
if (inputs.contains(latestKey)) { |
|||
return true; |
|||
} |
|||
} |
|||
|
|||
return false; |
|||
} |
|||
|
|||
public boolean matchesAttributesKeys(List<String> attributes, AttributeScope scope) { |
|||
if (attributes == null || attributes.isEmpty()) { |
|||
return false; |
|||
} |
|||
|
|||
for (String key : attributes) { |
|||
ReferencedEntityKey latestKey = new ReferencedEntityKey(key, ArgumentType.ATTRIBUTE, scope); |
|||
if (inputs.contains(latestKey)) { |
|||
return true; |
|||
} |
|||
} |
|||
|
|||
return false; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,52 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.cf.configuration.aggregation; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
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 java.util.List; |
|||
import java.util.Map; |
|||
|
|||
@Data |
|||
public class LatestValuesAggregationCalculatedFieldConfiguration implements CalculatedFieldConfiguration { |
|||
|
|||
private AggSource source; |
|||
private Map<String, ReferencedEntityKey> inputs; |
|||
private long deduplicationIntervalMillis; |
|||
private Map<String, AggMetric> metrics; |
|||
private Output output; |
|||
|
|||
@Override |
|||
public CalculatedFieldType getType() { |
|||
return CalculatedFieldType.LATEST_VALUES_AGGREGATION; |
|||
} |
|||
|
|||
@Override |
|||
public void validate() { |
|||
} |
|||
|
|||
public CfAggTrigger buildTrigger() { |
|||
return CfAggTrigger.builder() |
|||
.inputs(List.copyOf(inputs.values())) |
|||
.entityProfiles(source.getEntityProfiles()) |
|||
.build(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2025 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.script.api.tbel; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class TbelCfLatestValuesAggregation implements TbelCfArg { |
|||
|
|||
private final Object value; |
|||
|
|||
@JsonCreator |
|||
public TbelCfLatestValuesAggregation( |
|||
@JsonProperty("value") Object value |
|||
) { |
|||
this.value = value; |
|||
} |
|||
|
|||
|
|||
@Override |
|||
public String getType() { |
|||
return "LATEST_VALUES_AGGREGATION"; |
|||
} |
|||
|
|||
@Override |
|||
public long memorySize() { |
|||
return 32; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue