Viacheslav Klimov
11 months ago
committed by
GitHub
GitHub
87 changed files with 3308 additions and 235 deletions
@ -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.service.cf.ctx.state.aggregation.single; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
public class AggIntervalEntry { |
|||
|
|||
private Long startTs; |
|||
private Long endTs; |
|||
|
|||
public boolean belongsToInterval(long ts) { |
|||
return ts >= startTs && ts < endTs; |
|||
} |
|||
|
|||
public long getIntervalDuration() { |
|||
return endTs - startTs; |
|||
} |
|||
|
|||
} |
|||
@ -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.single; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class AggIntervalEntryStatus { |
|||
|
|||
private long lastArgsRefreshTs = -1; |
|||
|
|||
private long lastMetricsEvalTs = -1; |
|||
|
|||
public AggIntervalEntryStatus(long lastArgsRefreshTs) { |
|||
this.lastArgsRefreshTs = lastArgsRefreshTs; |
|||
} |
|||
|
|||
public boolean intervalPassed(long checkInterval) { |
|||
return lastMetricsEvalTs <= System.currentTimeMillis() - checkInterval; |
|||
} |
|||
|
|||
@JsonIgnore |
|||
public boolean argsUpdated() { |
|||
return lastArgsRefreshTs > -1; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,87 @@ |
|||
/** |
|||
* 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.single; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.script.api.tbel.TbelCfArg; |
|||
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; |
|||
|
|||
import java.util.Map; |
|||
|
|||
@Data |
|||
public class EntityAggregationArgumentEntry implements ArgumentEntry { |
|||
|
|||
private Map<AggIntervalEntry, AggIntervalEntryStatus> aggIntervals; |
|||
|
|||
private boolean forceResetPrevious; |
|||
|
|||
public EntityAggregationArgumentEntry(Map<AggIntervalEntry, AggIntervalEntryStatus> aggIntervals) { |
|||
this.aggIntervals = aggIntervals; |
|||
} |
|||
|
|||
@Override |
|||
public ArgumentEntryType getType() { |
|||
return ArgumentEntryType.ENTITY_AGGREGATION; |
|||
} |
|||
|
|||
@Override |
|||
public Object getValue() { |
|||
return aggIntervals; |
|||
} |
|||
|
|||
@Override |
|||
public boolean updateEntry(ArgumentEntry entry) { |
|||
boolean updated = false; |
|||
if (entry instanceof EntityAggregationArgumentEntry entityAggEntry) { |
|||
aggIntervals.putAll(entityAggEntry.getAggIntervals()); |
|||
} else if (entry instanceof SingleValueArgumentEntry singleValueArgEntry) { |
|||
long entryTs = singleValueArgEntry.getTs(); |
|||
long argUpdateTs = System.currentTimeMillis(); |
|||
for (Map.Entry<AggIntervalEntry, AggIntervalEntryStatus> aggIntervalEntry : aggIntervals.entrySet()) { |
|||
if (singleValueArgEntry.isForceResetPrevious()) { |
|||
aggIntervalEntry.getValue().setLastArgsRefreshTs(argUpdateTs); |
|||
updated = true; |
|||
continue; |
|||
} |
|||
if (aggIntervalEntry.getKey().belongsToInterval(entryTs)) { |
|||
aggIntervalEntry.getValue().setLastArgsRefreshTs(argUpdateTs); |
|||
return true; |
|||
} |
|||
} |
|||
} |
|||
return updated; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isEmpty() { |
|||
return aggIntervals.isEmpty(); |
|||
} |
|||
|
|||
@Override |
|||
public JsonNode jsonValue() { |
|||
return JacksonUtil.valueToTree(aggIntervals); |
|||
} |
|||
|
|||
@Override |
|||
public TbelCfArg toTbelCfArg() { |
|||
return null; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,271 @@ |
|||
/** |
|||
* 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.single; |
|||
|
|||
import com.fasterxml.jackson.databind.node.ArrayNode; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.script.api.tbel.TbUtils; |
|||
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.AggKeyInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.AggInterval; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.Watermark; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; |
|||
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 java.time.Instant; |
|||
import java.time.ZoneId; |
|||
import java.time.ZonedDateTime; |
|||
import java.util.ArrayList; |
|||
import java.util.Comparator; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultMetricArgumentEntry; |
|||
|
|||
public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldState { |
|||
|
|||
private AggInterval interval; |
|||
private long watermarkDuration; |
|||
private long checkInterval; |
|||
private Map<String, AggMetric> metrics; |
|||
|
|||
private CalculatedFieldProcessingService cfProcessingService; |
|||
|
|||
public EntityAggregationCalculatedFieldState(EntityId entityId) { |
|||
super(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) { |
|||
super.setCtx(ctx, actorCtx); |
|||
this.cfProcessingService = ctx.getCfProcessingService(); |
|||
var configuration = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); |
|||
Watermark watermark = configuration.getWatermark(); |
|||
watermarkDuration = watermark == null ? 0 : TimeUnit.SECONDS.toMillis(watermark.getDuration()); |
|||
checkInterval = TimeUnit.SECONDS.toMillis(ctx.getSystemContext().getCfCheckInterval()); |
|||
interval = configuration.getInterval(); |
|||
metrics = configuration.getMetrics(); |
|||
} |
|||
|
|||
@Override |
|||
public void init(boolean restored) { |
|||
super.init(restored); |
|||
if (restored) { |
|||
fillMissingIntervals(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public CalculatedFieldType getType() { |
|||
return CalculatedFieldType.ENTITY_AGGREGATION; |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx) throws Exception { |
|||
createIntervalIfNotExist(); |
|||
long now = System.currentTimeMillis(); |
|||
|
|||
Map<AggIntervalEntry, Map<String, ArgumentEntry>> results = new HashMap<>(); |
|||
List<AggIntervalEntry> expiredIntervals = new ArrayList<>(); |
|||
getIntervals().forEach((intervalEntry, argIntervalStatuses) -> { |
|||
processInterval(now, intervalEntry, argIntervalStatuses, expiredIntervals, results); |
|||
}); |
|||
removeExpiredIntervals(expiredIntervals); |
|||
|
|||
Output output = ctx.getOutput(); |
|||
ArrayNode result = toResult(results, output.getDecimalsByDefault()); |
|||
if (result.isEmpty()) { |
|||
return Futures.immediateFuture(TelemetryCalculatedFieldResult.EMPTY); |
|||
} |
|||
return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() |
|||
.type(output.getType()) |
|||
.scope(output.getScope()) |
|||
.result(result) |
|||
.build()); |
|||
} |
|||
|
|||
private void removeExpiredIntervals(List<AggIntervalEntry> expiredIntervals) { |
|||
expiredIntervals.forEach(expiredInterval -> { |
|||
arguments.values().stream() |
|||
.map(EntityAggregationArgumentEntry.class::cast) |
|||
.forEach(arg -> arg.getAggIntervals().remove(expiredInterval)); |
|||
}); |
|||
} |
|||
|
|||
private void createIntervalIfNotExist() { |
|||
AggIntervalEntry currentInterval = new AggIntervalEntry(interval.getCurrentIntervalStartTs(), interval.getCurrentIntervalEndTs()); |
|||
arguments.forEach((argName, argumentEntry) -> { |
|||
var entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; |
|||
entityAggEntry.getAggIntervals().computeIfAbsent(currentInterval, current -> new AggIntervalEntryStatus()); |
|||
}); |
|||
} |
|||
|
|||
private void fillMissingIntervals() { |
|||
ZoneId zoneId = interval.getZoneId(); |
|||
long currentIntervalEndTs = interval.getCurrentIntervalEndTs(); |
|||
|
|||
Map<AggIntervalEntry, Map<String, AggIntervalEntryStatus>> intervals = getIntervals(); |
|||
AggIntervalEntry lastIntervalEntry = intervals.keySet().stream().max(Comparator.comparing(AggIntervalEntry::getEndTs)).orElse(null); |
|||
if (lastIntervalEntry == null) { |
|||
return; |
|||
} |
|||
|
|||
ZonedDateTime nextStart = Instant.ofEpochMilli(lastIntervalEntry.getEndTs()).atZone(zoneId); |
|||
ZonedDateTime nextEnd = interval.getNextIntervalStart(nextStart); |
|||
|
|||
while (nextEnd.toInstant().toEpochMilli() <= currentIntervalEndTs) { |
|||
long nextStartTs = nextStart.toInstant().toEpochMilli(); |
|||
long nextEndTs = nextEnd.toInstant().toEpochMilli(); |
|||
AggIntervalEntry missing = new AggIntervalEntry(nextStartTs, nextEndTs); |
|||
|
|||
arguments.forEach((argName, argumentEntry) -> { |
|||
var entityAggEntry = (EntityAggregationArgumentEntry) argumentEntry; |
|||
AggIntervalEntryStatus intervalEntryStatus = new AggIntervalEntryStatus(System.currentTimeMillis()); |
|||
entityAggEntry.getAggIntervals().computeIfAbsent(missing, missingInterval -> intervalEntryStatus); |
|||
}); |
|||
|
|||
nextStart = nextEnd; |
|||
nextEnd = interval.getNextIntervalStart(nextStart); |
|||
} |
|||
} |
|||
|
|||
private Map<AggIntervalEntry, Map<String, AggIntervalEntryStatus>> getIntervals() { |
|||
Map<AggIntervalEntry, Map<String, AggIntervalEntryStatus>> intervals = new HashMap<>(); |
|||
arguments.forEach((argName, entry) -> { |
|||
var argEntry = (EntityAggregationArgumentEntry) entry; |
|||
argEntry.getAggIntervals().forEach((intervalEntry, status) -> |
|||
intervals.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(argName, status) |
|||
); |
|||
}); |
|||
return intervals; |
|||
} |
|||
|
|||
private void processInterval(long now, |
|||
AggIntervalEntry intervalEntry, |
|||
Map<String, AggIntervalEntryStatus> args, |
|||
List<AggIntervalEntry> expiredIntervals, |
|||
Map<AggIntervalEntry, Map<String, ArgumentEntry>> results) { |
|||
long startTs = intervalEntry.getStartTs(); |
|||
long endTs = intervalEntry.getEndTs(); |
|||
|
|||
if (now - endTs > watermarkDuration) { |
|||
handleExpiredInterval(intervalEntry, args, results); |
|||
expiredIntervals.add(intervalEntry); |
|||
} else if (now - startTs >= intervalEntry.getIntervalDuration()) { |
|||
handleActiveInterval(intervalEntry, args, results); |
|||
} |
|||
} |
|||
|
|||
private void handleExpiredInterval(AggIntervalEntry intervalEntry, |
|||
Map<String, AggIntervalEntryStatus> args, |
|||
Map<AggIntervalEntry, Map<String, ArgumentEntry>> results) { |
|||
args.forEach((argName, argEntryIntervalStatus) -> { |
|||
if (argEntryIntervalStatus.getLastArgsRefreshTs() > argEntryIntervalStatus.getLastMetricsEvalTs()) { |
|||
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); |
|||
processMetric(intervalEntry, argName, false, results); |
|||
} else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) { |
|||
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); |
|||
processMetric(intervalEntry, argName, true, results); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private void handleActiveInterval(AggIntervalEntry intervalEntry, |
|||
Map<String, AggIntervalEntryStatus> args, |
|||
Map<AggIntervalEntry, Map<String, ArgumentEntry>> results) { |
|||
args.forEach((argName, argEntryIntervalStatus) -> { |
|||
if (argEntryIntervalStatus.intervalPassed(checkInterval)) { |
|||
if (argEntryIntervalStatus.argsUpdated()) { |
|||
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); |
|||
argEntryIntervalStatus.setLastArgsRefreshTs(-1); |
|||
processMetric(intervalEntry, argName, false, results); |
|||
} else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) { |
|||
argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); |
|||
processMetric(intervalEntry, argName, true, results); |
|||
} |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private void processMetric(AggIntervalEntry intervalEntry, |
|||
String argName, |
|||
boolean useDefault, |
|||
Map<AggIntervalEntry, Map<String, ArgumentEntry>> results) { |
|||
String metricName = findMetricName(argName); |
|||
if (metricName != null) { |
|||
AggMetric metric = metrics.get(metricName); |
|||
String argKey = ctx.getArguments().get(argName).getRefEntityKey().getKey(); |
|||
ArgumentEntry metricEntry = useDefault |
|||
? createDefaultMetricArgumentEntry(argKey, metric) |
|||
: cfProcessingService.fetchMetricDuringInterval(ctx.getTenantId(), entityId, argKey, metric, intervalEntry); |
|||
if (!metricEntry.isEmpty()) { |
|||
results.computeIfAbsent(intervalEntry, i -> new HashMap<>()).put(metricName, metricEntry); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private String findMetricName(String argName) { |
|||
return metrics.entrySet().stream() |
|||
.filter(e -> ((AggKeyInput) e.getValue().getInput()).getKey().equals(argName)) |
|||
.map(Map.Entry::getKey) |
|||
.findFirst() |
|||
.orElse(null); |
|||
} |
|||
|
|||
protected ArrayNode toResult(Map<AggIntervalEntry, Map<String, ArgumentEntry>> results, Integer precision) { |
|||
ArrayNode result = JacksonUtil.newArrayNode(); |
|||
results.forEach((interval, args) -> { |
|||
ObjectNode metricsNode = JacksonUtil.newObjectNode(); |
|||
for (Map.Entry<String, ArgumentEntry> entry : args.entrySet()) { |
|||
String metricName = entry.getKey(); |
|||
ArgumentEntry argumentEntry = entry.getValue(); |
|||
if (!argumentEntry.isEmpty()) { |
|||
Object resultValue = argumentEntry.getValue() instanceof Number number |
|||
? TbUtils.roundResult(number.doubleValue(), precision) |
|||
: argumentEntry.getValue(); |
|||
metricsNode.put(metricName, JacksonUtil.toString(resultValue)); |
|||
} |
|||
} |
|||
if (!metricsNode.isEmpty()) { |
|||
ObjectNode resultNode = JacksonUtil.newObjectNode(); |
|||
resultNode.put("ts", interval.getEndTs() - 1); |
|||
resultNode.set("values", metricsNode); |
|||
result.add(resultNode); |
|||
} |
|||
}); |
|||
return result; |
|||
} |
|||
|
|||
@Override |
|||
public boolean isReady() { |
|||
return true; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,255 @@ |
|||
/** |
|||
* 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.springframework.test.annotation.DirtiesContext; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.Tenant; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.cf.CalculatedField; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
import org.thingsboard.server.common.data.cf.configuration.Argument; |
|||
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.AggKeyInput; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.AggInterval; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.CustomInterval; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.Watermark; |
|||
import org.thingsboard.server.common.data.debug.DebugSettings; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
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.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 |
|||
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) |
|||
@TestPropertySource(properties = { |
|||
"actors.calculated_fields.check_interval=1" |
|||
}) |
|||
public class EntityAggregationCalculatedFieldTest extends AbstractControllerTest { |
|||
|
|||
private Tenant savedTenant; |
|||
|
|||
@Before |
|||
public void beforeEach() throws Exception { |
|||
loginSysAdmin(); |
|||
|
|||
updateDefaultTenantProfileConfig(tenantProfileConfig -> { |
|||
tenantProfileConfig.setMinAllowedDeduplicationIntervalInSecForCF(1); |
|||
tenantProfileConfig.setMinAllowedAggregationIntervalInSecForCF(1); |
|||
}); |
|||
|
|||
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"); |
|||
} |
|||
|
|||
@After |
|||
public void afterTest() throws Exception { |
|||
loginSysAdmin(); |
|||
|
|||
deleteTenant(savedTenant.getId()); |
|||
} |
|||
|
|||
@Test |
|||
public void testCreateCfAndNoTelemetryDuringInterval_checkAggregation() throws Exception { |
|||
Device device = createDevice("Device", "1234567890111"); |
|||
|
|||
CustomInterval customInterval = new CustomInterval("Europe/Kyiv", 0L, 5L); |
|||
long intervalEndTs = customInterval.getCurrentIntervalEndTs(); |
|||
|
|||
CalculatedField totalConsumptionCF = createTotalConsumptionCF(device.getId(), customInterval, null); |
|||
long interval = customInterval.getCurrentIntervalDurationMillis(); |
|||
|
|||
await().alias("create CF and no telemetry during interval -> save metric with default value") |
|||
.atMost(2 * interval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode result = getLatestTelemetry(device.getId(), "consumption"); |
|||
assertThat(result).isNotNull(); |
|||
assertThat(result.get("consumption").get(0).get("value").asText()).isEqualTo("9999"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCreateCfWithoutWatermark_checkAggregation() throws Exception { |
|||
Device device = createDevice("Device", "1234567890111"); |
|||
|
|||
CustomInterval customInterval = new CustomInterval("Europe/Kyiv", 0L, 5L); |
|||
long currentIntervalStartTs = customInterval.getCurrentIntervalStartTs(); |
|||
long currentIntervalEndTs = customInterval.getCurrentIntervalEndTs(); |
|||
|
|||
long tsBeforeInterval = currentIntervalStartTs - 1000; |
|||
long tsInInterval_1 = currentIntervalStartTs + 1000; |
|||
long tsInInterval_2 = currentIntervalStartTs + 500; |
|||
long tsInInterval_3 = currentIntervalStartTs + 200; |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsBeforeInterval)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":100}}", tsInInterval_1)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":180}}", tsInInterval_2)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsInInterval_3)); |
|||
|
|||
long interval = customInterval.getCurrentIntervalDurationMillis(); |
|||
CalculatedField totalConsumptionCF = createTotalConsumptionCF(device.getId(), customInterval, null); |
|||
|
|||
await().alias("create CF -> perform aggregation after interval end") |
|||
.atMost(2 * interval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode result = getLatestTelemetry(device.getId(), "consumption"); |
|||
assertThat(result).isNotNull(); |
|||
assertThat(result.get("consumption").get(0).get("value").asText()).isEqualTo("400"); |
|||
}); |
|||
|
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":500}}", tsInInterval_1)); |
|||
|
|||
await().alias("update telemetry that belongs to previous interval -> no aggregation since watermark is not set ") |
|||
.atMost(2 * interval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode result = getLatestTelemetry(device.getId(), "consumption"); |
|||
assertThat(result).isNotNull(); |
|||
assertThat(result.get("consumption").get(0).get("value").asText()).isEqualTo("400"); |
|||
}); |
|||
} |
|||
|
|||
@Test |
|||
public void testCreateCfWithWatermark_checkAggregationDuringWatermark() throws Exception { |
|||
Device device = createDevice("Device", "1234567890111"); |
|||
|
|||
CustomInterval customInterval = new CustomInterval("Europe/Kyiv", 0L, 5L); |
|||
long currentIntervalStartTs = customInterval.getCurrentIntervalStartTs(); |
|||
long currentIntervalEndTs = customInterval.getCurrentIntervalEndTs(); |
|||
|
|||
long tsBeforeInterval = currentIntervalStartTs - 1000L; |
|||
long tsInInterval_1 = currentIntervalStartTs + 1000L; |
|||
long tsInInterval_2 = currentIntervalStartTs + 500L; |
|||
long tsInInterval_3 = currentIntervalStartTs + 200L; |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsBeforeInterval)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":100}}", tsInInterval_1)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":180}}", tsInInterval_2)); |
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsInInterval_3)); |
|||
|
|||
long interval = customInterval.getCurrentIntervalDurationMillis(); |
|||
Watermark watermark = new Watermark(10); |
|||
CalculatedField totalConsumptionCF = createTotalConsumptionCF(device.getId(), customInterval, watermark); |
|||
|
|||
await().alias("create CF -> perform aggregation after interval end") |
|||
.atMost(2 * interval, TimeUnit.MILLISECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode result = getLatestTelemetry(device.getId(), "consumption"); |
|||
assertThat(result).isNotNull(); |
|||
assertThat(result.get("consumption").get(0).get("value").asText()).isEqualTo("400"); |
|||
}); |
|||
|
|||
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":300}}", tsInInterval_1)); |
|||
|
|||
await().alias("update telemetry during watermark -> perform aggregation") |
|||
.atMost(2 * 10, TimeUnit.SECONDS) |
|||
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) |
|||
.untilAsserted(() -> { |
|||
ObjectNode result = getLatestTelemetry(device.getId(), "consumption"); |
|||
assertThat(result).isNotNull(); |
|||
assertThat(result.get("consumption").get(0).get("value").asText()).isEqualTo("600"); |
|||
}); |
|||
} |
|||
|
|||
private CalculatedField createTotalConsumptionCF(EntityId entityId, AggInterval aggInterval, Watermark watermark) { |
|||
Map<String, Argument> arguments = new HashMap<>(); |
|||
Argument argument = new Argument(); |
|||
argument.setRefEntityKey(new ReferencedEntityKey("energy", ArgumentType.TS_LATEST, null)); |
|||
arguments.put("en", argument); |
|||
|
|||
Map<String, AggMetric> aggMetrics = new HashMap<>(); |
|||
|
|||
AggMetric consumption = new AggMetric(); |
|||
consumption.setFunction(AggFunction.SUM); |
|||
consumption.setInput(new AggKeyInput("en")); |
|||
consumption.setDefaultValue(9999L); |
|||
aggMetrics.put("consumption", consumption); |
|||
|
|||
Output output = new Output(); |
|||
output.setType(OutputType.TIME_SERIES); |
|||
output.setDecimalsByDefault(0); |
|||
|
|||
return createAggCf("Consumption per minute", entityId, |
|||
aggInterval, |
|||
watermark, |
|||
arguments, |
|||
aggMetrics, |
|||
output); |
|||
} |
|||
|
|||
private CalculatedField createAggCf(String name, |
|||
EntityId entityId, |
|||
AggInterval aggInterval, |
|||
Watermark watermark, |
|||
Map<String, Argument> inputs, |
|||
Map<String, AggMetric> metrics, |
|||
Output output) { |
|||
CalculatedField calculatedField = new CalculatedField(); |
|||
calculatedField.setName(name); |
|||
calculatedField.setEntityId(entityId); |
|||
calculatedField.setType(CalculatedFieldType.ENTITY_AGGREGATION); |
|||
|
|||
EntityAggregationCalculatedFieldConfiguration configuration = new EntityAggregationCalculatedFieldConfiguration(); |
|||
|
|||
configuration.setArguments(inputs); |
|||
configuration.setMetrics(metrics); |
|||
configuration.setInterval(aggInterval); |
|||
if (watermark != null) { |
|||
configuration.setWatermark(watermark); |
|||
} |
|||
configuration.setOutput(output); |
|||
|
|||
calculatedField.setConfiguration(configuration); |
|||
calculatedField.setDebugSettings(DebugSettings.all()); |
|||
return saveCalculatedField(calculatedField); |
|||
} |
|||
|
|||
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,96 @@ |
|||
/** |
|||
* 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.single; |
|||
|
|||
import jakarta.validation.Valid; |
|||
import jakarta.validation.constraints.NotEmpty; |
|||
import jakarta.validation.constraints.NotNull; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
import org.thingsboard.server.common.data.cf.configuration.Argument; |
|||
import org.thingsboard.server.common.data.cf.configuration.ArgumentType; |
|||
import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration; |
|||
import org.thingsboard.server.common.data.cf.configuration.Output; |
|||
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.single.interval.AggInterval; |
|||
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.Watermark; |
|||
|
|||
import java.util.Map; |
|||
|
|||
@Data |
|||
public class EntityAggregationCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration { |
|||
|
|||
private Map<String, Argument> arguments; |
|||
@Valid |
|||
@NotEmpty |
|||
private Map<String, AggMetric> metrics; |
|||
@Valid |
|||
@NotNull |
|||
private AggInterval interval; |
|||
@Valid |
|||
private Watermark watermark; |
|||
@Valid |
|||
@NotNull |
|||
private Output output; |
|||
|
|||
@Override |
|||
public CalculatedFieldType getType() { |
|||
return CalculatedFieldType.ENTITY_AGGREGATION; |
|||
} |
|||
|
|||
@Override |
|||
public void validate() { |
|||
validateArguments(); |
|||
validateMetrics(); |
|||
validateInterval(); |
|||
} |
|||
|
|||
private void validateArguments() { |
|||
if (arguments.containsKey("ctx")) { |
|||
throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used."); |
|||
} |
|||
if (arguments.values().stream().anyMatch(argument -> !ArgumentType.TS_LATEST.equals(argument.getRefEntityKey().getType()))) { |
|||
throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' support only TS_LATEST arguments."); |
|||
} |
|||
} |
|||
|
|||
private void validateMetrics() { |
|||
if (metrics == null || metrics.isEmpty()) { |
|||
throw new IllegalArgumentException("Metrics map cannot be empty."); |
|||
} |
|||
|
|||
for (AggMetric metric : metrics.values()) { |
|||
if (metric.getInput() instanceof AggKeyInput aggKeyInput) { |
|||
if (!arguments.containsKey(aggKeyInput.getKey())) { |
|||
throw new IllegalArgumentException( |
|||
"Metric references unknown argument: '" + aggKeyInput.getKey() + "'." |
|||
); |
|||
} |
|||
} else { |
|||
throw new IllegalArgumentException("Metric key can only refer to argument."); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void validateInterval() { |
|||
if (interval == null) { |
|||
throw new IllegalArgumentException("Interval must be defined."); |
|||
} |
|||
interval.validate(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,67 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|||
import com.fasterxml.jackson.annotation.JsonSubTypes; |
|||
import com.fasterxml.jackson.annotation.JsonTypeInfo; |
|||
|
|||
import java.time.ZoneId; |
|||
import java.time.ZonedDateTime; |
|||
|
|||
@JsonTypeInfo( |
|||
use = JsonTypeInfo.Id.NAME, |
|||
include = JsonTypeInfo.As.PROPERTY, |
|||
property = "type" |
|||
) |
|||
@JsonSubTypes({ |
|||
@JsonSubTypes.Type(value = HourInterval.class, name = "HOUR"), |
|||
@JsonSubTypes.Type(value = DayInterval.class, name = "DAY"), |
|||
@JsonSubTypes.Type(value = WeekInterval.class, name = "WEEK"), |
|||
@JsonSubTypes.Type(value = WeekSunSatInterval.class, name = "WEEK_SUN_SAT"), |
|||
@JsonSubTypes.Type(value = MonthInterval.class, name = "MONTH"), |
|||
@JsonSubTypes.Type(value = QuarterInterval.class, name = "QUARTER"), |
|||
@JsonSubTypes.Type(value = YearInterval.class, name = "YEAR"), |
|||
@JsonSubTypes.Type(value = CustomInterval.class, name = "CUSTOM") |
|||
}) |
|||
@JsonIgnoreProperties(ignoreUnknown = true) |
|||
public interface AggInterval { |
|||
|
|||
@JsonIgnore |
|||
AggIntervalType getType(); |
|||
|
|||
@JsonIgnore |
|||
ZoneId getZoneId(); |
|||
|
|||
@JsonIgnore |
|||
long getCurrentIntervalDurationMillis(); |
|||
|
|||
@JsonIgnore |
|||
long getCurrentIntervalStartTs(); |
|||
|
|||
long getDateTimeIntervalStartTs(ZonedDateTime dateTime); |
|||
|
|||
@JsonIgnore |
|||
long getCurrentIntervalEndTs(); |
|||
|
|||
long getDateTimeIntervalEndTs(ZonedDateTime dateTime); |
|||
|
|||
ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart); |
|||
|
|||
void validate(); |
|||
|
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
public enum AggIntervalType { |
|||
|
|||
HOUR, |
|||
DAY, |
|||
WEEK, |
|||
WEEK_SUN_SAT, |
|||
MONTH, |
|||
QUARTER, |
|||
YEAR, |
|||
CUSTOM |
|||
|
|||
} |
|||
@ -0,0 +1,108 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonInclude; |
|||
import jakarta.validation.constraints.NotBlank; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.ZoneId; |
|||
import java.time.ZonedDateTime; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@Data |
|||
@JsonInclude(JsonInclude.Include.NON_NULL) |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public abstract class BaseAggInterval implements AggInterval { |
|||
|
|||
@NotBlank |
|||
protected String tz; |
|||
protected Long offsetSec; // delay seconds since start of interval
|
|||
|
|||
@Override |
|||
public ZoneId getZoneId() { |
|||
return ZoneId.of(tz); |
|||
} |
|||
|
|||
protected long getOffsetSafe() { |
|||
return offsetSec != null ? offsetSec : 0L; |
|||
} |
|||
|
|||
@Override |
|||
public long getCurrentIntervalDurationMillis() { |
|||
return getCurrentIntervalEndTs() - getCurrentIntervalStartTs(); |
|||
} |
|||
|
|||
@Override |
|||
public long getCurrentIntervalStartTs() { |
|||
ZoneId zoneId = getZoneId(); |
|||
ZonedDateTime now = ZonedDateTime.now(zoneId); |
|||
return getDateTimeIntervalStartTs(now); |
|||
} |
|||
|
|||
@Override |
|||
public long getDateTimeIntervalStartTs(ZonedDateTime dateTime) { |
|||
long offset = getOffsetSafe(); |
|||
ZonedDateTime shiftedNow = dateTime.minusSeconds(offset); |
|||
ZonedDateTime alignedStart = getAlignedBoundary(shiftedNow, false); |
|||
ZonedDateTime actualStart = alignedStart.plusSeconds(offset); |
|||
return actualStart.toInstant().toEpochMilli(); |
|||
} |
|||
|
|||
@Override |
|||
public long getCurrentIntervalEndTs() { |
|||
ZoneId zoneId = getZoneId(); |
|||
ZonedDateTime now = ZonedDateTime.now(zoneId); |
|||
return getDateTimeIntervalEndTs(now); |
|||
} |
|||
|
|||
@Override |
|||
public long getDateTimeIntervalEndTs(ZonedDateTime dateTime) { |
|||
long offset = getOffsetSafe(); |
|||
ZonedDateTime shiftedNow = dateTime.minusSeconds(offset); |
|||
ZonedDateTime alignedEnd = getAlignedBoundary(shiftedNow, true); |
|||
ZonedDateTime actualEnd = alignedEnd.plusSeconds(offset); |
|||
return actualEnd.toInstant().toEpochMilli(); |
|||
} |
|||
|
|||
protected abstract ZonedDateTime alignToIntervalStart(ZonedDateTime reference); |
|||
|
|||
protected ZonedDateTime getAlignedBoundary(ZonedDateTime reference, boolean next) { |
|||
ZonedDateTime base = alignToIntervalStart(reference); |
|||
return next ? getNextIntervalStart(base) : base; |
|||
} |
|||
|
|||
@Override |
|||
public void validate() { |
|||
try { |
|||
getZoneId(); |
|||
} catch (Exception ex) { |
|||
throw new IllegalArgumentException("Invalid timezone in interval: " + ex.getMessage()); |
|||
} |
|||
if (offsetSec != null) { |
|||
if (offsetSec < 0) { |
|||
throw new IllegalArgumentException("Offset cannot be negative."); |
|||
} |
|||
if (TimeUnit.SECONDS.toMillis(offsetSec) >= getCurrentIntervalDurationMillis()) { |
|||
throw new IllegalArgumentException("Offset must be greater than interval duration."); |
|||
} |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,68 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import jakarta.validation.constraints.Min; |
|||
import jakarta.validation.constraints.NotNull; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.Duration; |
|||
import java.time.ZonedDateTime; |
|||
|
|||
@EqualsAndHashCode(callSuper = true) |
|||
@Data |
|||
@NoArgsConstructor |
|||
public class CustomInterval extends BaseAggInterval { |
|||
|
|||
@NotNull |
|||
@Min(1) |
|||
private Long durationSec; |
|||
|
|||
public CustomInterval(String tz, Long offsetSec, Long durationSec) { |
|||
super(tz, offsetSec); |
|||
this.durationSec = durationSec; |
|||
} |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.CUSTOM; |
|||
} |
|||
|
|||
@Override |
|||
public long getCurrentIntervalDurationMillis() { |
|||
return getDurationMillis(); |
|||
} |
|||
|
|||
private long getDurationMillis() { |
|||
return Duration.ofSeconds(durationSec).toMillis(); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
ZonedDateTime localMidnight = reference.toLocalDate().atStartOfDay(reference.getZone()); |
|||
long secondsFromMidnight = Duration.between(localMidnight, reference).getSeconds(); |
|||
long alignedSecondsFromMidnight = (secondsFromMidnight / durationSec) * durationSec; |
|||
return localMidnight.plusSeconds(alignedSecondsFromMidnight); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusSeconds(durationSec); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.ZonedDateTime; |
|||
import java.time.temporal.ChronoUnit; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class DayInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.DAY; |
|||
} |
|||
|
|||
public DayInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return reference.truncatedTo(ChronoUnit.DAYS); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusDays(1); |
|||
} |
|||
|
|||
} |
|||
@ -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.common.data.cf.configuration.aggregation.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.ZonedDateTime; |
|||
import java.time.temporal.ChronoUnit; |
|||
|
|||
@EqualsAndHashCode(callSuper = true) |
|||
@Data |
|||
@NoArgsConstructor |
|||
public class HourInterval extends BaseAggInterval { |
|||
|
|||
public HourInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.HOUR; |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return reference.truncatedTo(ChronoUnit.HOURS); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusHours(1); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.ZonedDateTime; |
|||
import java.time.temporal.ChronoUnit; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class MonthInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.MONTH; |
|||
} |
|||
|
|||
public MonthInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return reference.withDayOfMonth(1).truncatedTo(ChronoUnit.DAYS); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusMonths(1); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.LocalDate; |
|||
import java.time.LocalTime; |
|||
import java.time.ZonedDateTime; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class QuarterInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.QUARTER; |
|||
} |
|||
|
|||
public QuarterInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
int month = reference.getMonthValue(); |
|||
int quarterStartMonth = ((month - 1) / 3) * 3 + 1; // 1, 4, 7, 10
|
|||
return ZonedDateTime.of( |
|||
LocalDate.of(reference.getYear(), quarterStartMonth, 1), |
|||
LocalTime.MIDNIGHT, |
|||
reference.getZone()); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusMonths(3); |
|||
} |
|||
|
|||
} |
|||
@ -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.single.interval; |
|||
|
|||
import jakarta.validation.constraints.Min; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public class Watermark { |
|||
|
|||
@Min(0) |
|||
private long duration; |
|||
|
|||
} |
|||
@ -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.common.data.cf.configuration.aggregation.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.DayOfWeek; |
|||
import java.time.ZonedDateTime; |
|||
import java.time.temporal.ChronoUnit; |
|||
import java.time.temporal.TemporalAdjusters; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class WeekInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.WEEK; |
|||
} |
|||
|
|||
public WeekInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return reference.with(TemporalAdjusters.previousOrSame(DayOfWeek.MONDAY)).truncatedTo(ChronoUnit.DAYS); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusWeeks(1); |
|||
} |
|||
|
|||
} |
|||
@ -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.common.data.cf.configuration.aggregation.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.DayOfWeek; |
|||
import java.time.ZonedDateTime; |
|||
import java.time.temporal.ChronoUnit; |
|||
import java.time.temporal.TemporalAdjusters; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class WeekSunSatInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.WEEK_SUN_SAT; |
|||
} |
|||
|
|||
public WeekSunSatInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return reference.with(TemporalAdjusters.previousOrSame(DayOfWeek.SUNDAY)).truncatedTo(ChronoUnit.DAYS); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusWeeks(1); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.time.LocalDate; |
|||
import java.time.LocalTime; |
|||
import java.time.ZonedDateTime; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
public class YearInterval extends BaseAggInterval { |
|||
|
|||
@Override |
|||
public AggIntervalType getType() { |
|||
return AggIntervalType.YEAR; |
|||
} |
|||
|
|||
public YearInterval(String tz, Long offsetSec) { |
|||
super(tz, offsetSec); |
|||
} |
|||
|
|||
@Override |
|||
protected ZonedDateTime alignToIntervalStart(ZonedDateTime reference) { |
|||
return ZonedDateTime.of( |
|||
LocalDate.of(reference.getYear(), 1, 1), |
|||
LocalTime.MIDNIGHT, |
|||
reference.getZone()); |
|||
} |
|||
|
|||
@Override |
|||
public ZonedDateTime getNextIntervalStart(ZonedDateTime currentStart) { |
|||
return currentStart.plusYears(1); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,128 @@ |
|||
/** |
|||
* 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.single; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.params.ParameterizedTest; |
|||
import org.junit.jupiter.params.provider.ValueSource; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
|||
import org.thingsboard.server.common.data.cf.configuration.Argument; |
|||
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.ReferencedEntityKey; |
|||
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.single.interval.HourInterval; |
|||
|
|||
import java.util.Map; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
|
|||
public class EntityAggregationCalculatedFieldConfigurationTest { |
|||
|
|||
@Test |
|||
void typeShouldBeEntityAggregation() { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
assertThat(cfg.getType()).isEqualTo(CalculatedFieldType.ENTITY_AGGREGATION); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@ValueSource(strings = {"ATTRIBUTE", "TS_ROLLING"}) |
|||
void validateShouldThrowWhenNotTsLatestArgumentUsed(String argumentType) { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
cfg.setArguments(Map.of("k", validArgument(ArgumentType.valueOf(argumentType)))); |
|||
assertThatThrownBy(cfg::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Calculated field with type: '" + cfg.getType() + "' support only TS_LATEST arguments."); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenMetricMapIsEmpty() { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
|
|||
cfg.setArguments(Map.of("k", validArgument(ArgumentType.TS_LATEST))); |
|||
cfg.setMetrics(Map.of()); |
|||
|
|||
assertThatThrownBy(cfg::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Metrics map cannot be empty."); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenMetricInputIsNotAggKeyInput() { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
|
|||
cfg.setArguments(Map.of("k", validArgument(ArgumentType.TS_LATEST))); |
|||
|
|||
AggMetric metric = new AggMetric(); |
|||
metric.setInput(new AggFunctionInput()); // cannot be function
|
|||
cfg.setMetrics(Map.of("m", metric)); |
|||
|
|||
cfg.setInterval(new HourInterval("Europe/Kiev", null)); |
|||
cfg.setOutput(new Output()); |
|||
|
|||
assertThatThrownBy(cfg::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Metric key can only refer to argument."); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenMetricReferencesUnknownArgument() { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
|
|||
cfg.setArguments(Map.of("k", validArgument(ArgumentType.TS_LATEST))); |
|||
|
|||
AggMetric metric = new AggMetric(); |
|||
metric.setInput(new AggKeyInput("unknown")); |
|||
cfg.setMetrics(Map.of("m", metric)); |
|||
|
|||
cfg.setInterval(new HourInterval("Europe/Kiev", null)); |
|||
cfg.setOutput(new Output()); |
|||
|
|||
assertThatThrownBy(cfg::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Metric references unknown argument: 'unknown'."); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenIntervalIsNull() { |
|||
var cfg = new EntityAggregationCalculatedFieldConfiguration(); |
|||
|
|||
cfg.setArguments(Map.of("k", validArgument(ArgumentType.TS_LATEST))); |
|||
cfg.setMetrics(Map.of("m", validMetric())); |
|||
cfg.setInterval(null); |
|||
cfg.setOutput(new Output()); |
|||
|
|||
assertThatThrownBy(cfg::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Interval must be defined."); |
|||
} |
|||
|
|||
private Argument validArgument(ArgumentType type) { |
|||
Argument a = new Argument(); |
|||
a.setRefEntityKey(new ReferencedEntityKey("key", type, null)); |
|||
return a; |
|||
} |
|||
|
|||
private AggMetric validMetric() { |
|||
AggMetric metric = new AggMetric(); |
|||
metric.setInput(new AggKeyInput("k")); |
|||
return metric; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,168 @@ |
|||
/** |
|||
* 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.single.interval; |
|||
|
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.params.ParameterizedTest; |
|||
import org.junit.jupiter.params.provider.Arguments; |
|||
import org.junit.jupiter.params.provider.MethodSource; |
|||
|
|||
import java.time.Duration; |
|||
import java.time.Instant; |
|||
import java.time.ZoneId; |
|||
import java.time.ZonedDateTime; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Function; |
|||
import java.util.function.LongFunction; |
|||
import java.util.stream.Stream; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.assertj.core.api.Assertions.assertThatThrownBy; |
|||
|
|||
public class AggIntervalTest { |
|||
|
|||
private static final String TZ = "Europe/Kiev"; |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenInvalidTimZone() { |
|||
AggInterval interval = new HourInterval("TimeZone", null); |
|||
|
|||
assertThatThrownBy(interval::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessageContaining("Invalid timezone in interval: "); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenOffsetIsNegative() { |
|||
AggInterval interval = new CustomInterval(TZ, -100L, TimeUnit.HOURS.toSeconds(2)); |
|||
|
|||
assertThatThrownBy(interval::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Offset cannot be negative."); |
|||
} |
|||
|
|||
@Test |
|||
void validateShouldThrowWhenOffsetGreaterThanIntervalDuration() { |
|||
AggInterval interval = new CustomInterval(TZ, TimeUnit.HOURS.toSeconds(2), TimeUnit.HOURS.toSeconds(2)); |
|||
|
|||
assertThatThrownBy(interval::validate) |
|||
.isInstanceOf(IllegalArgumentException.class) |
|||
.hasMessage("Offset must be greater than interval duration."); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@MethodSource("intervals") |
|||
void testGetStartAndEndWithoutOffset(LongFunction<AggInterval> intervalCreator) { |
|||
AggInterval interval = intervalCreator.apply(0L); |
|||
|
|||
ZonedDateTime dateTime = ZonedDateTime.of( |
|||
// 2025.11.11 00:00:00
|
|||
2025, 11, 11, 0, 0, 0, 0, ZoneId.of(TZ) |
|||
); |
|||
long startTs = interval.getDateTimeIntervalStartTs(dateTime); |
|||
long endTs = interval.getDateTimeIntervalEndTs(dateTime); |
|||
|
|||
assertThat(endTs).isGreaterThan(startTs); |
|||
assertThat(endTs - startTs).isEqualTo(interval.getCurrentIntervalDurationMillis()); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@MethodSource("intervals") |
|||
void testApplyOffset(LongFunction<AggInterval> intervalCreator) { |
|||
long offsetSec = TimeUnit.MINUTES.toSeconds(15); |
|||
AggInterval intervalWithOffset = intervalCreator.apply(offsetSec); |
|||
AggInterval intervalNoOffset = intervalCreator.apply(0L); |
|||
|
|||
ZonedDateTime dateTime = ZonedDateTime.of( |
|||
// 2025.11.11 11:20:00 - chosen so 15m offset shifts into a new interval
|
|||
2025, 11, 11, 11, 20, 0, 0, ZoneId.of(TZ) |
|||
); |
|||
|
|||
long startWithOffsetTs = intervalWithOffset.getDateTimeIntervalStartTs(dateTime); |
|||
long startNoOffsetTs = intervalNoOffset.getDateTimeIntervalStartTs(dateTime); |
|||
|
|||
ZonedDateTime startWithOffset = Instant.ofEpochMilli(startWithOffsetTs).atZone(intervalWithOffset.getZoneId()); |
|||
ZonedDateTime startNoOffset = Instant.ofEpochMilli(startNoOffsetTs).atZone(intervalNoOffset.getZoneId()); |
|||
|
|||
long actualOffset = Duration.between(startNoOffset, startWithOffset).toSeconds(); |
|||
assertThat(actualOffset).isEqualTo(offsetSec); |
|||
} |
|||
|
|||
private static Stream<Arguments> intervals() { |
|||
return Stream.of( |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new HourInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new DayInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new WeekInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new WeekSunSatInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new MonthInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new QuarterInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new YearInterval(TZ, offset)), |
|||
Arguments.of((LongFunction<AggInterval>) offset -> new CustomInterval(TZ, offset, TimeUnit.HOURS.toSeconds(4))) |
|||
); |
|||
} |
|||
|
|||
@ParameterizedTest |
|||
@MethodSource("nextIntervalFromExactDate") |
|||
void testNextIntervalFromExactDate(LongFunction<AggInterval> intervalCreator, Function<ZonedDateTime, ZonedDateTime> expectedDateTimeFunction) { |
|||
AggInterval interval = intervalCreator.apply(0L); |
|||
|
|||
ZonedDateTime currentStart = ZonedDateTime.of( |
|||
2025, 11, 11, 0, 0, 0, 0, ZoneId.of(TZ) |
|||
); |
|||
|
|||
ZonedDateTime nextStart = interval.getNextIntervalStart(currentStart); |
|||
|
|||
assertThat(nextStart).isEqualTo(expectedDateTimeFunction.apply(currentStart)); |
|||
} |
|||
|
|||
private static Stream<Arguments> nextIntervalFromExactDate() { |
|||
return Stream.of( |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new HourInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusHours(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new DayInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusDays(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new WeekInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusWeeks(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new WeekSunSatInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusWeeks(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new MonthInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusMonths(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new QuarterInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusMonths(3) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new YearInterval(TZ, offset), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusYears(1) |
|||
), |
|||
Arguments.of( |
|||
(LongFunction<AggInterval>) offset -> new CustomInterval(TZ, offset, TimeUnit.HOURS.toSeconds(4)), |
|||
(Function<ZonedDateTime, ZonedDateTime>) currentInterval -> currentInterval.plusHours(4) |
|||
) |
|||
); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,71 @@ |
|||
///
|
|||
/// 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.
|
|||
///
|
|||
|
|||
import { ChangeDetectorRef, Component, DestroyRef, forwardRef, Renderer2, ViewContainerRef, } from '@angular/core'; |
|||
import { FormBuilder, NG_VALIDATORS, NG_VALUE_ACCESSOR, } from '@angular/forms'; |
|||
import { TbPopoverService } from '@shared/components/popover.service'; |
|||
import { EntityService } from '@core/http/entity.service'; |
|||
import { Store } from '@ngrx/store'; |
|||
import { AppState } from '@core/core.state'; |
|||
import { |
|||
CalculatedFieldArgumentsTableComponent |
|||
} from '@home/components/calculated-fields/components/calculated-field-arguments/calculated-field-arguments-table.component'; |
|||
import { ArgumentEntityType } from '@shared/models/calculated-field.models'; |
|||
|
|||
@Component({ |
|||
selector: 'tb-entity-aggregation-arguments-table', |
|||
templateUrl: './calculated-field-arguments-table.component.html', |
|||
styleUrls: [`calculated-field-arguments-table.component.scss`], |
|||
providers: [ |
|||
{ |
|||
provide: NG_VALUE_ACCESSOR, |
|||
useExisting: forwardRef(() => EntityAggregationArgumentsTableComponent), |
|||
multi: true |
|||
}, |
|||
{ |
|||
provide: NG_VALIDATORS, |
|||
useExisting: forwardRef(() => EntityAggregationArgumentsTableComponent), |
|||
multi: true |
|||
} |
|||
], |
|||
}) |
|||
export class EntityAggregationArgumentsTableComponent extends CalculatedFieldArgumentsTableComponent { |
|||
|
|||
constructor( |
|||
protected fb: FormBuilder, |
|||
protected popoverService: TbPopoverService, |
|||
protected viewContainerRef: ViewContainerRef, |
|||
protected cd: ChangeDetectorRef, |
|||
protected renderer: Renderer2, |
|||
protected entityService: EntityService, |
|||
protected destroyRef: DestroyRef, |
|||
protected store: Store<AppState> |
|||
) { |
|||
super(fb, popoverService, viewContainerRef, cd, renderer, entityService, destroyRef, store); |
|||
|
|||
this.argumentNameColumn = 'calculated-fields.argument-name'; |
|||
this.displayColumns = ['name', 'type', 'key', 'actions']; |
|||
this.panelAdditionalCtx = { |
|||
hiddenEntityTypes: true, |
|||
argumentEntityTypes: [ArgumentEntityType.Current], |
|||
hint: 'calculated-fields.entity-aggregation.argument-setting-hint', |
|||
hiddenDefaultValue: true, |
|||
hiddenEntityKeyTypes: true, |
|||
}; |
|||
|
|||
this.isScript = false; |
|||
} |
|||
} |
|||
@ -0,0 +1,133 @@ |
|||
<!-- |
|||
|
|||
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. |
|||
|
|||
--> |
|||
<div [formGroup]="entityAggregationConfiguration" class="tb-form-panel no-border no-padding"> |
|||
<div class="tb-form-panel"> |
|||
<div class="tb-form-panel-title"> |
|||
{{ 'calculated-fields.arguments' | translate }} |
|||
</div> |
|||
<div class="tb-form-hint tb-primary-fill hint-container"> |
|||
{{ 'calculated-fields.entity-aggregation.argument-hint' | translate }} |
|||
</div> |
|||
<tb-entity-aggregation-arguments-table formControlName="arguments" |
|||
[entityId]="entityId" |
|||
[tenantId]="tenantId" |
|||
[entityName]="entityName"/> |
|||
</div> |
|||
<div class="tb-form-panel"> |
|||
<div class="tb-form-panel-title" tb-hint-tooltip-icon="{{ 'calculated-fields.hint.metrics' | translate }}"> |
|||
{{ 'calculated-fields.metrics.metrics' | translate }} |
|||
</div> |
|||
<tb-calculated-field-metrics-table formControlName="metrics" |
|||
simpleMode |
|||
[arguments]="arguments$ | async" |
|||
></tb-calculated-field-metrics-table> |
|||
</div> |
|||
<div class="tb-form-panel"> |
|||
<div class="tb-form-panel-title" tbTruncateWithTooltip tb-hint-tooltip-icon="{{ 'calculated-fields.entity-aggregation.aggregation-interval-hint' | translate }}"> |
|||
{{ 'calculated-fields.entity-aggregation.aggregation-interval' | translate }} |
|||
</div> |
|||
<ng-container formGroupName="interval"> |
|||
<div class="flex items-start gap-3 xs:flex-col xs:items-stretch"> |
|||
<mat-form-field class="flex-1" appearance="outline" subscriptSizing="dynamic"> |
|||
<mat-label translate>calculated-fields.aggregate-interval-type</mat-label> |
|||
<mat-select formControlName="type" required> |
|||
<mat-option *ngFor="let type of AggIntervalTypes" [value]="type"> |
|||
{{ AggIntervalTypeTranslations.get(type) | translate }} |
|||
</mat-option> |
|||
</mat-select> |
|||
</mat-form-field> |
|||
<tb-timezone-select |
|||
class="flex-1" |
|||
appearance="outline" |
|||
subscriptSizing="dynamic" |
|||
required |
|||
userTimezoneByDefault |
|||
formControlName="tz"> |
|||
</tb-timezone-select> |
|||
</div> |
|||
@if (entityAggregationConfiguration.get('interval.type').value === AggIntervalType.CUSTOM) { |
|||
<tb-time-unit-input required |
|||
[minTime]="minAllowedAggregationIntervalInSecForCF" |
|||
[stepMultipleOf]="DayInSec" |
|||
sameWidthInputs |
|||
appearance="outline" |
|||
subscriptSizing="dynamic" |
|||
containerClass="flex gap-3" |
|||
labelText="{{ 'calculated-fields.aggregate-interval-value' | translate }}" |
|||
minErrorText="{{ 'calculated-fields.aggregate-interval-value-min' | translate : {sec: minAllowedAggregationIntervalInSecForCF} }}" |
|||
requiredText="{{ 'calculated-fields.aggregate-interval-value-required' | translate }}" |
|||
stepMultipleOfErrorText="{{ 'calculated-fields.aggregate-interval-value-step-multiple-of' | translate }}" |
|||
formControlName="durationSec"> |
|||
</tb-time-unit-input> |
|||
} |
|||
<div class="tb-form-panel stroked"> |
|||
<mat-slide-toggle class="mat-slide flex" formControlName="allowOffsetSec"> |
|||
<div tb-hint-tooltip-icon="{{ 'calculated-fields.entity-aggregation.apply-offset-hint' | translate }}"> |
|||
{{ 'calculated-fields.entity-aggregation.apply-offset' | translate }} |
|||
</div> |
|||
</mat-slide-toggle> |
|||
@if (entityAggregationConfiguration.get('interval.allowOffsetSec').value) { |
|||
<tb-time-unit-input required |
|||
[minTime]="0" |
|||
[maxTime]="maxOffsetTime" |
|||
sameWidthInputs |
|||
appearance="outline" |
|||
subscriptSizing="dynamic" |
|||
containerClass="flex gap-3" |
|||
labelText="{{ 'calculated-fields.entity-aggregation.offset-value' | translate }}" |
|||
minErrorText="{{ 'calculated-fields.entity-aggregation.offset-value-min' | translate }}" |
|||
maxErrorText="{{ 'calculated-fields.entity-aggregation.offset-value-max' | translate }}" |
|||
requiredText="{{ 'calculated-fields.entity-aggregation.offset-value-required' | translate }}" |
|||
formControlName="offsetSec"> |
|||
</tb-time-unit-input> |
|||
<div class="tb-form-hint tb-primary-fill hint-container"> |
|||
{{ hint }} |
|||
</div> |
|||
} |
|||
</div> |
|||
</ng-container> |
|||
<div class="tb-form-panel stroked"> |
|||
<mat-slide-toggle class="mat-slide flex" formControlName="allowWatermark"> |
|||
<div tb-hint-tooltip-icon="{{ 'calculated-fields.entity-aggregation.wait-delay-hint' | translate }}"> |
|||
{{ 'calculated-fields.entity-aggregation.wait-delay' | translate }} |
|||
</div> |
|||
</mat-slide-toggle> |
|||
@if (entityAggregationConfiguration.get('allowWatermark').value) { |
|||
<ng-container formGroupName="watermark"> |
|||
<tb-time-unit-input required |
|||
[minTime]="60" |
|||
sameWidthInputs |
|||
appearance="outline" |
|||
containerClass="flex gap-3" |
|||
labelText="{{ 'calculated-fields.entity-aggregation.duration' | translate }}" |
|||
minErrorText="{{ 'calculated-fields.entity-aggregation.duration-min' | translate }}" |
|||
hintText="{{ 'calculated-fields.entity-aggregation.duration-hint' | translate }}" |
|||
requiredText="{{ 'calculated-fields.entity-aggregation.duration-required' | translate }}" |
|||
formControlName="duration"> |
|||
</tb-time-unit-input> |
|||
</ng-container> |
|||
} |
|||
</div> |
|||
</div> |
|||
<tb-calculate-field-output formControlName="output" |
|||
containerInputClass="grid items-start grid-cols-2 gap-3 xs:grid-cols-1" |
|||
simpleMode |
|||
hiddenName |
|||
disableType |
|||
[entityId]="entityId"></tb-calculate-field-output> |
|||
</div> |
|||
@ -0,0 +1,411 @@ |
|||
///
|
|||
/// 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.
|
|||
///
|
|||
|
|||
import { Component, forwardRef, Input } from '@angular/core'; |
|||
import { |
|||
ControlValueAccessor, |
|||
FormBuilder, |
|||
NG_VALIDATORS, |
|||
NG_VALUE_ACCESSOR, |
|||
ValidationErrors, |
|||
Validator, |
|||
Validators |
|||
} from '@angular/forms'; |
|||
import { EntityId } from '@shared/models/id/entity-id'; |
|||
import { |
|||
AggInterval, |
|||
AggIntervalType, |
|||
AggIntervalTypeTranslations, |
|||
CalculatedFieldEntityAggregationConfiguration, |
|||
CalculatedFieldOutput, |
|||
CalculatedFieldType, |
|||
notEmptyObjectValidator, |
|||
OutputType |
|||
} from '@shared/models/calculated-field.models'; |
|||
import { filter, map } from 'rxjs/operators'; |
|||
import { takeUntilDestroyed } from '@angular/core/rxjs-interop'; |
|||
import { AVG_MONTH, AVG_QUARTER, DAY, HOUR, MINUTE, SECOND, YEAR } from '@shared/models/time/time.models'; |
|||
import { deepClone, isDefinedAndNotNull } from '@core/utils'; |
|||
import { getCurrentAuthState } from '@core/auth/auth.selectors'; |
|||
import { Store } from '@ngrx/store'; |
|||
import { AppState } from '@core/core.state'; |
|||
import { merge } from 'rxjs'; |
|||
import { TranslateService } from '@ngx-translate/core'; |
|||
import _moment from 'moment'; |
|||
|
|||
interface CalculatedFieldEntityAggregationConfigurationValue extends CalculatedFieldEntityAggregationConfiguration { |
|||
interval: AggInterval & {allowOffsetSec?: boolean}; |
|||
allowWatermark: boolean; |
|||
} |
|||
|
|||
enum TimeCategory { |
|||
SECONDS = 'SECONDS', |
|||
MINUTES = 'MINUTES', |
|||
HOURS = 'HOURS', |
|||
DAYS = 'DAYS' |
|||
} |
|||
|
|||
@Component({ |
|||
selector: 'tb-entity-aggregation-component', |
|||
templateUrl: './entity-aggregation-component.component.html', |
|||
providers: [ |
|||
{ |
|||
provide: NG_VALUE_ACCESSOR, |
|||
useExisting: forwardRef(() => EntityAggregationComponentComponent), |
|||
multi: true |
|||
}, |
|||
{ |
|||
provide: NG_VALIDATORS, |
|||
useExisting: forwardRef(() => EntityAggregationComponentComponent), |
|||
multi: true |
|||
} |
|||
], |
|||
}) |
|||
export class EntityAggregationComponentComponent implements ControlValueAccessor, Validator { |
|||
|
|||
@Input({required: true}) |
|||
entityId: EntityId; |
|||
|
|||
@Input({required: true}) |
|||
tenantId: string; |
|||
|
|||
@Input({required: true}) |
|||
entityName: string; |
|||
|
|||
readonly minAllowedAggregationIntervalInSecForCF = getCurrentAuthState(this.store).minAllowedAggregationIntervalInSecForCF; |
|||
readonly DayInSec = DAY / SECOND; |
|||
|
|||
entityAggregationConfiguration = this.fb.group({ |
|||
arguments: this.fb.control({}, notEmptyObjectValidator()), |
|||
metrics: this.fb.control({}, notEmptyObjectValidator()), |
|||
interval: this.fb.group({ |
|||
type: [AggIntervalType.HOUR], |
|||
tz: ['', Validators.required], |
|||
durationSec: [this.minAllowedAggregationIntervalInSecForCF, Validators.required], |
|||
allowOffsetSec: [false], |
|||
offsetSec: [this.minAllowedAggregationIntervalInSecForCF > 60 ? MINUTE / SECOND : 1, Validators.required], |
|||
}), |
|||
allowWatermark: [false], |
|||
watermark: this.fb.group({ |
|||
duration: [HOUR/SECOND, Validators.required], |
|||
}), |
|||
output: this.fb.control<CalculatedFieldOutput>({ |
|||
type: OutputType.Timeseries, |
|||
}), |
|||
}); |
|||
|
|||
arguments$ = this.entityAggregationConfiguration.get('arguments').valueChanges.pipe( |
|||
map(argumentsObj => Object.keys(argumentsObj)) |
|||
); |
|||
|
|||
AggIntervalType = AggIntervalType; |
|||
AggIntervalTypes = Object.values(AggIntervalType) as AggIntervalType[]; |
|||
AggIntervalTypeTranslations = AggIntervalTypeTranslations; |
|||
|
|||
hint: string; |
|||
|
|||
private propagateChange: (config: CalculatedFieldEntityAggregationConfiguration) => void = () => { }; |
|||
|
|||
constructor(private fb: FormBuilder, |
|||
private store: Store<AppState>, |
|||
private translate: TranslateService,) { |
|||
|
|||
this.entityAggregationConfiguration.get('interval.type').valueChanges.pipe( |
|||
takeUntilDestroyed() |
|||
).subscribe((type: AggIntervalType) => { |
|||
this.checkAggIntervalType(type); |
|||
}); |
|||
|
|||
this.entityAggregationConfiguration.get('interval.allowOffsetSec').valueChanges.pipe( |
|||
takeUntilDestroyed() |
|||
).subscribe((allow: boolean) => { |
|||
this.checkIntervalDuration(allow); |
|||
}); |
|||
|
|||
this.entityAggregationConfiguration.get('allowWatermark').valueChanges.pipe( |
|||
takeUntilDestroyed() |
|||
).subscribe((allow: boolean) => { |
|||
this.checkWatermark(allow); |
|||
}); |
|||
|
|||
merge( |
|||
this.entityAggregationConfiguration.get('interval.type').valueChanges, |
|||
this.entityAggregationConfiguration.get('interval.durationSec').valueChanges, |
|||
this.entityAggregationConfiguration.get('interval.offsetSec').valueChanges, |
|||
this.entityAggregationConfiguration.get('interval.allowOffsetSec').valueChanges, |
|||
).pipe( |
|||
filter(() => this.entityAggregationConfiguration.get('interval.allowOffsetSec').value), |
|||
takeUntilDestroyed() |
|||
).subscribe(() => { |
|||
this.updatedOffsetHint(); |
|||
}); |
|||
|
|||
this.entityAggregationConfiguration.valueChanges.pipe( |
|||
takeUntilDestroyed() |
|||
).subscribe((value: CalculatedFieldEntityAggregationConfigurationValue) => { |
|||
this.updatedModel(deepClone(value)); |
|||
}); |
|||
} |
|||
|
|||
validate(): ValidationErrors | null { |
|||
return this.entityAggregationConfiguration.valid || this.entityAggregationConfiguration.disabled ? null : {invalidPropagateConfig: false}; |
|||
} |
|||
|
|||
writeValue(value: CalculatedFieldEntityAggregationConfiguration): void { |
|||
const data: CalculatedFieldEntityAggregationConfigurationValue = { |
|||
...value, |
|||
allowWatermark: isDefinedAndNotNull(value.watermark), |
|||
interval: {...value.interval, allowOffsetSec: isDefinedAndNotNull(value?.interval?.offsetSec)} |
|||
} |
|||
this.entityAggregationConfiguration.patchValue(data, {emitEvent: false}); |
|||
this.checkAggIntervalType(this.entityAggregationConfiguration.get('interval.type').value); |
|||
this.checkIntervalDuration(this.entityAggregationConfiguration.get('interval.allowOffsetSec').value); |
|||
this.checkWatermark(this.entityAggregationConfiguration.get('allowWatermark').value); |
|||
this.updatedOffsetHint(); |
|||
setTimeout(() => { |
|||
this.entityAggregationConfiguration.get('arguments').updateValueAndValidity({onlySelf: true}); |
|||
}); |
|||
} |
|||
|
|||
registerOnChange(fn: (config: CalculatedFieldEntityAggregationConfiguration) => void): void { |
|||
this.propagateChange = fn; |
|||
} |
|||
|
|||
registerOnTouched(_: any): void { } |
|||
|
|||
setDisabledState(isDisabled: boolean): void { |
|||
if (isDisabled) { |
|||
this.entityAggregationConfiguration.disable({emitEvent: false}); |
|||
} else { |
|||
this.entityAggregationConfiguration.enable({emitEvent: false}); |
|||
this.checkAggIntervalType(this.entityAggregationConfiguration.get('interval.type').value); |
|||
this.checkIntervalDuration(this.entityAggregationConfiguration.get('interval.allowOffsetSec').value); |
|||
this.checkWatermark(this.entityAggregationConfiguration.get('allowWatermark').value); |
|||
} |
|||
} |
|||
|
|||
get maxOffsetTime(): number { |
|||
switch (this.entityAggregationConfiguration.get('interval.type').value as AggIntervalType) { |
|||
case AggIntervalType.HOUR: |
|||
return HOUR / SECOND - 1; |
|||
case AggIntervalType.DAY: |
|||
return DAY / SECOND - 1; |
|||
case AggIntervalType.WEEK: |
|||
case AggIntervalType.WEEK_SUN_SAT: |
|||
return 7 * DAY / SECOND - 1; |
|||
case AggIntervalType.MONTH: |
|||
return AVG_MONTH / SECOND; |
|||
case AggIntervalType.QUARTER: |
|||
return AVG_QUARTER / SECOND - 1; |
|||
case AggIntervalType.YEAR: |
|||
return YEAR / SECOND - 1; |
|||
case AggIntervalType.CUSTOM: |
|||
return this.entityAggregationConfiguration.get('interval.durationSec').value - 1; |
|||
} |
|||
} |
|||
|
|||
private updatedModel(value: CalculatedFieldEntityAggregationConfigurationValue): void { |
|||
value.type = CalculatedFieldType.ENTITY_AGGREGATION; |
|||
if (!value.interval.allowOffsetSec) { |
|||
delete value.interval.offsetSec; |
|||
} |
|||
delete value.interval.allowOffsetSec; |
|||
if (!value.allowWatermark) { |
|||
delete value.watermark; |
|||
} |
|||
delete value.allowWatermark; |
|||
this.propagateChange(value); |
|||
} |
|||
|
|||
private checkAggIntervalType(type: AggIntervalType) { |
|||
if (type === AggIntervalType.CUSTOM) { |
|||
this.entityAggregationConfiguration.get('interval.durationSec').enable({emitEvent: false}); |
|||
} else { |
|||
this.entityAggregationConfiguration.get('interval.durationSec').disable({emitEvent: false}); |
|||
} |
|||
} |
|||
|
|||
private checkIntervalDuration(allow: boolean) { |
|||
if (allow) { |
|||
this.entityAggregationConfiguration.get('interval.offsetSec').enable({emitEvent: false}); |
|||
} else { |
|||
this.entityAggregationConfiguration.get('interval.offsetSec').disable({emitEvent: false}); |
|||
this.hint = ''; |
|||
} |
|||
} |
|||
|
|||
private checkWatermark(allow: boolean) { |
|||
if (allow) { |
|||
this.entityAggregationConfiguration.get('watermark').enable({emitEvent: false}); |
|||
} else { |
|||
this.entityAggregationConfiguration.get('watermark').disable({emitEvent: false}); |
|||
} |
|||
} |
|||
|
|||
private updatedOffsetHint(): void { |
|||
const offset = this.entityAggregationConfiguration.get('interval.offsetSec').value; |
|||
const intervalType = this.entityAggregationConfiguration.get('interval.type').value as AggIntervalType; |
|||
const durationSec = this.entityAggregationConfiguration.get('interval.durationSec').value; |
|||
const offsetCategory = this.getTimeCategory(offset); |
|||
const now = _moment.utc(); |
|||
let interval: string = ''; |
|||
if (intervalType === AggIntervalType.CUSTOM) { |
|||
const durationSecCategory = this.getTimeCategory(durationSec); |
|||
const formatString = this.getCustomFormatString(offsetCategory, durationSecCategory); |
|||
const intervals: string[] = []; |
|||
let allInterval = durationSec >= HOUR*6/SECOND && durationSec < DAY/SECOND; |
|||
now.startOf('year').add(offset, 'seconds'); |
|||
|
|||
let repeat = 2; |
|||
if (allInterval) { |
|||
repeat = Math.floor(DAY/SECOND/durationSec); |
|||
if (repeat > 4) { |
|||
repeat = 2; |
|||
allInterval = false; |
|||
} |
|||
} |
|||
|
|||
for (let i = 0; i < repeat; i++) { |
|||
const s1 = now.clone().add(i * durationSec, 'seconds').format(formatString); |
|||
const s2 = now.clone().add((i + 1) * durationSec, 'seconds').format(formatString); |
|||
intervals.push(`${s1} - ${s2}`); |
|||
} |
|||
interval = intervals.join('; '); |
|||
|
|||
if (allInterval) { |
|||
this.hint = this.translate.instant('calculated-fields.aggregate-period-hint-offset', {interval}); |
|||
} else { |
|||
interval += '…' |
|||
this.hint = this.translate.instant('calculated-fields.aggregate-period-hint-offset-and-so-on', {interval}); |
|||
} |
|||
} else { |
|||
interval = this.buildStandardIntervalString(now, intervalType, offset, offsetCategory); |
|||
this.hint = this.translate.instant('calculated-fields.aggregate-period-hint-offset-and-so-on', { interval }); |
|||
} |
|||
} |
|||
|
|||
private getTimeCategory(seconds: number): TimeCategory { |
|||
if (seconds % (DAY / SECOND) === 0) { |
|||
return TimeCategory.DAYS; |
|||
} |
|||
if (seconds % (HOUR / SECOND) === 0) { |
|||
return TimeCategory.HOURS; |
|||
} |
|||
if (seconds % (MINUTE / SECOND) === 0) { |
|||
return TimeCategory.MINUTES; |
|||
} |
|||
return TimeCategory.SECONDS; |
|||
} |
|||
|
|||
private getCustomFormatString(offsetCat: TimeCategory, durationCat: TimeCategory): string { |
|||
if (durationCat === TimeCategory.DAYS) { |
|||
if (offsetCat === TimeCategory.SECONDS) { |
|||
return '[Day] D, HH:mm:ss'; |
|||
} |
|||
if (offsetCat === TimeCategory.MINUTES || offsetCat === TimeCategory.HOURS) { |
|||
return '[Day] D, HH:mm'; |
|||
} |
|||
return '[Day] D'; |
|||
} else { |
|||
if (offsetCat === TimeCategory.SECONDS) { |
|||
return 'HH:mm:ss'; |
|||
} |
|||
return 'HH:mm'; |
|||
} |
|||
} |
|||
|
|||
private formatAdditiveInterval(now: _moment.Moment, addUnit: 'hour' | 'day' | 'month' | 'quarter', offsetCat: TimeCategory, |
|||
formats: { [key in TimeCategory]?: { s1: string, s2: string, s3: string } }): string { |
|||
const formatTs = formats[offsetCat] || formats[TimeCategory.SECONDS]; |
|||
|
|||
if (!formatTs) { |
|||
return ''; |
|||
} |
|||
|
|||
const s1 = now.format(formatTs.s1); |
|||
const s2 = now.clone().add(1, addUnit).format(formatTs.s2); |
|||
const s3 = now.clone().add(2, addUnit).format(formatTs.s3); |
|||
|
|||
return `${s1} - ${s2}; ${s2} - ${s3}…`; |
|||
} |
|||
|
|||
private formatNextInterval(now: _moment.Moment, offsetCat: TimeCategory, secFmt: string, minHourFmt: string, dayFmt: string): string { |
|||
let s1: string; |
|||
if (offsetCat === TimeCategory.SECONDS) { |
|||
s1 = now.format(secFmt); |
|||
} else if (offsetCat === TimeCategory.MINUTES || offsetCat === TimeCategory.HOURS) { |
|||
s1 = now.format(minHourFmt); |
|||
} else { |
|||
s1 = now.format(dayFmt); |
|||
} |
|||
|
|||
const s2 = `Next ${s1}`; |
|||
const s3 = `Following ${s1}`; |
|||
return `${s1} - ${s2}; ${s2} - ${s3}… `; |
|||
} |
|||
|
|||
private buildStandardIntervalString(now: _moment.Moment, type: AggIntervalType, offset: number, offsetCat: TimeCategory): string { |
|||
switch (type) { |
|||
case AggIntervalType.HOUR: |
|||
now.startOf('day').add(offset, 'seconds'); |
|||
return this.formatAdditiveInterval(now, 'hour', offsetCat, { |
|||
[TimeCategory.SECONDS]: { s1: 'HH:mm:ss', s2: 'HH:mm:ss', s3: 'HH:mm:ss' }, |
|||
[TimeCategory.MINUTES]: { s1: 'HH:mm:ss', s2: 'HH:mm', s3: 'HH:mm' } |
|||
}); |
|||
|
|||
case AggIntervalType.DAY: |
|||
now.startOf('month').add(offset, 'seconds'); |
|||
return this.formatAdditiveInterval(now, 'day', offsetCat, { |
|||
[TimeCategory.SECONDS]: { s1: '[Day] D, HH:mm:ss', s2: '[Day] D, HH:mm:ss', s3: '[Day] D, HH:mm:ss' }, |
|||
[TimeCategory.MINUTES]: { s1: '[Day] D, HH:mm:ss', s2: '[Day] D, HH:mm', s3: '[Day] D, HH:mm' }, |
|||
[TimeCategory.HOURS]: { s1: 'HH:mm:ss', s2: '[Day] D, HH:mm', s3: '[Day] D, HH:mm' } // Note: Original logic, s1 format is different
|
|||
}); |
|||
|
|||
case AggIntervalType.WEEK: |
|||
now.isoWeekday(1).startOf('isoWeek').add(offset, 'seconds'); |
|||
return this.formatNextInterval(now, offsetCat, 'ddd, HH:mm:ss', 'ddd, HH:mm', 'ddd'); |
|||
|
|||
case AggIntervalType.WEEK_SUN_SAT: |
|||
now.startOf('week').add(offset, 'seconds'); |
|||
return this.formatNextInterval(now, offsetCat, 'ddd, HH:mm:ss', 'ddd, HH:mm', 'ddd'); |
|||
|
|||
case AggIntervalType.MONTH: |
|||
now.startOf('year').add(offset, 'seconds'); |
|||
return this.formatAdditiveInterval(now, 'month', offsetCat, { |
|||
[TimeCategory.SECONDS]: { s1: 'Do [of month], HH:mm:ss', s2: '[Next] Do, HH:mm:ss', s3: '[Following] Do, HH:mm:ss' }, |
|||
[TimeCategory.MINUTES]: { s1: 'Do [of month], HH:mm', s2: '[Next] Do, HH:mm', s3: '[Following] Do, HH:mm' }, |
|||
[TimeCategory.HOURS]: { s1: 'Do [of month], HH:mm', s2: '[Next] Do, HH:mm', s3: '[Following] Do, HH:mm' }, |
|||
[TimeCategory.DAYS]: { s1: 'Do [of month]', s2: '[Next] Do', s3: '[Following] Do' } |
|||
}); |
|||
|
|||
case AggIntervalType.QUARTER: |
|||
now.startOf('year').add(offset, 'seconds'); |
|||
return this.formatAdditiveInterval(now, 'quarter', offsetCat, { |
|||
[TimeCategory.SECONDS]: { s1: 'MMM Do, HH:mm:ss', s2: 'MMM Do, HH:mm:ss', s3: 'MMM Do, HH:mm:ss' }, |
|||
[TimeCategory.MINUTES]: { s1: 'MMM Do, HH:mm', s2: 'MMM Do, HH:mm', s3: 'MMM Do, HH:mm' }, |
|||
[TimeCategory.HOURS]: { s1: 'MMM Do, HH:mm', s2: 'MMM Do, HH:mm', s3: 'MMM Do, HH:mm' }, |
|||
[TimeCategory.DAYS]: { s1: 'MMM Do', s2: 'MMM Do', s3: 'MMM Do' } |
|||
}); |
|||
|
|||
case AggIntervalType.YEAR: |
|||
now.startOf('year').add(offset, 'seconds'); |
|||
return this.formatNextInterval(now, offsetCat, 'MMM Do, HH:mm:ss', 'MMM Do, HH:mm', 'MMM Do'); |
|||
|
|||
default: |
|||
return ''; |
|||
} |
|||
} |
|||
} |
|||
@ -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.
|
|||
///
|
|||
|
|||
import { NgModule } from '@angular/core'; |
|||
import { CommonModule } from '@angular/common'; |
|||
import { SharedModule } from '@shared/shared.module'; |
|||
import { |
|||
CalculatedFieldOutputModule |
|||
} from '@home/components/calculated-fields/components/output/calculated-field-output.module'; |
|||
import { |
|||
CalculatedFieldArgumentsTableModule |
|||
} from '@home/components/calculated-fields/components/calculated-field-arguments/calculated-field-arguments-table.module'; |
|||
import { |
|||
EntityAggregationComponentComponent |
|||
} from '@home/components/calculated-fields/components/entity-aggregation-configuration/entity-aggregation-component.component'; |
|||
import { |
|||
CalculatedFieldMetricsTableModule |
|||
} from '@home/components/calculated-fields/components/metrics/calculated-field-metrics-table.module'; |
|||
|
|||
@NgModule({ |
|||
imports: [ |
|||
CommonModule, |
|||
SharedModule, |
|||
CalculatedFieldOutputModule, |
|||
CalculatedFieldArgumentsTableModule, |
|||
CalculatedFieldMetricsTableModule, |
|||
], |
|||
declarations: [ |
|||
EntityAggregationComponentComponent, |
|||
], |
|||
exports: [ |
|||
EntityAggregationComponentComponent, |
|||
] |
|||
}) |
|||
export class EntityAggregationComponentModule { |
|||
} |
|||
@ -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.
|
|||
///
|
|||
|
|||
import { NgModule } from '@angular/core'; |
|||
import { CommonModule } from '@angular/common'; |
|||
import { SharedModule } from '@shared/shared.module'; |
|||
import { |
|||
CalculatedFieldMetricsTableComponent |
|||
} from '@home/components/calculated-fields/components/metrics/calculated-field-metrics-table.component'; |
|||
import { |
|||
CalculatedFieldMetricsPanelComponent |
|||
} from '@home/components/calculated-fields/components/metrics/calculated-field-metrics-panel.component'; |
|||
|
|||
@NgModule({ |
|||
imports: [ |
|||
CommonModule, |
|||
SharedModule, |
|||
], |
|||
declarations: [ |
|||
CalculatedFieldMetricsTableComponent, |
|||
CalculatedFieldMetricsPanelComponent |
|||
], |
|||
exports: [ |
|||
CalculatedFieldMetricsTableComponent |
|||
] |
|||
}) |
|||
export class CalculatedFieldMetricsTableModule { |
|||
|
|||
} |
|||
Loading…
Reference in new issue