From c82f52af9b8629f584cbd1e56ae51418d3364c46 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 10 Sep 2026 13:27:28 +0200 Subject: [PATCH 01/13] Add Distro Energy day-ahead submission Submits the parent optimisation asset's net power forecast to the Distro Energy trader API, one POST per day for five days ahead. The API requires exactly one entry per quarter-hour of the market day, which is 92 on the spring-forward day and 100 on the fall-back day rather than the usual 96. Entries are therefore driven by a grid of real instants across the market day instead of by the number of rows the datapoint query returns, so the length follows the DST transition without special-casing it. Predicted datapoints are stored as JVM-local wall-clock rather than UTC (see openremote/openremote#3292), so each instant is mapped back into that frame to look up its value. On the fall-back day the repeated hour collapses onto a single stored row, so those positions reuse the surviving value and log a warning. Once predicted datapoints are stored in UTC the same code becomes exact without changing. Several defects in the original submission block are fixed along the way: - Interval query bounds are inclusive at both ends, so querying a full day returned 97 buckets instead of 96. The upper bound is now the start of the last interval. - Gap-filled buckets carry a null value, which threw NPE when unboxed into SubmissionData.volume. Missing intervals are submitted as 0.0 instead. - Positions were sent in descending order; the API requires ascending. - scheduleAtFixedRate received an absolute epoch-millis initial delay and a millisecond period, both interpreted as MINUTES, so the task never fired. The market timezone is configurable through DISTRO_ENERGY_TIMEZONE and defaults to Europe/Amsterdam. --- .../ems/agent/EmsDistroEnergyAsset.java | 50 ++++ .../distroenergy/DayAheadResource.java | 36 +++ .../distroenergy/DistroEnergyHandler.java | 275 ++++++++++++++++++ .../distroenergy/dto/DayAheadSubmission.java | 22 ++ .../distroenergy/dto/SubmissionData.java | 21 ++ .../DistroEnergyHandlerTest.groovy | 194 ++++++++++++ 6 files changed, 598 insertions(+) create mode 100644 ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java create mode 100644 ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DayAheadResource.java create mode 100644 ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java create mode 100644 ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/DayAheadSubmission.java create mode 100644 ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/SubmissionData.java create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy diff --git a/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java b/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java new file mode 100644 index 0000000..9a5b43d --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java @@ -0,0 +1,50 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.agent; + +import jakarta.persistence.Entity; +import java.util.Optional; +import org.openremote.model.asset.Asset; +import org.openremote.model.asset.AssetDescriptor; +import org.openremote.model.value.AttributeDescriptor; +import org.openremote.model.value.ValueType; + +@Entity +public class EmsDistroEnergyAsset extends Asset { + + public static final AttributeDescriptor PORTFOLIO = + new AttributeDescriptor<>("portfolio", ValueType.TEXT); + + public static final AssetDescriptor DESCRIPTOR = + new AssetDescriptor<>("transmission-tower", null, EmsDistroEnergyAsset.class); + + protected EmsDistroEnergyAsset() {} + + public EmsDistroEnergyAsset(String name) { + super(name); + } + + public Optional getPortfolio() { + return getAttributes().getValue(PORTFOLIO); + } + + public void setPortfolio(String portfolio) { + getAttributes().getOrCreate(PORTFOLIO).setValue(portfolio); + } +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DayAheadResource.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DayAheadResource.java new file mode 100644 index 0000000..2caff6d --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DayAheadResource.java @@ -0,0 +1,36 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy; + +import static jakarta.ws.rs.core.MediaType.APPLICATION_JSON; + +import jakarta.ws.rs.*; +import org.openremote.extension.ems.manager.distroenergy.dto.DayAheadSubmission; + +@Path("trader") +public interface DayAheadResource { + + @POST + @Consumes({APPLICATION_JSON}) + @Path("{portfolio}/day-ahead/data") + void postDayAhead( + @PathParam("portfolio") String portfolio, + @HeaderParam("x-client-key") String clientKey, + DayAheadSubmission dayAheadSubmission); +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java new file mode 100644 index 0000000..0a9ad87 --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -0,0 +1,275 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy; + +import static java.time.format.DateTimeFormatter.BASIC_ISO_DATE; +import static org.openremote.container.web.WebTargetBuilder.createClient; +import static org.openremote.model.syslog.SyslogCategory.API; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jboss.resteasy.client.jaxrs.ResteasyClient; +import org.openremote.container.timer.TimerService; +import org.openremote.extension.ems.manager.distroenergy.dto.DayAheadSubmission; +import org.openremote.extension.ems.manager.distroenergy.dto.SubmissionData; +import org.openremote.manager.datapoint.AssetPredictedDatapointService; +import org.openremote.model.Container; +import org.openremote.model.attribute.AttributeRef; +import org.openremote.model.datapoint.ValueDatapoint; +import org.openremote.model.datapoint.query.AssetDatapointIntervalQuery; +import org.openremote.model.syslog.SyslogCategory; + +public class DistroEnergyHandler { + + private static final Logger LOG = SyslogCategory.getLogger(API, DistroEnergyHandler.class); + public static final String DISTRO_ENERGY_CLIENT_KEY = "DISTRO_ENERGY_CLIENT_KEY"; + public static final String DISTRO_ENERGY_BASE_URL = "DISTRO_ENERGY_BASE_URL"; + public static final String DISTRO_ENERGY_BASE_URL_DEFAULT = "https://ibt.dev.distro.energy/"; + public static final String DISTRO_ENERGY_TIMEZONE = "DISTRO_ENERGY_TIMEZONE"; + public static final String DISTRO_ENERGY_TIMEZONE_DEFAULT = "Europe/Amsterdam"; + public static final String REQUEST_INTERVAL_MINUTES = "REQUEST_INTERVAL"; + public static final String REQUEST_INTERVAL_MINUTES_DEFAULT = "60"; + + /** Market settlement interval. One submission entry per ISP. */ + protected static final Duration ISP_DURATION = Duration.ofMinutes(15); + + protected static final int DAYS_AHEAD = 5; + + protected final AttributeRef powerNetAttributeRef; + protected final String distroEnergyBaseUrl; + protected final String portfolio; + protected final String clientKey; + protected final ZoneId marketZone; + protected final long requestIntervalMinutes; + protected final DayAheadResource dayAheadResource; + + protected final TimerService timerService; + protected final ScheduledExecutorService scheduledExecutorService; + protected final AssetPredictedDatapointService assetPredictedDatapointService; + protected ScheduledFuture nextRequestFuture; + + public static class Factory { + protected Container container; + + public Factory(Container container) { + this.container = container; + } + + public DistroEnergyHandler createHandler(AttributeRef powerNetAttributeRef, String portfolio) { + return new DistroEnergyHandler(powerNetAttributeRef, portfolio, container); + } + } + + public DistroEnergyHandler( + AttributeRef powerNetAttributeRef, String portfolio, Container container) { + this.powerNetAttributeRef = powerNetAttributeRef; + this.portfolio = portfolio; + + this.timerService = container.getService(TimerService.class); + this.scheduledExecutorService = container.getScheduledExecutor(); + this.assetPredictedDatapointService = + container.getService(AssetPredictedDatapointService.class); + + this.distroEnergyBaseUrl = + container.getConfig().getOrDefault(DISTRO_ENERGY_BASE_URL, DISTRO_ENERGY_BASE_URL_DEFAULT); + this.marketZone = + ZoneId.of( + container + .getConfig() + .getOrDefault(DISTRO_ENERGY_TIMEZONE, DISTRO_ENERGY_TIMEZONE_DEFAULT)); + this.requestIntervalMinutes = + Integer.parseInt( + container + .getConfig() + .getOrDefault(REQUEST_INTERVAL_MINUTES, REQUEST_INTERVAL_MINUTES_DEFAULT)); + this.clientKey = container.getConfig().get(DISTRO_ENERGY_CLIENT_KEY); + + if (clientKey == null) { + throw new RuntimeException( + DISTRO_ENERGY_CLIENT_KEY + " not defined, cannot use Distro Energy."); + } + + ResteasyClient client = createClient(org.openremote.container.Container.EXECUTOR); + this.dayAheadResource = client.target(this.distroEnergyBaseUrl).proxy(DayAheadResource.class); + + this.nextRequestFuture = + scheduledExecutorService.scheduleAtFixedRate( + this::submitDayAheadForecasts, + getFirstRequestDelayMillis(), + Duration.ofMinutes(requestIntervalMinutes).toMillis(), + TimeUnit.MILLISECONDS); + LOG.info( + "DistroEnergyHandler instance for distro energy deployed for portfolio: " + this.portfolio); + } + + protected void submitDayAheadForecasts() { + LocalDate firstDay = timerService.getNow().atZone(marketZone).toLocalDate().plusDays(1); + + for (int day = 0; day < DAYS_AHEAD; day++) { + LocalDate marketDate = firstDay.plusDays(day); + try { + submitDayAheadForecast(marketDate); + } catch (Exception e) { + // Keep going: one bad day must not lose the other four, nor kill the recurring task. + LOG.log( + Level.WARNING, + "Failed to submit day-ahead forecast for portfolio " + + portfolio + + " and day " + + marketDate, + e); + } + } + } + + protected void submitDayAheadForecast(LocalDate marketDate) { + ZoneId storageZone = ZoneId.systemDefault(); + ZonedDateTime dayStart = marketDate.atStartOfDay(marketZone); + ZonedDateTime dayEnd = dayStart.plusDays(1); + + // The interval query bounds are inclusive at both ends, so the upper bound is the start of the + // last ISP rather than the end of the day. Otherwise the query returns one bucket too many. + List> datapoints = + assetPredictedDatapointService.queryDatapoints( + powerNetAttributeRef.getId(), + powerNetAttributeRef.getName(), + new AssetDatapointIntervalQuery( + dayStart.withZoneSameInstant(storageZone).toLocalDateTime(), + dayEnd.minus(ISP_DURATION).withZoneSameInstant(storageZone).toLocalDateTime(), + "15 minutes", + AssetDatapointIntervalQuery.Formula.AVG, + true)); + + List submissionData = + buildSubmissionData(marketDate, marketZone, storageZone, datapoints); + + dayAheadResource.postDayAhead( + portfolio, + clientKey, + new DayAheadSubmission( + submissionData.toArray(new SubmissionData[0]), + Long.parseLong(marketDate.format(BASIC_ISO_DATE)), + timerService.getCurrentTimeMillis())); + } + + /** + * Builds one submission entry per ISP of the given market day, in ascending position order. + * + *

