Merge pull request #271 from ie3-institute/ms/#267-refactoring-data-connections

Refactoring data connections and container.

Author: staudtMarius

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 - Removed Jenkinsfile [#290](https://github.com/ie3-institute/simonaAPI/issues/290)
 - Adapted dependabot workflow and added CODEOWNERS [#294](https://github.com/ie3-institute/simonaAPI/issues/294)
+- Refactoring external data connection [#267](https://github.com/ie3-institute/simonaAPI/issues/267)
+- Refactoring data containers [#268](https://github.com/ie3-institute/simonaAPI/issues/268)
 
 ## [0.9.0] - 2025-05-09
 

docs/readthedocs/connections/connections.md

@@ -1 +1,40 @@
 # Data connections
+
+This API defines some data connections in order to exchange data between SIMONA and an external simulation.
+
+1. Input data connections
+2. Output data connections
+3. Bidirectional data connections
+
+## Input data connections
+
+These data connections are used to provide SIMONA with external data. Each input data connection contains two references
+for SIMONA actors. The first reference specifies the SIMONA service that will receive the external data. The second
+reference is for the external simulation adapter. This adapter receives a message to schedule the data service in SIMONA.
+
+The process of sending data to the service and requesting scheduling is handled by the `sendExtMsg` method.
+To send a message, simply call the method with the message as input.
+
+Currently, the following input data connections are provided:
+- ExtPrimaryDataConnection
+
+
+## Output data connections
+
+These data connections are used to provide SIMONA response messages to the external simulation. Each output data connection
+has a queue for messages sent by SIMONA. The result data connection itself cannot send messages to SIMONA.
+
+Currently, the following input data connections are provided:
+- ExtResultListener
+
+
+## Bidirectional data connections
+
+The bidirectional data connection combines the functionality of both input and output data connections. These data connections
+can be used to send data to SIMONA and receive responses. One additional feature is that bidirectional data connections
+can be used to request responses, e.g. SIMONA results.
+
+Currently, the following input data connections are provided:
+- ExtEmDataConnection
+- ExtEvDataConnection
+- ExtResultDataConnection

docs/readthedocs/simulations/externalsimulation.md

@@ -32,3 +32,20 @@ method should provide a new tick as long as there are future activities.
 
 The method `getDataConnections` returns all [data connections](/connections/connections) that are used to connect the
 external simulation to SIMONA.
+
+
+## Communication between SIMONA and an external simulation
+
+Depending on the connections used by the external simulation, the following communication structures are used:
+
+- Input connections: external simulation -> simonaAPI -> SIMONA
+- Output connections: SIMONA -> simonaAPI -> external simulation
+
+Bidirectional connections can use both structures.
+
+
+## Co-simulations
+
+This API provides an extension to the class `edu.ie3.simona.api.simulation.ExtSimulation` in the form of 
+`edu.ie3.simona.api.simulation.ExtCoSimulation`. This class contains some methods to simplify the definition of external
+co-simulations.

src/main/java/edu/ie3/simona/api/data/DataQueueExtSimulationExtSimulator.java

@@ -0,0 +1,117 @@
+/*
+ * © 2024. TU Dortmund University,
+ * Institute of Energy Systems, Energy Efficiency and Energy Economics,
+ * Research group Distribution grid planning and operation
+ */
+
+package edu.ie3.simona.api.data;
+
+import edu.ie3.simona.api.data.container.ExtDataContainer;
+import java.util.Optional;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+/** Data queue to allow data flow between SimonaAPI and an external simulation */
+public final class ExtDataContainerQueue<V extends ExtDataContainer> {
+  private final LinkedBlockingDeque<V> receiverTriggerDeque = new LinkedBlockingDeque<>();
+
+  /** Returns the number of elements in this queue. */
+  public int size() {
+    return receiverTriggerDeque.size();
+  }
+
+  /**
+   * Method for adding an {@link ExtDataContainer} to the queue.
+   *
+   * @param data to be added
+   * @throws InterruptedException if the thread running this has been interrupted during the
+   *     blocking operation
+   */
+  public void queueData(V data) throws InterruptedException {
+    receiverTriggerDeque.putLast(data);
+  }
+
+  /**
+   * Method to retrieve and remove an {@link ExtDataContainer} from the queue. This method waits
+   * (blocks) until data is added to the queue.
+   *
+   * @return a data container
+   * @throws InterruptedException if the thread running this has been interrupted during the
+   *     blocking operation
+   */
+  public V takeContainer() throws InterruptedException {
+    return receiverTriggerDeque.takeFirst();
+  }
+
+  /**
+   * Method to retrieve and remove an {@link ExtDataContainer} from the queue. This method waits
+   * (blocks) until either data is added to the queue or the specified wait time is reached.
+   *
+   * @param timeout maximal time to wait for data
+   * @param unit unit of the timeout
+   * @return an option for a data container
+   * @throws InterruptedException if the thread running this has been interrupted during the
+   *     blocking operation
+   */
+  public Optional<V> pollContainer(long timeout, TimeUnit unit) throws InterruptedException {
+    return Optional.ofNullable(receiverTriggerDeque.pollFirst(timeout, unit));
+  }
+
+  /**
+   * Method to retrieve only a part of a container from the queue. This method waits (blocks) until
+   * data is added to the queue.
+   *
+   * @param extractor function to extract a part of the container
+   * @return the extracted part
+   * @param <R> type of returned value
+   * @throws InterruptedException if the thread running this has been interrupted during the
+   *     blocking operation
+   */
+  public <R> R takeData(Function<V, R> extractor) throws InterruptedException {
+    // removes the first container from the queue
+    V data = receiverTriggerDeque.takeFirst();
+    R result = extractor.apply(data);
+
+    // if the container is not empty, it should remain in the queue.
+    // else the container needs to be removed
+    if (!data.isEmpty()) {
+      receiverTriggerDeque.putFirst(data);
+    }
+
+    return result;
+  }
+
+  /**
+   * Method to retrieve only a part of a container from the queue. This method waits (blocks) until
+   * either data is added to the queue or the specified wait time is reached.
+   *
+   * @param extractor function to extract a part of the container
+   * @param timeout maximal time to wait for data
+   * @param unit unit of the timeout
+   * @return an option for the extracted part
+   * @param <R> type of returned value
+   * @throws InterruptedException if the thread running this has been interrupted during the
+   *     blocking operation
+   */
+  public <R> Optional<R> pollData(Function<V, R> extractor, long timeout, TimeUnit unit)
+      throws InterruptedException {
+    // removes the first container from the queue
+    Optional<V> containerOption = pollContainer(timeout, unit);
+
+    if (containerOption.isPresent()) {
+      V data = containerOption.get();
+      R result = extractor.apply(data);
+
+      // if the container is not empty, it should remain in the queue.
+      // else the container needs to be removed
+      if (!data.isEmpty()) {
+        receiverTriggerDeque.putFirst(data);
+      }
+
+      return Optional.of(result);
+    }
+
+    return Optional.empty();
+  }
+}