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
 - Re-organizing test resources into their respective packages [#105](https://github.com/ie3-institute/simona/issues/105)

build.gradle

@@ -30,6 +30,8 @@ ext {
   tscfgVersion = '0.9.997'
   scapegoatVersion = '1.4.12'
 
+  testContainerVersion = '0.40.3'
+
   scriptsLocation = 'gradle' + File.separator + 'scripts' + File.separator // location of script plugins
 }
 
@@ -107,6 +109,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}"

gradle/scripts/tscfg.gradle

@@ -15,7 +15,7 @@ task genConfigClass {
       args = [
         "build/tscfg-${tscfgVersion}.jar",
         "--spec",
-        "src/main/resources/config/simona-config-template.conf",
+        "src/main/resources/config/config-template.conf",
         "--scala",
         "--durations",
         "--pn",

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

@@ -101,9 +101,7 @@ simona.input.primary = {
       jdbcUrl:  string
       userName: string
       password: string
-      weatherTableName: 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 +148,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,8 @@ object SimonaConfig {
             jdbcUrl: java.lang.String,
             password: java.lang.String,
             schemaName: java.lang.String,
-            timeColumnName: 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 +923,10 @@ object SimonaConfig {
               schemaName =
                 if (c.hasPathOrNull("schemaName")) c.getString("schemaName")
                 else "public",
-              timeColumnName =
-                $_reqStr(parentPath, c, "timeColumnName", $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 +1271,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 +1287,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/PrimaryServiceProxy.scala

@@ -9,9 +9,12 @@ package edu.ie3.simona.service.primary
 import akka.actor.{Actor, ActorRef, PoisonPill, Props}
 import edu.ie3.datamodel.io.csv.CsvIndividualTimeSeriesMetaInformation
 import edu.ie3.datamodel.io.naming.FileNamingStrategy
-import edu.ie3.datamodel.io.naming.timeseries.ColumnScheme
+import edu.ie3.datamodel.io.naming.timeseries.IndividualTimeSeriesMetaInformation
 import edu.ie3.datamodel.io.source.TimeSeriesMappingSource
-import edu.ie3.datamodel.io.source.csv.CsvTimeSeriesMappingSource
+import edu.ie3.datamodel.io.source.csv.{
+  CsvTimeSeriesMappingSource,
+  CsvTimeSeriesTypeSource
+}
 import edu.ie3.datamodel.models.value.Value
 import edu.ie3.simona.config.SimonaConfig
 import edu.ie3.simona.config.SimonaConfig.Simona.Input.Primary.CsvParams
@@ -51,7 +54,6 @@ import java.time.ZonedDateTime
 import java.util.UUID
 import scala.Option.when
 import scala.jdk.CollectionConverters._
-import scala.jdk.OptionConverters._
 import scala.util.{Failure, Success, Try}
 
 /** This actor has information on which models can be replaced by precalculated
@@ -137,27 +139,35 @@ case class PrimaryServiceProxy(
     ).filter(_.isDefined).flatten.headOption match {
       case Some(CsvParams(csvSep, folderPath, _)) =>
         // TODO: Configurable file naming strategy
+        val fileNamingStrategy = new FileNamingStrategy()
         val mappingSource = new CsvTimeSeriesMappingSource(
           csvSep,
           folderPath,
-          new FileNamingStrategy()
+          fileNamingStrategy
+        )
+        val typeSource = new CsvTimeSeriesTypeSource(
+          csvSep,
+          folderPath,
+          fileNamingStrategy
         )
         val modelToTimeSeries = mappingSource.getMapping.asScala.toMap
+        val timeSeriesMetaInformation =
+          typeSource.getTimeSeriesMetaInformation.asScala.toMap
+
         val timeSeriesToSourceRef = modelToTimeSeries.values
           .to(LazyList)
           .distinct
           .flatMap { timeSeriesUuid =>
-            mappingSource
-              .timeSeriesMetaInformation(timeSeriesUuid)
-              .toScala match {
+            timeSeriesMetaInformation
+              .get(timeSeriesUuid) match {
               case Some(metaInformation) =>
                 val columnScheme = metaInformation.getColumnScheme
                 /* Only register those entries, that meet the supported column schemes */
                 when(
                   PrimaryServiceWorker.supportedColumnSchemes
                     .contains(columnScheme)
                 ) {
-                  timeSeriesUuid -> SourceRef(columnScheme, None)
+                  timeSeriesUuid -> SourceRef(metaInformation, None)
                 }
               case None =>
                 log.warning(
@@ -251,14 +261,12 @@ case class PrimaryServiceProxy(
         /* There is yet a worker apparent. Register the requesting actor. The worker will reply to the original
          * requesting actor. */
         worker ! WorkerRegistrationMessage(requestingActor)
-      case Some(SourceRef(columnScheme, None)) =>
+      case Some(SourceRef(metaInformation, None)) =>
         /* There is NO worker apparent, yet. Spin one off. */
         initializeWorker(
-          columnScheme,
-          timeSeriesUuid,
+          metaInformation,
           stateData.simulationStart,
-          stateData.primaryConfig,
-          stateData.mappingSource
+          stateData.primaryConfig
         ) match {
           case Success(workerRef) =>
             /* Forward the registration request. The worker will reply about successful registration or not. */
@@ -289,33 +297,28 @@ case class PrimaryServiceProxy(
   /** Instantiate a new [[PrimaryServiceWorker]] and send initialization
     * information
     *
-    * @param columnScheme
-    *   Scheme of the data to expect
+    * @param metaInformation
+    *   Meta information (including column scheme) of the time series
+    * @param simulationStart
+    *   The time of the simulation start
     * @param primaryConfig
     *   Configuration for the primary config
-    * @param mappingSource
-    *   Source for time series mapping, that might deliver additional
-    *   information for the source initialization
     * @return
     *   The [[ActorRef]] to the worker
     */
   protected def initializeWorker(
-      columnScheme: ColumnScheme,
-      timeSeriesUuid: UUID,
+      metaInformation: IndividualTimeSeriesMetaInformation,
       simulationStart: ZonedDateTime,
-      primaryConfig: PrimaryConfig,
-      mappingSource: TimeSeriesMappingSource
+      primaryConfig: PrimaryConfig
   ): Try[ActorRef] = {
     val workerRef = classToWorkerRef(
-      columnScheme.getValueClass,
-      timeSeriesUuid.toString,
-      simulationStart
+      metaInformation.getColumnScheme.getValueClass,
+      metaInformation.getUuid.toString
     )
     toInitData(
-      primaryConfig,
-      mappingSource,
-      timeSeriesUuid,
-      simulationStart
+      metaInformation,
+      simulationStart,
+      primaryConfig
     ) match {
       case Success(initData) =>
         scheduler ! ScheduleTriggerMessage(
@@ -341,42 +344,36 @@ case class PrimaryServiceProxy(
     *   Class of the values to provide later on
     * @param timeSeriesUuid
     *   uuid of the time series the actor processes
-    * @param simulationStart
-    *   Wall clock time of first instant in simulation
     * @tparam V
     *   Type of the class to provide
     * @return
     *   The [[ActorRef]] to the spun off actor
     */
   protected def classToWorkerRef[V <: Value](
       valueClass: Class[V],
-      timeSeriesUuid: String,
-      simulationStart: ZonedDateTime
+      timeSeriesUuid: String
   ): ActorRef = {
     import edu.ie3.simona.actor.SimonaActorNaming._
     context.system.simonaActorOf(
-      PrimaryServiceWorker.props(scheduler, valueClass, simulationStart),
+      PrimaryServiceWorker.props(scheduler, valueClass),
       timeSeriesUuid
     )
   }
 
   /** Building proper init data for the worker
     *
-    * @param primaryConfig
-    *   Configuration for primary sources
-    * @param mappingSource
-    *   Source to get mapping information about time series
-    * @param timeSeriesUuid
-    *   Unique identifier for the time series
+    * @param metaInformation
+    *   Meta information (including column scheme) of the time series
     * @param simulationStart
-    *   Wall clock time of the first instant in simulation
+    *   The time of the simulation start
+    * @param primaryConfig
+    *   Configuration for the primary config
     * @return
     */
   private def toInitData(
-      primaryConfig: PrimaryConfig,
-      mappingSource: TimeSeriesMappingSource,
-      timeSeriesUuid: UUID,
-      simulationStart: ZonedDateTime
+      metaInformation: IndividualTimeSeriesMetaInformation,
+      simulationStart: ZonedDateTime,
+      primaryConfig: PrimaryConfig
   ): Try[InitPrimaryServiceStateData] =
     primaryConfig match {
       case PrimaryConfig(
@@ -385,29 +382,27 @@ case class PrimaryServiceProxy(
             None,
             None
           ) =>
-        /* The mapping and actual data sources are from csv. At first, get the file name of the file to read. */
-        Try(mappingSource.timeSeriesMetaInformation(timeSeriesUuid).get)
-          .flatMap {
-            /* Time series meta information could be successfully obtained */
-            case csvMetaData: CsvIndividualTimeSeriesMetaInformation =>
-              Success(
-                CsvInitPrimaryServiceStateData(
-                  timeSeriesUuid,
-                  simulationStart,
-                  csvSep,
-                  directoryPath,
-                  csvMetaData.getFullFilePath,
-                  new FileNamingStrategy(),
-                  timePattern
-                )
+        /* The actual data sources are from csv. Meta information have to match */
+        metaInformation match {
+          case csvMetaData: CsvIndividualTimeSeriesMetaInformation =>
+            Success(
+              CsvInitPrimaryServiceStateData(
+                csvMetaData.getUuid,
+                simulationStart,
+                csvSep,
+                directoryPath,
+                csvMetaData.getFullFilePath,
+                new FileNamingStrategy(),
+                timePattern
               )
-            case invalidMetaData =>
-              Failure(
-                new InitializationException(
-                  s"Expected '${classOf[CsvIndividualTimeSeriesMetaInformation]}', but got '$invalidMetaData'."
-                )
+            )
+          case invalidMetaData =>
+            Failure(
+              new InitializationException(
+                s"Expected '${classOf[CsvIndividualTimeSeriesMetaInformation]}', but got '$invalidMetaData'."
               )
-          }
+            )
+        }
       case unsupported =>
         Failure(
           new InitializationException(
@@ -489,14 +484,14 @@ object PrimaryServiceProxy {
 
   /** Giving reference to the target time series and source worker.
     *
-    * @param columnScheme
-    *   Column scheme of the time series to get
+    * @param metaInformation
+    *   Meta information (including column scheme) of the time series
     * @param worker
-    *   Optional reference to a yet existing worker providing information on
-    *   that time series
+    *   Optional reference to an already existing worker providing information
+    *   on that time series
     */
   final case class SourceRef(
-      columnScheme: ColumnScheme,
+      metaInformation: IndividualTimeSeriesMetaInformation,
       worker: Option[ActorRef]
   )