Implementation of typed two-parted scheduler

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Changed from ComparableQuantity to squants in power flow [#554](https://github.com/ie3-institute/simona/issues/554)
 - Reduce log level on missing diffuse irradiance [#629](https://github.com/ie3-institute/simona/issues/629)
 - Updated to gradle 8.4 [#648](https://github.com/ie3-institute/simona/issues/648)
+- Introducing new scheduling infrastructure:
+  - Two-parted scheduler in akka typed [#378](https://github.com/ie3-institute/simona/issues/378)
+  - Adapting to simonaAPI 0.3.0 (adapted message protocol)
 
 ## [3.0.0] - 2023-08-07
 

build.gradle

@@ -93,7 +93,7 @@ dependencies {
     exclude group: 'edu.ie3'
   }
 
-  implementation('com.github.ie3-institute:simonaAPI:0.2.0') {
+  implementation('com.github.ie3-institute:simonaAPI:0.3.0') {
     exclude group: 'org.apache.logging.log4j'
     exclude group: 'org.slf4j'
     /* Exclude our own nested dependencies */

docs/readthedocs/usersguide.md

@@ -151,7 +151,7 @@ In order to run an external simulation, several requirements have to be fulfille
 
 - The external simulation should be implemented in its own project (repository).
 - The project should include the *shadowJar* gradle plugin (``id "com.github.johnrengelman.shadow" version "x.y.z"``).
-- A class (called *main class* here) needs to extend ``edu.ie3.simona.api.schedule.ExtSimulation`` and thus implement the two methods ``List<Long> initialize()`` and ``List<Long> doActivity(long tick)``. The method ``initialize`` is called when the external simulation needs to be initialized whereas the method ``doActivity`` is called when time step ``tick`` is triggered. ``initialize`` and ``doActivity`` must return a list of subsequent new ticks that the sub simulation should be scheduled at.
+- A class (called *main class* here) needs to extend ``edu.ie3.simona.api.schedule.ExtSimulation`` and thus implement the two methods ``Optional<Long> initialize()`` and ``Optional<Long> doActivity(long tick)``. The method ``initialize`` is called when the external simulation needs to be initialized whereas the method ``doActivity`` is called when time step ``tick`` is triggered. ``initialize`` and ``doActivity`` can return a subsequent new tick that the sub simulation should be activated with next.
 - For each data stream, a sub-interface of ``edu.ie3.simona.api.data.ExtDataSimulation`` needs to be implemented, such as ``edu.ie3.simona.api.data.ev.ExtEvSimulation``, and all methods of the interface have to be implemented. The *main class* could be the implementing class here.
 - In order for SIMONA to use the external simulation, a class that extends ``edu.ie3.simona.api.ExtLinkInterface`` has to reside inside the project. The class has to implement the corresponding methods by returning the control stream and data stream implementations (could all be the same *main class*).
 - When loading the external simulations, SIMONA is looking for the corresponding service files of the class ``edu.ie3.simona.api.ExtLinkInterface``. Therefor every external simulation needs the following service file: ``src/main/resources/META-INF/services/edu.ie3.simona.api.ExtLinkInterface``. The service file needs to contain the relative path to the class that extends ``edu.ie3.simona.api.ExtLinkInterface``.

src/main/scala/edu/ie3/simona/actor/ActorUtil.scala

@@ -0,0 +1,20 @@
+/*
+ * © 2023. TU Dortmund University,
+ * Institute of Energy Systems, Energy Efficiency and Energy Economics,
+ * Research group Distribution grid planning and operation
+ */
+
+package edu.ie3.simona.actor
+
+import akka.actor.typed.Behavior
+import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
+
+object ActorUtil {
+  def stopOnError[M](
+      ctx: ActorContext[M],
+      msg: String
+  ): Behavior[M] = {
+    ctx.log.error(s"$msg. Stopping.")
+    Behaviors.stopped
+  }
+}

src/main/scala/edu/ie3/simona/agent/grid/DBFSAlgorithm.scala

@@ -72,8 +72,7 @@ trait DBFSAlgorithm extends PowerFlowSupport with GridResultsSupport {
     case Event(
           TriggerWithIdMessage(
             StartGridSimulationTrigger(currentTick),
-            triggerId,
-            _
+            triggerId
           ),
           gridAgentBaseData: GridAgentBaseData
         ) =>
@@ -452,11 +451,9 @@ trait DBFSAlgorithm extends PowerFlowSupport with GridResultsSupport {
       environmentRefs.scheduler ! CompletionMessage(
         simTriggerId,
         Some(
-          Vector(
-            ScheduleTriggerMessage(
-              ActivityStartTrigger(currentTick + resolution),
-              self
-            )
+          ScheduleTriggerMessage(
+            ActivityStartTrigger(currentTick + resolution),
+            self
           )
         )
       )
@@ -1011,8 +1008,8 @@ trait DBFSAlgorithm extends PowerFlowSupport with GridResultsSupport {
   /** Normally only reached by the superior (dummy) agent!
     *
     * Triggers a state transition to [[SimulateGrid]], informs the
-    * [[edu.ie3.simona.scheduler.SimScheduler]] about the finish of this sweep
-    * and requests a new trigger for itself for a new sweep (which means a new
+    * [[edu.ie3.simona.scheduler.Scheduler]] about the finish of this sweep and
+    * requests a new trigger for itself for a new sweep (which means a new
     * [[StartGridSimulationTrigger]])
     *
     * @param gridAgentBaseData
@@ -1032,9 +1029,7 @@ trait DBFSAlgorithm extends PowerFlowSupport with GridResultsSupport {
     environmentRefs.scheduler ! CompletionMessage(
       oldTrigger,
       Some(
-        Vector(
-          ScheduleTriggerMessage(StartGridSimulationTrigger(currentTick), self)
-        )
+        ScheduleTriggerMessage(StartGridSimulationTrigger(currentTick), self)
       )
     )
 

src/main/scala/edu/ie3/simona/agent/grid/GridAgent.scala

@@ -35,8 +35,6 @@ import edu.ie3.util.TimeUtil
 import java.time.ZonedDateTime
 import java.time.temporal.ChronoUnit
 import java.util.UUID
-import scala.concurrent.ExecutionContext
-import scala.concurrent.duration._
 import scala.language.postfixOps
 
 object GridAgent {
@@ -103,8 +101,7 @@ class GridAgent(
             InitializeGridAgentTrigger(
               gridAgentInitData: GridAgentInitData
             ),
-            triggerId,
-            _
+            triggerId
           ),
           _
         ) =>
@@ -174,12 +171,14 @@ class GridAgent(
 
       log.debug("Je suis initialized")
 
-      goto(Idle) using gridAgentBaseData replying CompletionMessage(
+      environmentRefs.scheduler ! CompletionMessage(
         triggerId,
         Some(
-          Vector(ScheduleTriggerMessage(ActivityStartTrigger(resolution), self))
+          ScheduleTriggerMessage(ActivityStartTrigger(resolution), self)
         )
       )
+
+      goto(Idle) using gridAgentBaseData
   }
 
   when(Idle) {
@@ -191,25 +190,25 @@ class GridAgent(
       stay()
 
     case Event(
-          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId, _),
+          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId),
           gridAgentBaseData: GridAgentBaseData
         ) =>
       log.debug("received activity start trigger {}", triggerId)
 
       unstashAll()
 
-      goto(SimulateGrid) using gridAgentBaseData replying CompletionMessage(
+      environmentRefs.scheduler ! CompletionMessage(
         triggerId,
         Some(
-          Vector(
-            ScheduleTriggerMessage(
-              StartGridSimulationTrigger(currentTick),
-              self
-            )
+          ScheduleTriggerMessage(
+            StartGridSimulationTrigger(currentTick),
+            self
           )
         )
       )
 
+      goto(SimulateGrid) using gridAgentBaseData
+
     case Event(StopMessage(_), data: GridAgentBaseData) =>
       // shutdown children
       data.gridEnv.nodeToAssetAgents.foreach { case (_, actors) =>

src/main/scala/edu/ie3/simona/agent/participant/ParticipantAgent.scala

@@ -111,8 +111,7 @@ abstract class ParticipantAgent[
                 outputConfig
               )
             ),
-            triggerId,
-            _
+            triggerId
           ),
           _: ParticipantUninitializedStateData[PD]
         ) =>
@@ -136,7 +135,7 @@ abstract class ParticipantAgent[
 
   when(Idle) {
     case Event(
-          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId, _),
+          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId),
           modelBaseStateData: ParticipantModelBaseStateData[PD, CD, M]
         ) if modelBaseStateData.services.isEmpty =>
       /* An activity start trigger is sent and no data is awaited (neither secondary nor primary). Therefore go straight
@@ -166,7 +165,7 @@ abstract class ParticipantAgent[
       )
 
     case Event(
-          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId, _),
+          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId),
           modelBaseStateData: ParticipantModelBaseStateData[PD, CD, M]
         ) =>
       /* An activity start trigger is sent, but I'm not sure yet, if secondary data will arrive. Figure out, if someone
@@ -179,7 +178,7 @@ abstract class ParticipantAgent[
       )
 
     case Event(
-          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId, _),
+          TriggerWithIdMessage(ActivityStartTrigger(currentTick), triggerId),
           fromOutsideBaseStateData: FromOutsideBaseStateData[M, PD]
         ) =>
       /* An activity start trigger is sent, but I'm still expecting primary data. Go to HandleInformation and wait for
@@ -285,8 +284,7 @@ abstract class ParticipantAgent[
     case Event(
           TriggerWithIdMessage(
             ActivityStartTrigger(activationTick),
-            triggerId,
-            _
+            triggerId
           ),
           stateData: DataCollectionStateData[PD]
         ) =>
@@ -409,7 +407,7 @@ abstract class ParticipantAgent[
       stay()
 
     case Event(_: ProvisionMessage[_], _) |
-        Event(TriggerWithIdMessage(ActivityStartTrigger(_), _, _), _) =>
+        Event(TriggerWithIdMessage(ActivityStartTrigger(_), _), _) =>
       /* I got faced to new data, also I'm not ready to handle it, yet OR I got asked to do something else, while I'm
        * still busy, I will put it aside and answer it later */
       stash()

src/main/scala/edu/ie3/simona/agent/participant/ParticipantAgentFundamentals.scala

@@ -139,12 +139,12 @@ protected trait ParticipantAgentFundamentals[
 
     /* Confirm final initialization */
     val (_, triggerId) = releaseTickAndTriggerId()
-    val newTriggerMessages =
-      ServiceActivationBaseStateData.tickToScheduleTriggerMessages(
+    val newTriggerMessage =
+      ServiceActivationBaseStateData.tickToScheduleTriggerMessage(
         senderToMaybeTick._2,
         self
       )
-    scheduler ! CompletionMessage(triggerId, newTriggerMessages)
+    scheduler ! CompletionMessage(triggerId, newTriggerMessage)
     goto(Idle) using stateData
   }
 
@@ -710,11 +710,11 @@ protected trait ParticipantAgentFundamentals[
       scheduler: ActorRef
   ): FSM.State[AgentState, ParticipantStateData[PD]] = {
     /* Determine the very next tick, where activation is required */
-    val (maybeActivationTriggers, updatedBaseStateData) =
+    val (maybeActivationTrigger, updatedBaseStateData) =
       popNextActivationTrigger(baseStateData)
 
     val (_, triggerId) = releaseTickAndTriggerId()
-    scheduler ! CompletionMessage(triggerId, maybeActivationTriggers)
+    scheduler ! CompletionMessage(triggerId, maybeActivationTrigger)
     unstashAll()
     goto(Idle) using updatedBaseStateData
   }
@@ -728,45 +728,43 @@ protected trait ParticipantAgentFundamentals[
     * @param baseStateData
     *   Base state data to be updated
     * @return
-    *   An [[Option]] to a [[Seq]] of new [[ScheduleTriggerMessage]] s as well
-    *   as the updated base state data. If the next activation tick is an
-    *   additional activation, this tick is removed from the list of desired
-    *   additional activations.
+    *   An [[Option]] of new [[ScheduleTriggerMessage]] as well as the updated
+    *   base state data. If the next activation tick is an additional
+    *   activation, this tick is removed from the list of desired additional
+    *   activations.
     */
   def popNextActivationTrigger(
       baseStateData: BaseStateData[PD]
-  ): (Option[Seq[ScheduleTriggerMessage]], BaseStateData[PD]) = {
+  ): (Option[ScheduleTriggerMessage], BaseStateData[PD]) = {
     /* Determine what comes next: An additional activation or new data - or both at once */
     val nextAdditionalActivation =
       baseStateData.additionalActivationTicks.headOption
     val nextDataTick =
       baseStateData.foreseenDataTicks.values.toSeq.sorted.headOption.flatten
 
-    /* return a [[Option]] to a [[Seq]] of [[ScheduleTriggerMessage]]s */
-    def toMessageSeq: (Long, ActorRef) => Option[Seq[ScheduleTriggerMessage]] =
+    /* return a [[Option]] of [[ScheduleTriggerMessage]]s */
+    def toMessage: (Long, ActorRef) => Option[ScheduleTriggerMessage] =
       (tick: Long, actorToBeTriggered: ActorRef) =>
         Some(
-          Seq(
-            ScheduleTriggerMessage(
-              ActivityStartTrigger(tick),
-              actorToBeTriggered
-            )
+          ScheduleTriggerMessage(
+            ActivityStartTrigger(tick),
+            actorToBeTriggered
           )
         )
 
     (nextAdditionalActivation, nextDataTick) match {
       case (None, Some(dataTick)) =>
         /* There is only a data tick available */
         (
-          toMessageSeq(dataTick, self),
+          toMessage(dataTick, self),
           baseStateData
         )
       case (Some(additionalTick), Some(dataTick))
           if dataTick < additionalTick =>
         /* The next foreseen activation will be based on foreseen data arrival. Do nothing else, then creating a
          * trigger. */
         (
-          toMessageSeq(dataTick, self),
+          toMessage(dataTick, self),
           baseStateData
         )
       case (Some(additionalTick), _) =>
@@ -784,7 +782,7 @@ protected trait ParticipantAgentFundamentals[
         )
 
         (
-          toMessageSeq(additionalTick, self),
+          toMessage(additionalTick, self),
           updatedBaseStateData
         )
       case (None, None) =>

src/main/scala/edu/ie3/simona/api/ExtMessageUtils.scala

@@ -17,23 +17,21 @@ import edu.ie3.simona.ontology.messages.SchedulerMessage.{
 }
 import edu.ie3.simona.ontology.trigger.Trigger.ActivityStartTrigger
 
-import scala.jdk.CollectionConverters.CollectionHasAsScala
+import scala.jdk.OptionConverters.RichOptional
 
 object ExtMessageUtils {
   implicit class RichExtCompletion(
       private val extCompl: ExtCompletionMessage
   ) {
     def toSimona(triggerId: Long, triggerActor: ActorRef): CompletionMessage = {
-      val newTriggers =
-        Option.when(!extCompl.newTriggers.isEmpty) {
-          extCompl.newTriggers.asScala.map { tick =>
-            ScheduleTriggerMessage(ActivityStartTrigger(tick), triggerActor)
-          }.toSeq
+      val newTrigger =
+        extCompl.nextActivation.toScala.map { tick =>
+          ScheduleTriggerMessage(ActivityStartTrigger(tick), triggerActor)
         }
 
       CompletionMessage(
         triggerId,
-        newTriggers
+        newTrigger
       )
     }
   }