The entries are driven by a grid of real instants stepping from the start to the end of the + * market day, so the count is 92, 96 or 100 depending on whether the day carries a DST + * transition. Each instant is mapped back into the frame the predicted datapoint table is written + * in, which is the JVM default zone (see {@code AbstractDatapointService}), and looked up there. + * + *

Under a JVM zone that observes DST the storage frame is not monotonic, so on the fall-back + * day the two instants of the repeated hour collapse onto a single stored row and both read the + * same value. That is a consequence of the naive primary key upstream + * (openremote/openremote#3292); once predicted datapoints are stored in UTC every instant maps to + * a distinct row and this method becomes exact without changing. + */ + static List buildSubmissionData( + LocalDate marketDate, + ZoneId marketZone, + ZoneId storageZone, + List> datapoints) { + + Map valuesByStorageKey = new HashMap<>(); + for (ValueDatapoint datapoint : datapoints) { + // Gap-filled buckets carry a null value; skip them so they fall through to the default below. + if (datapoint.getValue() instanceof Number value) { + valuesByStorageKey.put( + Instant.ofEpochMilli(datapoint.getTimestamp()).atZone(storageZone).toLocalDateTime(), + value.doubleValue()); + } + } + + ZonedDateTime dayStart = marketDate.atStartOfDay(marketZone); + ZonedDateTime dayEnd = dayStart.plusDays(1); + + List submissionData = new ArrayList<>(); + Set keysRead = new HashSet<>(); + int missing = 0; + int position = 1; + + for (ZonedDateTime isp = dayStart; isp.isBefore(dayEnd); isp = isp.plus(ISP_DURATION)) { + LocalDateTime storageKey = isp.withZoneSameInstant(storageZone).toLocalDateTime(); + Double value = valuesByStorageKey.get(storageKey); + + if (value == null) { + missing++; + } else if (!keysRead.add(storageKey)) { + LOG.warning( + "Day-ahead position " + + position + + " on " + + marketDate + + " reuses the value stored at " + + storageKey + + " because the repeated DST hour" + + " collapses onto one predicted datapoint row (openremote/openremote#3292)"); + } + + submissionData.add(new SubmissionData(position++, null, null, value != null ? value : 0.0)); + } + + if (missing > 0) { + LOG.warning( + "Day-ahead submission for " + + marketDate + + " has " + + missing + + " of " + + submissionData.size() + + " intervals without a forecast; submitted as 0.0"); + } + + return submissionData; + } + + protected long getFirstRequestDelayMillis() { + long firstRequestMillis = + timerService + .getNow() + .truncatedTo(ChronoUnit.HOURS) + .plus(30, ChronoUnit.MINUTES) + .toEpochMilli(); + return Math.max(0L, firstRequestMillis - timerService.getCurrentTimeMillis()); + } + + public void undeploy() { + if (nextRequestFuture != null) { + nextRequestFuture.cancel(true); + } + } +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/DayAheadSubmission.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/DayAheadSubmission.java new file mode 100644 index 0000000..b75c9ad --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/DayAheadSubmission.java @@ -0,0 +1,22 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy.dto; + +public record DayAheadSubmission( + SubmissionData[] submissionData, long day, long creationTimestamp) {} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/SubmissionData.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/SubmissionData.java new file mode 100644 index 0000000..91c2fb2 --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/SubmissionData.java @@ -0,0 +1,21 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy.dto; + +public record SubmissionData(int position, Double priceLow, Double priceHigh, double volume) {} diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy new file mode 100644 index 0000000..91ae35e --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy @@ -0,0 +1,194 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy + +import org.lfenergy.shapeshifter.api.datetime.DateTimeCalculation +import org.openremote.model.datapoint.ValueDatapoint +import spock.lang.Specification +import spock.lang.Unroll + +import java.time.Duration +import java.time.LocalDate +import java.time.ZoneId +import java.time.ZonedDateTime + +/** + * Unit test for {@link DistroEnergyHandler#buildSubmissionData}. + * + * Deliberately a plain {@code Specification} with no {@code ManagerContainerTrait}: that trait forces + * {@code TIMER_CLOCK_TYPE: PSEUDO}, whose {@code init()} calls {@code TimeZone.setDefault("UTC")} + * JVM-wide. Running under a forced-UTC JVM would hide exactly the DST behaviour under test here. + * + * The Distro Energy day-ahead API requires exactly one entry per quarter-hour of the market day: 96 + * normally, 92 on the spring-forward day and 100 on the fall-back day. Expected counts are + * cross-checked against shapeshifter's {@code DateTimeCalculation.numberOfIspsOnDay} so they are not + * computed by the same arithmetic under test. + */ +class DistroEnergyHandlerTest extends Specification { + + static final ZoneId MARKET = ZoneId.of("Europe/Amsterdam") + static final ZoneId AMSTERDAM_STORAGE = ZoneId.of("Europe/Amsterdam") + static final ZoneId UTC_STORAGE = ZoneId.of("UTC") + static final Duration ISP = Duration.ofMinutes(15) + + static final LocalDate NORMAL_DAY = LocalDate.of(2026, 9, 15) + static final LocalDate SPRING_FORWARD = LocalDate.of(2026, 3, 29) // last Sunday of March + static final LocalDate FALL_BACK = LocalDate.of(2026, 10, 25) // last Sunday of October + + /** Every ISP instant of the market day, which is what the forecast writer would have produced. */ + static List ispInstants(LocalDate marketDate) { + def start = marketDate.atStartOfDay(MARKET) + def end = start.plusDays(1) + def instants = [] + for (def isp = start; isp.isBefore(end); isp = isp.plus(ISP)) { + instants << isp + } + instants + } + + /** + * Simulates the predicted datapoint table for a full day of forecast. + * + * Datapoints come back keyed by epoch millis, but the row they were read from is keyed by the naive + * local time in the storage zone. Under a DST-observing storage zone two instants of the fall-back + * hour share one row, so this collapses them the way the upsert primary key does. + */ + static List> datapointsFor(LocalDate marketDate, ZoneId storageZone, + Closure valueFor = { int i -> + (i + 1) * 1.0d + }) { + Map> byStorageKey = [:] + ispInstants(marketDate).eachWithIndex { ZonedDateTime isp, int i -> + def storageKey = isp.withZoneSameInstant(storageZone).toLocalDateTime() + // Last write wins, exactly like ON CONFLICT ... DO UPDATE SET value = excluded.value + byStorageKey[storageKey] = new ValueDatapoint(isp.toInstant().toEpochMilli(), valueFor(i)) + } + new ArrayList<>(byStorageKey.values()) + } + + static long expectedIsps(LocalDate marketDate) { + DateTimeCalculation.numberOfIspsOnDay(marketDate, ISP, MARKET.id) + } + + @Unroll + def "submission has one entry per market ISP on #marketDate with #storageZone storage"() { + given: "a full day of predicted datapoints as stored in the given frame" + def datapoints = datapointsFor(marketDate, storageZone) + + when: + def data = DistroEnergyHandler.buildSubmissionData(marketDate, MARKET, storageZone, datapoints) + + then: "the length matches the market day, independent of the storage frame" + data.size() == expected + expected == expectedIsps(marketDate) + + and: "positions are 1-based and ascending, as the API requires" + data*.position == (1..expected).toList() + + where: + marketDate | storageZone || expected + NORMAL_DAY | AMSTERDAM_STORAGE || 96 + SPRING_FORWARD | AMSTERDAM_STORAGE || 92 + FALL_BACK | AMSTERDAM_STORAGE || 100 + NORMAL_DAY | UTC_STORAGE || 96 + SPRING_FORWARD | UTC_STORAGE || 92 + FALL_BACK | UTC_STORAGE || 100 + } + + def "UTC storage recovers every distinct value on the fall-back day"() { + given: + def datapoints = datapointsFor(FALL_BACK, UTC_STORAGE) + + when: + def data = DistroEnergyHandler.buildSubmissionData(FALL_BACK, MARKET, UTC_STORAGE, datapoints) + + then: "no row collision, so all 100 forecast values survive" + datapoints.size() == 100 + data*.volume == (1..100).collect { it * 1.0d } + } + + def "Amsterdam storage still fills 100 entries when the repeated hour collapses"() { + given: "the fall-back hour collapses onto four shared rows" + def datapoints = datapointsFor(FALL_BACK, AMSTERDAM_STORAGE) + + when: + def data = DistroEnergyHandler.buildSubmissionData(FALL_BACK, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "four quarter-hours were lost at write time" + datapoints.size() == 96 + + and: "the submission is still the full 100 entries the API requires" + data.size() == 100 + + and: "and none of them defaulted to 0.0, they reuse the surviving twin" + data*.volume.every { it != 0.0d } + } + + def "spring-forward day never looks up the non-existent local hour"() { + given: + def datapoints = datapointsFor(SPRING_FORWARD, AMSTERDAM_STORAGE) + + expect: "the missing hour simply is not part of the day" + datapoints.size() == 92 + + when: + def data = DistroEnergyHandler.buildSubmissionData(SPRING_FORWARD, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "so nothing is filled and every value is real" + data.size() == 92 + data*.volume == (1..92).collect { it * 1.0d } + } + + def "missing forecast intervals become 0.0 rather than throwing"() { + given: "only the first two intervals have a forecast" + def datapoints = datapointsFor(NORMAL_DAY, AMSTERDAM_STORAGE).take(2) + + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "the day is still complete, gaps filled with 0.0" + data.size() == 96 + data.findAll { it.volume == 0.0d }.size() == 94 + } + + def "gap-filled null buckets do not cause an NPE"() { + given: "the query returned buckets with null values, as gapFill does" + def datapoints = ispInstants(NORMAL_DAY).collect { + new ValueDatapoint(it.toInstant().toEpochMilli(), (Double) null) + } + + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: + noExceptionThrown() + data.size() == 96 + data*.volume.every { it == 0.0d } + } + + def "empty forecast still yields a well-formed submission"() { + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, []) + + then: + data.size() == 96 + data*.position == (1..96).toList() + data*.volume.every { it == 0.0d } + } +} From db896a83a4f80c1c267d9564b25cf6654300f2f4 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 10 Sep 2026 13:27:36 +0200 Subject: [PATCH 02/13] Deploy Distro Energy handlers from EmsOptimisationService Nothing constructed DistroEnergyHandler, so the integration was unreachable. Handlers are now created for existing EmsDistroEnergyAsset instances at startup and kept in sync with create, update and delete persistence events, following the pattern already used for the GOPACS handlers. Handlers are keyed by asset id rather than portfolio. GOPACS keys by EAN because inbound messages route on it, whereas this handler has no external routing key and the asset id survives a portfolio edit. Deployment is wrapped in try/catch because DistroEnergyHandler throws when DISTRO_ENERGY_CLIENT_KEY is unset, which would otherwise fail service startup for every deployment that does not use Distro Energy. The asset panel config is added so the portfolio attribute is editable in the UI. --- .../ems/manager/EmsOptimisationService.java | 77 +++++++++++++++++++ .../resources/ems/config/asset-types.json | 49 ++++++++++++ 2 files changed, 126 insertions(+) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index c98e97f..586ae41 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -33,13 +33,16 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.logging.Level; import java.util.logging.Logger; import org.apache.camel.builder.RouteBuilder; import org.openremote.container.message.MessageBrokerService; import org.openremote.container.timer.TimerService; +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset; import org.openremote.extension.ems.agent.EmsElectricityBatteryAsset; import org.openremote.extension.ems.agent.EmsEnergyOptimisationAsset; import org.openremote.extension.ems.agent.EmsGOPACSAsset; +import org.openremote.extension.ems.manager.distroenergy.DistroEnergyHandler; import org.openremote.extension.ems.manager.gopacs.GOPACSHandler; import org.openremote.extension.ems.manager.gopacs.GOPACSRedispatchHandler; import org.openremote.manager.asset.AssetProcessingService; @@ -55,6 +58,7 @@ import org.openremote.model.asset.AssetFilter; import org.openremote.model.attribute.Attribute; import org.openremote.model.attribute.AttributeEvent; +import org.openremote.model.attribute.AttributeRef; import org.openremote.model.datapoint.ValueDatapoint; import org.openremote.model.datapoint.query.AssetDatapointAllQuery; import org.openremote.model.query.AssetQuery; @@ -68,6 +72,7 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer protected GOPACSHandler.Factory gopacsHandlerFactory; protected GOPACSRedispatchHandler.Factory gopacsRedispatchHandlerFactory; + protected DistroEnergyHandler.Factory distroEnergyHandlerFactory; private final Map> energyOptimisationAssetsMap = new ConcurrentHashMap<>(); @@ -75,6 +80,10 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer private final Map gopacsHandlerMap = new HashMap<>(); private final Map gopacsRedispatchHandlerMap = new HashMap<>(); + // Keyed by asset id rather than portfolio: unlike GOPACS there is no external routing key, and + // the asset id stays stable when the portfolio is edited. + private final Map distroEnergyHandlerMap = new HashMap<>(); + @SuppressWarnings("unchecked") @Override public void configure() throws Exception { @@ -103,6 +112,7 @@ public void init(Container container) throws Exception { gopacsHandlerFactory = new GOPACSHandler.Factory(container); gopacsRedispatchHandlerFactory = new GOPACSRedispatchHandler.Factory(container); + distroEnergyHandlerFactory = new DistroEnergyHandler.Factory(container); } @Override @@ -169,6 +179,17 @@ public void start(Container container) throws Exception { } }); + // Start Distro Energy handler for all Distro Energy assets + services + .getAssetStorageService() + .findAll( + new AssetQuery() + .types(EmsDistroEnergyAsset.class) + .attributeName(EmsDistroEnergyAsset.PORTFOLIO.getName())) + .stream() + .map(asset -> (EmsDistroEnergyAsset) asset) + .forEach(this::startDistroEnergyHandler); + // List of asset types that are part of the core EMS service String[] assetTypes = { EmsElectricityBatteryAsset.DESCRIPTOR.getName(), @@ -189,6 +210,8 @@ public void start(Container container) throws Exception { public void stop(Container container) throws Exception { gopacsRedispatchHandlerMap.forEach((ean, handler) -> handler.stopPolling()); gopacsRedispatchHandlerMap.clear(); + distroEnergyHandlerMap.forEach((assetId, handler) -> handler.undeploy()); + distroEnergyHandlerMap.clear(); energyOptimisationAssetsMap.forEach((assetId, scheduledFuture) -> stopOptimisation(assetId)); energyOptimisationTimersMap.clear(); } @@ -352,6 +375,50 @@ private void stopRedispatchHandler(String contractedEan) { } } + private void startDistroEnergyHandler(EmsDistroEnergyAsset distroEnergyAsset) { + String assetId = distroEnergyAsset.getId(); + String portfolio = distroEnergyAsset.getPortfolio().orElse(""); + + if (portfolio.isBlank()) { + LOG.warning( + "Unable to deploy Distro Energy because portfolio is blank for asset: " + assetId); + return; + } + + // The forecast being submitted is the parent optimisation asset's net power. + String energyOptimisationAssetId = distroEnergyAsset.getParentId(); + if (energyOptimisationAssetId == null) { + LOG.warning( + String.format( + "Unable to deploy Distro Energy for portfolio '%s'; asset '%s' has no parent '%s'", + portfolio, assetId, EmsEnergyOptimisationAsset.class.getSimpleName())); + return; + } + + LOG.fine("Deploying Distro Energy for portfolio: " + portfolio); + try { + distroEnergyHandlerMap.put( + assetId, + distroEnergyHandlerFactory.createHandler( + new AttributeRef( + energyOptimisationAssetId, EmsEnergyOptimisationAsset.POWER_NET.getName()), + portfolio)); + } catch (Exception e) { + // A missing client key or an unusable base URL must not take down the rest of the EMS + // service. + LOG.log(Level.WARNING, "Failed to deploy Distro Energy for portfolio: " + portfolio, e); + return; + } + LOG.fine("Deployed Distro Energy for portfolio: " + portfolio); + } + + private void stopDistroEnergyHandler(String assetId) { + DistroEnergyHandler existing = distroEnergyHandlerMap.remove(assetId); + if (existing != null) { + existing.undeploy(); + } + } + protected void processAssetChange(PersistenceEvent persistenceEvent) { if (persistenceEvent.getEntity() instanceof EmsEnergyOptimisationAsset emsEnergyOptimisationAsset) { @@ -382,6 +449,16 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { // Redispatch handler is managed via attribute events (redispatchEnabled) } }); + } else if (persistenceEvent.getEntity() instanceof EmsDistroEnergyAsset emsDistroEnergyAsset) { + switch (persistenceEvent.getCause()) { + case DELETE -> stopDistroEnergyHandler(emsDistroEnergyAsset.getId()); + case CREATE -> startDistroEnergyHandler(emsDistroEnergyAsset); + // Redeploy so a changed portfolio or parent takes effect. + case UPDATE -> { + stopDistroEnergyHandler(emsDistroEnergyAsset.getId()); + startDistroEnergyHandler(emsDistroEnergyAsset); + } + } } } diff --git a/ems/src/main/resources/ems/config/asset-types.json b/ems/src/main/resources/ems/config/asset-types.json index 5ff07a9..9ac6e89 100644 --- a/ems/src/main/resources/ems/config/asset-types.json +++ b/ems/src/main/resources/ems/config/asset-types.json @@ -246,5 +246,54 @@ "column": 1 } ] + }, + "EmsDistroEnergyAsset": { + "viewerStyles": {}, + "panels": [ + { + "type": "info", + "hideOnMobile": true, + "properties": { + "include": [] + }, + "attributes": { + "include": ["notes"] + } + }, + { + "type": "info", + "column": 1, + "title": "location", + "properties": { + "include": [] + }, + "attributes": { + "include": ["location"], + "itemConfig": { + "location": { + "label": "" + } + } + } + }, + { + "type": "info", + "title": "Configuration", + "properties": { + "include": [] + }, + "attributes": { + "include": ["portfolio"] + } + }, + { + "type": "history", + "column": 1 + }, + { + "type": "linkedUsers", + "column": 1 + } + ] } } From afc6d9f9d019ba87addbc4430e961b6f27fbeb4f Mon Sep 17 00:00:00 2001 From: Michael Date: Fri, 11 Sep 2026 11:41:55 +0200 Subject: [PATCH 03/13] Start Distro Energy submission from deploy() instead of the constructor getFirstRequestDelayMillis() clamps the initial delay to zero, so any deploy after the half hour scheduled the recurring task with no delay at all. The task could then start on an executor thread while the constructor was still running, observing a partly constructed handler. Move the scheduling into a deploy() method that EmsOptimisationService calls after construction, matching how startRedispatchHandler calls startPolling() on GOPACSRedispatchHandler. The handler is registered in the map before deploy() runs, so one whose scheduling is rejected during shutdown is still reachable from stop(). --- .../ems/manager/EmsOptimisationService.java | 10 +++++++--- .../manager/distroenergy/DistroEnergyHandler.java | 14 +++++++++++--- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 586ae41..3865ad4 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -396,19 +396,23 @@ private void startDistroEnergyHandler(EmsDistroEnergyAsset distroEnergyAsset) { } LOG.fine("Deploying Distro Energy for portfolio: " + portfolio); + DistroEnergyHandler handler; try { - distroEnergyHandlerMap.put( - assetId, + handler = distroEnergyHandlerFactory.createHandler( new AttributeRef( energyOptimisationAssetId, EmsEnergyOptimisationAsset.POWER_NET.getName()), - portfolio)); + portfolio); } catch (Exception e) { // A missing client key or an unusable base URL must not take down the rest of the EMS // service. LOG.log(Level.WARNING, "Failed to deploy Distro Energy for portfolio: " + portfolio, e); return; } + // Registered before the schedule starts, so a handler whose deploy() is rejected during + // shutdown is still reachable from stop() and gets its client closed. + distroEnergyHandlerMap.put(assetId, handler); + handler.deploy(); LOG.fine("Deployed Distro Energy for portfolio: " + portfolio); } diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index 0a9ad87..3099850 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -123,15 +123,23 @@ public DistroEnergyHandler( ResteasyClient client = createClient(org.openremote.container.Container.EXECUTOR); this.dayAheadResource = client.target(this.distroEnergyBaseUrl).proxy(DayAheadResource.class); + } - this.nextRequestFuture = + /** + * Starts the recurring submission. + * + *

Separate from the constructor so the scheduled task cannot observe a partly constructed + * handler: {@link #getFirstRequestDelayMillis()} clamps to zero, so any deploy after the half hour + * would otherwise start the task on an executor thread while the constructor was still running. + */ + public void deploy() { + nextRequestFuture = scheduledExecutorService.scheduleAtFixedRate( this::submitDayAheadForecasts, getFirstRequestDelayMillis(), Duration.ofMinutes(requestIntervalMinutes).toMillis(), TimeUnit.MILLISECONDS); - LOG.info( - "DistroEnergyHandler instance for distro energy deployed for portfolio: " + this.portfolio); + LOG.info("Deployed Distro Energy handler for portfolio: " + portfolio); } protected void submitDayAheadForecasts() { From f878f18a234e4a53210e2e862dbe38079bf0938c Mon Sep 17 00:00:00 2001 From: Michael Date: Fri, 11 Sep 2026 11:43:05 +0200 Subject: [PATCH 04/13] Close the Distro Energy REST client when the handler is undeployed The handler created a ResteasyClient into a local variable and kept only the proxy, so undeploy() cancelled the scheduled task but left the client and its connection pool allocated. EmsOptimisationService recreates the handler on every update of the asset, so each edit leaked a client that had already made requests and held sockets. Store the client as a field and close it in undeploy(), cancelling the scheduled future with an interrupt first so an in-flight POST is aborted before the client goes away. This matches GOPACSRedispatchHandler. The constructor also guards the window between createClient and the proxy: client.target throws IllegalArgumentException on a malformed DISTRO_ENERGY_BASE_URL, and no reference escapes a constructor that threw, so undeploy() could never close that client. Closing is safe with the shared executor: WebTargetBuilder.createClient passes it through ResteasyClientBuilder.executorService(ExecutorService), which sets cleanupExecutor to false, so Container.EXECUTOR is untouched. --- .../manager/distroenergy/DistroEnergyHandler.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index 3099850..e97ed2f 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -73,6 +73,7 @@ public class DistroEnergyHandler { protected final String clientKey; protected final ZoneId marketZone; protected final long requestIntervalMinutes; + protected final ResteasyClient client; protected final DayAheadResource dayAheadResource; protected final TimerService timerService; @@ -121,8 +122,14 @@ public DistroEnergyHandler( DISTRO_ENERGY_CLIENT_KEY + " not defined, cannot use Distro Energy."); } - ResteasyClient client = createClient(org.openremote.container.Container.EXECUTOR); - this.dayAheadResource = client.target(this.distroEnergyBaseUrl).proxy(DayAheadResource.class); + this.client = createClient(org.openremote.container.Container.EXECUTOR); + try { + this.dayAheadResource = client.target(this.distroEnergyBaseUrl).proxy(DayAheadResource.class); + } catch (RuntimeException e) { + // No reference escapes a constructor that threw, so undeploy() can never close this client. + client.close(); + throw e; + } } /** @@ -277,7 +284,11 @@ protected long getFirstRequestDelayMillis() { public void undeploy() { if (nextRequestFuture != null) { + // Interrupt to abort any in-flight HTTP call before closing the client nextRequestFuture.cancel(true); + nextRequestFuture = null; } + client.close(); + LOG.info("Undeployed Distro Energy handler for portfolio: " + portfolio); } } From 2c595eb25870b345c9528eddd1b56f3157182f92 Mon Sep 17 00:00:00 2001 From: Michael Date: Fri, 11 Sep 2026 11:44:15 +0200 Subject: [PATCH 05/13] Verify the Distro Energy parent is an EmsEnergyOptimisationAsset The comment said the parent must be an EmsEnergyOptimisationAsset, but only the presence of a parent ID was checked. An asset placed under any other parent deployed cleanly and then queried powerNet on an asset with no such attribute, producing an empty forecast rather than an error. Resolve the parent at deploy time instead. find(id, false, type) returns null both for a missing asset and for one of the wrong type, which are the same problem here, so the message reports it as such rather than claiming the type is wrong. loadComplete is false because only the type is being read. --- .../ems/manager/EmsOptimisationService.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 3865ad4..35699dd 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -395,6 +395,23 @@ private void startDistroEnergyHandler(EmsDistroEnergyAsset distroEnergyAsset) { return; } + // find(..., EmsEnergyOptimisationAsset.class) returns null both for a missing asset and for one + // of the wrong type, which are the same problem here: there is no net power forecast to submit. + if (services + .getAssetStorageService() + .find(energyOptimisationAssetId, false, EmsEnergyOptimisationAsset.class) + == null) { + LOG.warning( + String.format( + "Unable to deploy Distro Energy for portfolio '%s'; parent '%s' of asset '%s' is not" + + " an existing '%s'", + portfolio, + energyOptimisationAssetId, + assetId, + EmsEnergyOptimisationAsset.class.getSimpleName())); + return; + } + LOG.fine("Deploying Distro Energy for portfolio: " + portfolio); DistroEnergyHandler handler; try { From 55d330ff11ea24cacf216b10a808a0befb0bcea5 Mon Sep 17 00:00:00 2001 From: Michael Date: Fri, 11 Sep 2026 11:49:14 +0200 Subject: [PATCH 06/13] Submit day-ahead forecasts through the actual horizon, not a fixed window The handler posted a fixed five days on every run and zero-filled every interval it had no data for. Nothing in this repository writes predicted powerNet; an external producer does. Days past that producer's horizon were therefore submitted as a full day of zeros, every hour, which is a real net power trading position for a day we know nothing about. The fixed five was also wrong in the other direction: a longer forecast was silently truncated. Submit from tomorrow forward until the first day with no forecast. A day that is only partly covered is still sent whole, zero-filled to midnight, because the API requires one entry per quarter-hour of the market day. Each run re-submits and overwrites, so a day is sent by the first run after the horizon reaches it and no catch-up state is needed. buildSubmissionData decides whether a day has a forecast, returning an empty list when it does not. That decision has to come from the ISP grid rather than from whatever the query returned: the two agree only because the query window is derived from the same day bounds, and widening that window would let a neighbouring day's value make this day look covered. An empty list is unambiguous because a submitted day is never shorter than 92 entries. A genuine all-zero forecast is still submitted. The distinction between a forecast 0.0 and a filled 0.0 only survives inside buildSubmissionData, where the grid lookup is a boxed Double and null means missing; it is erased at the boundary because SubmissionData.volume is a primitive. Gap logging is split accordingly. A partly covered final day is now the expected steady state and logs at FINE with the boundary position, so a horizon that is systematically short by a fixed number of intervals stays diagnosable. A gap inside the forecast still warns, because the producer writes every interval it covers and those positions go out as 0.0. MAX_DAYS_AHEAD is a defensive ceiling rather than a business window; the loop stops at the horizon, so it only bounds the damage if a stray far-future datapoint makes the horizon look unbounded. --- .../distroenergy/DistroEnergyHandler.java | 132 ++++++++++++++---- .../DistroEnergyHandlerTest.groovy | 106 ++++++++++++-- 2 files changed, 205 insertions(+), 33 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index e97ed2f..8d7bf1e 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -65,7 +65,12 @@ public class DistroEnergyHandler { /** Market settlement interval. One submission entry per ISP. */ protected static final Duration ISP_DURATION = Duration.ofMinutes(15); - protected static final int DAYS_AHEAD = 5; + /** + * Defensive ceiling on one run, not a business window. The loop stops at the first day with no + * forecast, so this only bounds the damage if a stray far-future datapoint makes the horizon look + * unbounded. + */ + protected static final int MAX_DAYS_AHEAD = 14; protected final AttributeRef powerNetAttributeRef; protected final String distroEnergyBaseUrl; @@ -136,8 +141,9 @@ public DistroEnergyHandler( * Starts the recurring submission. * *

Separate from the constructor so the scheduled task cannot observe a partly constructed - * handler: {@link #getFirstRequestDelayMillis()} clamps to zero, so any deploy after the half hour - * would otherwise start the task on an executor thread while the constructor was still running. + * handler: {@link #getFirstRequestDelayMillis()} clamps to zero, so any deploy after the half + * hour would otherwise start the task on an executor thread while the constructor was still + * running. */ public void deploy() { nextRequestFuture = @@ -151,13 +157,18 @@ public void deploy() { protected void submitDayAheadForecasts() { LocalDate firstDay = timerService.getNow().atZone(marketZone).toLocalDate().plusDays(1); + int submitted = 0; - for (int day = 0; day < DAYS_AHEAD; day++) { + for (int day = 0; day < MAX_DAYS_AHEAD; day++) { LocalDate marketDate = firstDay.plusDays(day); try { - submitDayAheadForecast(marketDate); + if (!submitDayAheadForecast(marketDate)) { + // First uncovered day is the end of the forecast horizon; nothing beyond it to send. + break; + } + submitted++; } catch (Exception e) { - // Keep going: one bad day must not lose the other four, nor kill the recurring task. + // Keep going: one bad day must not lose the days after it, nor kill the recurring task. LOG.log( Level.WARNING, "Failed to submit day-ahead forecast for portfolio " @@ -167,9 +178,27 @@ protected void submitDayAheadForecasts() { e); } } + + // Reaching the ceiling means the horizon looked unbounded, which the forecast producer cannot + // legitimately do; the run was truncated and later days were not sent. + if (submitted >= MAX_DAYS_AHEAD) { + LOG.warning( + "Day-ahead run for portfolio " + + portfolio + + " hit the ceiling of " + + MAX_DAYS_AHEAD + + " days; check the predicted " + + powerNetAttributeRef.getName() + + " data for far-future values"); + } else { + LOG.fine("Day-ahead run for portfolio " + portfolio + " submitted " + submitted + " day(s)"); + } } - protected void submitDayAheadForecast(LocalDate marketDate) { + /** + * @return whether a submission was actually POSTed for this day. + */ + protected boolean submitDayAheadForecast(LocalDate marketDate) { ZoneId storageZone = ZoneId.systemDefault(); ZonedDateTime dayStart = marketDate.atStartOfDay(marketZone); ZonedDateTime dayEnd = dayStart.plusDays(1); @@ -190,6 +219,19 @@ protected void submitDayAheadForecast(LocalDate marketDate) { List submissionData = buildSubmissionData(marketDate, marketZone, storageZone, datapoints); + if (submissionData.isEmpty()) { + // Beyond the forecast horizon. The task repeats and the API overwrites a day on every + // submission, so the day is sent by the first run after the horizon reaches it; there is no + // catch-up state to keep here. + LOG.fine( + "No forecast yet for portfolio " + + portfolio + + " and day " + + marketDate + + "; not sending"); + return false; + } + dayAheadResource.postDayAhead( portfolio, clientKey, @@ -197,21 +239,34 @@ protected void submitDayAheadForecast(LocalDate marketDate) { submissionData.toArray(new SubmissionData[0]), Long.parseLong(marketDate.format(BASIC_ISO_DATE)), timerService.getCurrentTimeMillis())); + return true; } /** - * Builds one submission entry per ISP of the given market day, in ascending position order. + * Builds one submission entry per ISP of the given market day, in ascending position order, or an + * empty list when the day carries no forecast at all. * *

The entries are driven by a grid of real instants stepping from the start to the end of the - * market day, so the count is 92, 96 or 100 depending on whether the day carries a DST - * transition. Each instant is mapped back into the frame the predicted datapoint table is written - * in, which is the JVM default zone (see {@code AbstractDatapointService}), and looked up there. + * market day, so a submitted day is always 92, 96 or 100 entries depending on whether the day + * carries a DST transition. Each instant is mapped back into the frame the predicted datapoint + * table is written in, which is the JVM default zone (see {@code AbstractDatapointService}), and + * looked up there. + * + *

An empty result means "nothing to submit" and cannot be confused with a real day, which is + * never shorter than 92 entries. A day beyond the external forecast producer's horizon + * legitimately has no data, and submitting it would post a zero net power trading position for a + * day we know nothing about. Within a day that does have a forecast every gap is filled with 0.0, + * interior and trailing alike, because the API requires the complete day. + * + *

The decision is taken from the ISP grid rather than from whatever the query returned, so a + * value belonging to a neighbouring day can never make this day look covered. * *

Under a JVM zone that observes DST the storage frame is not monotonic, so on the fall-back * day the two instants of the repeated hour collapse onto a single stored row and both read the * same value. That is a consequence of the naive primary key upstream * (openremote/openremote#3292); once predicted datapoints are stored in UTC every instant maps to - * a distinct row and this method becomes exact without changing. + * a distinct row and this method becomes exact without changing. The collapse can only duplicate + * a read, never erase one, so it cannot turn a day with a forecast into a skip. */ static List buildSubmissionData( LocalDate marketDate, @@ -235,6 +290,7 @@ static List buildSubmissionData( List submissionData = new ArrayList<>(); Set keysRead = new HashSet<>(); int missing = 0; + int lastRealPosition = 0; int position = 1; for (ZonedDateTime isp = dayStart; isp.isBefore(dayEnd); isp = isp.plus(ISP_DURATION)) { @@ -243,30 +299,56 @@ static List buildSubmissionData( if (value == null) { missing++; - } else if (!keysRead.add(storageKey)) { - LOG.warning( - "Day-ahead position " - + position - + " on " - + marketDate - + " reuses the value stored at " - + storageKey - + " because the repeated DST hour" - + " collapses onto one predicted datapoint row (openremote/openremote#3292)"); + } else { + lastRealPosition = position; + if (!keysRead.add(storageKey)) { + LOG.warning( + "Day-ahead position " + + position + + " on " + + marketDate + + " reuses the value stored at " + + storageKey + + " because the repeated DST hour" + + " collapses onto one predicted datapoint row (openremote/openremote#3292)"); + } } submissionData.add(new SubmissionData(position++, null, null, value != null ? value : 0.0)); } - if (missing > 0) { + // Nothing at all was forecast for this day. Hand the caller the empty sentinel and stay silent + // here: the caller knows the portfolio and owns the log line. + if (lastRealPosition == 0) { + return List.of(); + } + + int trailingMissing = submissionData.size() - lastRealPosition; + int interiorMissing = missing - trailingMissing; + + if (interiorMissing > 0) { + // A hole before the end of the forecast means the producer skipped intervals it did cover, + // and those positions go out as 0.0, which is a real trading value. LOG.warning( "Day-ahead submission for " + marketDate + " has " - + missing + + interiorMissing + + " of " + + submissionData.size() + + " intervals with a gap inside the forecast; submitted as 0.0"); + } + if (trailingMissing > 0) { + // Expected once the forecast horizon ends inside this day. Logged with the boundary so a + // horizon that is systematically short by a fixed number of ISPs is still diagnosable. + LOG.fine( + "Forecast for " + + marketDate + + " ends at position " + + lastRealPosition + " of " + submissionData.size() - + " intervals without a forecast; submitted as 0.0"); + + "; the remainder is filled with 0.0 up to midnight"); } return submissionData; diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy index 91ae35e..441e7e5 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerTest.groovy @@ -167,7 +167,7 @@ class DistroEnergyHandlerTest extends Specification { data.findAll { it.volume == 0.0d }.size() == 94 } - def "gap-filled null buckets do not cause an NPE"() { + def "a day of only gap-filled null buckets is not submitted"() { given: "the query returned buckets with null values, as gapFill does" def datapoints = ispInstants(NORMAL_DAY).collect { new ValueDatapoint(it.toInstant().toEpochMilli(), (Double) null) @@ -176,19 +176,109 @@ class DistroEnergyHandlerTest extends Specification { when: def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) - then: + then: "a non-empty result set with no real values still means no forecast" noExceptionThrown() - data.size() == 96 + datapoints.size() == 96 + data.isEmpty() + } + + def "a day with no forecast at all is not submitted"() { + when: "the forecast horizon has not reached this day" + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, []) + + then: "there is nothing to submit, rather than a day of zeros" + data.isEmpty() + } + + def "a genuine all-zero forecast is still submitted"() { + given: "the producer wrote 0.0 for every interval, which is a real trading position" + def datapoints = datapointsFor(NORMAL_DAY, AMSTERDAM_STORAGE, { int i -> 0.0d }) + + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "a forecast 0.0 and a filled 0.0 are not confused" + data.size() == expectedIsps(NORMAL_DAY) data*.volume.every { it == 0.0d } } - def "empty forecast still yields a well-formed submission"() { + def "a forecast that stops mid-day is filled to midnight rather than truncated"() { + given: "the forecast horizon ends after 40 of the 96 intervals" + def datapoints = datapointsFor(NORMAL_DAY, AMSTERDAM_STORAGE).take(40) + when: - def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, []) + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) - then: - data.size() == 96 + then: "the API still receives the complete day" + data.size() == expectedIsps(NORMAL_DAY) data*.position == (1..96).toList() - data*.volume.every { it == 0.0d } + + and: "the covered intervals keep their real values" + data[0..39]*.volume == (1..40).collect { it * 1.0d } + + and: "and the tail to midnight is filled with 0.0" + data[40..95]*.volume.every { it == 0.0d } + } + + @Unroll + def "a single real value at position #realPosition still submits the whole day"() { + given: "exactly one interval of the day was forecast" + def instant = ispInstants(NORMAL_DAY)[realPosition - 1] + def datapoints = [new ValueDatapoint(instant.toInstant().toEpochMilli(), 7.5d)] + + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "the day is complete, with that one value in place and the rest filled" + data.size() == 96 + data[realPosition - 1].volume == 7.5d + data.findAll { it.volume == 0.0d }.size() == 95 + + where: + realPosition << [1, 48, 96] + } + + def "a value outside the market day does not make the day submittable"() { + given: "the only datapoint lies one ISP before the start of the market day" + def beforeDay = NORMAL_DAY.atStartOfDay(MARKET).minus(ISP) + def datapoints = [new ValueDatapoint(beforeDay.toInstant().toEpochMilli(), 12.0d)] + + when: + def data = DistroEnergyHandler.buildSubmissionData(NORMAL_DAY, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "the decision follows the ISP grid, not whatever the query happened to return" + data.isEmpty() + } + + @Unroll + def "the submit-or-skip decision is independent of the DST arithmetic on #marketDate"() { + given: "the same day with and without a forecast" + def covered = datapointsFor(marketDate, AMSTERDAM_STORAGE) + def uncovered = ispInstants(marketDate).collect { + new ValueDatapoint(it.toInstant().toEpochMilli(), (Double) null) + } + + expect: "an uncovered day is skipped, and a covered one is still exactly the right length" + DistroEnergyHandler.buildSubmissionData(marketDate, MARKET, AMSTERDAM_STORAGE, uncovered).isEmpty() + DistroEnergyHandler.buildSubmissionData(marketDate, MARKET, AMSTERDAM_STORAGE, covered).size() == + expectedIsps(marketDate) + + where: + marketDate << [NORMAL_DAY, SPRING_FORWARD, FALL_BACK] + } + + def "the fall-back collapse cannot turn a day with a forecast into a skip"() { + given: "the only value sits in the repeated hour, where two ISPs share one stored row" + def repeatedHour = ispInstants(FALL_BACK)[8] // 02:00 CEST, the first pass of the repeat + def datapoints = [new ValueDatapoint(repeatedHour.toInstant().toEpochMilli(), 3.25d)] + + when: + def data = DistroEnergyHandler.buildSubmissionData(FALL_BACK, MARKET, AMSTERDAM_STORAGE, datapoints) + + then: "the day is submitted in full" + data.size() == 100 + + and: "both passes read the surviving row, so the collapse only ever adds a read" + data.findAll { it.volume == 3.25d }*.position == [9, 13] } } From 374a38e49809f63a5b335b7f46d9683f2140911a Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 13:51:07 +0200 Subject: [PATCH 07/13] Add FINE logging to DistroEnergyHandler for debugging Logs config on construction, first-run schedule time and interval on deploy, run start with market day and horizon, per-day datapoint count, and submission confirmation. --- .../distroenergy/DistroEnergyHandler.java | 47 ++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index 8d7bf1e..138af63 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -135,6 +135,16 @@ public DistroEnergyHandler( client.close(); throw e; } + + LOG.fine( + "Configured Distro Energy handler for portfolio " + + portfolio + + ": baseUrl=" + + distroEnergyBaseUrl + + ", marketZone=" + + marketZone + + ", requestIntervalMinutes=" + + requestIntervalMinutes); } /** @@ -146,10 +156,21 @@ public DistroEnergyHandler( * running. */ public void deploy() { + long firstRequestDelayMillis = getFirstRequestDelayMillis(); + LOG.fine( + "First Distro Energy day-ahead submission for portfolio " + + portfolio + + " scheduled at " + + Instant.ofEpochMilli(timerService.getCurrentTimeMillis() + firstRequestDelayMillis) + + " (delay " + + firstRequestDelayMillis + + "ms), repeating every " + + requestIntervalMinutes + + " minute(s)"); nextRequestFuture = scheduledExecutorService.scheduleAtFixedRate( this::submitDayAheadForecasts, - getFirstRequestDelayMillis(), + firstRequestDelayMillis, Duration.ofMinutes(requestIntervalMinutes).toMillis(), TimeUnit.MILLISECONDS); LOG.info("Deployed Distro Energy handler for portfolio: " + portfolio); @@ -157,6 +178,14 @@ public void deploy() { protected void submitDayAheadForecasts() { LocalDate firstDay = timerService.getNow().atZone(marketZone).toLocalDate().plusDays(1); + LOG.fine( + "Starting day-ahead submission run for portfolio " + + portfolio + + "; first market day " + + firstDay + + ", horizon up to " + + MAX_DAYS_AHEAD + + " day(s)"); int submitted = 0; for (int day = 0; day < MAX_DAYS_AHEAD; day++) { @@ -216,6 +245,14 @@ protected boolean submitDayAheadForecast(LocalDate marketDate) { AssetDatapointIntervalQuery.Formula.AVG, true)); + LOG.fine( + "Queried " + + datapoints.size() + + " predicted datapoint(s) for portfolio " + + portfolio + + " and day " + + marketDate); + List submissionData = buildSubmissionData(marketDate, marketZone, storageZone, datapoints); @@ -239,6 +276,14 @@ protected boolean submitDayAheadForecast(LocalDate marketDate) { submissionData.toArray(new SubmissionData[0]), Long.parseLong(marketDate.format(BASIC_ISO_DATE)), timerService.getCurrentTimeMillis())); + LOG.fine( + "Submitted day-ahead forecast for portfolio " + + portfolio + + " and day " + + marketDate + + " (" + + submissionData.size() + + " interval(s))"); return true; } From 79a5e8fe5d4b8045ae49c092f734c6e20c4efc8f Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 13:51:28 +0200 Subject: [PATCH 08/13] Fix first day-ahead submission to always land on :30 getFirstRequestDelayMillis clamped to zero and fired immediately whenever deploy happened after the current hour's :30 mark, instead of waiting for the next one. Roll forward to the next hour's :30 when already past it. --- .../distroenergy/DistroEnergyHandler.java | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index 138af63..f6c88ca 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -400,17 +400,19 @@ static List buildSubmissionData( } protected long getFirstRequestDelayMillis() { - long firstRequestMillis = - timerService - .getNow() - .truncatedTo(ChronoUnit.HOURS) - .plus(30, ChronoUnit.MINUTES) - .toEpochMilli(); - return Math.max(0L, firstRequestMillis - timerService.getCurrentTimeMillis()); + Instant now = timerService.getNow(); + Instant nextHalfHour = now.truncatedTo(ChronoUnit.HOURS).plus(30, ChronoUnit.MINUTES); + // Already past this hour's :30 mark: roll to next hour's, so the first run always lands on :30 + // rather than firing immediately. + if (!nextHalfHour.isAfter(now)) { + nextHalfHour = nextHalfHour.plus(1, ChronoUnit.HOURS); + } + return Math.max(0L, nextHalfHour.toEpochMilli() - timerService.getCurrentTimeMillis()); } public void undeploy() { if (nextRequestFuture != null) { + LOG.fine("Cancelling scheduled day-ahead submissions for portfolio " + portfolio); // Interrupt to abort any in-flight HTTP call before closing the client nextRequestFuture.cancel(true); nextRequestFuture = null; From 3f60b087cda8e96bbcdb8fe1740bfb8732d399fe Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 13:53:27 +0200 Subject: [PATCH 09/13] Add attribute event processing for Distro energy asset --- .../ems/manager/EmsOptimisationService.java | 61 ++++++++++++++++--- 1 file changed, 51 insertions(+), 10 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 35699dd..2fe629a 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -194,7 +194,8 @@ public void start(Container container) throws Exception { String[] assetTypes = { EmsElectricityBatteryAsset.DESCRIPTOR.getName(), EmsEnergyOptimisationAsset.DESCRIPTOR.getName(), - EmsGOPACSAsset.DESCRIPTOR.getName() + EmsGOPACSAsset.DESCRIPTOR.getName(), + EmsDistroEnergyAsset.DESCRIPTOR.getName(), }; // Listen to attribute events of listed asset types @@ -471,15 +472,22 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { } }); } else if (persistenceEvent.getEntity() instanceof EmsDistroEnergyAsset emsDistroEnergyAsset) { - switch (persistenceEvent.getCause()) { - case DELETE -> stopDistroEnergyHandler(emsDistroEnergyAsset.getId()); - case CREATE -> startDistroEnergyHandler(emsDistroEnergyAsset); - // Redeploy so a changed portfolio or parent takes effect. - case UPDATE -> { - stopDistroEnergyHandler(emsDistroEnergyAsset.getId()); - startDistroEnergyHandler(emsDistroEnergyAsset); - } - } + emsDistroEnergyAsset + .getPortfolio() + .ifPresent( + portfolio -> { + if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { + stopDistroEnergyHandler(portfolio); + } + if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { + startDistroEnergyHandler(emsDistroEnergyAsset); + } + if (persistenceEvent.getCause() == PersistenceEvent.Cause.UPDATE) { + stopDistroEnergyHandler(portfolio); + startDistroEnergyHandler(emsDistroEnergyAsset); + } + } + ); } } @@ -495,6 +503,10 @@ private void processAttributeEvent(AttributeEvent attributeEvent) { processAttributeEventEmsGOPACSAsset(attributeEvent); return; } + + if (assetType.equals(EmsDistroEnergyAsset.DESCRIPTOR.getName())) { + processAttributeEventEmsDistroEnergyAsset(attributeEvent); + } } private void processAttributeEventEmsEnergyOptimisationAsset(AttributeEvent attributeEvent) { @@ -892,6 +904,35 @@ private void processAttributeEventEmsGOPACSAsset(AttributeEvent attributeEvent) } } + private void processAttributeEventEmsDistroEnergyAsset(AttributeEvent attributeEvent) { + String assetId = attributeEvent.getId(); + + // Get asset from database + EmsDistroEnergyAsset emsDistroEnergyAsset = (EmsDistroEnergyAsset) services.getAssetStorageService().find(assetId); + + // Check if asset exists + if (emsDistroEnergyAsset == null) { + return; + } + + String attributeName = attributeEvent.getName(); + + if (attributeName.equals(EmsDistroEnergyAsset.PORTFOLIO.getName())) { + attributeEvent + .getOldValue(String.class) + .ifPresent( + oldPortfolio -> { + stopDistroEnergyHandler(assetId); + }); + attributeEvent + .getValue(String.class) + .ifPresent( + portfolio -> { + startDistroEnergyHandler(emsDistroEnergyAsset); + }); + } + } + private void updatePowerLimitProfileManualForecasts( EmsEnergyOptimisationAsset energyOptimisationAsset) { String logPrefix = From 5b063a741158ba748b5812228be564f75767bd28 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 14:24:03 +0200 Subject: [PATCH 10/13] Fix spotless formatting in EmsOptimisationService CI's spotlessJavaCheck failed on the attribute-event handler for EmsDistroEnergyAsset: an over-length line and over-indented lambda bodies. Applied via spotlessJavaApply, no behaviour change. --- .../ems/manager/EmsOptimisationService.java | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 2fe629a..2f4e2af 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -908,7 +908,8 @@ private void processAttributeEventEmsDistroEnergyAsset(AttributeEvent attributeE String assetId = attributeEvent.getId(); // Get asset from database - EmsDistroEnergyAsset emsDistroEnergyAsset = (EmsDistroEnergyAsset) services.getAssetStorageService().find(assetId); + EmsDistroEnergyAsset emsDistroEnergyAsset = + (EmsDistroEnergyAsset) services.getAssetStorageService().find(assetId); // Check if asset exists if (emsDistroEnergyAsset == null) { @@ -919,17 +920,17 @@ private void processAttributeEventEmsDistroEnergyAsset(AttributeEvent attributeE if (attributeName.equals(EmsDistroEnergyAsset.PORTFOLIO.getName())) { attributeEvent - .getOldValue(String.class) - .ifPresent( - oldPortfolio -> { - stopDistroEnergyHandler(assetId); - }); + .getOldValue(String.class) + .ifPresent( + oldPortfolio -> { + stopDistroEnergyHandler(assetId); + }); attributeEvent - .getValue(String.class) - .ifPresent( - portfolio -> { - startDistroEnergyHandler(emsDistroEnergyAsset); - }); + .getValue(String.class) + .ifPresent( + portfolio -> { + startDistroEnergyHandler(emsDistroEnergyAsset); + }); } } From 390b1267ad925255fe02235bc564f2c622f018e7 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 14:24:26 +0200 Subject: [PATCH 11/13] Fix Distro Energy handler leak on delete and update processAssetChange stopped the handler by portfolio, but distroEnergyHandlerMap is keyed by asset id (same as the map's own put/remove and the attribute-event path). Map.remove with the wrong key is a silent no-op, so: - DELETE never undeployed the handler: the ResteasyClient and scheduled task leaked, and it kept POSTing forecasts for an asset that no longer exists. - UPDATE never stopped the old handler before creating a new one: both kept submitting concurrently, one of them under the stale portfolio if that's what changed. Stop by asset id instead, which is always available from the entity regardless of whether the portfolio attribute happens to be present, so DELETE no longer needs to be gated on it either. Added EmsOptimisationServiceDistroEnergyTest to cover CREATE/UPDATE/ DELETE handler lifecycle. Uses a recording DistroEnergyHandler subclass (same pattern as RecordingGOPACSHandler in GOPACSHandlerTest) rather than mocking the class directly, so no extra test dependency is needed for a concrete class with no no-arg constructor: the fake Container just has to be enough for the real constructor to succeed, and deploy()/undeploy() are overridden to record calls instead of scheduling or closing a real client. --- .../ems/manager/EmsOptimisationService.java | 30 ++-- ...OptimisationServiceDistroEnergyTest.groovy | 157 ++++++++++++++++++ 2 files changed, 171 insertions(+), 16 deletions(-) create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 2f4e2af..cb0f8e6 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -472,22 +472,20 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { } }); } else if (persistenceEvent.getEntity() instanceof EmsDistroEnergyAsset emsDistroEnergyAsset) { - emsDistroEnergyAsset - .getPortfolio() - .ifPresent( - portfolio -> { - if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { - stopDistroEnergyHandler(portfolio); - } - if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { - startDistroEnergyHandler(emsDistroEnergyAsset); - } - if (persistenceEvent.getCause() == PersistenceEvent.Cause.UPDATE) { - stopDistroEnergyHandler(portfolio); - startDistroEnergyHandler(emsDistroEnergyAsset); - } - } - ); + // distroEnergyHandlerMap is keyed by asset id, not portfolio: stop by asset id so this + // actually finds the handler to remove, and so DELETE cleans up even if the portfolio + // attribute is absent on the entity snapshot. + String assetId = emsDistroEnergyAsset.getId(); + if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { + stopDistroEnergyHandler(assetId); + } + if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { + startDistroEnergyHandler(emsDistroEnergyAsset); + } + if (persistenceEvent.getCause() == PersistenceEvent.Cause.UPDATE) { + stopDistroEnergyHandler(assetId); + startDistroEnergyHandler(emsDistroEnergyAsset); + } } } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy new file mode 100644 index 0000000..0283176 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy @@ -0,0 +1,157 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager + +import org.openremote.container.timer.TimerService +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset +import org.openremote.extension.ems.agent.EmsEnergyOptimisationAsset +import org.openremote.extension.ems.manager.distroenergy.DistroEnergyHandler +import org.openremote.manager.asset.AssetStorageService +import org.openremote.manager.datapoint.AssetPredictedDatapointService +import org.openremote.model.Container +import org.openremote.model.PersistenceEvent +import org.openremote.model.attribute.AttributeRef +import org.openremote.model.util.ValueUtil +import spock.lang.Specification + +import java.util.concurrent.ScheduledExecutorService + +/** + * Exercises {@link EmsOptimisationService#processAssetChange} for {@link EmsDistroEnergyAsset} + * persistence events, via a recording {@code DistroEnergyHandler} subclass -- the same pattern + * {@code GOPACSHandlerTest} uses for {@code GOPACSHandler} -- rather than mocking the class. + * {@code DistroEnergyHandler} builds a real {@code ResteasyClient} from its {@code Container} in + * the constructor, so the fake {@code Container} just needs to be enough for that constructor to + * succeed; {@code deploy()}/{@code undeploy()} are overridden so nothing is actually scheduled or + * submitted. + * + * {@code distroEnergyHandlerMap} is keyed by asset id (it has no external routing key, and the + * asset id survives a portfolio edit -- see the map's own comment). These tests catch stop calls + * made with the wrong key: a wrong key makes {@code Map.remove} a silent no-op, so the old + * handler is never undeployed and keeps submitting concurrently with its replacement. + */ +class EmsOptimisationServiceDistroEnergyTest extends Specification { + + static final String ASSET_ID = "distroAsset1" + static final String PARENT_ID = "optimisationAsset1" + + List createdHandlers + EmsOptimisationService service + + def setupSpec() { + // Populates the asset model registry (asset type -> attribute descriptors) from this + // extension's own AssetModelProvider SPI registration, which real asset construction needs. + // A plain Specification has no container to do this at startup, unlike production and unlike + // ManagerContainerTrait-based tests. + ValueUtil.initialise(null) + } + + def setup() { + def assetStorageService = Mock(AssetStorageService) + assetStorageService.find(PARENT_ID, false, EmsEnergyOptimisationAsset.class) >> + new EmsEnergyOptimisationAsset("parent") + + def handlerContainer = Stub(Container) { + getConfig() >> [(DistroEnergyHandler.DISTRO_ENERGY_CLIENT_KEY): "test-key"] + getService(TimerService) >> Stub(TimerService) + getScheduledExecutor() >> Stub(ScheduledExecutorService) + getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) + } + + createdHandlers = [] + + service = new EmsOptimisationService() + service.services = Services.builder().withAssetStorageService(assetStorageService).build() + service.distroEnergyHandlerFactory = new DistroEnergyHandler.Factory(handlerContainer) { + @Override + DistroEnergyHandler createHandler(AttributeRef powerNetAttributeRef, String portfolio) { + def handler = + new RecordingDistroEnergyHandler(powerNetAttributeRef, portfolio, handlerContainer) + createdHandlers << handler + return handler + } + } + } + + private static EmsDistroEnergyAsset distroAsset(String portfolio = "portfolio-a") { + def asset = new EmsDistroEnergyAsset("distro").setId(ASSET_ID).setParentId(PARENT_ID) + asset.setPortfolio(portfolio) + return asset + } + + private static PersistenceEvent event( + PersistenceEvent.Cause cause, EmsDistroEnergyAsset asset) { + return new PersistenceEvent<>(cause, asset, null, null, null) + } + + def "CREATE deploys a handler for the asset"() { + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distroAsset())) + + then: + createdHandlers.size() == 1 + createdHandlers[0].deployCount == 1 + } + + def "DELETE undeploys the handler keyed by asset id, not portfolio"() { + given: "a deployed handler" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distroAsset())) + def original = createdHandlers[0] + + when: + service.processAssetChange(event(PersistenceEvent.Cause.DELETE, distroAsset())) + + then: "the handler for this asset id is undeployed rather than leaked" + original.undeployCount == 1 + } + + def "UPDATE undeploys the old handler before deploying its replacement"() { + given: "a deployed handler for the original portfolio" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distroAsset("portfolio-a"))) + def original = createdHandlers[0] + + when: "the portfolio changes" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, distroAsset("portfolio-b"))) + + then: "the old handler is stopped and exactly one new handler replaces it" + original.undeployCount == 1 + createdHandlers.size() == 2 + createdHandlers[1].deployCount == 1 + } + + // Records deploy()/undeploy() calls instead of scheduling submissions or closing a real client. + static class RecordingDistroEnergyHandler extends DistroEnergyHandler { + int deployCount = 0 + int undeployCount = 0 + + RecordingDistroEnergyHandler(AttributeRef powerNetAttributeRef, String portfolio, Container container) { + super(powerNetAttributeRef, portfolio, container) + } + + @Override + void deploy() { + deployCount++ + } + + @Override + void undeploy() { + undeployCount++ + } + } +} From bbb6abc2ce4e771c57d3f52c40328c9a97e27abb Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:05:07 +0200 Subject: [PATCH 12/13] Report submission status on the Distro Energy asset Adds two READ_ONLY attributes to EmsDistroEnergyAsset so a stalled handler is visible in the UI without reading logs: - lastSubmission (TIMESTAMP): when the last run that submitted at least one market day finished. - daysSubmitted (POSITIVE_INTEGER): how many market days the last run submitted, which is the forecast horizon in days. This was only a FINE log line before. DistroEnergyHandler writes daysSubmitted at the end of every run and lastSubmission only when something was sent, so a run that found no forecast at all leaves a fresh daysSubmitted of 0 next to a stale lastSubmission, and a handler that stopped running leaves both stale. The handler now takes the Distro Energy asset id, since it reports on that asset rather than on the parent it reads the forecast from, and resolves AssetProcessingService from the container like GOPACSHandler. DistroEnergyHandlerStatusTest covers both cases through a subclass that reports a fixed horizon without touching the API. --- .../ems/agent/EmsDistroEnergyAsset.java | 12 ++ .../ems/manager/EmsOptimisationService.java | 1 + .../distroenergy/DistroEnergyHandler.java | 30 ++++- .../resources/ems/config/asset-types.json | 10 ++ ...OptimisationServiceDistroEnergyTest.groovy | 14 ++- .../DistroEnergyHandlerStatusTest.groovy | 114 ++++++++++++++++++ 6 files changed, 173 insertions(+), 8 deletions(-) create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerStatusTest.groovy diff --git a/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java b/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java index 9a5b43d..4735de1 100644 --- a/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java +++ b/ems/src/main/java/org/openremote/extension/ems/agent/EmsDistroEnergyAsset.java @@ -22,7 +22,9 @@ import java.util.Optional; import org.openremote.model.asset.Asset; import org.openremote.model.asset.AssetDescriptor; +import org.openremote.model.attribute.MetaItem; import org.openremote.model.value.AttributeDescriptor; +import org.openremote.model.value.MetaItemType; import org.openremote.model.value.ValueType; @Entity @@ -31,6 +33,16 @@ public class EmsDistroEnergyAsset extends Asset { public static final AttributeDescriptor PORTFOLIO = new AttributeDescriptor<>("portfolio", ValueType.TEXT); + /** When the last run that submitted at least one market day finished. */ + public static final AttributeDescriptor LAST_SUBMISSION = + new AttributeDescriptor<>( + "lastSubmission", ValueType.TIMESTAMP, new MetaItem<>(MetaItemType.READ_ONLY)); + + /** Number of market days the last run submitted, i.e. the forecast horizon in days. */ + public static final AttributeDescriptor DAYS_SUBMITTED = + new AttributeDescriptor<>( + "daysSubmitted", ValueType.POSITIVE_INTEGER, new MetaItem<>(MetaItemType.READ_ONLY)); + public static final AssetDescriptor DESCRIPTOR = new AssetDescriptor<>("transmission-tower", null, EmsDistroEnergyAsset.class); diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index cb0f8e6..efcec78 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -418,6 +418,7 @@ private void startDistroEnergyHandler(EmsDistroEnergyAsset distroEnergyAsset) { try { handler = distroEnergyHandlerFactory.createHandler( + assetId, new AttributeRef( energyOptimisationAssetId, EmsEnergyOptimisationAsset.POWER_NET.getName()), portfolio); diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index f6c88ca..4251add 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -42,14 +42,18 @@ import java.util.logging.Logger; import org.jboss.resteasy.client.jaxrs.ResteasyClient; import org.openremote.container.timer.TimerService; +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset; import org.openremote.extension.ems.manager.distroenergy.dto.DayAheadSubmission; import org.openremote.extension.ems.manager.distroenergy.dto.SubmissionData; +import org.openremote.manager.asset.AssetProcessingService; import org.openremote.manager.datapoint.AssetPredictedDatapointService; import org.openremote.model.Container; +import org.openremote.model.attribute.AttributeEvent; import org.openremote.model.attribute.AttributeRef; import org.openremote.model.datapoint.ValueDatapoint; import org.openremote.model.datapoint.query.AssetDatapointIntervalQuery; import org.openremote.model.syslog.SyslogCategory; +import org.openremote.model.value.AttributeDescriptor; public class DistroEnergyHandler { @@ -72,6 +76,9 @@ public class DistroEnergyHandler { */ protected static final int MAX_DAYS_AHEAD = 14; + /** The {@link EmsDistroEnergyAsset} this handler reports its status on. */ + protected final String assetId; + protected final AttributeRef powerNetAttributeRef; protected final String distroEnergyBaseUrl; protected final String portfolio; @@ -84,6 +91,7 @@ public class DistroEnergyHandler { protected final TimerService timerService; protected final ScheduledExecutorService scheduledExecutorService; protected final AssetPredictedDatapointService assetPredictedDatapointService; + protected final AssetProcessingService assetProcessingService; protected ScheduledFuture nextRequestFuture; public static class Factory { @@ -93,13 +101,15 @@ public Factory(Container container) { this.container = container; } - public DistroEnergyHandler createHandler(AttributeRef powerNetAttributeRef, String portfolio) { - return new DistroEnergyHandler(powerNetAttributeRef, portfolio, container); + public DistroEnergyHandler createHandler( + String assetId, AttributeRef powerNetAttributeRef, String portfolio) { + return new DistroEnergyHandler(assetId, powerNetAttributeRef, portfolio, container); } } public DistroEnergyHandler( - AttributeRef powerNetAttributeRef, String portfolio, Container container) { + String assetId, AttributeRef powerNetAttributeRef, String portfolio, Container container) { + this.assetId = assetId; this.powerNetAttributeRef = powerNetAttributeRef; this.portfolio = portfolio; @@ -107,6 +117,7 @@ public DistroEnergyHandler( this.scheduledExecutorService = container.getScheduledExecutor(); this.assetPredictedDatapointService = container.getService(AssetPredictedDatapointService.class); + this.assetProcessingService = container.getService(AssetProcessingService.class); this.distroEnergyBaseUrl = container.getConfig().getOrDefault(DISTRO_ENERGY_BASE_URL, DISTRO_ENERGY_BASE_URL_DEFAULT); @@ -208,6 +219,14 @@ protected void submitDayAheadForecasts() { } } + // Status on the asset, so a stalled handler is visible without the logs. daysSubmitted is + // written on every run; lastSubmission only when something was sent, so a run that found no + // forecast at all leaves a fresh daysSubmitted of 0 next to a stale lastSubmission. + sendAttributeEvent(EmsDistroEnergyAsset.DAYS_SUBMITTED, submitted); + if (submitted > 0) { + sendAttributeEvent(EmsDistroEnergyAsset.LAST_SUBMISSION, timerService.getCurrentTimeMillis()); + } + // Reaching the ceiling means the horizon looked unbounded, which the forecast producer cannot // legitimately do; the run was truncated and later days were not sent. if (submitted >= MAX_DAYS_AHEAD) { @@ -399,6 +418,11 @@ static List buildSubmissionData( return submissionData; } + protected void sendAttributeEvent(AttributeDescriptor attribute, T value) { + assetProcessingService.sendAttributeEvent( + new AttributeEvent(assetId, attribute.getName(), value), getClass().getSimpleName()); + } + protected long getFirstRequestDelayMillis() { Instant now = timerService.getNow(); Instant nextHalfHour = now.truncatedTo(ChronoUnit.HOURS).plus(30, ChronoUnit.MINUTES); diff --git a/ems/src/main/resources/ems/config/asset-types.json b/ems/src/main/resources/ems/config/asset-types.json index 9ac6e89..be21477 100644 --- a/ems/src/main/resources/ems/config/asset-types.json +++ b/ems/src/main/resources/ems/config/asset-types.json @@ -286,6 +286,16 @@ "include": ["portfolio"] } }, + { + "type": "info", + "title": "Status", + "properties": { + "include": [] + }, + "attributes": { + "include": ["lastSubmission", "daysSubmitted"] + } + }, { "type": "history", "column": 1 diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy index 0283176..8a91678 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceDistroEnergyTest.groovy @@ -22,6 +22,7 @@ import org.openremote.container.timer.TimerService import org.openremote.extension.ems.agent.EmsDistroEnergyAsset import org.openremote.extension.ems.agent.EmsEnergyOptimisationAsset import org.openremote.extension.ems.manager.distroenergy.DistroEnergyHandler +import org.openremote.manager.asset.AssetProcessingService import org.openremote.manager.asset.AssetStorageService import org.openremote.manager.datapoint.AssetPredictedDatapointService import org.openremote.model.Container @@ -72,6 +73,7 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { getService(TimerService) >> Stub(TimerService) getScheduledExecutor() >> Stub(ScheduledExecutorService) getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) + getService(AssetProcessingService) >> Stub(AssetProcessingService) } createdHandlers = [] @@ -80,9 +82,10 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { service.services = Services.builder().withAssetStorageService(assetStorageService).build() service.distroEnergyHandlerFactory = new DistroEnergyHandler.Factory(handlerContainer) { @Override - DistroEnergyHandler createHandler(AttributeRef powerNetAttributeRef, String portfolio) { - def handler = - new RecordingDistroEnergyHandler(powerNetAttributeRef, portfolio, handlerContainer) + DistroEnergyHandler createHandler( + String assetId, AttributeRef powerNetAttributeRef, String portfolio) { + def handler = new RecordingDistroEnergyHandler( + assetId, powerNetAttributeRef, portfolio, handlerContainer) createdHandlers << handler return handler } @@ -140,8 +143,9 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { int deployCount = 0 int undeployCount = 0 - RecordingDistroEnergyHandler(AttributeRef powerNetAttributeRef, String portfolio, Container container) { - super(powerNetAttributeRef, portfolio, container) + RecordingDistroEnergyHandler( + String assetId, AttributeRef powerNetAttributeRef, String portfolio, Container container) { + super(assetId, powerNetAttributeRef, portfolio, container) } @Override diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerStatusTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerStatusTest.groovy new file mode 100644 index 0000000..34ed167 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandlerStatusTest.groovy @@ -0,0 +1,114 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager.distroenergy + +import org.openremote.container.timer.TimerService +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.datapoint.AssetPredictedDatapointService +import org.openremote.model.Container +import org.openremote.model.attribute.AttributeEvent +import org.openremote.model.attribute.AttributeRef +import spock.lang.Specification + +import java.time.Instant +import java.time.LocalDate +import java.util.concurrent.ScheduledExecutorService + +/** + * Checks the status attributes a submission run writes to the Distro Energy asset. The HTTP side + * is bypassed by overriding {@link DistroEnergyHandler#submitDayAheadForecast} to report a horizon + * of a fixed number of days, so only the bookkeeping around the loop is under test. + */ +class DistroEnergyHandlerStatusTest extends Specification { + + static final String ASSET_ID = "distroAsset1" + static final long NOW = Instant.parse("2026-09-16T10:00:00Z").toEpochMilli() + + AssetProcessingService assetProcessingService + Container container + FixedHorizonHandler handler + + def setup() { + assetProcessingService = Mock(AssetProcessingService) + def timerService = Stub(TimerService) { + getCurrentTimeMillis() >> NOW + getNow() >> Instant.ofEpochMilli(NOW) + } + container = Stub(Container) { + getConfig() >> [(DistroEnergyHandler.DISTRO_ENERGY_CLIENT_KEY): "test-key"] + getService(TimerService) >> timerService + getScheduledExecutor() >> Stub(ScheduledExecutorService) + getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) + getService(AssetProcessingService) >> assetProcessingService + } + } + + def cleanup() { + // Closes the real ResteasyClient the constructor built. + handler?.undeploy() + } + + def "a run that submitted days writes daysSubmitted and lastSubmission"() { + given: "a handler whose forecast horizon is two market days" + handler = new FixedHorizonHandler(2, container) + + when: + handler.submitDayAheadForecasts() + + then: + 1 * assetProcessingService.sendAttributeEvent({ AttributeEvent e -> + e.id == ASSET_ID && e.name == EmsDistroEnergyAsset.DAYS_SUBMITTED.name && e.value.get() == 2 + }, _) + 1 * assetProcessingService.sendAttributeEvent({ AttributeEvent e -> + e.id == ASSET_ID && e.name == EmsDistroEnergyAsset.LAST_SUBMISSION.name && e.value.get() == NOW + }, _) + 0 * assetProcessingService._ + } + + def "a run that found no forecast writes daysSubmitted 0 and leaves lastSubmission alone"() { + given: "a handler whose forecast horizon is empty" + handler = new FixedHorizonHandler(0, container) + + when: + handler.submitDayAheadForecasts() + + then: + 1 * assetProcessingService.sendAttributeEvent({ AttributeEvent e -> + e.name == EmsDistroEnergyAsset.DAYS_SUBMITTED.name && e.value.get() == 0 + }, _) + 0 * assetProcessingService._ + } + + // Reports a forecast covering exactly `horizonDays` market days without touching the API. + static class FixedHorizonHandler extends DistroEnergyHandler { + final int horizonDays + int calls = 0 + + FixedHorizonHandler(int horizonDays, Container container) { + super(ASSET_ID, new AttributeRef("parent", "powerNet"), "portfolio-a", container) + this.horizonDays = horizonDays + } + + @Override + protected boolean submitDayAheadForecast(LocalDate marketDate) { + return calls++ Date: Wed, 16 Sep 2026 17:05:07 +0200 Subject: [PATCH 13/13] Log the DST-hour collapse once per day instead of once per position buildSubmissionData warned for every position whose storage key had already been read, so on the fall-back day it emitted a line per quarter-hour in the repeated hour, per portfolio, per run. Count the collapsed positions and log a single summary for the day, with the count, so the condition stays visible without flooding the log. The root cause is upstream (openremote/openremote#3292). --- .../distroenergy/DistroEnergyHandler.java | 25 ++++++++++++------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java index 4251add..5f641cf 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroEnergyHandler.java @@ -354,6 +354,7 @@ static List buildSubmissionData( List submissionData = new ArrayList<>(); Set keysRead = new HashSet<>(); int missing = 0; + int collapsed = 0; int lastRealPosition = 0; int position = 1; @@ -366,15 +367,7 @@ static List buildSubmissionData( } else { lastRealPosition = position; if (!keysRead.add(storageKey)) { - LOG.warning( - "Day-ahead position " - + position - + " on " - + marketDate - + " reuses the value stored at " - + storageKey - + " because the repeated DST hour" - + " collapses onto one predicted datapoint row (openremote/openremote#3292)"); + collapsed++; } } @@ -387,6 +380,20 @@ static List buildSubmissionData( return List.of(); } + if (collapsed > 0) { + // One line per day rather than per position: on the fall-back day every position in the + // repeated hour reads the same stored row. + LOG.warning( + "Day-ahead submission for " + + marketDate + + " has " + + collapsed + + " of " + + submissionData.size() + + " positions reusing a value from the repeated DST hour, which collapses onto one" + + " predicted datapoint row (openremote/openremote#3292)"); + } + int trailingMissing = submissionData.size() - lastRealPosition; int interiorMissing = missing - trailingMissing;