diff --git a/ems/docs/DistroEnergyIntegration.md b/ems/docs/DistroEnergyIntegration.md index 4773fac..3f26591 100644 --- a/ems/docs/DistroEnergyIntegration.md +++ b/ems/docs/DistroEnergyIntegration.md @@ -6,8 +6,14 @@ participants. Within Distro each participant is a **portfolio**. A portfolio submits day-ahead orders (volumes per 15-minute interval, optionally with a price range) and Distro places them on the market. -OpenRemote uses the Distro **Day-Ahead Order Submissions** API to submit the net power forecast of an EMS site as -day-ahead market orders. The API rules that shape the integration: +OpenRemote uses two Distro APIs, each optional and independent of the other: + +- **Day-Ahead Order Submissions** submits the net power forecast of an EMS site as day-ahead market orders (this + section) +- **GOPACS Connector** places redispatch orders in response to GOPACS congestion announcements (see + [GOPACS Orders](#gopacs-orders)) + +The day-ahead API rules that shape the integration: - Orders may be submitted up to five days in advance - A new submission replaces any earlier submission for the same delivery day @@ -156,6 +162,148 @@ portfolio … and day …` and the run continues with the next day. The schedule - Stopping the handler cancels the schedule, interrupts any request in flight and closes the HTTP client - Use `daysSubmitted` and `lastSubmission` to spot a stalled integration without reading the logs +### GOPACS Orders + +Distro relays orders placed through its GOPACS Connector API to ETPA, where GOPACS can match them to solve grid +congestion. OpenRemote does not poll GOPACS for this: the announcement comes from an `Ems GOPACS Asset` with +redispatch enabled (see [GOPACS Integration](GopacsIntegration.md)), and `EmsOptimisationService` copies it to the +Distro asset next to it. Neither handler references the other, so the Distro asset needs no portfolio or day-ahead +setup. + +#### Configuration + +| Variable | Required | Description | +| ----------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------- | +| `DISTRO_GOPACS_CLIENT_ID` | Yes | OAuth client ID provided by Distro | +| `DISTRO_GOPACS_CLIENT_SECRET` | Yes | OAuth client secret provided by Distro | +| `DISTRO_GOPACS_TENANT_ID` | Yes | Distro tenant, e.g. `slimstrandnet` | +| `DISTRO_GOPACS_URL` | No | Base URL of the API (default: `https://api.distro.energy/v1`, development: `https://api.dev.distro.energy/v1`) | +| `DISTRO_GOPACS_TOKEN_URL` | No | OAuth token endpoint (default: `https://auth.distro.energy/oauth2/token`, development: `https://auth.dev.distro.energy/oauth2/token`) | + +The handler does not start without the three required variables. Distro must also whitelist the manager's outbound IP +address. Tokens are requested with client credentials and scope `distro-api/gopacs.access`, cached and refreshed +shortly before they expire. + +#### Asset Setup + +1. Place the `Ems Distro Energy Asset` under the same `Ems Energy Optimisation Asset` as the `Ems GOPACS Asset`. The + announcement is copied from that GOPACS asset, and quantities are pre-filled from that parent's batteries +2. Set `gopacsEnabled` to `true` on the `Ems Distro Energy Asset` + +These attributes are copied from the `Ems GOPACS Asset`: + +| `Ems GOPACS Asset` attribute | Copied to `Ems Distro Energy Asset` attribute | +| ---------------------------- | --------------------------------------------- | +| `contractedEAN` | `gopacsEan` | +| `redispatchAnnouncementId` | `gopacsAnnouncementId` | +| `redispatchStartTime` | `gopacsStartTime` | +| `redispatchEndTime` | `gopacsEndTime` | +| `redispatchBidValidityEnd` | `gopacsBidValidityEnd` | +| `redispatchEanEffectivity` | `gopacsEanEffectivity` | + +They are copied when one of them changes on the GOPACS asset, when either asset is saved, and when the manager starts. +Only values that differ are written, so the redispatch poller rewriting the same announcement on each poll leaves the +Distro asset alone. Nothing is copied while the parent has more than one `Ems GOPACS Asset`. Attribute links set up +for an earlier version are no longer needed and can be removed. + +| Attribute | Value Type | Read-only | Purpose | +| ------------------------------------ | ---------------- | --------- | ------------------------------------------------------------------------------------------------ | +| `gopacsEnabled` | Boolean | | Starts or stops the GOPACS order handler. Stopping cancels the open orders | +| `gopacsEan` … `gopacsEanEffectivity` | Copied | ✓ | Announcement inputs, see the table above | +| `gopacsOrderType` | Enum | | `BUY` or `SELL` (dropdown). Pre-filled from the announcement, may be overridden | +| `gopacsOrderQuantity` | Number (kW) | | Constant power over the whole window. Pre-filled from the batteries, may be overridden | +| `gopacsOrderPrice` | Number (EUR/MWh) | | Set by the operator, 0 or more. Changing it replaces the open orders of the current announcement | +| `gopacsConfirmOrder` | Boolean | | Set to `true` to place the order; reset by the handler | +| `gopacsCancelOrders` | Boolean | | Set to `true` to cancel every open order; reset by the handler | +| `gopacsOrders` | JSON object | ✓ | Orders still being tracked, by reference, as last returned by Distro | +| `gopacsOrderHistory` | JSON object | ✓ | One data point per order action, status change, refused confirmation or price change | +| `gopacsSettlementHistory` | JSON object | ✓ | One data point per settled trade | +| `gopacsLastSync` | Timestamp | ✓ | When the tracked orders were last refreshed | + +#### Workflow + +```mermaid +sequenceDiagram + participant G as Ems GOPACS Asset + participant D as Ems Distro Energy Asset + actor Op as Operator + participant API as Distro GOPACS API + + G->>D: copied by EmsOptimisationService (announcement, window, effectivity) + D->>D: Pre-fill gopacsOrderType and gopacsOrderQuantity + Op->>D: Set gopacsOrderPrice, toggle gopacsConfirmOrder + D->>API: POST /gopacs/{tenant}/orders + loop Every 5 minutes while orders are tracked + D->>API: GET /gopacs/{tenant}/orders/{reference} + D->>API: GET /settlements/gopacs/{tenant} (once SETTLED) + end + Op->>D: Change gopacsOrderPrice + D->>API: PUT /gopacs/{tenant}/orders/{reference} + G->>D: announcement closes or changes + D->>API: DELETE /gopacs/{tenant}/orders/{reference} +``` + +#### Order Sizing + +**Order type** comes from the GOPACS EAN effectivity category, `BUY__SELL_`, ranked +`NONE` < `VERY_LOW` < `LOW` < `MODERATE` < `HIGH` < `VERY_HIGH` < `MAXIMUM`. The more effective side wins. +`BUY_NONE_SELL_NONE` suggests nothing. On `BUY_MODERATE_SELL_MODERATE` the side the batteries can offer more of wins, +`BUY` on a draw. `BUY` means consume more (charge) and `SELL` means consume less (discharge). + +**Quantity** is the constant power the batteries under the parent optimisation asset can hold for the whole window: + +- Only batteries allowed to charge (`BUY`) or discharge (`SELL`) and not `disconnected` count +- Power: the sum of `chargePowerMaximum` or `dischargePowerMaximum` +- Energy: the sum of the headroom to `energyLevelPercentageMaximum` (`BUY`, divided by `chargeEfficiency`) or to + `energyLevelPercentageMinimum` (`SELL`, multiplied by `dischargeEfficiency`), divided by the window duration +- The lower of the two, rounded down to 100 kW steps. Distro's minimum order is 0.1 MW + +#### Placing an Order + +On `gopacsConfirmOrder` the handler rounds the quantity down to 100 kW steps and the price to cents, as Distro takes at +most one decimal in MW and two in EUR/MWh. It then checks that there is an announcement, an 18-digit EAN, a `BUY` or +`SELL` type, a quantity of at least 100 kW, a price, and a non-empty window that starts more than 15 +minutes from now (Distro's latest expiry). When `gopacsBidValidityEnd` is known it must not have passed, as GOPACS no +longer takes bids for the announcement after it. A failed check is logged and recorded as `REFUSED` in `gopacsOrderHistory`. +The order is sent with: + +| Field | Source | +| ---------------------- | ----------------------------------------------------------------------------------------------------------------------------------------- | +| `reference` | A new random UUID | +| `quantity` | `gopacsOrderQuantity` in MW | +| `start`, `end` | `gopacsStartTime`, `gopacsEndTime` in Unix seconds | +| `metadata` | `{"announcementId": }`, which links the order to its announcement | +| `customExpirationTime` | `gopacsBidValidityEnd` rounded down to a quarter-hour, unless it is unknown, passed, or after Distro's default of 15 minutes before start | + +The order is tracked as `PENDING` before it is posted. When Distro rejects it with a `4xx` it is untracked again. On a +timeout or a `5xx` it stays tracked, because Distro may have accepted it: the next poll picks up its real status, or +untracks it when Distro answers `404`. Either way the failure is recorded as `CREATE_FAILED`. + +#### Order Lifecycle + +`PENDING` and `SUBMITTED` orders are open: they are cancelled when their announcement closes or is replaced by another, +or when the operator sets `gopacsCancelOrders`. A price change replaces the `SUBMITTED` orders of the current +announcement. A `LOCKED` order cannot be changed +(Distro answers `423`) and keeps its old price. `MATCHED` orders stay tracked until they are `SETTLED`, at which point +all their trades are recorded in `gopacsSettlementHistory`. Distro makes the trades available as soon as an order is +`SETTLED`, so they are fetched once; a failed fetch is recorded as `SETTLEMENTS_FAILED` and not retried. Orders that are `SETTLED`, `REVERTED`, `REJECTED` or `CANCELED` +are no longer tracked. The tracked orders are saved on the asset, so a restart picks them up again. + +Setting `gopacsEnabled` to `false` or deleting the asset cancels every open order. `LOCKED` and `MATCHED` orders cannot +be cancelled and stay at Distro. After a disable they are tracked again once the handler is re-enabled. After a delete +nothing tracks them any more and their trades are not recorded, so check for them in Distro before deleting an asset. + +#### Components + +``` +distroenergy/ + DistroGopacsHandler.java Pre-fills, places, replaces, cancels and polls orders for one asset + DistroGopacsResource.java RESTEasy client proxy for the GOPACS Connector API + dto/GopacsOrderRequest.java Create and replace body + dto/GopacsOrder.java Order as stored by Distro + dto/GopacsSettlement*.java Settled trades +``` + ### Testing To test against Distro's development environment, set `DISTRO_ENERGY_BASE_URL` to @@ -165,4 +313,6 @@ Automated tests: - `DistroEnergyHandlerTest` — payload building, volume conversion, DST days and gap filling - `DistroEnergyHandlerStatusTest` — status attributes written after a run and the gate closure boundary -- `EmsOptimisationServiceDistroEnergyTest` — handler lifecycle in `EmsOptimisationService` +- `DistroGopacsHandlerTest` — order type and quantity rules, expiration, and the create, replace, cancel and poll flows +- `DistroGopacsHandlerHttpTest` — OAuth token request, bearer reuse and the JSON wire format against WireMock +- `EmsOptimisationServiceDistroEnergyTest` — day-ahead and GOPACS handler lifecycle in `EmsOptimisationService` diff --git a/ems/docs/GopacsIntegration.md b/ems/docs/GopacsIntegration.md index b884dd2..e406017 100755 --- a/ems/docs/GopacsIntegration.md +++ b/ems/docs/GopacsIntegration.md @@ -210,7 +210,7 @@ are read-only. | Announcement | `redispatchEndTime` | Timestamp | - | ✓ | End of the problem period. | | Announcement | `redispatchBidValidityEnd` | Timestamp | - | ✓ | Latest moment a bid can still be submitted for this announcement. | | Announcement | `redispatchRequestedPower` | Number | kW | ✓ | Remaining problem profile, written as predicted data points (15-min ISP grid, 7-day retention). | -| Announcement | `redispatchEanEffectivity` | Text | - | ✓ | Effectivity category in which the contracted EAN was matched (e.g. `THREE_PHASE_NETWORK_REDUCE`). | +| Announcement | `redispatchEanEffectivity` | Text | - | ✓ | Effectivity category in which the contracted EAN was matched (e.g. `BUY_LOW_SELL_HIGH`). | | Announcement | `redispatchRequestAreaBuy` | Text | - | ✓ | DSO-supplied area description for buy orders. | | Announcement | `redispatchRequestAreaSell` | Text | - | ✓ | DSO-supplied area description for sell orders. | | Announcement | `redispatchLastPoll` | timestamp | - | ✓ | Timestamp of the last completed poll cycle (only updated when the API responded). | 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 4735de1..ca58969 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 @@ -18,6 +18,8 @@ */ package org.openremote.extension.ems.agent; +import static org.openremote.model.Constants.*; + import jakarta.persistence.Entity; import java.util.Optional; import org.openremote.model.asset.Asset; @@ -25,7 +27,9 @@ import org.openremote.model.attribute.MetaItem; import org.openremote.model.value.AttributeDescriptor; import org.openremote.model.value.MetaItemType; +import org.openremote.model.value.ValueDescriptor; import org.openremote.model.value.ValueType; +import org.openremote.model.value.ValueType.ObjectMap; @Entity public class EmsDistroEnergyAsset extends Asset { @@ -43,6 +47,94 @@ public class EmsDistroEnergyAsset extends Asset { new AttributeDescriptor<>( "daysSubmitted", ValueType.POSITIVE_INTEGER, new MetaItem<>(MetaItemType.READ_ONLY)); + // --- GOPACS orders: configuration --- + + public static final AttributeDescriptor GOPACS_ENABLED = + new AttributeDescriptor<>("gopacsEnabled", ValueType.BOOLEAN); + + // --- GOPACS orders: announcement inputs, copied from the EmsGOPACSAsset with the same parent --- + + public static final AttributeDescriptor GOPACS_EAN = + new AttributeDescriptor<>( + "gopacsEan", ValueType.TEXT, new MetaItem<>(MetaItemType.READ_ONLY)); + + public static final AttributeDescriptor GOPACS_ANNOUNCEMENT_ID = + new AttributeDescriptor<>( + "gopacsAnnouncementId", ValueType.TEXT, new MetaItem<>(MetaItemType.READ_ONLY)); + + public static final AttributeDescriptor GOPACS_START_TIME = + new AttributeDescriptor<>( + "gopacsStartTime", ValueType.TIMESTAMP, new MetaItem<>(MetaItemType.READ_ONLY)); + + public static final AttributeDescriptor GOPACS_END_TIME = + new AttributeDescriptor<>( + "gopacsEndTime", ValueType.TIMESTAMP, new MetaItem<>(MetaItemType.READ_ONLY)); + + public static final AttributeDescriptor GOPACS_BID_VALIDITY_END = + new AttributeDescriptor<>( + "gopacsBidValidityEnd", ValueType.TIMESTAMP, new MetaItem<>(MetaItemType.READ_ONLY)); + + public static final AttributeDescriptor GOPACS_EAN_EFFECTIVITY = + new AttributeDescriptor<>( + "gopacsEanEffectivity", ValueType.TEXT, new MetaItem<>(MetaItemType.READ_ONLY)); + + // --- GOPACS orders: next order, pre-filled by the handler and operator-overridable --- + + public enum GopacsOrderTypeValueType { + BUY, + SELL + } + + public static final ValueDescriptor GOPACS_ORDER_TYPE_VALUE_TYPE = + new ValueDescriptor<>("gopacsOrderTypeValueType", GopacsOrderTypeValueType.class); + + public static final AttributeDescriptor GOPACS_ORDER_TYPE = + new AttributeDescriptor<>("gopacsOrderType", GOPACS_ORDER_TYPE_VALUE_TYPE); + + public static final AttributeDescriptor GOPACS_ORDER_QUANTITY = + new AttributeDescriptor<>("gopacsOrderQuantity", ValueType.NUMBER) + .withUnits(UNITS_KILO, UNITS_WATT); + + public static final AttributeDescriptor GOPACS_ORDER_PRICE = + new AttributeDescriptor<>("gopacsOrderPrice", ValueType.POSITIVE_NUMBER) + .withUnits("EUR", UNITS_PER, UNITS_MEGA, UNITS_WATT, UNITS_HOUR); + + // --- GOPACS orders: operator actions, reset by the handler --- + + public static final AttributeDescriptor GOPACS_CONFIRM_ORDER = + new AttributeDescriptor<>("gopacsConfirmOrder", ValueType.BOOLEAN); + + public static final AttributeDescriptor GOPACS_CANCEL_ORDERS = + new AttributeDescriptor<>("gopacsCancelOrders", ValueType.BOOLEAN); + + // --- GOPACS orders: status --- + + /** Orders still being tracked, keyed by reference. Terminal orders drop out. */ + public static final AttributeDescriptor GOPACS_ORDERS = + new AttributeDescriptor<>( + "gopacsOrders", ValueType.JSON_OBJECT, new MetaItem<>(MetaItemType.READ_ONLY)); + + /** One data point per order action and status change. */ + public static final AttributeDescriptor GOPACS_ORDER_HISTORY = + new AttributeDescriptor<>( + "gopacsOrderHistory", + ValueType.JSON_OBJECT, + new MetaItem<>(MetaItemType.READ_ONLY), + new MetaItem<>(MetaItemType.STORE_DATA_POINTS)); + + /** One data point per settled trade. */ + public static final AttributeDescriptor GOPACS_SETTLEMENT_HISTORY = + new AttributeDescriptor<>( + "gopacsSettlementHistory", + ValueType.JSON_OBJECT, + new MetaItem<>(MetaItemType.READ_ONLY), + new MetaItem<>(MetaItemType.STORE_DATA_POINTS)); + + /** When the tracked orders were last synced with Distro. */ + public static final AttributeDescriptor GOPACS_LAST_SYNC = + new AttributeDescriptor<>( + "gopacsLastSync", ValueType.TIMESTAMP, new MetaItem<>(MetaItemType.READ_ONLY)); + public static final AssetDescriptor DESCRIPTOR = new AssetDescriptor<>("transmission-tower", null, EmsDistroEnergyAsset.class); @@ -59,4 +151,62 @@ public Optional getPortfolio() { public void setPortfolio(String portfolio) { getAttributes().getOrCreate(PORTFOLIO).setValue(portfolio); } + + // --- GOPACS getters --- + + public Optional getGopacsEnabled() { + return getAttributes().getValue(GOPACS_ENABLED); + } + + public Optional getGopacsEan() { + return getAttributes().getValue(GOPACS_EAN); + } + + public Optional getGopacsAnnouncementId() { + return getAttributes().getValue(GOPACS_ANNOUNCEMENT_ID); + } + + public Optional getGopacsStartTime() { + return getAttributes().getValue(GOPACS_START_TIME); + } + + public Optional getGopacsEndTime() { + return getAttributes().getValue(GOPACS_END_TIME); + } + + public Optional getGopacsBidValidityEnd() { + return getAttributes().getValue(GOPACS_BID_VALIDITY_END); + } + + public Optional getGopacsEanEffectivity() { + return getAttributes().getValue(GOPACS_EAN_EFFECTIVITY); + } + + public Optional getGopacsOrderType() { + return getAttributes().getValue(GOPACS_ORDER_TYPE); + } + + public Optional getGopacsOrderQuantity() { + return getAttributes().getValue(GOPACS_ORDER_QUANTITY); + } + + public Optional getGopacsOrderPrice() { + return getAttributes().getValue(GOPACS_ORDER_PRICE); + } + + public Optional getGopacsOrders() { + return getAttributes().getValue(GOPACS_ORDERS); + } + + // --- GOPACS setters --- + + public EmsDistroEnergyAsset setGopacsEnabled(Boolean enabled) { + getAttributes().getOrCreate(GOPACS_ENABLED).setValue(enabled); + return this; + } + + public EmsDistroEnergyAsset setGopacsOrderPrice(Double price) { + getAttributes().getOrCreate(GOPACS_ORDER_PRICE).setValue(price); + return this; + } } 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 4db9b54..3f49f96 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 @@ -43,6 +43,7 @@ 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.distroenergy.DistroGopacsHandler; import org.openremote.extension.ems.manager.gopacs.GOPACSHandler; import org.openremote.extension.ems.manager.gopacs.GOPACSRedispatchHandler; import org.openremote.manager.asset.AssetProcessingService; @@ -63,6 +64,7 @@ import org.openremote.model.datapoint.query.AssetDatapointAllQuery; import org.openremote.model.query.AssetQuery; import org.openremote.model.syslog.SyslogCategory; +import org.openremote.model.value.AttributeDescriptor; public class EmsOptimisationService extends RouteBuilder implements ContainerService { @@ -70,9 +72,27 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer SyslogCategory.getLogger(DATA, EmsOptimisationService.class.getName()); protected Services services; + /** + * Announcement inputs of the Distro GOPACS order handler, copied from the {@link EmsGOPACSAsset} + * that shares the Distro asset's parent. The GOPACS asset stays the only place they are edited. + */ + protected static final Map, AttributeDescriptor> + GOPACS_TO_DISTRO_INPUTS = + Map.of( + EmsGOPACSAsset.CONTRACTED_EAN, EmsDistroEnergyAsset.GOPACS_EAN, + EmsGOPACSAsset.REDISPATCH_ANNOUNCEMENT_ID, + EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID, + EmsGOPACSAsset.REDISPATCH_START_TIME, EmsDistroEnergyAsset.GOPACS_START_TIME, + EmsGOPACSAsset.REDISPATCH_END_TIME, EmsDistroEnergyAsset.GOPACS_END_TIME, + EmsGOPACSAsset.REDISPATCH_BID_VALIDITY_END, + EmsDistroEnergyAsset.GOPACS_BID_VALIDITY_END, + EmsGOPACSAsset.REDISPATCH_EAN_EFFECTIVITY, + EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY); + protected GOPACSHandler.Factory gopacsHandlerFactory; protected GOPACSRedispatchHandler.Factory gopacsRedispatchHandlerFactory; protected DistroEnergyHandler.Factory distroEnergyHandlerFactory; + protected DistroGopacsHandler.Factory distroGopacsHandlerFactory; private final Map> energyOptimisationAssetsMap = new ConcurrentHashMap<>(); @@ -91,6 +111,10 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer // 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<>(); + // Keyed by asset id. Reconciled from the persistence route and the attribute event subscription, + // so the check-then-start runs under a lock for the same reason as gopacsLifecycleLock. + private final Map distroGopacsHandlerMap = new ConcurrentHashMap<>(); + private final Object distroGopacsLifecycleLock = new Object(); @SuppressWarnings("unchecked") @Override @@ -121,6 +145,7 @@ public void init(Container container) throws Exception { gopacsHandlerFactory = new GOPACSHandler.Factory(container); gopacsRedispatchHandlerFactory = new GOPACSRedispatchHandler.Factory(container); distroEnergyHandlerFactory = new DistroEnergyHandler.Factory(container); + distroGopacsHandlerFactory = new DistroGopacsHandler.Factory(container); } @Override @@ -171,7 +196,11 @@ public void start(Container container) throws Exception { .attributeName(EmsGOPACSAsset.CONTRACTED_EAN.getName())) .stream() .map(asset -> (EmsGOPACSAsset) asset) - .forEach(this::reconcileGopacsAsset); + .forEach( + asset -> { + reconcileGopacsAsset(asset); + syncDistroGopacsInputsFrom(asset); + }); // Start Distro Energy handler for all Distro Energy assets services @@ -184,6 +213,17 @@ public void start(Container container) throws Exception { .map(asset -> (EmsDistroEnergyAsset) asset) .forEach(this::startDistroEnergyHandler); + // Start Distro GOPACS order handler for every Distro Energy asset that has it enabled + services + .getAssetStorageService() + .findAll( + new AssetQuery() + .types(EmsDistroEnergyAsset.class) + .attributeName(EmsDistroEnergyAsset.GOPACS_ENABLED.getName())) + .stream() + .map(asset -> (EmsDistroEnergyAsset) asset) + .forEach(this::reconcileDistroGopacsHandler); + // List of asset types that are part of the core EMS service String[] assetTypes = { EmsElectricityBatteryAsset.DESCRIPTOR.getName(), @@ -211,6 +251,11 @@ public void stop(Container container) throws Exception { } distroEnergyHandlerMap.forEach((assetId, handler) -> handler.undeploy()); distroEnergyHandlerMap.clear(); + synchronized (distroGopacsLifecycleLock) { + // Open orders stay at Distro: the next start restores and keeps tracking them. + distroGopacsHandlerMap.forEach((assetId, handler) -> handler.undeploy(false)); + distroGopacsHandlerMap.clear(); + } energyOptimisationAssetsMap.forEach((assetId, scheduledFuture) -> stopOptimisation(assetId)); energyOptimisationTimersMap.clear(); } @@ -555,6 +600,55 @@ private void stopDistroEnergyHandler(String assetId) { } } + /** + * Runs one Distro GOPACS order handler for the asset while {@code gopacsEnabled} is true and none + * otherwise. A running handler is left alone: its tracked orders are restored from the asset on + * deploy, but a redeploy would still drop an in-flight poll. + */ + private void reconcileDistroGopacsHandler(EmsDistroEnergyAsset asset) { + synchronized (distroGopacsLifecycleLock) { + String assetId = asset.getId(); + boolean enabled = asset.getGopacsEnabled().orElse(false); + if (enabled == distroGopacsHandlerMap.containsKey(assetId)) { + LOG.fine( + String.format( + "Distro GOPACS handler for asset %s already %s", + assetId, enabled ? "running" : "stopped")); + return; + } + if (!enabled) { + LOG.info("gopacsEnabled is off, stopping Distro GOPACS handler for asset: " + assetId); + stopDistroGopacsHandler(assetId); + return; + } + LOG.info("gopacsEnabled is on, starting Distro GOPACS handler for asset: " + assetId); + DistroGopacsHandler handler; + try { + handler = distroGopacsHandlerFactory.createHandler(assetId); + } catch (Exception e) { + // Missing credentials or an unusable URL must not take down the rest of the EMS service. + LOG.log(Level.SEVERE, "Failed to deploy Distro GOPACS handler for asset: " + assetId, e); + return; + } + // Registered before deploy() for the same reason as the day-ahead handler. + distroGopacsHandlerMap.put(assetId, handler); + handler.deploy(); + } + } + + /** + * Stops the handler for good and cancels its open orders, so none can still be matched at Distro + * once nothing tracks them. + */ + private void stopDistroGopacsHandler(String assetId) { + synchronized (distroGopacsLifecycleLock) { + DistroGopacsHandler existing = distroGopacsHandlerMap.remove(assetId); + if (existing != null) { + existing.undeploy(true); + } + } + } + protected void processAssetChange(PersistenceEvent persistenceEvent) { if (persistenceEvent.getEntity() instanceof EmsEnergyOptimisationAsset emsEnergyOptimisationAsset) { @@ -566,7 +660,10 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { // Stopping goes by asset id because the entity carries the EAN as saved, which is not // necessarily the key the running handler was registered under. case DELETE -> stopGopacsAsset(emsGOPACSAsset.getId()); - case CREATE, UPDATE -> reconcileGopacsAsset(emsGOPACSAsset); + case CREATE, UPDATE -> { + reconcileGopacsAsset(emsGOPACSAsset); + syncDistroGopacsInputsFrom(emsGOPACSAsset); + } } } else if (persistenceEvent.getEntity() instanceof EmsDistroEnergyAsset emsDistroEnergyAsset) { // distroEnergyHandlerMap is keyed by asset id, not portfolio: stop by asset id so this @@ -575,6 +672,11 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { String assetId = emsDistroEnergyAsset.getId(); if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { stopDistroEnergyHandler(assetId); + stopDistroGopacsHandler(assetId); + } else { + reconcileDistroGopacsHandler(emsDistroEnergyAsset); + // Also covers the Distro asset being created after, or moved next to, its GOPACS sibling. + syncDistroGopacsInputsTo(emsDistroEnergyAsset); } if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { startDistroEnergyHandler(emsDistroEnergyAsset); @@ -938,6 +1040,11 @@ private void processAttributeEventEmsGOPACSAsset(AttributeEvent attributeEvent) reconcileGopacsAsset(gopacsAsset); } + if (GOPACS_TO_DISTRO_INPUTS.keySet().stream() + .anyMatch(source -> source.getName().equals(attributeName))) { + syncDistroGopacsInputsFrom(gopacsAsset); + } + // Handle bid confirmation if (attributeName.equals(EmsGOPACSAsset.REDISPATCH_CONFIRM_BID.getName())) { boolean confirmed = (Boolean) attributeEvent.getValue().orElse(false); @@ -998,6 +1105,22 @@ private void processAttributeEventEmsDistroEnergyAsset(AttributeEvent attributeE startDistroEnergyHandler(emsDistroEnergyAsset); }); } + + if (attributeName.equals(EmsDistroEnergyAsset.GOPACS_ENABLED.getName())) { + reconcileDistroGopacsHandler(emsDistroEnergyAsset); + } else if (DistroGopacsHandler.INPUTS.contains(attributeName)) { + DistroGopacsHandler handler = distroGopacsHandlerMap.get(assetId); + if (handler != null) { + handler.onAttributeEvent(attributeEvent); + } else if (attributeEvent.getValue().filter(Boolean.TRUE::equals).isPresent()) { + // Only the operator flags: announcement inputs are copied in whether a handler runs or not. + LOG.warning( + String.format( + "Ignoring %s for asset %s: no Distro GOPACS handler is running. Check that" + + " gopacsEnabled is on and that the handler deployed (DISTRO_GOPACS_* config)", + attributeName, assetId)); + } + } } private void updatePowerLimitProfileManualForecasts( @@ -1165,22 +1288,24 @@ private void updatePowerLimitProfileTotalForecasts( } private EmsGOPACSAsset getGopacsAsset(EmsEnergyOptimisationAsset energyOptimisationAsset) { - EmsGOPACSAsset gopacsAsset = null; String logPrefix = String.format( "assetType='%s', assetId='%s', assetName='%s'", energyOptimisationAsset.getAssetType(), energyOptimisationAsset.getId(), energyOptimisationAsset.getAssetName()); + return getGopacsAsset(energyOptimisationAsset.getId(), logPrefix); + } + + /** The one GOPACS asset under the parent, or null if there is none or more than one. */ + private EmsGOPACSAsset getGopacsAsset(String parentId, String logPrefix) { + EmsGOPACSAsset gopacsAsset = null; // Check for GOPACS assets List gopacsAssets = services .getAssetStorageService() - .findAll( - new AssetQuery() - .parents(energyOptimisationAsset.getId()) - .types(EmsGOPACSAsset.class)) + .findAll(new AssetQuery().parents(parentId).types(EmsGOPACSAsset.class)) .stream() .map(asset -> (EmsGOPACSAsset) asset) .toList(); @@ -1201,6 +1326,67 @@ private EmsGOPACSAsset getGopacsAsset(EmsEnergyOptimisationAsset energyOptimisat return gopacsAsset; } + /** Copies the announcement inputs from the GOPACS asset to every Distro asset next to it. */ + private void syncDistroGopacsInputsFrom(EmsGOPACSAsset gopacsAsset) { + if (gopacsAsset.getParentId() == null) { + return; + } + services + .getAssetStorageService() + .findAll( + new AssetQuery().parents(gopacsAsset.getParentId()).types(EmsDistroEnergyAsset.class)) + .forEach(asset -> syncDistroGopacsInputs(gopacsAsset, (EmsDistroEnergyAsset) asset)); + } + + /** Copies the announcement inputs to the Distro asset from the GOPACS asset next to it. */ + private void syncDistroGopacsInputsTo(EmsDistroEnergyAsset distroEnergyAsset) { + if (distroEnergyAsset.getParentId() == null) { + return; + } + String logPrefix = + String.format( + "assetType='%s', assetId='%s', assetName='%s'", + distroEnergyAsset.getType(), distroEnergyAsset.getId(), distroEnergyAsset.getName()); + EmsGOPACSAsset gopacsAsset = getGopacsAsset(distroEnergyAsset.getParentId(), logPrefix); + if (gopacsAsset != null) { + syncDistroGopacsInputs(gopacsAsset, distroEnergyAsset); + } + } + + /** + * Writes every announcement input whose value differs from the GOPACS asset's. Unchanged ones are + * skipped, so the redispatch poller rewriting the same announcement on each poll does not rewrite + * the Distro asset too. + */ + private void syncDistroGopacsInputs( + EmsGOPACSAsset gopacsAsset, EmsDistroEnergyAsset distroEnergyAsset) { + GOPACS_TO_DISTRO_INPUTS.forEach( + (source, target) -> { + Optional> targetAttribute = + distroEnergyAsset.getAttribute(target.getName()); + if (targetAttribute.isEmpty()) { + LOG.fine( + String.format( + "Distro asset %s has no %s attribute, not copying %s from GOPACS asset %s", + distroEnergyAsset.getId(), + target.getName(), + source.getName(), + gopacsAsset.getId())); + return; + } + Object value = + gopacsAsset.getAttribute(source.getName()).flatMap(Attribute::getValue).orElse(null); + if (Objects.equals(value, targetAttribute.get().getValue().orElse(null))) { + return; + } + services + .getAssetProcessingService() + .sendAttributeEvent( + new AttributeEvent(distroEnergyAsset.getId(), target.getName(), value), + getClass().getSimpleName()); + }); + } + private String generatePowerLimitProfile(double powerLimit) { LocalDateTime startOfDay = LocalDateTime.now().toLocalDate().atStartOfDay(); LocalDateTime endOfDay = startOfDay.plusDays(1); diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandler.java new file mode 100644 index 0000000..a4dce9a --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandler.java @@ -0,0 +1,941 @@ +/* + * 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 org.openremote.container.web.WebTargetBuilder.createClient; +import static org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType.BUY; +import static org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType.SELL; +import static org.openremote.model.syslog.SyslogCategory.API; + +import com.fasterxml.jackson.core.type.TypeReference; +import jakarta.ws.rs.WebApplicationException; +import jakarta.ws.rs.core.Response; +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +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 java.util.stream.Collectors; +import java.util.stream.Stream; +import org.jboss.resteasy.client.jaxrs.ResteasyClient; +import org.openremote.container.timer.TimerService; +import org.openremote.container.web.WebTargetBuilder; +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset; +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType; +import org.openremote.extension.ems.agent.EmsElectricityBatteryAsset; +import org.openremote.extension.ems.agent.EmsElectricityBatteryAsset.EmsElectricityBatteryConnectionStatusValueType; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrder; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrderRequest; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsSettlement; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsSettlementPage; +import org.openremote.manager.asset.AssetProcessingService; +import org.openremote.manager.asset.AssetStorageService; +import org.openremote.model.Container; +import org.openremote.model.attribute.AttributeEvent; +import org.openremote.model.auth.OAuthClientCredentialsGrant; +import org.openremote.model.query.AssetQuery; +import org.openremote.model.syslog.SyslogCategory; +import org.openremote.model.util.ValueUtil; +import org.openremote.model.value.AttributeDescriptor; + +/** + * Places GOPACS redispatch orders through the Distro Energy GOPACS Connector API for one {@link + * EmsDistroEnergyAsset}. + * + *

The announcement arrives on the asset's {@code gopacs*} input attributes, which {@code + * EmsOptimisationService} copies from the {@code EmsGOPACSAsset} with the same parent, so this + * handler never talks to GOPACS itself. From those inputs it pre-fills an order type and quantity, + * which the operator may override before setting a price and confirming. Tracked orders are polled + * until they reach a final status. + */ +public class DistroGopacsHandler { + + private static final Logger LOG = SyslogCategory.getLogger(API, DistroGopacsHandler.class); + + public static final String DISTRO_GOPACS_CLIENT_ID = "DISTRO_GOPACS_CLIENT_ID"; + public static final String DISTRO_GOPACS_CLIENT_SECRET = "DISTRO_GOPACS_CLIENT_SECRET"; + public static final String DISTRO_GOPACS_TENANT_ID = "DISTRO_GOPACS_TENANT_ID"; + public static final String DISTRO_GOPACS_URL = "DISTRO_GOPACS_URL"; + public static final String DISTRO_GOPACS_URL_DEFAULT = "https://api.distro.energy/v1"; + public static final String DISTRO_GOPACS_TOKEN_URL = "DISTRO_GOPACS_TOKEN_URL"; + public static final String DISTRO_GOPACS_TOKEN_URL_DEFAULT = + "https://auth.distro.energy/oauth2/token"; + + protected static final String SCOPE = "distro-api/gopacs.access"; + protected static final Duration POLL_INTERVAL = Duration.ofMinutes(5); + protected static final long ISP_SECONDS = Duration.ofMinutes(15).toSeconds(); + + /** Distro takes quantities of at least 0.1 MW with at most one decimal, so steps of 100 kW. */ + protected static final double MIN_QUANTITY_KW = 100; + + /** Largest page size the settlements endpoint accepts. */ + protected static final int SETTLEMENT_PAGE_SIZE = 100; + + /** Distro's error bodies are short JSON; anything longer is cut off in logs and history. */ + private static final int MAX_ERROR_BODY_LENGTH = 500; + + public static final String METADATA_ANNOUNCEMENT_ID = "announcementId"; + + /** Orders that can still be cancelled. Only SUBMITTED ones can be replaced. */ + protected static final Set OPEN_STATUSES = Set.of("PENDING", "SUBMITTED"); + + protected static final Set REPLACEABLE_STATUSES = Set.of("SUBMITTED"); + + /** Statuses after which an order no longer changes and is no longer tracked. */ + protected static final Set FINAL_STATUSES = + Set.of("SETTLED", "REVERTED", "REJECTED", "CANCELED"); + + /** Relative effectivities in a GOPACS EAN effectivity category name, weakest first. */ + protected static final List EFFECTIVITY_RANKS = + List.of("NONE", "VERY_LOW", "LOW", "MODERATE", "HIGH", "VERY_HIGH", "MAXIMUM"); + + /** Inputs the pre-filled order type and quantity are derived from. */ + protected static final Set SUGGESTION_INPUTS = + Set.of( + EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.getName(), + EmsDistroEnergyAsset.GOPACS_START_TIME.getName(), + EmsDistroEnergyAsset.GOPACS_END_TIME.getName(), + EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY.getName()); + + /** + * Every attribute {@link #onAttributeEvent} reacts to. The handler's own writes are left out, so + * they are not routed back to it. + */ + public static final Set INPUTS = + Stream.concat( + SUGGESTION_INPUTS.stream(), + Stream.of( + EmsDistroEnergyAsset.GOPACS_ORDER_PRICE.getName(), + EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.getName(), + EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS.getName())) + .collect(Collectors.toUnmodifiableSet()); + + protected final String assetId; + protected final String tenantId; + protected final ResteasyClient client; + protected final DistroGopacsResource resource; + + protected final TimerService timerService; + protected final ScheduledExecutorService scheduledExecutorService; + protected final AssetStorageService assetStorageService; + protected final AssetProcessingService assetProcessingService; + + /** Tracked orders by reference, mirrored to {@link EmsDistroEnergyAsset#GOPACS_ORDERS}. */ + protected final Map orders = new LinkedHashMap<>(); + + protected ScheduledFuture pollFuture; + + // Tasks already queued on the shared executor still run after undeploy(); they check this first. + protected volatile boolean undeployed; + + public static class Factory { + protected Container container; + + public Factory(Container container) { + this.container = container; + } + + public DistroGopacsHandler createHandler(String assetId) { + return new DistroGopacsHandler(assetId, container, null); + } + } + + /** + * @param resource used as is when not null, which is how tests stand in for the HTTP client; + * otherwise one is built from the container config. + */ + protected DistroGopacsHandler( + String assetId, Container container, DistroGopacsResource resource) { + this.assetId = assetId; + this.timerService = container.getService(TimerService.class); + this.scheduledExecutorService = container.getScheduledExecutor(); + this.assetStorageService = container.getService(AssetStorageService.class); + this.assetProcessingService = container.getService(AssetProcessingService.class); + + Map config = container.getConfig(); + this.tenantId = requireConfig(config, DISTRO_GOPACS_TENANT_ID); + + if (resource != null) { + this.client = null; + this.resource = resource; + } else { + String clientId = requireConfig(config, DISTRO_GOPACS_CLIENT_ID); + String clientSecret = requireConfig(config, DISTRO_GOPACS_CLIENT_SECRET); + String baseUrl = config.getOrDefault(DISTRO_GOPACS_URL, DISTRO_GOPACS_URL_DEFAULT); + String tokenUrl = + config.getOrDefault(DISTRO_GOPACS_TOKEN_URL, DISTRO_GOPACS_TOKEN_URL_DEFAULT); + this.client = createClient(org.openremote.container.Container.EXECUTOR); + try { + this.resource = createResource(client, baseUrl, tokenUrl, clientId, clientSecret); + } catch (RuntimeException e) { + // No reference escapes a constructor that threw, so undeploy() can never close this client. + client.close(); + throw e; + } + // The client ID is not a secret; the secret itself is never logged. + LOG.info( + String.format( + "Configured Distro GOPACS handler for asset %s: baseUrl=%s, tokenUrl=%s, tenant=%s," + + " clientId=%s", + assetId, baseUrl, tokenUrl, tenantId, clientId)); + } + } + + /** + * Builds the API proxy. OpenRemote's OAuth filter fetches a client-credentials token with HTTP + * Basic authentication, caches it and refreshes it shortly before it expires. + */ + static DistroGopacsResource createResource( + ResteasyClient client, + String baseUrl, + String tokenUrl, + String clientId, + String clientSecret) { + return new WebTargetBuilder(client, URI.create(baseUrl)) + .setOAuthAuthentication( + new OAuthClientCredentialsGrant(tokenUrl, clientId, clientSecret, SCOPE) + .setBasicAuthHeader(true)) + .build() + .proxy(DistroGopacsResource.class); + } + + private static String requireConfig(Map config, String key) { + String value = config.get(key); + if (value == null || value.isBlank()) { + throw new IllegalStateException(key + " not defined, cannot place Distro GOPACS orders."); + } + return value; + } + + /** Restores the tracked orders saved on the asset and starts polling them. */ + public synchronized void deploy() { + EmsDistroEnergyAsset asset = findAsset(); + if (asset != null) { + asset + .getGopacsOrders() + .ifPresent( + saved -> + orders.putAll( + ValueUtil.JSON.convertValue( + saved, new TypeReference>() {}))); + // An undeploy within a second of a toggle skips its reset, and a flag left true would make + // the next toggle an unchanged value that is ignored. + for (AttributeDescriptor flag : + List.of( + EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER, + EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS)) { + if (asset.getAttributes().getValue(flag).orElse(false)) { + sendAttributeEvent(flag, false); + } + } + } + long pollMillis = POLL_INTERVAL.toMillis(); + pollFuture = + scheduledExecutorService.scheduleAtFixedRate( + this::pollOrders, pollMillis, pollMillis, TimeUnit.MILLISECONDS); + LOG.info( + "Deployed Distro GOPACS handler for asset " + assetId + " (" + orders.size() + " tracked)"); + } + + /** + * Stops polling and closes the client. + * + * @param cancelOpenOrders cancels the open orders first, for when the handler stops for good + * (disabled or asset deleted) rather than for a manager restart, which picks them up again. + * Runs on the executor because the caller is an event thread and cancelling calls Distro. + */ + public void undeploy(boolean cancelOpenOrders) { + undeployed = true; + if (pollFuture != null) { + // Interrupt to abort any in-flight HTTP call before closing the client + pollFuture.cancel(true); + pollFuture = null; + } + if (cancelOpenOrders) { + scheduledExecutorService.execute( + () -> { + // Waits for a running order action, like the close below. + synchronized (this) { + try { + cancelOrders(null); + } catch (RuntimeException e) { + LOG.log( + Level.WARNING, + "Failed to cancel open Distro GOPACS orders for asset " + assetId, + e); + } finally { + closeClient(); + } + } + }); + } else { + // Wait for a running order action to finish so it never uses a closed client. + synchronized (this) { + closeClient(); + } + } + LOG.info("Undeployed Distro GOPACS handler for asset " + assetId); + } + + private void closeClient() { + if (client != null) { + client.close(); + } + } + + /** + * Reacts to a changed {@link #INPUTS} attribute of the asset. Runs on the executor because every + * reaction may call Distro, and the caller is the client event thread. + * + *

Events whose value did not change are ignored: a save of the asset republishes all of its + * attributes, and an attribute link on the inputs would forward every unchanged update the + * redispatch poller writes on each poll. + */ + public void onAttributeEvent(AttributeEvent event) { + Object oldValue = event.getOldValue().orElse(null); + Object value = event.getValue().orElse(null); + if (Objects.equals(oldValue, value)) { + if (Boolean.TRUE.equals(value) && isFlag(event.getName())) { + LOG.info( + String.format( + "Ignoring %s for Distro GOPACS asset %s: it is already true, set it to false first", + event.getName(), assetId)); + } else { + LOG.finest( + String.format( + "Ignoring unchanged %s for Distro GOPACS asset %s", event.getName(), assetId)); + } + return; + } + LOG.fine( + String.format( + "Distro GOPACS asset %s: %s changed from %s to %s", + assetId, event.getName(), oldValue, value)); + scheduledExecutorService.execute( + () -> { + try { + processAttributeChange(event.getName(), oldValue, value); + } catch (Exception e) { + LOG.log( + Level.SEVERE, + "Failed to process " + event.getName() + " for Distro GOPACS asset " + assetId, + e); + } + }); + } + + private static boolean isFlag(String attributeName) { + return attributeName.equals(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.getName()) + || attributeName.equals(EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS.getName()); + } + + protected synchronized void processAttributeChange( + String attributeName, Object oldValue, Object value) { + if (undeployed) { + LOG.fine( + String.format( + "Dropping %s for Distro GOPACS asset %s: the handler is undeployed", + attributeName, assetId)); + return; + } + if (attributeName.equals(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.getName()) + && oldValue instanceof String oldAnnouncementId) { + // The announcement closed or was superseded: what is still open for it can no longer help. + cancelOrders(oldAnnouncementId); + } + + if (SUGGESTION_INPUTS.contains(attributeName)) { + updateSuggestion(); + } else if (attributeName.equals(EmsDistroEnergyAsset.GOPACS_ORDER_PRICE.getName()) + && value instanceof Number price) { + replaceOrders(price.doubleValue()); + } else if (attributeName.equals(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.getName()) + && Boolean.TRUE.equals(value)) { + LOG.info("Order confirmed for Distro GOPACS asset " + assetId); + placeOrder(); + resetFlag(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER); + } else if (attributeName.equals(EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS.getName()) + && Boolean.TRUE.equals(value)) { + LOG.info("Cancel of all open orders requested for Distro GOPACS asset " + assetId); + cancelOrders(null); + resetFlag(EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS); + } + } + + /** Pre-fills order type and quantity for the current announcement, or clears both. */ + protected void updateSuggestion() { + EmsDistroEnergyAsset asset = findAsset(); + if (asset == null) { + return; + } + String category = asset.getGopacsEanEffectivity().orElse(null); + Long start = asset.getGopacsStartTime().orElse(null); + Long end = asset.getGopacsEndTime().orElse(null); + + GopacsOrderTypeValueType orderType = null; + Double quantityKw = null; + if (asset.getGopacsAnnouncementId().isPresent() + && category != null + && start != null + && end != null) { + double windowHours = (end - start) / 3_600_000.0; + List batteries = findBatteries(asset.getParentId()); + double buyKw = quantityKw(batteries, BUY, windowHours); + double sellKw = quantityKw(batteries, SELL, windowHours); + orderType = orderTypeFor(category, buyKw, sellKw); + double sideKw = orderType == BUY ? buyKw : sellKw; + if (orderType == null) { + LOG.info( + String.format( + "No order suggested for Distro GOPACS asset %s: effectivity category '%s' favours" + + " neither side", + assetId, category)); + } else if (sideKw < MIN_QUANTITY_KW) { + LOG.info( + String.format( + "No order suggested for Distro GOPACS asset %s: the batteries can hold %s kW to %s," + + " below the 100 kW minimum", + assetId, sideKw, orderType)); + orderType = null; + } else { + quantityKw = sideKw; + LOG.info( + String.format( + "Suggested %s of %s kW for Distro GOPACS asset %s (category %s, buy/sell %s/%s kW)", + orderType, quantityKw, assetId, category, buyKw, sellKw)); + } + } else { + LOG.fine( + String.format( + "No order suggested for Distro GOPACS asset %s: announcement incomplete (id=%s," + + " category=%s, start=%s, end=%s)", + assetId, asset.getGopacsAnnouncementId().orElse(null), category, start, end)); + } + sendAttributeEvent(EmsDistroEnergyAsset.GOPACS_ORDER_TYPE, orderType); + sendAttributeEvent(EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY, quantityKw); + } + + protected List findBatteries(String parentId) { + if (parentId == null) { + return List.of(); + } + return assetStorageService + .findAll(new AssetQuery().parents(parentId).types(EmsElectricityBatteryAsset.class)) + .stream() + .map(asset -> (EmsElectricityBatteryAsset) asset) + .toList(); + } + + /** + * Maps a GOPACS EAN effectivity category ({@code BUY__SELL_}) to the + * side an order on the EAN helps most. When both sides are equally effective the side the + * batteries can offer more of wins, BUY on a draw. + * + * @return BUY, SELL, or null when neither side helps or the category is not recognised. + */ + static GopacsOrderTypeValueType orderTypeFor(String category, double buyKw, double sellKw) { + if (category == null || !category.startsWith("BUY_")) { + return null; + } + int separator = category.indexOf("_SELL_"); + if (separator < 0) { + return null; + } + int buy = EFFECTIVITY_RANKS.indexOf(category.substring("BUY_".length(), separator)); + int sell = EFFECTIVITY_RANKS.indexOf(category.substring(separator + "_SELL_".length())); + if (buy < 0 || sell < 0 || (buy == 0 && sell == 0)) { + return null; + } + if (buy != sell) { + return buy > sell ? BUY : SELL; + } + return buyKw >= sellKw ? BUY : SELL; + } + + /** + * The constant power the batteries can hold for the whole window on one side: their combined + * maximum power, limited so the energy over the window fits their state-of-charge headroom (seen + * from the grid, so through the charge or discharge efficiency). Batteries that are not allowed + * on that side, are disconnected or lack the data to size them are left out. Rounded down to the + * 0.1 MW steps Distro accepts. + */ + static double quantityKw( + List batteries, + GopacsOrderTypeValueType orderType, + double windowHours) { + if (windowHours <= 0) { + return 0; + } + boolean buy = orderType == BUY; + double powerKw = 0; + double energyKwh = 0; + + for (EmsElectricityBatteryAsset battery : batteries) { + boolean allowed = + (buy ? battery.getAllowCharging() : battery.getAllowDischarging()).orElse(false); + boolean disconnected = + battery.getConnectionStatus().orElse(null) + == EmsElectricityBatteryConnectionStatusValueType.disconnected; + Double capacityKwh = battery.getEnergyCapacity().orElse(null); + Double levelPercentage = battery.getEnergyLevelPercentage().orElse(null); + Integer limitPercentage = + (buy + ? battery.getEnergyLevelPercentageMaximum() + : battery.getEnergyLevelPercentageMinimum()) + .orElse(null); + if (!allowed + || disconnected + || capacityKwh == null + || levelPercentage == null + || limitPercentage == null) { + continue; + } + + double headroomKwh = + capacityKwh + * (buy ? limitPercentage - levelPercentage : levelPercentage - limitPercentage) + / 100.0; + if (headroomKwh <= 0) { + continue; + } + double efficiency = + (buy ? battery.getChargeEfficiency() : battery.getDischargeEfficiency()).orElse(100) + / 100.0; + // Charging draws more from the grid than ends up stored; discharging delivers less. + energyKwh += buy ? headroomKwh / efficiency : headroomKwh * efficiency; + // Discharge power is stored as a negative number. + powerKw += + Math.abs( + (buy ? battery.getChargePowerMaximum() : battery.getDischargePowerMaximum()) + .orElse(0.0)); + } + + return roundDownToStep(Math.min(powerKw, energyKwh / windowHours)); + } + + /** Rounded down to the 100 kW steps Distro accepts, so an order never offers more than asked. */ + static double roundDownToStep(double kw) { + return Math.floor(kw / MIN_QUANTITY_KW) * MIN_QUANTITY_KW; + } + + /** Distro takes prices with at most two decimals. */ + static double roundToCents(double price) { + return Math.round(price * 100) / 100.0; + } + + /** + * The bid validity end as order expiration, rounded down to a quarter-hour. Left to Distro's + * default (15 minutes before start) when it is unknown, already passed or later than that. + */ + static Long expirationSeconds(long startSeconds, Long validityEndSeconds, long nowSeconds) { + if (validityEndSeconds == null) { + return null; + } + long expiration = validityEndSeconds - Math.floorMod(validityEndSeconds, ISP_SECONDS); + return expiration > nowSeconds && expiration <= startSeconds - ISP_SECONDS ? expiration : null; + } + + /** + * Checks an order against what Distro accepts. The window comes from the GOPACS announcement, so + * it is trusted to be quarter-hour aligned. + * + * @return why the order cannot be placed, or null when it is valid. + */ + static String validateInput( + String announcementId, + String ean, + GopacsOrderTypeValueType orderType, + Double enteredKw, + Double price, + Long start, + Long end, + Long validityEndSeconds, + long nowSeconds) { + if (announcementId == null) { + return "no announcement"; + } + if (ean == null || !ean.matches("\\d{18}")) { + return "EAN '" + ean + "' is not an 18-digit EAN"; + } + if (orderType == null) { + return "no order type"; + } + if (enteredKw == null || roundDownToStep(enteredKw) < MIN_QUANTITY_KW) { + return "quantity " + enteredKw + " kW is below the 100 kW minimum"; + } + if (price == null) { + return "no price"; + } + if (start == null || end == null || end <= start) { + return "window " + start + " to " + end + " is empty"; + } + // Distro needs the order to expire in the future and at least 15 minutes before start. + if (start / 1000 - ISP_SECONDS <= nowSeconds) { + return "window starts in less than 15 minutes"; + } + // GOPACS no longer takes bids for the announcement. + if (validityEndSeconds != null && validityEndSeconds <= nowSeconds) { + return "bid validity end has passed"; + } + return null; + } + + /** Places an order for the current announcement from the asset's order attributes. */ + protected void placeOrder() { + EmsDistroEnergyAsset asset = findAsset(); + if (asset == null) { + return; + } + String announcementId = asset.getGopacsAnnouncementId().orElse(null); + String ean = asset.getGopacsEan().orElse(null); + GopacsOrderTypeValueType orderType = asset.getGopacsOrderType().orElse(null); + Double enteredKw = asset.getGopacsOrderQuantity().orElse(null); + Double price = asset.getGopacsOrderPrice().map(DistroGopacsHandler::roundToCents).orElse(null); + Long start = asset.getGopacsStartTime().orElse(null); + Long end = asset.getGopacsEndTime().orElse(null); + Long validityEndSeconds = asset.getGopacsBidValidityEnd().map(ms -> ms / 1000).orElse(null); + long nowSeconds = timerService.getNow().getEpochSecond(); + + String invalid = + validateInput( + announcementId, + ean, + orderType, + enteredKw, + price, + start, + end, + validityEndSeconds, + nowSeconds); + if (invalid != null) { + LOG.warning("Not placing Distro GOPACS order for asset " + assetId + ": " + invalid); + recordHistory("REFUSED", null, invalid); + return; + } + + long startSeconds = start / 1000; + GopacsOrderRequest request = + new GopacsOrderRequest( + UUID.randomUUID().toString(), + orderType.name(), + roundDownToStep(enteredKw) / 1000, + startSeconds, + end / 1000, + price, + ean, + Map.of(METADATA_ANNOUNCEMENT_ID, announcementId), + expirationSeconds(startSeconds, validityEndSeconds, nowSeconds)); + LOG.info( + String.format( + "Placing Distro GOPACS order for asset %s (entered %s kW): %s", + assetId, enteredKw, ValueUtil.asJSON(request).orElse(request.toString()))); + // Tracked before posting: a timeout can hide an order Distro did accept. Polling then finds + // its real status, or untracks it when Distro answers 404. + orders.put(request.reference(), GopacsOrder.pending(request)); + try { + GopacsOrder order = resource.createOrder(tenantId, request); + LOG.info( + String.format( + "Placed Distro GOPACS order %s for asset %s: status %s", + order.reference(), assetId, order.status())); + orders.put(order.reference(), order); + recordHistory("CREATED", order, null); + } catch (RuntimeException e) { + String failure = logFailure("Failed to place Distro GOPACS order for asset " + assetId, e); + if (e instanceof WebApplicationException wae + && wae.getResponse().getStatus() >= 400 + && wae.getResponse().getStatus() < 500) { + // Rejected outright, so Distro holds no order for this reference. + orders.remove(request.reference()); + } + recordHistory("CREATE_FAILED", orders.get(request.reference()), failure); + } + saveOrders(); + } + + /** Re-submits the replaceable orders for the current announcement at the new price. */ + protected void replaceOrders(double price) { + EmsDistroEnergyAsset asset = findAsset(); + String announcementId = asset == null ? null : asset.getGopacsAnnouncementId().orElse(null); + if (announcementId == null) { + return; + } + List toReplace = ordersFor(announcementId, REPLACEABLE_STATUSES); + if (toReplace.isEmpty()) { + LOG.fine( + String.format( + "Price change for Distro GOPACS asset %s: no replaceable orders for announcement %s", + assetId, announcementId)); + return; + } + price = roundToCents(price); + for (GopacsOrder order : toReplace) { + GopacsOrderRequest request = order.toRequest(price); + try { + GopacsOrder replaced = resource.replaceOrder(tenantId, order.reference(), request); + LOG.info( + String.format( + "Replaced Distro GOPACS order %s for asset %s at %s EUR/MWh", + order.reference(), assetId, price)); + orders.put(replaced.reference(), replaced); + recordHistory("REPLACED", replaced, null); + } catch (RuntimeException e) { + // 423 means the order is locked for matching; it keeps its old price. + String failure = + logFailure( + "Failed to replace Distro GOPACS order " + + order.reference() + + " for asset " + + assetId, + e); + recordHistory("REPLACE_FAILED", order, failure); + } + } + saveOrders(); + } + + /** + * Cancels the open orders placed for the given announcement, or all open orders when it is null. + */ + protected void cancelOrders(String announcementId) { + List toCancel = ordersFor(announcementId, OPEN_STATUSES); + LOG.fine( + String.format( + "Cancelling %d open Distro GOPACS orders for asset %s (announcement %s)", + toCancel.size(), assetId, announcementId == null ? "any" : announcementId)); + for (GopacsOrder order : toCancel) { + try { + resource.cancelOrder(tenantId, order.reference()); + // Also the only record once the asset is deleted, as the writes below then go nowhere. + LOG.info("Cancelled Distro GOPACS order " + order.reference() + " for asset " + assetId); + orders.remove(order.reference()); + recordHistory("CANCELED", order, null); + } catch (RuntimeException e) { + String failure = + logFailure( + "Failed to cancel Distro GOPACS order " + + order.reference() + + " for asset " + + assetId, + e); + recordHistory("CANCEL_FAILED", order, failure); + } + } + if (!toCancel.isEmpty()) { + saveOrders(); + } + } + + /** The tracked orders in one of the statuses, for the announcement or for any when null. */ + protected List ordersFor(String announcementId, Set statuses) { + List matching = new ArrayList<>(); + for (GopacsOrder order : orders.values()) { + String orderAnnouncementId = + order.metadata() == null ? null : order.metadata().get(METADATA_ANNOUNCEMENT_ID); + if (statuses.contains(order.status()) + && (announcementId == null || announcementId.equals(orderAnnouncementId))) { + matching.add(order); + } + } + return matching; + } + + /** + * Refreshes every tracked order. A status change is recorded in the history; a settled order + * records its trades; an order in a final status stops being tracked. + */ + protected synchronized void pollOrders() { + if (undeployed || orders.isEmpty()) { + return; + } + LOG.fine(String.format("Polling %d Distro GOPACS orders for asset %s", orders.size(), assetId)); + for (GopacsOrder tracked : List.copyOf(orders.values())) { + try { + GopacsOrder order = resource.getOrder(tenantId, tracked.reference()); + if (!Objects.equals(order.status(), tracked.status())) { + LOG.info( + String.format( + "Distro GOPACS order %s for asset %s went from %s to %s%s", + order.reference(), + assetId, + tracked.status(), + order.status(), + order.reason() == null ? "" : " (" + order.reason() + ")")); + recordHistory("STATUS", order, order.reason()); + } + if (FINAL_STATUSES.contains(order.status())) { + if ("SETTLED".equals(order.status())) { + recordSettlements(order); + } + orders.remove(order.reference()); + } else { + orders.put(order.reference(), order); + } + } catch (WebApplicationException e) { + if (e.getResponse().getStatus() == 404) { + LOG.warning( + "Distro no longer knows GOPACS order " + tracked.reference() + "; untracking it"); + orders.remove(tracked.reference()); + recordHistory("NOT_FOUND", tracked, null); + } else { + logFailure("Failed to refresh Distro GOPACS order " + tracked.reference(), e); + } + } catch (RuntimeException e) { + logFailure("Failed to refresh Distro GOPACS order " + tracked.reference(), e); + } + } + saveOrders(); + sendAttributeEvent(EmsDistroEnergyAsset.GOPACS_LAST_SYNC, timerService.getCurrentTimeMillis()); + } + + /** + * Records each trade of a settled order. A settled order's trades no longer change and the order + * is untracked right after, so each trade is recorded once, unless the manager stops between + * recording them and saving the tracked orders. + * + *

Distro makes the trades available as soon as the order is SETTLED, so they are fetched once. + * A failed fetch is recorded and not retried. + */ + protected void recordSettlements(GopacsOrder order) { + // Fetched in full before recording, so a failed page never leaves the trades half recorded. + List settlements = new ArrayList<>(); + try { + GopacsSettlementPage page; + do { + page = + resource.listSettlements( + tenantId, + order.start(), + order.end(), + order.reference(), + SETTLEMENT_PAGE_SIZE, + settlements.size()); + if (page.items() != null) { + settlements.addAll(page.items()); + } + } while (page.items() != null + && !page.items().isEmpty() + && settlements.size() < page.total()); + } catch (RuntimeException e) { + String failure = + logFailure( + "Failed to fetch the trades of settled Distro GOPACS order " + order.reference(), e); + recordHistory("SETTLEMENTS_FAILED", order, failure); + return; + } + + for (GopacsSettlement settlement : settlements) { + Map entry = ValueUtil.JSON.convertValue(settlement, new TypeReference<>() {}); + entry.put("recordedAt", timerService.getCurrentTimeMillis()); + sendAttributeEvent(EmsDistroEnergyAsset.GOPACS_SETTLEMENT_HISTORY, entry); + } + } + + protected void recordHistory(String action, GopacsOrder order, String detail) { + Map entry = new HashMap<>(); + entry.put("action", action); + if (order != null) { + entry.putAll(ValueUtil.JSON.convertValue(order, new TypeReference>() {})); + } + if (detail != null) { + entry.put("detail", detail); + } + entry.put("timestamp", timerService.getCurrentTimeMillis()); + sendAttributeEvent(EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY, entry); + } + + protected void saveOrders() { + sendAttributeEvent( + EmsDistroEnergyAsset.GOPACS_ORDERS, + ValueUtil.JSON.convertValue(orders, new TypeReference>() {})); + } + + /** + * Logs a failed Distro call and returns a short description for the history. An HTTP error is + * logged with Distro's error body and without a stack trace, which adds nothing for it. + */ + private static String logFailure(String message, RuntimeException e) { + String failure = describe(e); + if (e instanceof WebApplicationException) { + LOG.warning(message + ": " + failure); + } else { + LOG.log(Level.WARNING, message, e); + } + return failure; + } + + /** The HTTP status and the error body Distro sent, or the exception without a response. */ + static String describe(RuntimeException e) { + if (!(e instanceof WebApplicationException wae)) { + return e.getClass().getSimpleName() + ": " + e.getMessage(); + } + Response response = wae.getResponse(); + String body = null; + try { + body = response.hasEntity() ? response.readEntity(String.class) : null; + } catch (RuntimeException ignored) { + // Already consumed or not readable: the status alone has to do. + } + if (body == null || body.isBlank()) { + return "HTTP " + response.getStatus(); + } + body = body.strip(); + return "HTTP " + + response.getStatus() + + ": " + + (body.length() > MAX_ERROR_BODY_LENGTH + ? body.substring(0, MAX_ERROR_BODY_LENGTH) + "..." + : body); + } + + private void resetFlag(AttributeDescriptor flag) { + // Delayed like the redispatch confirmation so the UI shows the toggle before it springs back. + scheduledExecutorService.schedule( + () -> { + if (!undeployed) { + sendAttributeEvent(flag, false); + } + }, + 1, + TimeUnit.SECONDS); + } + + protected EmsDistroEnergyAsset findAsset() { + EmsDistroEnergyAsset asset = (EmsDistroEnergyAsset) assetStorageService.find(assetId); + if (asset == null) { + LOG.warning("Distro GOPACS asset not found: " + assetId); + } + return asset; + } + + protected void sendAttributeEvent(AttributeDescriptor attribute, Object value) { + assetProcessingService.sendAttributeEvent( + new AttributeEvent(assetId, attribute.getName(), value), getClass().getSimpleName()); + } +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsResource.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsResource.java new file mode 100644 index 0000000..700305b --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/DistroGopacsResource.java @@ -0,0 +1,71 @@ +/* + * 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.GopacsOrder; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrderRequest; +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsSettlementPage; + +/** + * Distro Energy GOPACS Connector API. All timestamps are epoch seconds. A non-2xx response surfaces + * as a {@link WebApplicationException} carrying the status, e.g. 423 for a locked order. + */ +@Path("/") +@Consumes(APPLICATION_JSON) +@Produces(APPLICATION_JSON) +public interface DistroGopacsResource { + + /** 202 for a new order, 200 with the stored order when the reference already exists. */ + @POST + @Path("gopacs/{tenantId}/orders") + GopacsOrder createOrder(@PathParam("tenantId") String tenantId, GopacsOrderRequest request); + + @GET + @Path("gopacs/{tenantId}/orders/{reference}") + GopacsOrder getOrder( + @PathParam("tenantId") String tenantId, @PathParam("reference") String reference); + + /** Only quantity, price, metadata and expiration may differ from the stored order. */ + @PUT + @Path("gopacs/{tenantId}/orders/{reference}") + GopacsOrder replaceOrder( + @PathParam("tenantId") String tenantId, + @PathParam("reference") String reference, + GopacsOrderRequest request); + + /** Synchronous: returns only once Distro has confirmed the cancellation. */ + @DELETE + @Path("gopacs/{tenantId}/orders/{reference}") + void cancelOrder( + @PathParam("tenantId") String tenantId, @PathParam("reference") String reference); + + /** Matched settlements overlapping [start, end), one per effective trade. */ + @GET + @Path("settlements/gopacs/{tenantId}") + GopacsSettlementPage listSettlements( + @PathParam("tenantId") String tenantId, + @QueryParam("start") long start, + @QueryParam("end") long end, + @QueryParam("orderReference") String orderReference, + @QueryParam("limit") int limit, + @QueryParam("offset") int offset); +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrder.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrder.java new file mode 100644 index 0000000..25cae7a --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrder.java @@ -0,0 +1,66 @@ +/* + * 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; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import java.util.Map; + +/** + * Order as stored by Distro. {@code status} stays a string so a status added to the draft API does + * not break deserialisation; {@code orderId} is absent until ETPA acknowledged the order and {@code + * reason} is only present for REJECTED or LOCKED. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record GopacsOrder( + String reference, + String orderId, + String status, + String orderType, + double quantity, + double price, + long start, + long end, + String ean, + Map metadata, + Long customExpirationTime, + String reason) { + + /** The order as tracked before Distro has answered the create call. */ + public static GopacsOrder pending(GopacsOrderRequest request) { + return new GopacsOrder( + request.reference(), + null, + "PENDING", + request.orderType(), + request.quantity(), + request.price(), + request.start(), + request.end(), + request.ean(), + request.metadata(), + request.customExpirationTime(), + null); + } + + /** The replace body that keeps everything but the price. */ + public GopacsOrderRequest toRequest(double price) { + return new GopacsOrderRequest( + reference, orderType, quantity, start, end, price, ean, metadata, customExpirationTime); + } +} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrderRequest.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrderRequest.java new file mode 100644 index 0000000..8e4fd3a --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsOrderRequest.java @@ -0,0 +1,37 @@ +/* + * 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; + +import java.util.Map; + +/** + * Body of a create or replace call. Quantity is in MW, price in EUR/MWh, timestamps in epoch + * seconds on quarter-hour boundaries. A null {@code customExpirationTime} is left out of the JSON, + * so Distro applies its default of 15 minutes before start. + */ +public record GopacsOrderRequest( + String reference, + String orderType, + double quantity, + long start, + long end, + double price, + String ean, + Map metadata, + Long customExpirationTime) {} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlement.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlement.java new file mode 100644 index 0000000..b5eb460 --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlement.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.dto; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; + +/** One effective ETPA trade. Capacity in MW, volume in MWh, price in EUR/MWh, cost in EUR. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public record GopacsSettlement( + String tradeId, + String orderReference, + String direction, + long start, + long end, + double capacity, + double volume, + double price, + double cost, + long matchedAt, + long settledAt) {} diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlementPage.java b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlementPage.java new file mode 100644 index 0000000..fe38b39 --- /dev/null +++ b/ems/src/main/java/org/openremote/extension/ems/manager/distroenergy/dto/GopacsSettlementPage.java @@ -0,0 +1,25 @@ +/* + * 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; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import java.util.List; + +@JsonIgnoreProperties(ignoreUnknown = true) +public record GopacsSettlementPage(List items, int total) {} diff --git a/ems/src/main/resources/ems/config/asset-types.json b/ems/src/main/resources/ems/config/asset-types.json index be21477..f292080 100644 --- a/ems/src/main/resources/ems/config/asset-types.json +++ b/ems/src/main/resources/ems/config/asset-types.json @@ -278,7 +278,7 @@ }, { "type": "info", - "title": "Configuration", + "title": "Day-ahead — Configuration", "properties": { "include": [] }, @@ -288,7 +288,7 @@ }, { "type": "info", - "title": "Status", + "title": "Day-ahead — Status", "properties": { "include": [] }, @@ -296,6 +296,72 @@ "include": ["lastSubmission", "daysSubmitted"] } }, + { + "type": "info", + "column": 1, + "title": "GOPACS — Configuration", + "properties": { + "include": [] + }, + "attributes": { + "include": ["gopacsEnabled"] + } + }, + { + "type": "info", + "title": "GOPACS — Announcement", + "properties": { + "include": [] + }, + "attributes": { + "include": [ + "gopacsEan", + "gopacsAnnouncementId", + "gopacsStartTime", + "gopacsEndTime", + "gopacsBidValidityEnd", + "gopacsEanEffectivity" + ] + } + }, + { + "type": "info", + "title": "GOPACS — Order", + "properties": { + "include": [] + }, + "attributes": { + "include": [ + "gopacsOrderType", + "gopacsOrderQuantity", + "gopacsOrderPrice", + "gopacsConfirmOrder", + "gopacsCancelOrders" + ] + } + }, + { + "type": "info", + "column": 1, + "title": "GOPACS — Status", + "properties": { + "include": [] + }, + "attributes": { + "include": ["gopacsOrders", "gopacsLastSync"] + } + }, + { + "type": "info", + "column": 1, + "title": "GOPACS — History", + "properties": { + "include": [] + }, + "attributes": { + "include": ["gopacsOrderHistory", "gopacsSettlementHistory"] + } + }, { "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 40ab7be..915d3c7 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,11 +22,14 @@ 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.extension.ems.manager.distroenergy.DistroGopacsHandler +import org.openremote.extension.ems.manager.distroenergy.DistroGopacsResource import org.openremote.manager.asset.AssetProcessingService 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.AttributeEvent import org.openremote.model.attribute.AttributeRef import org.openremote.model.util.ValueUtil import spock.lang.Specification @@ -53,7 +56,10 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { static final String PARENT_ID = "optimisationAsset1" List createdHandlers + List createdGopacsHandlers + Map gopacsConfig Map handlerConfig + AssetStorageService assetStorageService EmsOptimisationService service def setupSpec() { @@ -65,9 +71,11 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { } def setup() { - def assetStorageService = Mock(AssetStorageService) + assetStorageService = Mock(AssetStorageService) assetStorageService.find(PARENT_ID, false, EmsEnergyOptimisationAsset.class) >> new EmsEnergyOptimisationAsset("parent") + // Saving a Distro asset looks for a GOPACS sibling to copy the announcement inputs from. + assetStorageService.findAll(_) >> [] handlerConfig = [(DistroEnergyHandler.DISTRO_ENERGY_CLIENT_KEY): "test-key"] def handlerContainer = Stub(Container) { @@ -79,6 +87,16 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { } createdHandlers = [] + createdGopacsHandlers = [] + + gopacsConfig = [(DistroGopacsHandler.DISTRO_GOPACS_TENANT_ID): "tenant"] + def gopacsContainer = Stub(Container) { + getConfig() >> { gopacsConfig } + getService(TimerService) >> Stub(TimerService) + getScheduledExecutor() >> Stub(ScheduledExecutorService) + getService(AssetStorageService) >> Stub(AssetStorageService) + getService(AssetProcessingService) >> Stub(AssetProcessingService) + } service = new EmsOptimisationService() service.services = Services.builder().withAssetStorageService(assetStorageService).build() @@ -92,6 +110,20 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { return handler } } + def gopacsResource = Stub(DistroGopacsResource) + service.distroGopacsHandlerFactory = new DistroGopacsHandler.Factory(gopacsContainer) { + @Override + DistroGopacsHandler createHandler(String assetId) { + def handler = new RecordingDistroGopacsHandler(assetId, gopacsContainer, gopacsResource) + createdGopacsHandlers << handler + return handler + } + } + } + + private static EmsDistroEnergyAsset gopacsAsset(Boolean enabled) { + return new EmsDistroEnergyAsset("distro").setId(ASSET_ID).setParentId(PARENT_ID) + .setGopacsEnabled(enabled) } private static EmsDistroEnergyAsset distroAsset(String portfolio = "portfolio-a") { @@ -159,6 +191,110 @@ class EmsOptimisationServiceDistroEnergyTest extends Specification { "non-numeric interval" | DistroEnergyHandler.DISTRO_ENERGY_REQUEST_INTERVAL | "hourly" } + def "the GOPACS order handler runs while gopacsEnabled is true, independent of day-ahead"() { + when: "a GOPACS-only asset without a portfolio is created" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(true))) + + then: "only the GOPACS handler is deployed" + createdHandlers.isEmpty() + createdGopacsHandlers.size() == 1 + createdGopacsHandlers[0].deployCount == 1 + + when: "the asset is saved again while still enabled" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(true))) + + then: "the running handler is left alone" + createdGopacsHandlers.size() == 1 + createdGopacsHandlers[0].undeployCalls.isEmpty() + + when: "it is disabled" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(false))) + + then: "it is stopped for good, cancelling its open orders" + createdGopacsHandlers[0].undeployCalls == [true] + } + + def "deleting the asset undeploys its GOPACS order handler"() { + given: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(true))) + + when: + service.processAssetChange(event(PersistenceEvent.Cause.DELETE, gopacsAsset(true))) + + then: "it is stopped for good, cancelling its open orders" + createdGopacsHandlers[0].undeployCalls == [true] + } + + def "stopping the service leaves the open orders for the next start"() { + given: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(true))) + + when: + service.stop(null) + + then: + createdGopacsHandlers[0].undeployCalls == [false] + } + + def "only the GOPACS order inputs are routed to the handler, not its own writes"() { + given: + def asset = gopacsAsset(true) + assetStorageService.find(ASSET_ID) >> asset + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, asset)) + + when: + [ + EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER, + EmsDistroEnergyAsset.GOPACS_ORDERS, + EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY, + EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY + ].each { + service.processAttributeEventEmsDistroEnergyAsset(new AttributeEvent(ASSET_ID, it.name, null)) + } + + then: + createdGopacsHandlers[0].events == [EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name] + } + + def "a missing GOPACS tenant is caught by the service rather than failing it"() { + given: + gopacsConfig.clear() + + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(true))) + + then: + noExceptionThrown() + createdGopacsHandlers.isEmpty() + } + + // Records deploy()/undeploy() calls instead of polling Distro. + static class RecordingDistroGopacsHandler extends DistroGopacsHandler { + int deployCount = 0 + List undeployCalls = [] + List events = [] + + RecordingDistroGopacsHandler( + String assetId, Container container, DistroGopacsResource resource) { + super(assetId, container, resource) + } + + @Override + void deploy() { + deployCount++ + } + + @Override + void undeploy(boolean cancelOpenOrders) { + undeployCalls << cancelOpenOrders + } + + @Override + void onAttributeEvent(AttributeEvent event) { + events << event.name + } + } + // Records deploy()/undeploy() calls instead of scheduling submissions or closing a real client. static class RecordingDistroEnergyHandler extends DistroEnergyHandler { int deployCount = 0 diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsInputsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsInputsTest.groovy new file mode 100644 index 0000000..83ca5a7 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsInputsTest.groovy @@ -0,0 +1,237 @@ +/* + * 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.extension.ems.agent.EmsDistroEnergyAsset +import org.openremote.extension.ems.agent.EmsGOPACSAsset +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.asset.AssetStorageService +import org.openremote.model.PersistenceEvent +import org.openremote.model.asset.Asset +import org.openremote.model.attribute.AttributeEvent +import org.openremote.model.query.AssetQuery +import org.openremote.model.util.ValueUtil +import spock.lang.Specification + +/** + * Exercises how {@link EmsOptimisationService} copies the announcement inputs from an + * {@link EmsGOPACSAsset} to the {@link EmsDistroEnergyAsset}s that share its parent. Storage hands + * out the assets in {@code storedAssets}, answering {@code findAll} by parent and type, and every + * attribute event the service sends is recorded in {@code sent}. + */ +class EmsOptimisationServiceGopacsInputsTest extends Specification { + + static final String PARENT_ID = "optimisationAsset1" + static final String GOPACS_ID = "gopacsAsset1" + static final String DISTRO_ID = "distroAsset1" + static final String OTHER_DISTRO_ID = "distroAsset2" + static final String EAN = "ean.265987182507322951" + static final String ANNOUNCEMENT_ID = "35b6767a-7be2-49d4-a743-2045674b7138" + static final long START = 1_800_000_000_000L + static final long END = START + 1_800_000L + static final long BID_VALIDITY_END = START - 3_600_000L + static final String EFFECTIVITY = "BUY_LOW_SELL_HIGH" + + List> storedAssets + List sent + EmsOptimisationService service + + def setupSpec() { + // Populates the asset model registry from this extension's own AssetModelProvider SPI + // registration, which real asset construction needs. + ValueUtil.initialise(null) + } + + def setup() { + storedAssets = [] + sent = [] + def assetStorageService = Stub(AssetStorageService) { + find(_ as String) >> { String assetId -> storedAssets.find { it.id == assetId } } + findAll(_ as AssetQuery) >> { AssetQuery query -> + storedAssets.findAll { + it.type in query.types && it.parentId in query.parents*.id + } + } + } + def assetProcessingService = Stub(AssetProcessingService) { + sendAttributeEvent(_ as AttributeEvent, _ as String) >> { AttributeEvent event, String source -> + sent << event + } + } + service = new EmsOptimisationService() + service.services = Services.builder() + .withAssetStorageService(assetStorageService) + .withAssetProcessingService(assetProcessingService) + .build() + } + + private static EmsGOPACSAsset gopacsAsset() { + def asset = new EmsGOPACSAsset("gopacs").setId(GOPACS_ID).setParentId(PARENT_ID) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.CONTRACTED_EAN).setValue(EAN) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_ANNOUNCEMENT_ID) + .setValue(ANNOUNCEMENT_ID) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_START_TIME).setValue(START) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_END_TIME).setValue(END) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_BID_VALIDITY_END) + .setValue(BID_VALIDITY_END) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_EAN_EFFECTIVITY) + .setValue(EFFECTIVITY) + return asset + } + + /** A Distro asset with every announcement input present and empty. */ + private static EmsDistroEnergyAsset distroAsset(String assetId = DISTRO_ID) { + def asset = new EmsDistroEnergyAsset("distro").setId(assetId).setParentId(PARENT_ID) + EmsOptimisationService.GOPACS_TO_DISTRO_INPUTS.values().each { + asset.getAttributes().getOrCreate(it) + } + return asset + } + + private static PersistenceEvent> event(PersistenceEvent.Cause cause, Asset asset) { + return new PersistenceEvent<>(cause, asset, null, null, null) + } + + private Map sentTo(String assetId) { + return sent.findAll { it.id == assetId }.collectEntries { [(it.name): it.value.orElse(null)] } + } + + static final Map ALL_INPUTS = [ + (EmsDistroEnergyAsset.GOPACS_EAN.name): EAN, + (EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.name): ANNOUNCEMENT_ID, + (EmsDistroEnergyAsset.GOPACS_START_TIME.name): START, + (EmsDistroEnergyAsset.GOPACS_END_TIME.name): END, + (EmsDistroEnergyAsset.GOPACS_BID_VALIDITY_END.name): BID_VALIDITY_END, + (EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY.name): EFFECTIVITY, + ] + + def "a GOPACS announcement change is copied to every Distro asset next to it"() { + given: + def gopacs = gopacsAsset() + storedAssets << gopacs << distroAsset() << distroAsset(OTHER_DISTRO_ID) + + when: + service.processAttributeEventEmsGOPACSAsset( + new AttributeEvent(GOPACS_ID, EmsGOPACSAsset.REDISPATCH_ANNOUNCEMENT_ID.name, + ANNOUNCEMENT_ID)) + + then: + sentTo(DISTRO_ID) == ALL_INPUTS + sentTo(OTHER_DISTRO_ID) == ALL_INPUTS + } + + def "inputs that already match are not written again"() { + given: + def distro = distroAsset() + distro.getAttributes().getOrCreate(EmsDistroEnergyAsset.GOPACS_EAN).setValue(EAN) + distro.getAttributes().getOrCreate(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID) + .setValue(ANNOUNCEMENT_ID) + storedAssets << gopacsAsset() << distro + + when: + service.processAttributeEventEmsGOPACSAsset( + new AttributeEvent(GOPACS_ID, EmsGOPACSAsset.REDISPATCH_START_TIME.name, START)) + + then: + sentTo(DISTRO_ID).keySet() == ALL_INPUTS.keySet() - [ + EmsDistroEnergyAsset.GOPACS_EAN.name, + EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.name + ] + } + + def "a closed announcement clears the Distro inputs"() { + given: + def gopacs = gopacsAsset() + gopacs.getAttributes().getOrCreate(EmsGOPACSAsset.REDISPATCH_ANNOUNCEMENT_ID).setValue(null) + def distro = distroAsset() + distro.getAttributes().getOrCreate(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID) + .setValue(ANNOUNCEMENT_ID) + storedAssets << gopacs << distro + + when: + service.processAttributeEventEmsGOPACSAsset( + new AttributeEvent(GOPACS_ID, EmsGOPACSAsset.REDISPATCH_ANNOUNCEMENT_ID.name, null)) + + then: + sentTo(DISTRO_ID).containsKey(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.name) + sentTo(DISTRO_ID)[EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.name] == null + } + + def "a GOPACS attribute outside the announcement inputs copies nothing"() { + given: + storedAssets << gopacsAsset() << distroAsset() + + when: + service.processAttributeEventEmsGOPACSAsset( + new AttributeEvent(GOPACS_ID, EmsGOPACSAsset.REDISPATCH_BID_PRICE.name, 50d)) + + then: + sent.isEmpty() + } + + def "a Distro asset created next to a GOPACS asset gets its inputs"() { + given: + def distro = distroAsset() + storedAssets << gopacsAsset() << distro + + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distro)) + + then: + sentTo(DISTRO_ID) == ALL_INPUTS + } + + def "a GOPACS asset created next to a Distro asset fills its inputs"() { + given: + def gopacs = gopacsAsset() + storedAssets << gopacs << distroAsset() + + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacs)) + + then: + sentTo(DISTRO_ID) == ALL_INPUTS + } + + def "nothing is copied while the Distro asset has more than one GOPACS sibling"() { + given: + def distro = distroAsset() + def otherGopacs = gopacsAsset().setId("gopacsAsset2") + storedAssets << gopacsAsset() << otherGopacs << distro + + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distro)) + + then: + sent.isEmpty() + } + + def "an input the Distro asset lacks is skipped and the others are still copied"() { + given: + def distro = distroAsset() + distro.getAttributes().remove(EmsDistroEnergyAsset.GOPACS_EAN.name) + storedAssets << gopacsAsset() << distro + + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, distro)) + + then: + sentTo(DISTRO_ID) == ALL_INPUTS - [(EmsDistroEnergyAsset.GOPACS_EAN.name): EAN] + } +} diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerHttpTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerHttpTest.groovy new file mode 100644 index 0000000..876cdeb --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerHttpTest.groovy @@ -0,0 +1,204 @@ +/* + * 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 com.github.tomakehurst.wiremock.WireMockServer +import jakarta.ws.rs.WebApplicationException +import org.jboss.resteasy.client.jaxrs.ResteasyClient +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrderRequest +import spock.lang.Shared +import spock.lang.Specification + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse +import static com.github.tomakehurst.wiremock.client.WireMock.containing +import static com.github.tomakehurst.wiremock.client.WireMock.delete +import static com.github.tomakehurst.wiremock.client.WireMock.deleteRequestedFor +import static com.github.tomakehurst.wiremock.client.WireMock.equalTo +import static com.github.tomakehurst.wiremock.client.WireMock.equalToJson +import static com.github.tomakehurst.wiremock.client.WireMock.get +import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor +import static com.github.tomakehurst.wiremock.client.WireMock.okJson +import static com.github.tomakehurst.wiremock.client.WireMock.post +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor +import static com.github.tomakehurst.wiremock.client.WireMock.put +import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo +import static org.openremote.container.web.WebTargetBuilder.createClient + +/** + * HTTP contract of the Distro GOPACS client built by {@link DistroGopacsHandler#createResource}: + * the client-credentials token request, bearer reuse across calls, the JSON wire format of an order + * and how error statuses surface. The unit spec covers what the handler does with the results. + */ +class DistroGopacsHandlerHttpTest extends Specification { + + static final String TENANT = "slimstrandnet" + static final String TOKEN_PATH = "/oauth2/token" + static final String ORDERS_PATH = "/v1/gopacs/${TENANT}/orders" + static final String ORDER_JSON = '''{ + "reference": "ref-1", "orderId": "etpa-1", "tenantId": "slimstrandnet", "status": "PENDING", + "orderType": "SELL", "quantity": 0.2, "price": 45, "start": 1792677600, "end": 1792684800, + "ean": "871685920001768800", "metadata": {"announcementId": "a-1"}, + "createdAt": 1792663200, "updatedAt": 1792663200, "lastSyncedAt": 1792663200, + "someFutureField": true + }''' + + @Shared + WireMockServer wireMock + ResteasyClient client + DistroGopacsResource resource + + def setupSpec() { + wireMock = new WireMockServer(0) + wireMock.start() + } + + def cleanupSpec() { + wireMock?.stop() + } + + def setup() { + wireMock.resetAll() + wireMock.stubFor(post(urlPathEqualTo(TOKEN_PATH)) + .willReturn(okJson('{"access_token": "token-1", "token_type": "Bearer", "expires_in": 3600}'))) + String base = "http://localhost:${wireMock.port()}" + client = createClient(org.openremote.container.Container.EXECUTOR) + resource = DistroGopacsHandler.createResource( + client, "${base}/v1", "${base}${TOKEN_PATH}", "client-id", "client-secret") + } + + def cleanup() { + client?.close() + } + + def "an order is posted as JSON in epoch seconds with a cached bearer token"() { + given: + wireMock.stubFor(post(urlPathEqualTo(ORDERS_PATH)) + .willReturn(aResponse().withStatus(202).withHeader("Content-Type", "application/json") + .withBody(ORDER_JSON))) + wireMock.stubFor(get(urlPathEqualTo("${ORDERS_PATH}/ref-1")).willReturn(okJson(ORDER_JSON))) + def request = new GopacsOrderRequest("ref-1", "SELL", 0.2, 1792677600, 1792684800, 45, + "871685920001768800", [announcementId: "a-1"], null) + + when: + def created = resource.createOrder(TENANT, request) + def fetched = resource.getOrder(TENANT, "ref-1") + + then: "the response is read, unknown fields ignored" + created.status() == "PENDING" + created.orderId() == "etpa-1" + created.metadata() == [announcementId: "a-1"] + fetched.reference() == "ref-1" + + and: "the body carries the order and leaves out the unset expiration" + wireMock.verify(postRequestedFor(urlPathEqualTo(ORDERS_PATH)) + .withHeader("Authorization", equalTo("Bearer token-1")) + .withRequestBody(equalToJson('''{ + "reference": "ref-1", "orderType": "SELL", "quantity": 0.2, + "start": 1792677600, "end": 1792684800, "price": 45.0, + "ean": "871685920001768800", "metadata": {"announcementId": "a-1"} + }'''))) + + and: "the token is fetched once with HTTP Basic client credentials and the GOPACS scope" + wireMock.verify(1, postRequestedFor(urlPathEqualTo(TOKEN_PATH)) + .withHeader("Authorization", equalTo("Basic " + "client-id:client-secret".bytes.encodeBase64())) + .withRequestBody(containing("grant_type=client_credentials")) + .withRequestBody(containing("scope=distro-api%2Fgopacs.access"))) + wireMock.verify(getRequestedFor(urlPathEqualTo("${ORDERS_PATH}/ref-1")) + .withHeader("Authorization", equalTo("Bearer token-1"))) + } + + def "posting an existing reference returns the stored order"() { + given: + wireMock.stubFor(post(urlPathEqualTo(ORDERS_PATH)).willReturn(okJson(ORDER_JSON))) + + when: + def stored = resource.createOrder(TENANT, new GopacsOrderRequest("ref-1", "SELL", 0.2, + 1792677600, 1792684800, 45, "871685920001768800", [announcementId: "a-1"], null)) + + then: + stored.reference() == "ref-1" + stored.status() == "PENDING" + } + + def "a locked order surfaces its 423 status"() { + given: + wireMock.stubFor(put(urlPathEqualTo("${ORDERS_PATH}/ref-1")).willReturn(aResponse() + .withStatus(423).withHeader("Content-Type", "application/json") + .withBody('{"code": "ORDER_LOCKED", "message": "locked", "details": [], "requestId": "r"}'))) + + when: + resource.replaceOrder(TENANT, "ref-1", new GopacsOrderRequest("ref-1", "SELL", 0.2, + 1792677600, 1792684800, 60, "871685920001768800", null, null)) + + then: + def e = thrown(WebApplicationException) + e.response.status == 423 + } + + def "a rejected call is described with the error body Distro sent"() { + given: + wireMock.stubFor(post(urlPathEqualTo(ORDERS_PATH)).willReturn(aResponse() + .withStatus(403).withHeader("Content-Type", "application/json") + .withBody('{"message": "Forbidden"}'))) + + when: + resource.createOrder(TENANT, new GopacsOrderRequest("ref-1", "SELL", 0.2, + 1792677600, 1792684800, 45, "871685920001768800", null, null)) + + then: + def e = thrown(WebApplicationException) + DistroGopacsHandler.describe(e) == 'HTTP 403: {"message": "Forbidden"}' + } + + def "a cancel is a DELETE answered with 204"() { + given: + wireMock.stubFor(delete(urlPathEqualTo("${ORDERS_PATH}/ref-1")).willReturn(aResponse().withStatus(204))) + + when: + resource.cancelOrder(TENANT, "ref-1") + + then: + wireMock.verify(deleteRequestedFor(urlPathEqualTo("${ORDERS_PATH}/ref-1"))) + } + + def "settlements are listed for one order over its delivery window"() { + given: + wireMock.stubFor(get(urlPathEqualTo("/v1/settlements/gopacs/${TENANT}")).willReturn(okJson('''{ + "items": [{"tradeId": "trade-1", "orderReference": "ref-1", "tenantId": "slimstrandnet", + "orderId": "etpa-1", "ean": "871685920001768800", "direction": "SELL", + "start": 1792677600, "end": 1792684800, "capacity": 0.1, "volume": 0.2, "price": 45, + "cost": 9, "volumeUnit": "MWH", "capacityUnit": "MW", "priceUnit": "EUR/MWH", + "currency": "EUR", "matchedAt": 1792670000, "settledAt": 1792690000}], + "limit": 25, "offset": 0, "total": 1}'''))) + + when: + def page = resource.listSettlements(TENANT, 1792677600, 1792684800, "ref-1", 100, 0) + + then: + page.total() == 1 + page.items()*.tradeId() == ["trade-1"] + page.items()[0].cost() == 9d + wireMock.verify(getRequestedFor(urlPathEqualTo("/v1/settlements/gopacs/${TENANT}")) + .withQueryParam("start", equalTo("1792677600")) + .withQueryParam("end", equalTo("1792684800")) + .withQueryParam("orderReference", equalTo("ref-1")) + .withQueryParam("limit", equalTo("100")) + .withQueryParam("offset", equalTo("0"))) + } +} diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerTest.groovy new file mode 100644 index 0000000..1607b78 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/distroenergy/DistroGopacsHandlerTest.groovy @@ -0,0 +1,548 @@ +/* + * 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 jakarta.ws.rs.NotFoundException +import jakarta.ws.rs.ProcessingException +import jakarta.ws.rs.WebApplicationException +import org.openremote.container.timer.TimerService +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset +import org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType +import org.openremote.extension.ems.agent.EmsElectricityBatteryAsset +import org.openremote.extension.ems.agent.EmsElectricityBatteryAsset.EmsElectricityBatteryConnectionStatusValueType +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrder +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsOrderRequest +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsSettlement +import org.openremote.extension.ems.manager.distroenergy.dto.GopacsSettlementPage +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.asset.AssetStorageService +import org.openremote.model.Container +import org.openremote.model.attribute.AttributeEvent +import org.openremote.model.util.ValueUtil +import org.openremote.model.value.AttributeDescriptor +import spock.lang.Specification + +import java.time.Instant +import java.util.concurrent.ScheduledExecutorService +import java.util.concurrent.ScheduledFuture + +import static org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType.BUY +import static org.openremote.extension.ems.agent.EmsDistroEnergyAsset.GopacsOrderTypeValueType.SELL + +/** + * Covers the order sizing rules and the attribute-driven order lifecycle of {@link + * DistroGopacsHandler}. The Distro API is a Spock mock of {@link DistroGopacsResource} handed to the + * test-support constructor, and the executor runs submitted work inline, so every reaction happens + * on the test thread. + */ +class DistroGopacsHandlerTest extends Specification { + + static final String ASSET_ID = "distroAsset1" + static final String PARENT_ID = "optimisationAsset1" + static final String TENANT = "slimstrandnet" + static final String EAN = "871685920001768800" + static final String ANNOUNCEMENT = "4f1c2b3a-0000-4000-8000-000000000001" + static final long NOW = Instant.parse("2026-10-22T10:00:00Z").toEpochMilli() + static final long START = Instant.parse("2026-10-22T14:00:00Z").toEpochMilli() + static final long END = Instant.parse("2026-10-22T16:00:00Z").toEpochMilli() + static final long VALIDITY_END = Instant.parse("2026-10-22T12:07:00Z").toEpochMilli() + + DistroGopacsResource resource + EmsDistroEnergyAsset asset + List batteries + List sent + DistroGopacsHandler handler + + def setupSpec() { + // Asset construction needs the attribute descriptors of this extension's asset types. + ValueUtil.initialise(null) + } + + def setup() { + resource = Mock(DistroGopacsResource) + asset = new EmsDistroEnergyAsset("distro").setId(ASSET_ID).setParentId(PARENT_ID) + batteries = [battery()] + sent = [] + + def assetProcessingService = Stub(AssetProcessingService) { + sendAttributeEvent(_, _) >> { AttributeEvent event, String source -> sent << event } + } + def assetStorageService = Stub(AssetStorageService) { + find(ASSET_ID) >> { asset } + findAll(_) >> { batteries } + } + def timerService = Stub(TimerService) { + getCurrentTimeMillis() >> NOW + getNow() >> Instant.ofEpochMilli(NOW) + } + def executor = Stub(ScheduledExecutorService) { + execute(_) >> { Runnable task -> task.run() } + scheduleAtFixedRate(*_) >> Stub(ScheduledFuture) + } + def container = Stub(Container) { + getConfig() >> [(DistroGopacsHandler.DISTRO_GOPACS_TENANT_ID): TENANT] + getService(TimerService) >> timerService + getService(AssetStorageService) >> assetStorageService + getService(AssetProcessingService) >> assetProcessingService + getScheduledExecutor() >> executor + } + handler = new DistroGopacsHandler(ASSET_ID, container, resource) + } + + // --- sizing rules --- + + def "category #category with buy #buyKw kW and sell #sellKw kW favours #expected"() { + expect: + DistroGopacsHandler.orderTypeFor(category, buyKw, sellKw) == expected + + where: + category | buyKw | sellKw || expected + "BUY_MAXIMUM_SELL_NONE" | 0 | 500 || BUY + "BUY_VERY_HIGH_SELL_VERY_LOW" | 0 | 500 || BUY + "BUY_HIGH_SELL_LOW" | 0 | 500 || BUY + "BUY_LOW_SELL_HIGH" | 500 | 0 || SELL + "BUY_VERY_LOW_SELL_VERY_HIGH" | 500 | 0 || SELL + "BUY_NONE_SELL_MAXIMUM" | 500 | 0 || SELL + "BUY_MODERATE_SELL_MODERATE" | 300 | 200 || BUY + "BUY_MODERATE_SELL_MODERATE" | 200 | 300 || SELL + "BUY_MODERATE_SELL_MODERATE" | 200 | 200 || BUY + "BUY_NONE_SELL_NONE" | 500 | 500 || null + "THREE_PHASE_NETWORK_REDUCE" | 500 | 500 || null + "BUY_SOME_SELL_NONE" | 500 | 500 || null + null | 500 | 500 || null + } + + def "quantity is the battery power the state of charge can hold for the window: #label"() { + expect: + DistroGopacsHandler.quantityKw(fleet, orderType, hours) == expectedKw + + where: + label | fleet | orderType | hours || expectedKw + "BUY limited by energy headroom" | [battery()] | BUY | 2 || 200 + "BUY limited by power" | [battery()] | BUY | 0.5 || 500 + "SELL uses the negative discharge" | [battery()] | SELL | 0.5 || 400 + "SELL limited by energy headroom" | [battery()] | SELL | 2 || 200 + "charge efficiency raises grid energy" | [battery(chargeEfficiency: 80)] | BUY | 2 || 200 + "discharge efficiency lowers it" | [battery(dischargeEfficiency: 50)] | SELL | 2 || 100 + "batteries add up" | [battery(), battery()] | BUY | 2 || 400 + "rounded down to 100 kW steps" | [battery(level: 45d)] | SELL | 2 || 100 + "charging not allowed" | [battery(allowCharging: false)] | BUY | 2 || 0 + "disconnected" | [battery(connection: EmsElectricityBatteryConnectionStatusValueType.disconnected)] | BUY | 2 || 0 + "already full" | [battery(level: 90d)] | BUY | 2 || 0 + "no state of charge" | [battery(level: null)] | BUY | 2 || 0 + "empty window" | [battery()] | BUY | 0 || 0 + } + + def "expiration #label"() { + expect: + DistroGopacsHandler.expirationSeconds(seconds(START), validityEnd, seconds(NOW)) == expected + + where: + label | validityEnd || expected + "is the bid validity end on a quarter" | seconds("2026-10-22T12:00:00Z") || seconds("2026-10-22T12:00:00Z") + "is rounded down to a quarter-hour" | seconds("2026-10-22T12:07:00Z") || seconds("2026-10-22T12:00:00Z") + "is left to Distro when unknown" | null || null + "may be 15 minutes before start" | seconds("2026-10-22T13:50:00Z") || seconds("2026-10-22T13:45:00Z") + "is left to Distro when after its default" | seconds("2026-10-22T14:10:00Z") || null + "is left to Distro when already passed" | seconds("2026-10-22T09:00:00Z") || null + } + + // --- suggestion --- + + def "a complete announcement pre-fills order type and quantity"() { + given: + announcement("BUY_LOW_SELL_HIGH") + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY.name, null, "BUY_LOW_SELL_HIGH") + + then: + sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_TYPE) == SELL + sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY) == 200d + } + + def "an announcement neither side helps clears the suggestion"() { + given: + announcement("BUY_NONE_SELL_NONE") + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY.name, null, "BUY_NONE_SELL_NONE") + + then: + sent.find { it.name == EmsDistroEnergyAsset.GOPACS_ORDER_TYPE.name }.value.isEmpty() + sent.find { it.name == EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY.name }.value.isEmpty() + } + + def "batteries that cannot hold the 100 kW minimum clear the suggestion"() { + given: + announcement("BUY_LOW_SELL_HIGH") + batteries = [battery(level: 12d)] + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY.name, null, "BUY_LOW_SELL_HIGH") + + then: + sent.find { it.name == EmsDistroEnergyAsset.GOPACS_ORDER_TYPE.name }.value.isEmpty() + sent.find { it.name == EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY.name }.value.isEmpty() + } + + def "an unchanged republish of an attribute does nothing"() { + given: + announcement("BUY_LOW_SELL_HIGH") + def attribute = asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID) + + when: + handler.onAttributeEvent(new AttributeEvent(asset, attribute, "AttributeLinkingService", + ANNOUNCEMENT, NOW, ANNOUNCEMENT, NOW - 1)) + + then: + sent.isEmpty() + 0 * resource._ + } + + // --- order lifecycle --- + + def "confirming posts the order in MW and epoch seconds and tracks it"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, 200, 45) + GopacsOrderRequest request = null + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + + then: + 1 * resource.createOrder(TENANT, _) >> { String tenant, GopacsOrderRequest body -> + request = body + return stored(body.reference(), "PENDING", ANNOUNCEMENT) + } + request.reference() ==~ /[a-z0-9-]{36}/ + request.orderType() == "SELL" + request.quantity() == 0.2d + request.start() == seconds(START) + request.end() == seconds(END) + request.price() == 45d + request.ean() == EAN + request.metadata() == [announcementId: ANNOUNCEMENT] + request.customExpirationTime() == seconds("2026-10-22T12:00:00Z") + handler.orders.keySet() == [request.reference()] as Set + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDERS) as Map).containsKey(request.reference()) + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY) as Map).action == "CREATED" + } + + def "confirming rounds the quantity down to 100 kW steps and the price to cents"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, 250, 45.678) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + + then: + 1 * resource.createOrder(TENANT, { + it.quantity() == 0.2d && it.price() == 45.68d + }) >> { String tenant, GopacsOrderRequest body -> + stored(body.reference(), "SUBMITTED", ANNOUNCEMENT) + } + } + + def "confirming refuses what Distro would reject: #label"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, quantityKw, price) + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_START_TIME).setValue(start) + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_BID_VALIDITY_END).setValue(validityEnd) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + + then: + 0 * resource.createOrder(*_) + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY) as Map).detail == detail + + where: + label | quantityKw | price | start | validityEnd || detail + "a quantity rounded to 0" | 99 | 45 | START | VALIDITY_END || "quantity 99.0 kW is below the 100 kW minimum" + "less than 15 minutes left" | 200 | 45 | Instant.parse("2026-10-22T10:15:00Z").toEpochMilli() | VALIDITY_END || "window starts in less than 15 minutes" + "bid validity end passed" | 200 | 45 | START | NOW || "bid validity end has passed" + } + + def "a failed create is recorded and nothing is tracked"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, 200, 45) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + + then: + 1 * resource.createOrder(TENANT, _) >> { throw new WebApplicationException(403) } + handler.orders.isEmpty() + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY) as Map).detail == "HTTP 403" + } + + def "a create that times out stays tracked as pending so polling can find it"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, 200, 45) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + + then: + 1 * resource.createOrder(TENANT, _) >> { throw new ProcessingException("Read timed out") } + handler.orders.size() == 1 + handler.orders.values().first().status() == "PENDING" + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDERS) as Map).size() == 1 + (sentValue(EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY) as Map).action == "CREATE_FAILED" + } + + def "a price change replaces the submitted orders of the current announcement only"() { + given: + announcement("BUY_LOW_SELL_HIGH") + track(stored("open-current", "SUBMITTED", ANNOUNCEMENT)) + track(stored("pending-current", "PENDING", ANNOUNCEMENT)) + track(stored("locked-current", "LOCKED", ANNOUNCEMENT)) + track(stored("open-other", "SUBMITTED", "other-announcement")) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_ORDER_PRICE.name, 45d, 60.004d) + + then: + 1 * resource.replaceOrder(TENANT, "open-current", { + it.price() == 60d && it.reference() == "open-current" + }) >> + stored("open-current", "SUBMITTED", ANNOUNCEMENT, 60) + 0 * resource.replaceOrder(*_) + handler.orders["open-current"].price() == 60d + } + + def "a new announcement cancels the open orders placed for the previous one"() { + given: + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID).setValue("next") + track(stored("open-previous", "PENDING", ANNOUNCEMENT)) + track(stored("matched-previous", "MATCHED", ANNOUNCEMENT)) + track(stored("open-next", "PENDING", "next")) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID.name, ANNOUNCEMENT, "next") + + then: + 1 * resource.cancelOrder(TENANT, "open-previous") + 0 * resource.cancelOrder(*_) + handler.orders.keySet() == ["matched-previous", "open-next"] as Set + } + + def "the cancel flag cancels every open order and is reset"() { + given: + track(stored("a", "PENDING", ANNOUNCEMENT)) + track(stored("b", "SUBMITTED", "other-announcement")) + track(stored("c", "LOCKED", ANNOUNCEMENT)) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS.name, false, true) + + then: + 1 * resource.cancelOrder(TENANT, "a") + 1 * resource.cancelOrder(TENANT, "b") + 0 * resource.cancelOrder(*_) + handler.orders.keySet() == ["c"] as Set + } + + def "polling keeps live orders, records settled trades and drops final orders"() { + given: + track(stored("live", "SUBMITTED", ANNOUNCEMENT)) + track(stored("settled", "MATCHED", ANNOUNCEMENT)) + track(stored("gone", "PENDING", ANNOUNCEMENT)) + def trade = new GopacsSettlement("trade-1", "settled", "BUY", seconds(START), seconds(END), + 0.2, 0.4, 45, 18, seconds(NOW), seconds(NOW)) + + when: + handler.pollOrders() + + then: + 1 * resource.getOrder(TENANT, "live") >> stored("live", "MATCHED", ANNOUNCEMENT) + 1 * resource.getOrder(TENANT, "settled") >> stored("settled", "SETTLED", ANNOUNCEMENT) + 1 * resource.getOrder(TENANT, "gone") >> { throw new NotFoundException() } + 1 * resource.listSettlements(TENANT, seconds(START), seconds(END), "settled", 100, 0) >> + new GopacsSettlementPage([trade], 1) + handler.orders.keySet() == ["live"] as Set + handler.orders["live"].status() == "MATCHED" + sent.findAll { + it.name == EmsDistroEnergyAsset.GOPACS_SETTLEMENT_HISTORY.name + }*.value*.get()*.tradeId == ["trade-1"] + sentValue(EmsDistroEnergyAsset.GOPACS_LAST_SYNC) == NOW + } + + def "a failed settlement fetch is recorded and the settled order dropped"() { + given: + track(stored("settled", "MATCHED", ANNOUNCEMENT)) + + when: + handler.pollOrders() + + then: + 1 * resource.getOrder(TENANT, "settled") >> stored("settled", "SETTLED", ANNOUNCEMENT) + 1 * resource.listSettlements(*_) >> { throw new NotFoundException() } + handler.orders.isEmpty() + sent.findAll { + it.name == EmsDistroEnergyAsset.GOPACS_ORDER_HISTORY.name + }*.value*.get()*.action == + ["STATUS", "SETTLEMENTS_FAILED"] + sent.findAll { it.name == EmsDistroEnergyAsset.GOPACS_SETTLEMENT_HISTORY.name }.isEmpty() + } + + def "settlements are read page by page until the total is reached"() { + given: + track(stored("settled", "MATCHED", ANNOUNCEMENT)) + def trade = { String id -> + new GopacsSettlement(id, "settled", "BUY", seconds(START), seconds(END), + 0.1, 0.2, 45, 9, seconds(NOW), seconds(NOW)) + } + + when: + handler.pollOrders() + + then: + 1 * resource.getOrder(TENANT, "settled") >> stored("settled", "SETTLED", ANNOUNCEMENT) + 1 * resource.listSettlements(TENANT, seconds(START), seconds(END), "settled", 100, 0) >> + new GopacsSettlementPage((1..100).collect { trade("t-$it") }, 101) + 1 * resource.listSettlements(TENANT, seconds(START), seconds(END), "settled", 100, 100) >> + new GopacsSettlementPage([trade("t-101")], 101) + sent.findAll { it.name == EmsDistroEnergyAsset.GOPACS_SETTLEMENT_HISTORY.name }.size() == 101 + } + + def "an undeployed handler ignores queued work"() { + given: + announcement("BUY_LOW_SELL_HIGH") + order(SELL, 200, 45) + track(stored("live", "SUBMITTED", ANNOUNCEMENT)) + handler.undeploy(false) + + when: + handler.processAttributeChange(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name, false, true) + handler.pollOrders() + + then: + 0 * resource._ + sent.isEmpty() + } + + def "undeploying for good cancels the open orders and keeps the rest"() { + given: + track(stored("pending", "PENDING", ANNOUNCEMENT)) + track(stored("submitted", "SUBMITTED", "other-announcement")) + track(stored("matched", "MATCHED", ANNOUNCEMENT)) + + when: + handler.undeploy(true) + + then: + 1 * resource.cancelOrder(TENANT, "pending") + 1 * resource.cancelOrder(TENANT, "submitted") + 0 * resource._ + handler.orders.keySet() == ["matched"] as Set + } + + def "deploy restores the orders saved on the asset"() { + given: + def saved = stored("saved", "SUBMITTED", ANNOUNCEMENT) + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ORDERS) + .setValue(ValueUtil.JSON.convertValue([saved: saved], org.openremote.model.value.ValueType.ObjectMap)) + + when: + handler.deploy() + + then: + handler.orders == [saved: saved] + } + + def "deploy resets a confirm flag left true by an earlier undeploy"() { + given: + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER).setValue(true) + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_CANCEL_ORDERS).setValue(false) + + when: + handler.deploy() + + then: + sent*.name == [EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER.name] + sentValue(EmsDistroEnergyAsset.GOPACS_CONFIRM_ORDER) == false + } + + // --- helpers --- + + static long seconds(long millis) { + return millis.intdiv(1000) + } + + static long seconds(String instant) { + return Instant.parse(instant).epochSecond + } + + static EmsElectricityBatteryAsset battery(Map overrides = [:]) { + Map values = [allowCharging : true, allowDischarging: true, capacity: 1000d, level: 50d, + max : 90, min: 10, chargeMax: 500d, dischargeMax: -400d, + chargeEfficiency : 100, dischargeEfficiency: 100, + connection : EmsElectricityBatteryConnectionStatusValueType.connected] + overrides + def battery = new EmsElectricityBatteryAsset("battery") + def attributes = battery.attributes + attributes.getOrCreate(EmsElectricityBatteryAsset.ALLOW_CHARGING).setValue(values.allowCharging) + attributes.getOrCreate(EmsElectricityBatteryAsset.ALLOW_DISCHARGING).setValue(values.allowDischarging) + attributes.getOrCreate(EmsElectricityBatteryAsset.ENERGY_CAPACITY).setValue(values.capacity) + attributes.getOrCreate(EmsElectricityBatteryAsset.ENERGY_LEVEL_PERCENTAGE).setValue(values.level) + attributes.getOrCreate(EmsElectricityBatteryAsset.ENERGY_LEVEL_PERCENTAGE_MAXIMUM).setValue(values.max) + attributes.getOrCreate(EmsElectricityBatteryAsset.ENERGY_LEVEL_PERCENTAGE_MINIMUM).setValue(values.min) + attributes.getOrCreate(EmsElectricityBatteryAsset.CHARGE_POWER_MAXIMUM).setValue(values.chargeMax) + attributes.getOrCreate(EmsElectricityBatteryAsset.DISCHARGE_POWER_MAXIMUM).setValue(values.dischargeMax) + attributes.getOrCreate(EmsElectricityBatteryAsset.CHARGE_EFFICIENCY).setValue(values.chargeEfficiency) + attributes.getOrCreate(EmsElectricityBatteryAsset.DISCHARGE_EFFICIENCY).setValue(values.dischargeEfficiency) + attributes.getOrCreate(EmsElectricityBatteryAsset.CONNECTION_STATUS).setValue(values.connection) + return battery + } + + void announcement(String category) { + def attributes = asset.attributes + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_EAN).setValue(EAN) + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ANNOUNCEMENT_ID).setValue(ANNOUNCEMENT) + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_START_TIME).setValue(START) + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_END_TIME).setValue(END) + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_BID_VALIDITY_END).setValue(VALIDITY_END) + attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_EAN_EFFECTIVITY).setValue(category) + } + + void order(GopacsOrderTypeValueType orderType, double quantityKw, double price) { + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ORDER_TYPE).setValue(orderType) + asset.attributes.getOrCreate(EmsDistroEnergyAsset.GOPACS_ORDER_QUANTITY).setValue(quantityKw) + asset.setGopacsOrderPrice(price) + } + + static GopacsOrder stored(String reference, String status, String announcementId, double price = 45) { + return new GopacsOrder(reference, null, status, "SELL", 0.2, price, seconds(START), seconds(END), + EAN, [announcementId: announcementId], null, null) + } + + void track(GopacsOrder order) { + handler.orders.put(order.reference(), order) + } + + Object sentValue(AttributeDescriptor attribute) { + return sent.reverse().find { it.name == attribute.name }?.value?.orElse(null) + } +}