git.basschouten.com Git - openhab-addons.git/blob

   1 /**
   2  * Copyright (c) 2010-2024 Contributors to the openHAB project
   3  *
   4  * See the NOTICE file(s) distributed with this work for additional
   5  * information.
   6  *
   7  * This program and the accompanying materials are made available under the
   8  * terms of the Eclipse Public License 2.0 which is available at
   9  * http://www.eclipse.org/legal/epl-2.0
  10  *
  11  * SPDX-License-Identifier: EPL-2.0
  12  */
  13 package org.openhab.binding.energidataservice.internal.handler;
  14
  15 import static org.openhab.binding.energidataservice.internal.EnergiDataServiceBindingConstants.*;
  16 import static org.openhab.core.types.TimeSeries.Policy.REPLACE;
  17
  18 import java.math.BigDecimal;
  19 import java.time.Duration;
  20 import java.time.Instant;
  21 import java.time.LocalDateTime;
  22 import java.time.LocalTime;
  23 import java.time.format.DateTimeFormatter;
  24 import java.time.temporal.ChronoUnit;
  25 import java.util.Arrays;
  26 import java.util.Collection;
  27 import java.util.Currency;
  28 import java.util.HashMap;
  29 import java.util.List;
  30 import java.util.Map;
  31 import java.util.Map.Entry;
  32 import java.util.Set;
  33 import java.util.concurrent.ScheduledFuture;
  34 import java.util.concurrent.TimeUnit;
  35
  36 import javax.measure.Unit;
  37
  38 import org.eclipse.jdt.annotation.NonNullByDefault;
  39 import org.eclipse.jdt.annotation.Nullable;
  40 import org.eclipse.jetty.client.HttpClient;
  41 import org.eclipse.jetty.http.HttpStatus;
  42 import org.openhab.binding.energidataservice.internal.ApiController;
  43 import org.openhab.binding.energidataservice.internal.CacheManager;
  44 import org.openhab.binding.energidataservice.internal.DatahubTariff;
  45 import org.openhab.binding.energidataservice.internal.action.EnergiDataServiceActions;
  46 import org.openhab.binding.energidataservice.internal.api.ChargeType;
  47 import org.openhab.binding.energidataservice.internal.api.ChargeTypeCode;
  48 import org.openhab.binding.energidataservice.internal.api.DatahubTariffFilter;
  49 import org.openhab.binding.energidataservice.internal.api.DatahubTariffFilterFactory;
  50 import org.openhab.binding.energidataservice.internal.api.Dataset;
  51 import org.openhab.binding.energidataservice.internal.api.DateQueryParameter;
  52 import org.openhab.binding.energidataservice.internal.api.DateQueryParameterType;
  53 import org.openhab.binding.energidataservice.internal.api.GlobalLocationNumber;
  54 import org.openhab.binding.energidataservice.internal.api.dto.CO2EmissionRecord;
  55 import org.openhab.binding.energidataservice.internal.api.dto.DatahubPricelistRecord;
  56 import org.openhab.binding.energidataservice.internal.api.dto.ElspotpriceRecord;
  57 import org.openhab.binding.energidataservice.internal.config.DatahubPriceConfiguration;
  58 import org.openhab.binding.energidataservice.internal.config.EnergiDataServiceConfiguration;
  59 import org.openhab.binding.energidataservice.internal.exception.DataServiceException;
  60 import org.openhab.binding.energidataservice.internal.retry.RetryPolicyFactory;
  61 import org.openhab.binding.energidataservice.internal.retry.RetryStrategy;
  62 import org.openhab.core.i18n.TimeZoneProvider;
  63 import org.openhab.core.library.types.DecimalType;
  64 import org.openhab.core.library.types.QuantityType;
  65 import org.openhab.core.library.unit.CurrencyUnits;
  66 import org.openhab.core.library.unit.Units;
  67 import org.openhab.core.thing.Channel;
  68 import org.openhab.core.thing.ChannelUID;
  69 import org.openhab.core.thing.Thing;
  70 import org.openhab.core.thing.ThingStatus;
  71 import org.openhab.core.thing.ThingStatusDetail;
  72 import org.openhab.core.thing.binding.BaseThingHandler;
  73 import org.openhab.core.thing.binding.ThingHandlerService;
  74 import org.openhab.core.types.Command;
  75 import org.openhab.core.types.RefreshType;
  76 import org.openhab.core.types.State;
  77 import org.openhab.core.types.TimeSeries;
  78 import org.openhab.core.types.UnDefType;
  79 import org.slf4j.Logger;
  80 import org.slf4j.LoggerFactory;
  81
  82 /**
  83  * The {@link EnergiDataServiceHandler} is responsible for handling commands, which are
  84  * sent to one of the channels.
  85  *
  86  * @author Jacob Laursen - Initial contribution
  87  */
  88 @NonNullByDefault
  89 public class EnergiDataServiceHandler extends BaseThingHandler {
  90
  91     private static final Duration emissionPrognosisJobInterval = Duration.ofMinutes(15);
  92     private static final Duration emissionRealtimeJobInterval = Duration.ofMinutes(5);
  93
  94     private final Logger logger = LoggerFactory.getLogger(EnergiDataServiceHandler.class);
  95     private final TimeZoneProvider timeZoneProvider;
  96     private final ApiController apiController;
  97     private final CacheManager cacheManager;
  98
  99     private EnergiDataServiceConfiguration config;
 100     private RetryStrategy retryPolicy = RetryPolicyFactory.initial();
 101     private boolean realtimeEmissionsFetchedFirstTime = false;
 102     private @Nullable ScheduledFuture<?> refreshPriceFuture;
 103     private @Nullable ScheduledFuture<?> refreshEmissionPrognosisFuture;
 104     private @Nullable ScheduledFuture<?> refreshEmissionRealtimeFuture;
 105     private @Nullable ScheduledFuture<?> priceUpdateFuture;
 106
 107     public EnergiDataServiceHandler(Thing thing, HttpClient httpClient, TimeZoneProvider timeZoneProvider) {
 108         super(thing);
 109         this.timeZoneProvider = timeZoneProvider;
 110         this.apiController = new ApiController(httpClient, timeZoneProvider);
 111         this.cacheManager = new CacheManager();
 112
 113         // Default configuration
 114         this.config = new EnergiDataServiceConfiguration();
 115     }
 116
 117     @Override
 118     public void handleCommand(ChannelUID channelUID, Command command) {
 119         if (!(command instanceof RefreshType)) {
 120             return;
 121         }
 122
 123         String channelId = channelUID.getId();
 124         if (ELECTRICITY_CHANNELS.contains(channelId)) {
 125             refreshElectricityPrices();
 126         } else if (CHANNEL_CO2_EMISSION_PROGNOSIS.equals(channelId)) {
 127             rescheduleEmissionPrognosisJob();
 128         } else if (CHANNEL_CO2_EMISSION_REALTIME.equals(channelId)) {
 129             realtimeEmissionsFetchedFirstTime = false;
 130             rescheduleEmissionRealtimeJob();
 131         }
 132     }
 133
 134     @Override
 135     public void initialize() {
 136         config = getConfigAs(EnergiDataServiceConfiguration.class);
 137
 138         if (config.priceArea.isBlank()) {
 139             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR,
 140                     "@text/offline.conf-error.no-price-area");
 141             return;
 142         }
 143         GlobalLocationNumber gln = config.getGridCompanyGLN();
 144         if (!gln.isEmpty() && !gln.isValid()) {
 145             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR,
 146                     "@text/offline.conf-error.invalid-grid-company-gln");
 147             return;
 148         }
 149         gln = config.getEnerginetGLN();
 150         if (!gln.isEmpty() && !gln.isValid()) {
 151             updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.CONFIGURATION_ERROR,
 152                     "@text/offline.conf-error.invalid-energinet-gln");
 153             return;
 154         }
 155
 156         updateStatus(ThingStatus.UNKNOWN);
 157
 158         refreshPriceFuture = scheduler.schedule(this::refreshElectricityPrices, 0, TimeUnit.SECONDS);
 159
 160         if (isLinked(CHANNEL_CO2_EMISSION_PROGNOSIS)) {
 161             rescheduleEmissionPrognosisJob();
 162         }
 163         if (isLinked(CHANNEL_CO2_EMISSION_REALTIME)) {
 164             rescheduleEmissionRealtimeJob();
 165         }
 166     }
 167
 168     @Override
 169     public void dispose() {
 170         ScheduledFuture<?> refreshPriceFuture = this.refreshPriceFuture;
 171         if (refreshPriceFuture != null) {
 172             refreshPriceFuture.cancel(true);
 173             this.refreshPriceFuture = null;
 174         }
 175         ScheduledFuture<?> refreshEmissionPrognosisFuture = this.refreshEmissionPrognosisFuture;
 176         if (refreshEmissionPrognosisFuture != null) {
 177             refreshEmissionPrognosisFuture.cancel(true);
 178             this.refreshEmissionPrognosisFuture = null;
 179         }
 180         ScheduledFuture<?> refreshEmissionRealtimeFuture = this.refreshEmissionRealtimeFuture;
 181         if (refreshEmissionRealtimeFuture != null) {
 182             refreshEmissionRealtimeFuture.cancel(true);
 183             this.refreshEmissionRealtimeFuture = null;
 184         }
 185         ScheduledFuture<?> priceUpdateFuture = this.priceUpdateFuture;
 186         if (priceUpdateFuture != null) {
 187             priceUpdateFuture.cancel(true);
 188             this.priceUpdateFuture = null;
 189         }
 190
 191         cacheManager.clear();
 192     }
 193
 194     @Override
 195     public Collection<Class<? extends ThingHandlerService>> getServices() {
 196         return Set.of(EnergiDataServiceActions.class);
 197     }
 198
 199     @Override
 200     public void channelUnlinked(ChannelUID channelUID) {
 201         super.channelUnlinked(channelUID);
 202
 203         if (CHANNEL_CO2_EMISSION_PROGNOSIS.equals(channelUID.getId()) && !isLinked(CHANNEL_CO2_EMISSION_PROGNOSIS)) {
 204             logger.debug("No more items linked to channel '{}', stopping emission prognosis refresh job",
 205                     channelUID.getId());
 206             ScheduledFuture<?> refreshEmissionPrognosisFuture = this.refreshEmissionPrognosisFuture;
 207             if (refreshEmissionPrognosisFuture != null) {
 208                 refreshEmissionPrognosisFuture.cancel(true);
 209                 this.refreshEmissionPrognosisFuture = null;
 210             }
 211         } else if (CHANNEL_CO2_EMISSION_REALTIME.contains(channelUID.getId())
 212                 && !isLinked(CHANNEL_CO2_EMISSION_REALTIME)) {
 213             logger.debug("No more items linked to channel '{}', stopping realtime emission refresh job",
 214                     channelUID.getId());
 215             ScheduledFuture<?> refreshEmissionRealtimeFuture = this.refreshEmissionRealtimeFuture;
 216             if (refreshEmissionRealtimeFuture != null) {
 217                 refreshEmissionRealtimeFuture.cancel(true);
 218                 this.refreshEmissionRealtimeFuture = null;
 219             }
 220         }
 221     }
 222
 223     private void refreshElectricityPrices() {
 224         RetryStrategy retryPolicy;
 225         try {
 226             if (isLinked(CHANNEL_SPOT_PRICE)) {
 227                 downloadSpotPrices();
 228             }
 229
 230             for (DatahubTariff datahubTariff : DatahubTariff.values()) {
 231                 if (isLinked(datahubTariff.getChannelId())) {
 232                     downloadTariffs(datahubTariff);
 233                 }
 234             }
 235
 236             updateStatus(ThingStatus.ONLINE);
 237             updatePrices();
 238             updateTimeSeries();
 239
 240             if (isLinked(CHANNEL_SPOT_PRICE)) {
 241                 if (cacheManager.getNumberOfFutureSpotPrices() < 13) {
 242                     retryPolicy = RetryPolicyFactory.whenExpectedSpotPriceDataMissing(DAILY_REFRESH_TIME_CET,
 243                             NORD_POOL_TIMEZONE);
 244                 } else {
 245                     retryPolicy = RetryPolicyFactory.atFixedTime(DAILY_REFRESH_TIME_CET, NORD_POOL_TIMEZONE);
 246                 }
 247             } else {
 248                 retryPolicy = RetryPolicyFactory.atFixedTime(LocalTime.MIDNIGHT, timeZoneProvider.getTimeZone());
 249             }
 250         } catch (DataServiceException e) {
 251             if (e.getHttpStatus() != 0) {
 252                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR,
 253                         HttpStatus.getCode(e.getHttpStatus()).getMessage());
 254             } else {
 255                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR, e.getMessage());
 256             }
 257             if (e.getCause() != null) {
 258                 logger.debug("Error retrieving prices", e);
 259             }
 260             retryPolicy = RetryPolicyFactory.fromThrowable(e);
 261         } catch (InterruptedException e) {
 262             logger.debug("Refresh job interrupted");
 263             Thread.currentThread().interrupt();
 264             return;
 265         }
 266
 267         reschedulePriceRefreshJob(retryPolicy);
 268     }
 269
 270     private void downloadSpotPrices() throws InterruptedException, DataServiceException {
 271         if (cacheManager.areSpotPricesFullyCached()) {
 272             logger.debug("Cached spot prices still valid, skipping download.");
 273             return;
 274         }
 275         DateQueryParameter start;
 276         if (cacheManager.areHistoricSpotPricesCached()) {
 277             start = DateQueryParameter.of(DateQueryParameterType.UTC_NOW);
 278         } else {
 279             start = DateQueryParameter.of(DateQueryParameterType.UTC_NOW,
 280                     Duration.ofHours(-CacheManager.NUMBER_OF_HISTORIC_HOURS));
 281         }
 282         Map<String, String> properties = editProperties();
 283         ElspotpriceRecord[] spotPriceRecords = apiController.getSpotPrices(config.priceArea, config.getCurrency(),
 284                 start, properties);
 285         cacheManager.putSpotPrices(spotPriceRecords, config.getCurrency());
 286         updateProperties(properties);
 287     }
 288
 289     private void downloadTariffs(DatahubTariff datahubTariff) throws InterruptedException, DataServiceException {
 290         GlobalLocationNumber globalLocationNumber = switch (datahubTariff) {
 291             case GRID_TARIFF -> config.getGridCompanyGLN();
 292             default -> config.getEnerginetGLN();
 293         };
 294         if (globalLocationNumber.isEmpty()) {
 295             return;
 296         }
 297         if (cacheManager.areTariffsValidTomorrow(datahubTariff)) {
 298             logger.debug("Cached tariffs of type {} still valid, skipping download.", datahubTariff);
 299             cacheManager.updateTariffs(datahubTariff);
 300         } else {
 301             DatahubTariffFilter filter = switch (datahubTariff) {
 302                 case GRID_TARIFF -> getGridTariffFilter();
 303                 case SYSTEM_TARIFF -> DatahubTariffFilterFactory.getSystemTariff();
 304                 case TRANSMISSION_GRID_TARIFF -> DatahubTariffFilterFactory.getTransmissionGridTariff();
 305                 case ELECTRICITY_TAX -> DatahubTariffFilterFactory.getElectricityTax();
 306                 case REDUCED_ELECTRICITY_TAX -> DatahubTariffFilterFactory.getReducedElectricityTax();
 307             };
 308             cacheManager.putTariffs(datahubTariff, downloadPriceLists(globalLocationNumber, filter));
 309         }
 310     }
 311
 312     private Collection<DatahubPricelistRecord> downloadPriceLists(GlobalLocationNumber globalLocationNumber,
 313             DatahubTariffFilter filter) throws InterruptedException, DataServiceException {
 314         Map<String, String> properties = editProperties();
 315         Collection<DatahubPricelistRecord> records = apiController.getDatahubPriceLists(globalLocationNumber,
 316                 ChargeType.Tariff, filter, properties);
 317         updateProperties(properties);
 318
 319         return records;
 320     }
 321
 322     private DatahubTariffFilter getGridTariffFilter() {
 323         Channel channel = getThing().getChannel(CHANNEL_GRID_TARIFF);
 324         if (channel == null) {
 325             return DatahubTariffFilterFactory.getGridTariffByGLN(config.gridCompanyGLN);
 326         }
 327
 328         DatahubPriceConfiguration datahubPriceConfiguration = channel.getConfiguration()
 329                 .as(DatahubPriceConfiguration.class);
 330
 331         if (!datahubPriceConfiguration.hasAnyFilterOverrides()) {
 332             return DatahubTariffFilterFactory.getGridTariffByGLN(config.gridCompanyGLN);
 333         }
 334
 335         DateQueryParameter start = datahubPriceConfiguration.getStart();
 336         if (start == null) {
 337             logger.warn("Invalid channel configuration parameter 'start' or 'offset': {} (offset: {})",
 338                     datahubPriceConfiguration.start, datahubPriceConfiguration.offset);
 339             return DatahubTariffFilterFactory.getGridTariffByGLN(config.gridCompanyGLN);
 340         }
 341
 342         Set<ChargeTypeCode> chargeTypeCodes = datahubPriceConfiguration.getChargeTypeCodes();
 343         Set<String> notes = datahubPriceConfiguration.getNotes();
 344         DatahubTariffFilter filter;
 345         if (!chargeTypeCodes.isEmpty() || !notes.isEmpty()) {
 346             // Completely override filter.
 347             filter = new DatahubTariffFilter(chargeTypeCodes, notes, start);
 348         } else {
 349             // Only override start date in pre-configured filter.
 350             filter = new DatahubTariffFilter(DatahubTariffFilterFactory.getGridTariffByGLN(config.gridCompanyGLN),
 351                     start);
 352         }
 353
 354         return new DatahubTariffFilter(filter, DateQueryParameter.of(filter.getDateQueryParameter(),
 355                 Duration.ofHours(-CacheManager.NUMBER_OF_HISTORIC_HOURS)));
 356     }
 357
 358     private void refreshCo2EmissionPrognosis() {
 359         try {
 360             updateCo2Emissions(Dataset.CO2EmissionPrognosis, CHANNEL_CO2_EMISSION_PROGNOSIS,
 361                     DateQueryParameter.of(DateQueryParameterType.UTC_NOW, Duration.ofMinutes(-5)));
 362             updateStatus(ThingStatus.ONLINE);
 363         } catch (DataServiceException e) {
 364             if (e.getHttpStatus() != 0) {
 365                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR,
 366                         HttpStatus.getCode(e.getHttpStatus()).getMessage());
 367             } else {
 368                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR, e.getMessage());
 369             }
 370             if (e.getCause() != null) {
 371                 logger.debug("Error retrieving CO2 emission prognosis", e);
 372             }
 373         } catch (InterruptedException e) {
 374             logger.debug("Emission prognosis refresh job interrupted");
 375             Thread.currentThread().interrupt();
 376             return;
 377         }
 378     }
 379
 380     private void refreshCo2EmissionRealtime() {
 381         try {
 382             updateCo2Emissions(Dataset.CO2Emission, CHANNEL_CO2_EMISSION_REALTIME,
 383                     DateQueryParameter.of(DateQueryParameterType.UTC_NOW,
 384                             realtimeEmissionsFetchedFirstTime ? Duration.ofMinutes(-5) : Duration.ofHours(-24)));
 385             realtimeEmissionsFetchedFirstTime = true;
 386             updateStatus(ThingStatus.ONLINE);
 387         } catch (DataServiceException e) {
 388             if (e.getHttpStatus() != 0) {
 389                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR,
 390                         HttpStatus.getCode(e.getHttpStatus()).getMessage());
 391             } else {
 392                 updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.OFFLINE.COMMUNICATION_ERROR, e.getMessage());
 393             }
 394             if (e.getCause() != null) {
 395                 logger.debug("Error retrieving CO2 realtime emissions", e);
 396             }
 397         } catch (InterruptedException e) {
 398             logger.debug("Emission realtime refresh job interrupted");
 399             Thread.currentThread().interrupt();
 400             return;
 401         }
 402     }
 403
 404     private void updateCo2Emissions(Dataset dataset, String channelId, DateQueryParameter dateQueryParameter)
 405             throws InterruptedException, DataServiceException {
 406         Map<String, String> properties = editProperties();
 407         CO2EmissionRecord[] emissionRecords = apiController.getCo2Emissions(dataset, config.priceArea,
 408                 dateQueryParameter, properties);
 409         updateProperties(properties);
 410
 411         TimeSeries timeSeries = new TimeSeries(REPLACE);
 412         Instant now = Instant.now();
 413
 414         if (dataset == Dataset.CO2Emission && emissionRecords.length > 0) {
 415             // Records are sorted descending, first record is current.
 416             updateState(channelId, new QuantityType<>(emissionRecords[0].emission(), Units.GRAM_PER_KILOWATT_HOUR));
 417         }
 418
 419         for (CO2EmissionRecord emissionRecord : emissionRecords) {
 420             State state = new QuantityType<>(emissionRecord.emission(), Units.GRAM_PER_KILOWATT_HOUR);
 421             timeSeries.add(emissionRecord.start(), state);
 422
 423             if (dataset == Dataset.CO2EmissionPrognosis && now.compareTo(emissionRecord.start()) >= 0
 424                     && now.compareTo(emissionRecord.end()) < 0) {
 425                 updateState(channelId, state);
 426             }
 427         }
 428         sendTimeSeries(channelId, timeSeries);
 429     }
 430
 431     private void updatePrices() {
 432         cacheManager.cleanup();
 433
 434         updateCurrentSpotPrice();
 435         Arrays.stream(DatahubTariff.values())
 436                 .forEach(tariff -> updateCurrentTariff(tariff.getChannelId(), cacheManager.getTariff(tariff)));
 437
 438         reschedulePriceUpdateJob();
 439     }
 440
 441     private void updateCurrentSpotPrice() {
 442         if (!isLinked(CHANNEL_SPOT_PRICE)) {
 443             return;
 444         }
 445         BigDecimal spotPrice = cacheManager.getSpotPrice();
 446         updatePriceState(CHANNEL_SPOT_PRICE, spotPrice, config.getCurrency());
 447     }
 448
 449     private void updateCurrentTariff(String channelId, @Nullable BigDecimal tariff) {
 450         if (!isLinked(channelId)) {
 451             return;
 452         }
 453         updatePriceState(channelId, tariff, CURRENCY_DKK);
 454     }
 455
 456     private void updatePriceState(String channelID, @Nullable BigDecimal price, Currency currency) {
 457         updateState(channelID, price != null ? getEnergyPrice(price, currency) : UnDefType.UNDEF);
 458     }
 459
 460     private State getEnergyPrice(BigDecimal price, Currency currency) {
 461         String currencyCode = currency.getCurrencyCode();
 462         Unit<?> unit = CurrencyUnits.getInstance().getUnit(currencyCode);
 463         if (unit == null) {
 464             logger.trace("Currency {} is unknown, falling back to DecimalType", currency.getCurrencyCode());
 465             return new DecimalType(price);
 466         }
 467         try {
 468             return new QuantityType<>(price + " " + currencyCode + "/kWh");
 469         } catch (IllegalArgumentException e) {
 470             logger.debug("Unable to create QuantityType, falling back to DecimalType", e);
 471             return new DecimalType(price);
 472         }
 473     }
 474
 475     private void updateTimeSeries() {
 476         TimeSeries spotPriceTimeSeries = new TimeSeries(REPLACE);
 477         Map<DatahubTariff, TimeSeries> datahubTimeSeriesMap = new HashMap<>();
 478         Map<DatahubTariff, BigDecimal> datahubPreviousTariff = new HashMap<>();
 479         for (DatahubTariff datahubTariff : DatahubTariff.values()) {
 480             datahubTimeSeriesMap.put(datahubTariff, new TimeSeries(REPLACE));
 481         }
 482
 483         Map<Instant, BigDecimal> spotPriceMap = cacheManager.getSpotPrices();
 484         List<Entry<Instant, BigDecimal>> spotPrices = spotPriceMap.entrySet().stream()
 485                 .sorted(Map.Entry.comparingByKey()).toList();
 486         for (Entry<Instant, BigDecimal> spotPrice : spotPrices) {
 487             Instant hourStart = spotPrice.getKey();
 488             if (isLinked(CHANNEL_SPOT_PRICE)) {
 489                 spotPriceTimeSeries.add(hourStart, getEnergyPrice(spotPrice.getValue(), config.getCurrency()));
 490             }
 491             for (Map.Entry<DatahubTariff, TimeSeries> entry : datahubTimeSeriesMap.entrySet()) {
 492                 DatahubTariff datahubTariff = entry.getKey();
 493                 String channelId = datahubTariff.getChannelId();
 494                 if (!isLinked(channelId)) {
 495                     continue;
 496                 }
 497                 BigDecimal tariff = cacheManager.getTariff(datahubTariff, hourStart);
 498                 if (tariff != null) {
 499                     BigDecimal previousTariff = datahubPreviousTariff.get(datahubTariff);
 500                     if (previousTariff != null && tariff.equals(previousTariff)) {
 501                         // Skip redundant states.
 502                         continue;
 503                     }
 504                     TimeSeries timeSeries = entry.getValue();
 505                     timeSeries.add(hourStart, getEnergyPrice(tariff, CURRENCY_DKK));
 506                     datahubPreviousTariff.put(datahubTariff, tariff);
 507                 }
 508             }
 509         }
 510         if (spotPriceTimeSeries.size() > 0) {
 511             sendTimeSeries(CHANNEL_SPOT_PRICE, spotPriceTimeSeries);
 512         }
 513         for (Map.Entry<DatahubTariff, TimeSeries> entry : datahubTimeSeriesMap.entrySet()) {
 514             DatahubTariff datahubTariff = entry.getKey();
 515             String channelId = datahubTariff.getChannelId();
 516             if (!isLinked(channelId)) {
 517                 continue;
 518             }
 519             TimeSeries timeSeries = entry.getValue();
 520             if (timeSeries.size() > 0) {
 521                 sendTimeSeries(channelId, timeSeries);
 522             }
 523         }
 524     }
 525
 526     /**
 527      * Get the configured {@link Currency} for spot prices.
 528      *
 529      * @return Spot price currency
 530      */
 531     public Currency getCurrency() {
 532         return config.getCurrency();
 533     }
 534
 535     /**
 536      * Get cached spot prices or try once to download them if not cached
 537      * (usually if no items are linked).
 538      *
 539      * @return Map of future spot prices
 540      */
 541     public Map<Instant, BigDecimal> getSpotPrices() {
 542         try {
 543             downloadSpotPrices();
 544         } catch (DataServiceException e) {
 545             if (logger.isDebugEnabled()) {
 546                 logger.warn("Error retrieving spot prices", e);
 547             } else {
 548                 logger.warn("Error retrieving spot prices: {}", e.getMessage());
 549             }
 550         } catch (InterruptedException e) {
 551             Thread.currentThread().interrupt();
 552         }
 553
 554         return cacheManager.getSpotPrices();
 555     }
 556
 557     /**
 558      * Return cached tariffs or try once to download them if not cached
 559      * (usually if no items are linked).
 560      *
 561      * @return Map of future tariffs
 562      */
 563     public Map<Instant, BigDecimal> getTariffs(DatahubTariff datahubTariff) {
 564         try {
 565             downloadTariffs(datahubTariff);
 566         } catch (DataServiceException e) {
 567             if (logger.isDebugEnabled()) {
 568                 logger.warn("Error retrieving tariffs", e);
 569             } else {
 570                 logger.warn("Error retrieving tariffs of type {}: {}", datahubTariff, e.getMessage());
 571             }
 572         } catch (InterruptedException e) {
 573             Thread.currentThread().interrupt();
 574         }
 575
 576         return cacheManager.getTariffs(datahubTariff);
 577     }
 578
 579     /**
 580      * Return whether reduced electricity tax is set in configuration.
 581      *
 582      * @return true if reduced electricity tax applies
 583      */
 584     public boolean isReducedElectricityTax() {
 585         return config.reducedElectricityTax;
 586     }
 587
 588     private void reschedulePriceUpdateJob() {
 589         ScheduledFuture<?> priceUpdateJob = this.priceUpdateFuture;
 590         if (priceUpdateJob != null) {
 591             // Do not interrupt ourselves.
 592             priceUpdateJob.cancel(false);
 593             this.priceUpdateFuture = null;
 594         }
 595
 596         Instant now = Instant.now();
 597         long millisUntilNextClockHour = Duration
 598                 .between(now, now.plus(1, ChronoUnit.HOURS).truncatedTo(ChronoUnit.HOURS)).toMillis() + 1;
 599         this.priceUpdateFuture = scheduler.schedule(this::updatePrices, millisUntilNextClockHour,
 600                 TimeUnit.MILLISECONDS);
 601         logger.debug("Price update job rescheduled in {} milliseconds", millisUntilNextClockHour);
 602     }
 603
 604     private void reschedulePriceRefreshJob(RetryStrategy retryPolicy) {
 605         // Preserve state of previous retry policy when configuration is the same.
 606         if (!retryPolicy.equals(this.retryPolicy)) {
 607             this.retryPolicy = retryPolicy;
 608         }
 609
 610         ScheduledFuture<?> refreshJob = this.refreshPriceFuture;
 611
 612         long secondsUntilNextRefresh = this.retryPolicy.getDuration().getSeconds();
 613         Instant timeOfNextRefresh = Instant.now().plusSeconds(secondsUntilNextRefresh);
 614         this.refreshPriceFuture = scheduler.schedule(this::refreshElectricityPrices, secondsUntilNextRefresh,
 615                 TimeUnit.SECONDS);
 616         logger.debug("Price refresh job rescheduled in {} seconds: {}", secondsUntilNextRefresh, timeOfNextRefresh);
 617         DateTimeFormatter formatter = DateTimeFormatter.ofPattern(PROPERTY_DATETIME_FORMAT);
 618         updateProperty(PROPERTY_NEXT_CALL, LocalDateTime.ofInstant(timeOfNextRefresh, timeZoneProvider.getTimeZone())
 619                 .truncatedTo(ChronoUnit.SECONDS).format(formatter));
 620
 621         if (refreshJob != null) {
 622             refreshJob.cancel(true);
 623         }
 624     }
 625
 626     private void rescheduleEmissionPrognosisJob() {
 627         logger.debug("Scheduling emission prognosis refresh job now and every {}", emissionPrognosisJobInterval);
 628
 629         ScheduledFuture<?> refreshEmissionPrognosisFuture = this.refreshEmissionPrognosisFuture;
 630         if (refreshEmissionPrognosisFuture != null) {
 631             refreshEmissionPrognosisFuture.cancel(true);
 632         }
 633
 634         this.refreshEmissionPrognosisFuture = scheduler.scheduleWithFixedDelay(this::refreshCo2EmissionPrognosis, 0,
 635                 emissionPrognosisJobInterval.toSeconds(), TimeUnit.SECONDS);
 636     }
 637
 638     private void rescheduleEmissionRealtimeJob() {
 639         logger.debug("Scheduling emission realtime refresh job now and every {}", emissionRealtimeJobInterval);
 640
 641         ScheduledFuture<?> refreshEmissionFuture = this.refreshEmissionRealtimeFuture;
 642         if (refreshEmissionFuture != null) {
 643             refreshEmissionFuture.cancel(true);
 644         }
 645
 646         this.refreshEmissionRealtimeFuture = scheduler.scheduleWithFixedDelay(this::refreshCo2EmissionRealtime, 0,
 647                 emissionRealtimeJobInterval.toSeconds(), TimeUnit.SECONDS);
 648     }
 649 }