Introducing primary data from SQL

CHANGELOG.md

@@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+
+### Added
+- Implement SQL source for primary data [#34](https://github.com/ie3-institute/simona/issues/34)
+
 ### Changed
 - Improving code readability in EvcsAgent by moving FreeLotsRequest to separate methods
 

build.gradle

@@ -29,6 +29,8 @@ ext {
   akkaVersion = '2.6.18'
   tscfgVersion = '0.9.996'
 
+  testContainerVersion = '0.39.12'
+
   scriptsLocation = 'gradle' + File.separator + 'scripts' + File.separator // location of script plugins
 }
 
@@ -67,7 +69,7 @@ dependencies {
     /* Exclude our own nested dependencies */
     exclude group: 'com.github.ie3-institute'
   }
-  implementation('com.github.ie3-institute:PowerSystemDataModel:2.1.0') {
+  implementation('com.github.ie3-institute:PowerSystemDataModel:3.0-SNAPSHOT') {
     exclude group: 'org.apache.logging.log4j'
     exclude group: 'org.slf4j'
     /* Exclude our own nested dependencies */
@@ -100,6 +102,10 @@ dependencies {
   testImplementation group: 'org.pegdown', name: 'pegdown', version: '1.6.0'
   testImplementation "com.typesafe.akka:akka-testkit_${scalaVersion}:${akkaVersion}" // akka testkit
 
+  // testcontainers
+  testImplementation "com.dimafeng:testcontainers-scala-scalatest_${scalaVersion}:${testContainerVersion}"
+  testImplementation "com.dimafeng:testcontainers-scala-postgresql_${scalaVersion}:${testContainerVersion}"
+
   /* --- Scala libs --- */
   /* CORE Scala */
   implementation "org.scala-lang:scala-library:${scalaBinaryVersion}"

src/main/resources/config/config-template.conf

@@ -101,9 +101,8 @@ simona.input.primary = {
       jdbcUrl:  string
       userName: string
       password: string
-      weatherTableName: string
+      tableName: string
       schemaName: string | "public"
-      timeColumnName: string
       timePattern: string | "yyyy-MM-dd'T'HH:mm:ss[.S[S][S]]'Z'" # default pattern from PSDM:TimeBasedSimpleValueFactory
     }
     #@optional
@@ -150,9 +149,9 @@ simona.input.weather.datasource = {
       jdbcUrl:  string
       userName: string
       password: string
-      weatherTableName: string
+      tableName: string
       schemaName: string | "public"
-      timeColumnName: string
+      timePattern: string | "yyyy-MM-dd'T'HH:mm:ss[.S[S][S]]'Z'" # default pattern from PSDM:TimeBasedSimpleValueFactory
     }
     #@optional
     couchbaseParams = {

src/main/scala/edu/ie3/simona/config/SimonaConfig.scala

@@ -1,5 +1,5 @@
 /*
- * © 2021. TU Dortmund University,
+ * © 2022. TU Dortmund University,
  * Institute of Energy Systems, Energy Efficiency and Energy Economics,
  * Research group Distribution grid planning and operation
  */
@@ -908,10 +908,9 @@ object SimonaConfig {
             jdbcUrl: java.lang.String,
             password: java.lang.String,
             schemaName: java.lang.String,
-            timeColumnName: java.lang.String,
+            tableName: java.lang.String,
             timePattern: java.lang.String,
-            userName: java.lang.String,
-            weatherTableName: java.lang.String
+            userName: java.lang.String
         )
         object SqlParams {
           def apply(
@@ -925,14 +924,11 @@ object SimonaConfig {
               schemaName =
                 if (c.hasPathOrNull("schemaName")) c.getString("schemaName")
                 else "public",
-              timeColumnName =
-                $_reqStr(parentPath, c, "timeColumnName", $tsCfgValidator),
+              tableName = $_reqStr(parentPath, c, "tableName", $tsCfgValidator),
               timePattern =
                 if (c.hasPathOrNull("timePattern")) c.getString("timePattern")
                 else "yyyy-MM-dd'T'HH:mm:ss[.S[S][S]]'Z'",
-              userName = $_reqStr(parentPath, c, "userName", $tsCfgValidator),
-              weatherTableName =
-                $_reqStr(parentPath, c, "weatherTableName", $tsCfgValidator)
+              userName = $_reqStr(parentPath, c, "userName", $tsCfgValidator)
             )
           }
           private def $_reqStr(
@@ -1277,9 +1273,9 @@ object SimonaConfig {
               jdbcUrl: java.lang.String,
               password: java.lang.String,
               schemaName: java.lang.String,
-              timeColumnName: java.lang.String,
-              userName: java.lang.String,
-              weatherTableName: java.lang.String
+              tableName: java.lang.String,
+              timePattern: java.lang.String,
+              userName: java.lang.String
           )
           object SqlParams {
             def apply(
@@ -1293,11 +1289,12 @@ object SimonaConfig {
                 schemaName =
                   if (c.hasPathOrNull("schemaName")) c.getString("schemaName")
                   else "public",
-                timeColumnName =
-                  $_reqStr(parentPath, c, "timeColumnName", $tsCfgValidator),
-                userName = $_reqStr(parentPath, c, "userName", $tsCfgValidator),
-                weatherTableName =
-                  $_reqStr(parentPath, c, "weatherTableName", $tsCfgValidator)
+                tableName =
+                  $_reqStr(parentPath, c, "tableName", $tsCfgValidator),
+                timePattern =
+                  if (c.hasPathOrNull("timePattern")) c.getString("timePattern")
+                  else "yyyy-MM-dd'T'HH:mm:ss[.S[S][S]]'Z'",
+                userName = $_reqStr(parentPath, c, "userName", $tsCfgValidator)
               )
             }
             private def $_reqStr(

src/main/scala/edu/ie3/simona/service/primary/PrimaryServiceWorker.scala

@@ -7,14 +7,17 @@
 package edu.ie3.simona.service.primary
 
 import akka.actor.{ActorRef, Props}
+import edu.ie3.datamodel.io.connectors.SqlConnector
 import edu.ie3.datamodel.io.csv.timeseries.ColumnScheme
 import edu.ie3.datamodel.io.factory.timeseries.TimeBasedSimpleValueFactory
 import edu.ie3.datamodel.io.naming.FileNamingStrategy
 import edu.ie3.datamodel.io.source.TimeSeriesSource
 import edu.ie3.datamodel.io.source.csv.CsvTimeSeriesSource
+import edu.ie3.datamodel.io.source.sql.SqlTimeSeriesSource
 import edu.ie3.datamodel.models.value.Value
 import edu.ie3.simona.agent.participant.data.Data.PrimaryData
 import edu.ie3.simona.agent.participant.data.Data.PrimaryData.RichValue
+import edu.ie3.simona.config.SimonaConfig.Simona.Input.Primary.SqlParams
 import edu.ie3.simona.exceptions.InitializationException
 import edu.ie3.simona.exceptions.WeatherServiceException.InvalidRegistrationRequestException
 import edu.ie3.simona.ontology.messages.SchedulerMessage
@@ -24,11 +27,11 @@ import edu.ie3.simona.service.ServiceStateData.{
   InitializeServiceStateData,
   ServiceActivationBaseStateData
 }
-import edu.ie3.simona.service.{ServiceStateData, SimonaService}
 import edu.ie3.simona.service.primary.PrimaryServiceWorker.{
   PrimaryServiceInitializedStateData,
   ProvidePrimaryDataMessage
 }
+import edu.ie3.simona.service.{ServiceStateData, SimonaService}
 import edu.ie3.simona.util.TickUtil.{RichZonedDateTime, TickLong}
 import edu.ie3.util.scala.collection.immutable.SortedDistinctSeq
 
@@ -61,68 +64,94 @@ final case class PrimaryServiceWorker[V <: Value](
         PrimaryServiceInitializedStateData[V],
         Option[Seq[SchedulerMessage.ScheduleTriggerMessage]]
     )
-  ] = initServiceData match {
-    case PrimaryServiceWorker.CsvInitPrimaryServiceStateData(
-          timeSeriesUuid,
-          simulationStart,
-          csvSep,
-          directoryPath,
-          filePath,
-          fileNamingStrategy,
-          timePattern
-        ) =>
-      /* Got the right data. Attempt to set up a source and acquire information */
-      implicit val startDateTime: ZonedDateTime = simulationStart
+  ] = {
+    (initServiceData match {
+      case PrimaryServiceWorker.CsvInitPrimaryServiceStateData(
+            timeSeriesUuid,
+            simulationStart,
+            csvSep,
+            directoryPath,
+            filePath,
+            fileNamingStrategy,
+            timePattern
+          ) =>
+        Try {
+          /* Set up source and acquire information */
+          val factory = new TimeBasedSimpleValueFactory(valueClass, timePattern)
+          val source = new CsvTimeSeriesSource(
+            csvSep,
+            directoryPath,
+            fileNamingStrategy,
+            timeSeriesUuid,
+            filePath,
+            valueClass,
+            factory
+          )
+          (source, simulationStart)
+        }
+      case PrimaryServiceWorker.SqlInitPrimaryServiceStateData(
+            sqlParams: SqlParams,
+            timeSeriesUuid: UUID,
+            simulationStart: ZonedDateTime
+          ) =>
+        Try {
+          val factory =
+            new TimeBasedSimpleValueFactory(valueClass, sqlParams.timePattern)
 
-      Try {
-        /* Set up source and acquire information */
-        val factory = new TimeBasedSimpleValueFactory(valueClass, timePattern)
-        val source = new CsvTimeSeriesSource(
-          csvSep,
-          directoryPath,
-          fileNamingStrategy,
-          timeSeriesUuid,
-          filePath,
-          valueClass,
-          factory
-        )
-        /* This seems not to be very efficient, but it is as efficient as possible. The getter method points to a
-         * final attribute within the source implementation. */
-        val (maybeNextTick, furtherActivationTicks) = SortedDistinctSeq(
-          source.getTimeSeries.getEntries.asScala
-            .filter { timeBasedValue =>
-              val dateTime = timeBasedValue.getTime
-              dateTime.isEqual(simulationStart) || dateTime.isAfter(
-                simulationStart
-              )
-            }
-            .map(timeBasedValue => timeBasedValue.getTime.toTick)
-            .toSeq
-            .sorted
-        ).pop
+          val sqlConnector = new SqlConnector(
+            sqlParams.jdbcUrl,
+            sqlParams.userName,
+            sqlParams.password
+          )
 
-        /* Set up the state data and determine the next activation tick. */
-        val initializedStateData =
-          PrimaryServiceInitializedStateData(
-            maybeNextTick,
-            furtherActivationTicks,
-            simulationStart,
-            source
+          val source = new SqlTimeSeriesSource(
+            sqlConnector,
+            sqlParams.schemaName,
+            sqlParams.tableName,
+            timeSeriesUuid,
+            valueClass,
+            factory
           )
-        val triggerMessage =
-          ServiceActivationBaseStateData.tickToScheduleTriggerMessages(
-            maybeNextTick,
-            self
+
+          (source, simulationStart)
+        }
+      case unsupported =>
+        /* Got the wrong init data */
+        Failure(
+          new InitializationException(
+            s"Provided init data '${unsupported.getClass.getSimpleName}' for primary service are invalid!"
           )
-        (initializedStateData, triggerMessage)
-      }
-    case unsupported =>
-      /* Got the wrong init data */
-      Failure(
-        new InitializationException(
-          s"Provided init data '${unsupported.getClass.getSimpleName}' for primary service are invalid!"
         )
-      )
+    }).map { case (source, simulationStart) =>
+      val (maybeNextTick, furtherActivationTicks) = SortedDistinctSeq(
+        // Note: The whole data set is used here, which might be inefficient depending on the source implementation.
+        source.getTimeSeries.getEntries.asScala
+          .filter { timeBasedValue =>
+            val dateTime = timeBasedValue.getTime
+            dateTime.isEqual(simulationStart) || dateTime.isAfter(
+              simulationStart
+            )
+          }
+          .map(timeBasedValue => timeBasedValue.getTime.toTick)
+          .toSeq
+          .sorted
+      ).pop
+
+      /* Set up the state data and determine the next activation tick. */
+      val initializedStateData =
+        PrimaryServiceInitializedStateData(
+          maybeNextTick,
+          furtherActivationTicks,
+          simulationStart,
+          source
+        )
+      val triggerMessage =
+        ServiceActivationBaseStateData.tickToScheduleTriggerMessages(
+          maybeNextTick,
+          self
+        )
+      (initializedStateData, triggerMessage)
+    }
   }
 
   /** Handle a request to register for information from this service
@@ -348,6 +377,22 @@ case object PrimaryServiceWorker {
       timePattern: String
   ) extends InitPrimaryServiceStateData
 
+  /** Specific implementation of [[InitPrimaryServiceStateData]], if the source
+    * to use utilizes an SQL database.
+    *
+    * @param sqlParams
+    *   Parameters regarding SQL connection and table selection
+    * @param timeSeriesUuid
+    *   Unique identifier of the time series to read
+    * @param simulationStart
+    *   Wall clock time of the beginning of simulation time
+    */
+  final case class SqlInitPrimaryServiceStateData(
+      sqlParams: SqlParams,
+      override val timeSeriesUuid: UUID,
+      override val simulationStart: ZonedDateTime
+  ) extends InitPrimaryServiceStateData
+
   /** Class carrying the state of a fully initialized [[PrimaryServiceWorker]]
     *
     * @param maybeNextActivationTick

src/main/scala/edu/ie3/simona/service/weather/WeatherSourceWrapper.scala

@@ -328,7 +328,7 @@ private[weather] object WeatherSourceWrapper extends LazyLogging {
       sqlConnector,
       idCoordinateSource,
       sqlParams.schemaName,
-      sqlParams.weatherTableName,
+      sqlParams.tableName,
       buildFactory(timestampPattern, scheme)
     )
     logger.info(

src/main/scala/edu/ie3/simona/util/ConfigUtil.scala

@@ -321,11 +321,7 @@ object ConfigUtil {
         logger.info(
           "Password for SQL weather source is empty. This is allowed, but not common. Please check if this an intended setting."
         )
-      if (sql.timeColumnName.isEmpty)
-        throw new InvalidConfigParameterException(
-          "Time column for SQL weather source cannot be empty"
-        )
-      if (sql.weatherTableName.isEmpty)
+      if (sql.tableName.isEmpty)
         throw new InvalidConfigParameterException(
           "Weather table name for SQL weather source cannot be empty"
         )

src/test/resources/edu/ie3/simona/service/primary/timeseries/its_p_9185b8c1-86ba-4a16-8dea-5ac898e8caa5.sql

@@ -0,0 +1,17 @@
+CREATE TABLE public."its_p_9185b8c1-86ba-4a16-8dea-5ac898e8caa5"
+(
+    time timestamp with time zone,
+    p double precision,
+    uuid uuid,
+    CONSTRAINT its_p_pkey PRIMARY KEY (uuid)
+)
+    WITH (
+        OIDS = FALSE
+    )
+    TABLESPACE pg_default;
+
+INSERT INTO
+    public."its_p_9185b8c1-86ba-4a16-8dea-5ac898e8caa5" (uuid, time, p)
+VALUES
+('0245d599-9a5c-4c32-9613-5b755fac8ca0', '2020-01-01 00:00:00+0', 1000.0),
+('a5e27652-9024-4a93-9d2a-590fbc3ab5a1', '2020-01-01 00:15:00+0', 1250.0);

src/test/resources/edu/ie3/simona/service/primary/timeseries/its_pqh_46be1e57-e4ed-4ef7-95f1-b2b321cb2047.sql

@@ -0,0 +1,19 @@
+CREATE TABLE public."its_pqh_46be1e57-e4ed-4ef7-95f1-b2b321cb2047"
+(
+    time timestamp with time zone,
+    p double precision,
+    q double precision,
+    heat_demand double precision,
+    uuid uuid,
+    CONSTRAINT its_pqh_pkey PRIMARY KEY (uuid)
+)
+    WITH (
+        OIDS = FALSE
+    )
+    TABLESPACE pg_default;
+
+INSERT INTO
+    public."its_pqh_46be1e57-e4ed-4ef7-95f1-b2b321cb2047" (uuid, time, p, q, heat_demand)
+VALUES
+('661ac594-47f0-4442-8d82-bbeede5661f7', '2020-01-01 00:00:00+0', 1000.0, 329.0, 8.0),
+('5adcd6c5-a903-433f-b7b5-5fe669a3ed30', '2020-01-01 00:15:00+0', 1250.0, 411.0, 12.0);