diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index b75946fd00..6d3116970b 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -30,7 +30,9 @@ import org.thingsboard.server.common.data.alarm.Alarm; 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.CalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -53,6 +55,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; @@ -440,9 +443,14 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM GeofencingCalculatedFieldState geofencingState = (GeofencingCalculatedFieldState) state; geofencingState.updateLastDynamicArgumentsRefreshTs(); } - - Map arguments = fetchArguments(ctx); - state.update(arguments, ctx); + if (ctx.getCfType() == CalculatedFieldType.ENTITY_AGGREGATION) { + var configuration = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); + long delayUntilIntervalEnd = configuration.getInterval().getDelayUntilIntervalEnd(); + ctx.scheduleReevaluation(delayUntilIntervalEnd, actorCtx); + } else { + Map arguments = fetchArguments(ctx); + state.update(arguments, ctx); + } state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize()); states.put(ctx.getCfId(), state); diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index 4f0e323b99..65cedf443f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -34,6 +34,7 @@ import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -93,6 +94,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private final Map> entityIdCalculatedFields = new HashMap<>(); private final Map> entityIdCalculatedFieldLinks = new HashMap<>(); private final Map> ownerEntities = new HashMap<>(); + private final Map entityAggCalculatedFields = new HashMap<>(); private ScheduledFuture cfsReevaluationTask; private final CalculatedFieldProcessingService cfExecService; @@ -370,6 +372,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } calculatedFields.put(cf.getId(), cfCtx); + if (cf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { + entityAggCalculatedFields.put(cf.getId(), cfCtx); + } // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); @@ -401,6 +406,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(newCfCtx).eventEntity(newCfCtx.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } finally { calculatedFields.put(newCf.getId(), newCfCtx); + if (newCf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { + entityAggCalculatedFields.put(newCf.getId(), newCfCtx); + } List oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); List newCfList = new CopyOnWriteArrayList<>(); boolean found = false; @@ -767,6 +775,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } finally { calculatedFields.put(cf.getId(), cfCtx); + if (cf.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfig) { + entityAggCalculatedFields.put(cf.getId(), cfCtx); + } // We use copy on write lists to safely pass the reference to another actor for the iteration. // Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead) entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 945792ebcd..b1abfdf36c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -27,7 +27,11 @@ import org.thingsboard.common.util.ThingsBoardExecutors; 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.RelationPathQueryDynamicSourceConfiguration; +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.RelatedEntitiesAggregationCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.Aggregation; @@ -48,14 +52,18 @@ import org.thingsboard.server.dao.usagerecord.ApiLimitService; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; @@ -98,6 +106,7 @@ public abstract class AbstractCalculatedFieldProcessingService { case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts); case SIMPLE, SCRIPT, ALARM, PROPAGATION -> getBaseCalculatedFieldArguments(ctx, entityId, ts); case RELATED_ENTITIES_AGGREGATION -> fetchRelatedEntitiesAggArguments(ctx, entityId, ts); + case ENTITY_AGGREGATION -> null; }; if (ctx.getCfType() == PROPAGATION) { argFutures.put(PROPAGATION_CONFIG_ARGUMENT, fetchPropagationCalculatedFieldArgument(ctx, entityId)); @@ -117,6 +126,15 @@ public abstract class AbstractCalculatedFieldProcessingService { return futures; } + private Map> getEntityArgumentsDuringInterval(CalculatedFieldCtx ctx, EntityId entityId, long ts) { + Map> futures = new HashMap<>(); + for (var entry : ctx.getArguments().entrySet()) { + var argValueFuture = fetchArgumentValue(ctx.getTenantId(), entityId, entry.getValue(), ts); + futures.put(entry.getKey(), argValueFuture); + } + return futures; + } + protected EntityId resolveEntityId(TenantId tenantId, EntityId entityId, Argument argument) { if (argument.getRefEntityId() != null) { return argument.getRefEntityId(); @@ -276,6 +294,57 @@ public abstract class AbstractCalculatedFieldProcessingService { }; } + protected Map fetchArgumentValuesDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception { + var config = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); + Map argumentValues = new HashMap<>(); + + for (Entry entry : config.getMetrics().entrySet()) { + String metricName = entry.getKey(); + AggMetric metric = entry.getValue(); + AggFunction function = metric.getFunction(); + BaseReadTsKvQuery query = new BaseReadTsKvQuery(((AggKeyInput) metric.getInput()).getKey(), interval.getStartTs(), interval.getEndTs(), 0, 1, Aggregation.valueOf(function.name())); + log.trace("[{}][{}] Fetching timeseries for query {}", ctx.getTenantId(), entityId, query); + ListenableFuture> tsFuture = timeseriesService.findAll(ctx.getTenantId(), entityId, List.of(query)); + ListenableFuture argumentEntryFut = Futures.transform(tsFuture, timeSeries -> { + log.debug("[{}][{}] Fetched {} timeseries for query {}", ctx.getTenantId(), entityId, timeSeries == null ? 0 : timeSeries.size(), query); + if (timeSeries == null || timeSeries.isEmpty()) { + return new SingleValueArgumentEntry(); + } + return ArgumentEntry.createSingleValueArgument(timeSeries.get(0)); + }, calculatedFieldCallbackExecutor); + + // Ugly but necessary. We do not expect to often fetch data from DB. Only once per pair lifetime. + // This call happens while processing the CF pack from the queue consumer. So the timeout should be relatively low. + // Alternatively, we can fetch the state outside the actor system and push separate command to create this actor, + // but this will significantly complicate the code. + ArgumentEntry argumentEntry = argumentEntryFut.get(1, TimeUnit.MINUTES); + argumentValues.put(metricName, argumentEntry); + } + + return argumentValues; + } + +// protected ListenableFuture fetchArgumentValuesDuringInterval(TenantId tenantId, EntityId entityId, Argument argument, AggInterval interval, long startTs) { +// return switch (argument.getRefEntityKey().getType()) { +// case ATTRIBUTE -> fetchAttribute(tenantId, entityId, argument, startTs); +// case TS_LATEST -> fetchTsLatest(tenantId, entityId, argument, startTs); +// default -> throw new IllegalStateException("Unsupported argument key type for entity aggregation calculated field: " + argument.getRefEntityKey().getType()); +// }; +// } +// +// private ListenableFuture fetchTimeSeries(TenantId tenantId, EntityId entityId, Argument argument, AggInterval interval) { +// long startInterval = System.currentTimeMillis() - interval.getIntervalDuration(); +// +// ReadTsKvQuery query = new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startInterval, System.currentTimeMillis(), 0, 1, Aggregation.NONE); +// +// log.trace("[{}][{}] Fetching timeseries for query {}", tenantId, entityId, query); +// ListenableFuture> fetchedTelemetryFut = timeseriesService.findAll(tenantId, entityId, List.of(query)); +// return Futures.transform(fetchedTelemetryFut, telemetry -> { +// log.debug("[{}][{}] Fetched {} timeseries for query {}", tenantId, entityId, telemetry == null ? 0 : telemetry.size(), query); +// return new SingleValueArgumentEntry(); +// }, calculatedFieldCallbackExecutor); +// } + private ListenableFuture fetchTsRolling(TenantId tenantId, EntityId entityId, Argument argument, long queryEndTs) { long argTimeWindow = argument.getTimeWindow() == 0 ? queryEndTs : argument.getTimeWindow(); long startInterval = queryEndTs - argTimeWindow; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java index a9139572b8..3d75be44a7 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java @@ -25,6 +25,7 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry; import java.util.List; import java.util.Map; @@ -37,6 +38,8 @@ public interface CalculatedFieldProcessingService { Map fetchArgsFromDb(TenantId tenantId, EntityId entityId, Map arguments); + Map fetchArgumentValuesDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception; + void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult result, List cfIds, TbCallback callback); void pushMsgToLinks(CalculatedFieldTelemetryMsg msg, List linkedCalculatedFields, TbCallback callback); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index 52393d0ffe..d347b9701a 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -47,6 +47,7 @@ import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.AggIntervalEntry; import java.util.ArrayList; import java.util.Collections; @@ -111,6 +112,11 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF return resolveArgumentFutures(argFutures); } + @Override + public Map fetchArgumentValuesDuringInterval(EntityId entityId, AggIntervalEntry interval, CalculatedFieldCtx ctx) throws Exception { + return super.fetchArgumentValuesDuringInterval(entityId, interval, ctx); + } + @Override public void pushMsgToRuleEngine(TenantId tenantId, EntityId entityId, CalculatedFieldResult result, List cfIds, TbCallback callback) { if (!(result instanceof PropagationCalculatedFieldResult propagationCalculatedFieldResult)) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 927787eae1..6245435b92 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -45,6 +45,7 @@ import org.thingsboard.server.common.data.cf.configuration.ScheduledUpdateSuppor import org.thingsboard.server.common.data.cf.configuration.SimpleCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.AggFunctionInput; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.geofencing.GeofencingCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; @@ -56,7 +57,9 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import org.thingsboard.server.common.data.util.CollectionsUtil; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.relation.RelationService; +import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; +import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; import org.thingsboard.server.service.telemetry.AlarmSubscriptionService; @@ -98,6 +101,7 @@ public class CalculatedFieldCtx implements Closeable { private TbelInvokeService tbelInvokeService; private RelationService relationService; private AlarmSubscriptionService alarmService; + private CalculatedFieldProcessingService cfProcessingService; private Map tbelExpressions; private Map> simpleExpressions; @@ -190,6 +194,9 @@ public class CalculatedFieldCtx implements Closeable { if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) { this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : -1L; } + if (calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration entityAggregationConfig) { + this.scheduledUpdateIntervalMillis = entityAggregationConfig.getInterval().getIntervalDuration(); + } this.requiresScheduledReevaluation = calculatedField.getConfiguration().requiresScheduledReevaluation(); if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) { this.useLatestTs = aggConfig.isUseLatestTs(); @@ -198,6 +205,7 @@ public class CalculatedFieldCtx implements Closeable { this.tbelInvokeService = systemContext.getTbelInvokeService(); this.relationService = systemContext.getRelationService(); this.alarmService = systemContext.getAlarmService(); + this.cfProcessingService = systemContext.getCalculatedFieldProcessingService(); this.maxDataPointsPerRollingArg = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg); // fixme why tenant profile update is not handled?? this.maxStateSize = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java index 9598cc2b49..0872d98ec0 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.service.cf.CalculatedFieldResult; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; @@ -44,7 +45,8 @@ import static org.thingsboard.server.utils.CalculatedFieldUtils.toSingleValueArg @Type(value = GeofencingCalculatedFieldState.class, name = "GEOFENCING"), @Type(value = AlarmCalculatedFieldState.class, name = "ALARM"), @Type(value = PropagationCalculatedFieldState.class, name = "PROPAGATION"), - @Type(value = RelatedEntitiesAggregationCalculatedFieldState.class, name = "RELATED_ENTITIES_AGGREGATION") + @Type(value = RelatedEntitiesAggregationCalculatedFieldState.class, name = "RELATED_ENTITIES_AGGREGATION"), + @Type(value = EntityAggregationCalculatedFieldState.class, name = "ENTITY_AGGREGATION") }) public interface CalculatedFieldState extends Closeable { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntry.java new file mode 100644 index 0000000000..657cd002ac --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/AggIntervalEntry.java @@ -0,0 +1,26 @@ +/** + * 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.Data; + +@Data +public class AggIntervalEntry { + + public Long startTs; + public Long endTs; + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java new file mode 100644 index 0000000000..6349c635e0 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java @@ -0,0 +1,116 @@ +/** + * 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.ObjectNode; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import lombok.Setter; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.actors.TbActorRef; +import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; +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.util.HashMap; +import java.util.Map; + +import static java.util.concurrent.TimeUnit.SECONDS; + +public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldState { + + private Map aggIntervals = new HashMap<>(); + + @Setter + private long lastArgsRefreshTs = -1; + @Setter + private long lastMetricsEvalTs = -1; + + private long deduplicationIntervalMs = -1; + + 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(); + deduplicationIntervalMs = SECONDS.toMillis(configuration.getDeduplicationIntervalInSec()); + } + + @Override + public CalculatedFieldType getType() { + return CalculatedFieldType.ENTITY_AGGREGATION; + } + + @Override + public ListenableFuture performCalculation(Map updatedArgs, CalculatedFieldCtx ctx) throws Exception { + long endTs = System.currentTimeMillis(); + long startTs = endTs - 1000; + AggIntervalEntry interval = new AggIntervalEntry(); + + Map metrics = cfProcessingService.fetchArgumentValuesDuringInterval(entityId, interval, ctx); + + Output output = ctx.getOutput(); + lastMetricsEvalTs = System.currentTimeMillis(); + ctx.scheduleReevaluation(deduplicationIntervalMs, actorCtx); + ObjectNode result = toResult(endTs, metrics); + if (result != null) { + return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() + .type(output.getType()) + .scope(output.getScope()) + .result(result) + .build()); + } + return Futures.immediateFuture(TelemetryCalculatedFieldResult.EMPTY); + } + + + protected ObjectNode toResult(long endTs, Map metrics) { + ObjectNode metricsNode = JacksonUtil.newObjectNode(); + for (Map.Entry entry : metrics.entrySet()) { + String metricName = entry.getKey(); + ArgumentEntry argumentEntry = entry.getValue(); + if (!argumentEntry.isEmpty()) { + metricsNode.put(metricName, JacksonUtil.toString(argumentEntry.getValue())); + } + } + ObjectNode resultNode = JacksonUtil.newObjectNode(); + if (!metricsNode.isEmpty()) { + resultNode.put("ts", endTs); + resultNode.set("values", metricsNode); + } + return resultNode; + } + + + @Override + public boolean isReady() { + return true; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java index 0c0d401688..72b5b73471 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java @@ -34,6 +34,7 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.single.EntityAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; @@ -83,6 +84,7 @@ public class CalculatedFieldArgumentUtils { case ALARM -> new AlarmCalculatedFieldState(entityId); case PROPAGATION -> new PropagationCalculatedFieldState(entityId); case RELATED_ENTITIES_AGGREGATION -> new RelatedEntitiesAggregationCalculatedFieldState(entityId); + case ENTITY_AGGREGATION -> new EntityAggregationCalculatedFieldState(entityId); }; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldType.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldType.java index 4463c835db..de36b43a40 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedFieldType.java @@ -26,7 +26,8 @@ public enum CalculatedFieldType { GEOFENCING, ALARM, PROPAGATION, - RELATED_ENTITIES_AGGREGATION; + RELATED_ENTITIES_AGGREGATION, + ENTITY_AGGREGATION; public static final Set all = Collections.unmodifiableSet(EnumSet.allOf(CalculatedFieldType.class)); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java index 3df9a32dcc..0ab402ff91 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonSubTypes.Type; import com.fasterxml.jackson.annotation.JsonTypeInfo; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.geofencing.GeofencingCalculatedFieldConfiguration; import org.thingsboard.server.common.data.id.CalculatedFieldId; @@ -42,7 +43,8 @@ import java.util.stream.Collectors; @Type(value = GeofencingCalculatedFieldConfiguration.class, name = "GEOFENCING"), @Type(value = AlarmCalculatedFieldConfiguration.class, name = "ALARM"), @Type(value = PropagationCalculatedFieldConfiguration.class, name = "PROPAGATION"), - @Type(value = RelatedEntitiesAggregationCalculatedFieldConfiguration.class, name = "RELATED_ENTITIES_AGGREGATION") + @Type(value = RelatedEntitiesAggregationCalculatedFieldConfiguration.class, name = "RELATED_ENTITIES_AGGREGATION"), + @Type(value = EntityAggregationCalculatedFieldConfiguration.class, name = "ENTITY_AGGREGATION") }) @JsonIgnoreProperties(ignoreUnknown = true) public interface CalculatedFieldConfiguration { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java new file mode 100644 index 0000000000..69f4b30d46 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java @@ -0,0 +1,59 @@ +/** + * 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 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.ArgumentsBasedCalculatedFieldConfiguration; +import org.thingsboard.server.common.data.cf.configuration.Output; +import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric; +import org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval.AggInterval; + +import java.util.Map; + +@Data +public class EntityAggregationCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration { + + private Map arguments; + @Valid + @NotEmpty + private Map metrics; + + private AggInterval interval; + private long deduplicationIntervalInSec; + private long watermark; + + private Output output; + + @Override + public CalculatedFieldType getType() { + return CalculatedFieldType.ENTITY_AGGREGATION; + } + + @Override + public void validate() { + if (arguments.containsKey("ctx")) { + throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used."); + } + if (arguments.values().stream().anyMatch(Argument::hasTsRollingArgument)) { + throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support TS_ROLLING arguments."); + } + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java new file mode 100644 index 0000000000..cf6f825560 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggInterval.java @@ -0,0 +1,44 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +@JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.PROPERTY, + property = "type" +) +@JsonSubTypes({ + @JsonSubTypes.Type(value = 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 = YearInterval.class, name = "YEAR"), + @JsonSubTypes.Type(value = CustomInterval.class, name = "CUSTOM"), + @JsonSubTypes.Type(value = SpecificTimeInterval.class, name = "SPECIFIC_TIME"), +}) +@JsonIgnoreProperties(ignoreUnknown = true) +public interface AggInterval { + + AggIntervalType getType(); + + long getDelayUntilIntervalEnd(); + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java new file mode 100644 index 0000000000..a5c29b4edf --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/AggIntervalType.java @@ -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, + YEAR, + CUSTOM, + SPECIFIC_TIME + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java new file mode 100644 index 0000000000..5790d7545d --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/BaseAggInterval.java @@ -0,0 +1,54 @@ +/** + * 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 java.time.DayOfWeek; +import java.time.Duration; +import java.time.ZonedDateTime; +import java.time.temporal.ChronoUnit; + +public abstract class BaseAggInterval implements AggInterval { + + protected long offsetMillis; // delay millis since start of interval + + @Override + public long getDelayUntilIntervalEnd() { + return getDelayUntilIntervalEnd(getType(), 1); + } + + protected long getDelayUntilIntervalEnd(AggIntervalType type, long multiplier) { + ZonedDateTime now = ZonedDateTime.now(); + ZonedDateTime next; + + switch (getType()) { + case HOUR -> next = now.plusHours(multiplier).truncatedTo(ChronoUnit.HOURS); + case DAY -> next = now.plusDays(multiplier).truncatedTo(ChronoUnit.DAYS); + case WEEK -> next = now.plusWeeks(multiplier).with(DayOfWeek.MONDAY).truncatedTo(ChronoUnit.DAYS); + case WEEK_SUN_SAT -> next = now.plusWeeks(multiplier).with(DayOfWeek.SUNDAY).truncatedTo(ChronoUnit.DAYS); + case MONTH -> next = now.plusMonths(multiplier).withDayOfMonth(1).truncatedTo(ChronoUnit.DAYS); + case YEAR -> next = now.plusYears(multiplier).withDayOfYear(1).truncatedTo(ChronoUnit.DAYS); + default -> throw new IllegalArgumentException("Unsupported type: " + getType()); + } + + long delayMillis = Duration.between(now, next).toMillis(); + if (offsetMillis > 0) { + delayMillis += offsetMillis; + } + + return Math.max(delayMillis, 0); + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java new file mode 100644 index 0000000000..585e11d955 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/CustomInterval.java @@ -0,0 +1,33 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval; + +public class CustomInterval extends BaseAggInterval { + + private int multiplier; // number of base units (e.g. 2 hours, 5 days) + private AggIntervalType internalIntervalType; + + @Override + public AggIntervalType getType() { + return AggIntervalType.CUSTOM; + } + + @Override + public long getDelayUntilIntervalEnd() { + return super.getDelayUntilIntervalEnd(internalIntervalType, multiplier); + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/DayInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/DayInterval.java new file mode 100644 index 0000000000..19620c6fda --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/DayInterval.java @@ -0,0 +1,25 @@ +/** + * 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 class DayInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.DAY; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/HourInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/HourInterval.java new file mode 100644 index 0000000000..303ff8305e --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/HourInterval.java @@ -0,0 +1,25 @@ +/** + * 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 class HourInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.HOUR; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MonthInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MonthInterval.java new file mode 100644 index 0000000000..1d7eaf14bc --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/MonthInterval.java @@ -0,0 +1,26 @@ +/** + * 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 class MonthInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.MONTH; + } + + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/SpecificTimeInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/SpecificTimeInterval.java new file mode 100644 index 0000000000..2b44366f8b --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/SpecificTimeInterval.java @@ -0,0 +1,42 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.cf.configuration.aggregation.single.interval; + +import java.time.LocalTime; + +public class SpecificTimeInterval implements AggInterval { + + public Long startMillis; // start millis since start of day + public Long endMillis; // end millis since start of day + + @Override + public AggIntervalType getType() { + return AggIntervalType.SPECIFIC_TIME; + } + + @Override + public long getDelayUntilIntervalEnd() { + long nowMillis = LocalTime.now().toNanoOfDay() / 1_000_000L; + long delayMillis; + if (nowMillis < endMillis) { + delayMillis = endMillis - nowMillis; // later today + } else { + delayMillis = (24 * 60 * 60 * 1000L - nowMillis) + endMillis; // next day + } + return delayMillis; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekInterval.java new file mode 100644 index 0000000000..4dd10a2747 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekInterval.java @@ -0,0 +1,25 @@ +/** + * 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 class WeekInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.WEEK; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekSunSatInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekSunSatInterval.java new file mode 100644 index 0000000000..c12e7dc584 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/WeekSunSatInterval.java @@ -0,0 +1,25 @@ +/** + * 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 class WeekSunSatInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.WEEK_SUN_SAT; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/YearInterval.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/YearInterval.java new file mode 100644 index 0000000000..24c6c72ed1 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/YearInterval.java @@ -0,0 +1,25 @@ +/** + * 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 class YearInterval extends BaseAggInterval { + + @Override + public AggIntervalType getType() { + return AggIntervalType.YEAR; + } + +}