Changed data structure for weather params and aggregating
This commit is contained in:
Guntis Smaukstelis
@@ -24,3 +24,4 @@ dependency-reduced-pom.xml
|
||||
*.log
|
||||
logs/
|
||||
.DS_Store
|
||||
data/*
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
package parse
|
||||
|
||||
import cats.implicits.{catsSyntaxOptionId, toFoldableOps}
|
||||
import io.circe.generic.semiauto.deriveEncoder
|
||||
import io.circe.syntax.EncoderOps
|
||||
import io.circe.{Encoder, Json}
|
||||
import parse.Parser.parseLine
|
||||
|
||||
object Aggregate {
|
||||
|
||||
case class UserQuery(
|
||||
cities: List[String],
|
||||
field: String,
|
||||
key: AggregateKey
|
||||
)
|
||||
|
||||
sealed trait AggregateKey
|
||||
object AggregateKey {
|
||||
case object Min extends AggregateKey
|
||||
case object Max extends AggregateKey
|
||||
case object Avg extends AggregateKey
|
||||
case object Sum extends AggregateKey
|
||||
case object List extends AggregateKey
|
||||
case object Distinct extends AggregateKey
|
||||
|
||||
def fromString(str: String): Option[AggregateKey] = str match {
|
||||
case "min" => Some(AggregateKey.Min)
|
||||
case "max" => Some(AggregateKey.Max)
|
||||
case "avg" => Some(AggregateKey.Avg)
|
||||
case "sum" => Some(AggregateKey.Sum)
|
||||
case "list" => Some(AggregateKey.List)
|
||||
case "distinct" => Some(AggregateKey.Distinct)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
def getKeys: List[String] = {
|
||||
val runtimeMirror = scala.reflect.runtime.currentMirror
|
||||
val classSymbol = runtimeMirror.classSymbol(classOf[AggregateKey])
|
||||
|
||||
// Get all case classes that extend the AggregateKey trait
|
||||
val keys = classSymbol.knownDirectSubclasses.map(_.name.toString).toList
|
||||
|
||||
keys
|
||||
}
|
||||
}
|
||||
|
||||
sealed trait AggregateValue
|
||||
case class DoubleValue(value: Double) extends AggregateValue
|
||||
case class ListOptionValue(list: List[Option[Double]]) extends AggregateValue
|
||||
case class StringListList(list: List[List[String]]) extends AggregateValue
|
||||
case class DistinctStringList(list: List[String]) extends AggregateValue
|
||||
object AggregateValue {
|
||||
def getKeys: List[String] = {
|
||||
val runtimeMirror = scala.reflect.runtime.currentMirror
|
||||
val classSymbol = runtimeMirror.classSymbol(classOf[AggregateValue])
|
||||
|
||||
// Get all case classes that extend the AggregateValue trait
|
||||
val keys = classSymbol.knownDirectSubclasses.map(_.name.toString).toList
|
||||
|
||||
keys
|
||||
}
|
||||
}
|
||||
|
||||
object AggregateValueImplicits {
|
||||
implicit val aggregateValueEncoder: Encoder[AggregateValue] = Encoder.instance {
|
||||
case doubleValue: DoubleValue => doubleValue.asJson
|
||||
case listOptionValue: ListOptionValue => listOptionValue.asJson
|
||||
case stringListList: StringListList => stringListList.asJson
|
||||
case distinctStringList: DistinctStringList => distinctStringList.asJson
|
||||
}
|
||||
|
||||
implicit val doubleValueEncoder: Encoder[DoubleValue] = Encoder.instance {
|
||||
case DoubleValue(value) if !value.isNaN => Json.fromDouble(value).get
|
||||
case _ => Json.Null
|
||||
}
|
||||
implicit val listOptionValueEncoder: Encoder[ListOptionValue] = deriveEncoder[ListOptionValue]
|
||||
implicit val stringListListEncoder: Encoder[StringListList] = deriveEncoder[StringListList]
|
||||
implicit val distinctStringListEncoder: Encoder[DistinctStringList] = deriveEncoder[DistinctStringList]
|
||||
}
|
||||
|
||||
def extractDoubleFieldValues(field: String, weatherData: List[WeatherData]): List[Option[Double]] =
|
||||
field match {
|
||||
case "tempMax" => weatherData.map(_.tempMax)
|
||||
case "tempMin" => weatherData.map(_.tempMin)
|
||||
case "tempAvg" => weatherData.map(_.tempAvg)
|
||||
case "precipitation" => weatherData.map(_.precipitation)
|
||||
case "windAvg" => weatherData.map(_.windAvg)
|
||||
case "windMax" => weatherData.map(_.windMax)
|
||||
case "visibilityMin" => weatherData.map(_.visibilityMin)
|
||||
case "visibilityAvg" => weatherData.map(_.visibilityAvg)
|
||||
case "snowAvg" => weatherData.map(_.snowAvg)
|
||||
case "atmPressire" => weatherData.map(_.atmPressure)
|
||||
case "dewPoint" => weatherData.map(_.dewPoint)
|
||||
case "humidity" => weatherData.map(_.humidity)
|
||||
case "sunDuration" => weatherData.map(_.sunDuration)
|
||||
// case "phenomena" => weatherData.map(_..phenomena)
|
||||
case _ => List.empty
|
||||
}
|
||||
|
||||
def aggregateDoubleValues(
|
||||
AggKey: AggregateKey,
|
||||
values: List[Option[Double]]
|
||||
): Option[AggregateValue] = {
|
||||
val flatValues = values.flatten
|
||||
AggKey match {
|
||||
case AggregateKey.Min => flatValues.minimumOption.map(DoubleValue)
|
||||
case AggregateKey.Max => flatValues.maximumOption.map(DoubleValue)
|
||||
case AggregateKey.Avg => flatValues.reduceOption(_ + _).map(sum => DoubleValue(sum / flatValues.length))
|
||||
case AggregateKey.Sum => flatValues.sum.some.map(DoubleValue)
|
||||
case AggregateKey.List => ListOptionValue(values).some
|
||||
case AggregateKey.Distinct => None
|
||||
}
|
||||
}
|
||||
|
||||
def aggregatePhenomenaValues(
|
||||
aggregationType: AggregateKey,
|
||||
values: List[List[String]]
|
||||
): Option[AggregateValue] =
|
||||
aggregationType match {
|
||||
case AggregateKey.List => StringListList(values).some
|
||||
case AggregateKey.Distinct => DistinctStringList(values.flatten.distinct).some
|
||||
case _ => None
|
||||
}
|
||||
}
|
||||
@@ -1,189 +0,0 @@
|
||||
package parse
|
||||
|
||||
import io.circe._
|
||||
import io.circe.syntax.EncoderOps
|
||||
|
||||
import java.time.LocalDateTime
|
||||
import java.time.format.DateTimeFormatter
|
||||
|
||||
sealed trait AggregateValue
|
||||
|
||||
|
||||
object AggregateValue {
|
||||
val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMdd-HHmm")
|
||||
|
||||
case class tempMax(value: Double) extends AggregateValue
|
||||
case class tempMaxList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class tempMin(value: Double) extends AggregateValue
|
||||
case class tempMinList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class tempAvg(value: Double) extends AggregateValue
|
||||
case class tempAvgList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class precipitationSum(value: Double) extends AggregateValue
|
||||
case class precipitationList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class windAvg(value: Double) extends AggregateValue
|
||||
case class windAvgList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class windMax(value: Double) extends AggregateValue
|
||||
case class windMaxList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class visibilityMin(value: Double) extends AggregateValue
|
||||
case class visibilityMinList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class visibilityAvg(value: Double) extends AggregateValue
|
||||
case class visibilityAvgList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class snowAvgList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class snowAvg(value: Double) extends AggregateValue
|
||||
case class snowMax(value: Double) extends AggregateValue
|
||||
case class atmPressureList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class atmPressureMin(value: Double) extends AggregateValue
|
||||
case class atmPressureMax(value: Double) extends AggregateValue
|
||||
case class atmPressureAvg(value: Double) extends AggregateValue
|
||||
case class dewList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class dewMin(value: Double) extends AggregateValue
|
||||
case class dewMax(value: Double) extends AggregateValue
|
||||
case class dewAvg(value: Double) extends AggregateValue
|
||||
case class humidityAvg(value: Double) extends AggregateValue
|
||||
case class humidityAvgList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class sunDurationList(value: Map[LocalDateTime, Double]) extends AggregateValue
|
||||
case class sunDurationSum(value: Double) extends AggregateValue
|
||||
|
||||
private def encodeDateTimeMap(map: Map[LocalDateTime, Double]): Json = {
|
||||
map.map { case (dateTime, value) =>
|
||||
Json.obj(dateFormatter.format(dateTime) -> Json.fromDoubleOrNull(value))
|
||||
}.toList.asJson
|
||||
}
|
||||
|
||||
implicit val encodeAggregateValue: Encoder[AggregateValue] = Encoder.instance {
|
||||
case key: tempMax => Json.fromDoubleOrNull(key.value)
|
||||
case key: tempMaxList => encodeDateTimeMap(key.value)
|
||||
case key: tempMin => Json.fromDoubleOrNull(key.value)
|
||||
case key: tempMinList => encodeDateTimeMap(key.value)
|
||||
case key: tempAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: tempAvgList => encodeDateTimeMap(key.value)
|
||||
case key: precipitationSum => Json.fromDoubleOrNull(key.value)
|
||||
case key: precipitationList => encodeDateTimeMap(key.value)
|
||||
case key: windAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: windAvgList => encodeDateTimeMap(key.value)
|
||||
case key: windMax => Json.fromDoubleOrNull(key.value)
|
||||
case key: windMaxList => encodeDateTimeMap(key.value)
|
||||
case key: visibilityMin => Json.fromDoubleOrNull(key.value)
|
||||
case key: visibilityMinList => encodeDateTimeMap(key.value)
|
||||
case key: visibilityAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: visibilityAvgList => encodeDateTimeMap(key.value)
|
||||
case key: snowAvgList => encodeDateTimeMap(key.value)
|
||||
case key: snowAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: snowMax => Json.fromDoubleOrNull(key.value)
|
||||
case key: atmPressureList => encodeDateTimeMap(key.value)
|
||||
case key: atmPressureMin => Json.fromDoubleOrNull(key.value)
|
||||
case key: atmPressureMax => Json.fromDoubleOrNull(key.value)
|
||||
case key: atmPressureAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: dewList => encodeDateTimeMap(key.value)
|
||||
case key: dewMin => Json.fromDoubleOrNull(key.value)
|
||||
case key: dewMax => Json.fromDoubleOrNull(key.value)
|
||||
case key: dewAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: humidityAvg => Json.fromDoubleOrNull(key.value)
|
||||
case key: humidityAvgList => encodeDateTimeMap(key.value)
|
||||
case key: sunDurationList => encodeDateTimeMap(key.value)
|
||||
case key: sunDurationSum => Json.fromDoubleOrNull(key.value)
|
||||
}
|
||||
|
||||
def getKeys: List[String] = {
|
||||
val runtimeMirror = scala.reflect.runtime.currentMirror
|
||||
val weatherParamClassSymbol = runtimeMirror.classSymbol(classOf[AggregateValue])
|
||||
|
||||
// Get all case classes that extend the WeatherParameter trait
|
||||
val weatherParameterCases = weatherParamClassSymbol.knownDirectSubclasses.map(_.name.toString).toList
|
||||
|
||||
weatherParameterCases
|
||||
}
|
||||
}
|
||||
|
||||
sealed trait AggregateKey
|
||||
|
||||
object AggregateKey {
|
||||
def stringToAggregateParam(strParam: String): Option[AggregateKey] = strParam match {
|
||||
case "tempMax" => Some(AggregateKey.tempMax)
|
||||
case "tempMaxList" => Some(AggregateKey.tempMaxList)
|
||||
case "tempMin" => Some(AggregateKey.tempMin)
|
||||
case "tempMinList" => Some(AggregateKey.tempMinList)
|
||||
case "tempAvg" => Some(AggregateKey.tempAvg)
|
||||
case "tempAvgList" => Some(AggregateKey.tempAvgList)
|
||||
case "precipitationSum" => Some(AggregateKey.precipitationSum)
|
||||
case "precipitationList" => Some(AggregateKey.precipitationList)
|
||||
case "windAvg" => Some(AggregateKey.windAvg)
|
||||
case "windAvgList" => Some(AggregateKey.windAvgList)
|
||||
case "windMax" => Some(AggregateKey.windMax)
|
||||
case "windMaxList" => Some(AggregateKey.windMaxList)
|
||||
case "visibilityMin" => Some(AggregateKey.visibilityMin)
|
||||
case "visibilityMinList" => Some(AggregateKey.visibilityMinList)
|
||||
case "visibilityAvg" => Some(AggregateKey.visibilityAvg)
|
||||
case "visibilityAvgList" => Some(AggregateKey.visibilityAvgList)
|
||||
case "snowAvgList" => Some(AggregateKey.snowAvgList)
|
||||
case "snowAvg" => Some(AggregateKey.snowAvg)
|
||||
case "snowMax" => Some(AggregateKey.snowMax)
|
||||
case "atmPressureList" => Some(AggregateKey.atmPressureList)
|
||||
case "atmPressureMin" => Some(AggregateKey.atmPressureMin)
|
||||
case "atmPressureMax" => Some(AggregateKey.atmPressureMax)
|
||||
case "atmPressureAvg" => Some(AggregateKey.atmPressureAvg)
|
||||
case "dewList" => Some(AggregateKey.dewList)
|
||||
case "dewMin" => Some(AggregateKey.dewMin)
|
||||
case "dewMax" => Some(AggregateKey.dewMax)
|
||||
case "dewAvg" => Some(AggregateKey.dewAvg)
|
||||
case "humidityAvg" => Some(AggregateKey.humidityAvg)
|
||||
case "humidityAvgList" => Some(AggregateKey.humidityAvgList)
|
||||
case "sunDurationList" => Some(AggregateKey.sunDurationList)
|
||||
case "sunDurationSum" => Some(AggregateKey.sunDurationSum)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
def getKeys: List[String] = {
|
||||
val runtimeMirror = scala.reflect.runtime.currentMirror
|
||||
val weatherParamClassSymbol = runtimeMirror.classSymbol(classOf[AggregateKey])
|
||||
|
||||
// Get all case classes that extend the WeatherParameter trait
|
||||
val weatherParameterCases = weatherParamClassSymbol.knownDirectSubclasses.map(_.name.toString).toList
|
||||
|
||||
weatherParameterCases
|
||||
}
|
||||
|
||||
case object tempMax extends AggregateKey
|
||||
case object tempMaxList extends AggregateKey
|
||||
|
||||
case object tempMin extends AggregateKey
|
||||
case object tempMinList extends AggregateKey
|
||||
|
||||
case object tempAvg extends AggregateKey
|
||||
case object tempAvgList extends AggregateKey
|
||||
|
||||
case object precipitationSum extends AggregateKey
|
||||
case object precipitationList extends AggregateKey
|
||||
|
||||
case object windAvg extends AggregateKey
|
||||
case object windAvgList extends AggregateKey
|
||||
|
||||
case object windMax extends AggregateKey
|
||||
case object windMaxList extends AggregateKey
|
||||
|
||||
case object visibilityMin extends AggregateKey
|
||||
case object visibilityMinList extends AggregateKey
|
||||
|
||||
case object visibilityAvg extends AggregateKey
|
||||
case object visibilityAvgList extends AggregateKey
|
||||
|
||||
case object snowAvg extends AggregateKey
|
||||
case object snowAvgList extends AggregateKey
|
||||
case object snowMax extends AggregateKey
|
||||
|
||||
case object atmPressureList extends AggregateKey
|
||||
case object atmPressureMin extends AggregateKey
|
||||
case object atmPressureMax extends AggregateKey
|
||||
case object atmPressureAvg extends AggregateKey
|
||||
|
||||
case object dewList extends AggregateKey
|
||||
case object dewMin extends AggregateKey
|
||||
case object dewMax extends AggregateKey
|
||||
case object dewAvg extends AggregateKey
|
||||
|
||||
case object humidityAvg extends AggregateKey
|
||||
case object humidityAvgList extends AggregateKey
|
||||
|
||||
case object sunDurationList extends AggregateKey
|
||||
case object sunDurationSum extends AggregateKey
|
||||
}
|
||||
@@ -3,6 +3,8 @@ package parse
|
||||
import cats.effect.IO
|
||||
import cats.effect.unsafe.implicits.global
|
||||
import io.circe.syntax.EncoderOps
|
||||
import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder
|
||||
import parse.Aggregate.{AggregateKey, UserQuery}
|
||||
|
||||
import java.time.LocalDateTime
|
||||
import java.time.format.DateTimeFormatter
|
||||
@@ -14,11 +16,13 @@ object Main {
|
||||
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
|
||||
val from = LocalDateTime.parse("20230414_2200", formatter)
|
||||
val to = LocalDateTime.parse("20230501_1230", formatter)
|
||||
val userQuery = UserQuery(List("Liepāja", "Rēzekne", "randomstr"), "tempMax", AggregateKey.Max)
|
||||
|
||||
for {
|
||||
lines <- db.DBService.getInRange(from, to)
|
||||
parsed <- IO.pure(Parser.queryData(lines, List("Liepāja", "Rēzekne", "randomstr"), AggregateKey.tempMax))
|
||||
parsed <- IO.pure(Parser.queryData(userQuery, lines))
|
||||
_ <- IO.println(parsed.asJson)
|
||||
_ <- IO.println(None.asJson)
|
||||
} yield ()
|
||||
}
|
||||
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
package parse
|
||||
|
||||
import parse.Aggregate.{AggregateValue, UserQuery, aggregateDoubleValues, aggregatePhenomenaValues, extractDoubleFieldValues}
|
||||
|
||||
import java.time.LocalDateTime
|
||||
import java.time.format.DateTimeFormatter
|
||||
import scala.util.Try
|
||||
|
||||
object Parser {
|
||||
def parseLine(line: String): Option[WeatherStationData] = {
|
||||
val paramCount = MeteoData.getCount
|
||||
val paramCount = WeatherData.getDoubleParamCount
|
||||
|
||||
def parseTimestamp(timestampStr: String): Option[LocalDateTime] = {
|
||||
val formatter = DateTimeFormatter.ofPattern("yyyydd.MM HH:mm")
|
||||
@@ -27,90 +29,30 @@ object Parser {
|
||||
for {
|
||||
(city, timestampStr, rest) <- splitParts(line.trim.split(";", -1).toList)
|
||||
timestamp <- parseTimestamp(timestampStr)
|
||||
(strList, meteoPhenomenaList) = rest.splitAt(paramCount)
|
||||
meteoData <- MeteoData.fromDoubles(strList.map(_.toDoubleOption))
|
||||
phenomena <- Some(meteoPhenomenaList.map(_.trim))
|
||||
} yield WeatherStationData(city, timestamp, meteoData, phenomena)
|
||||
(strList, phenomenaList) = rest.splitAt(paramCount)
|
||||
phenomena <- Some(phenomenaList.map(_.trim))
|
||||
weatherData <- WeatherData.fromDoubles(strList.map(_.toDoubleOption), phenomena)
|
||||
} yield WeatherStationData(city, timestamp, weatherData)
|
||||
}
|
||||
|
||||
def queryData(
|
||||
lines: List[String],
|
||||
cities: List[String],
|
||||
aggregator: AggregateKey,
|
||||
): Map[String, AggregateValue] = {
|
||||
def queryData(userQuery: UserQuery, lines: List[String]): Map[String, Option[AggregateValue]] = {
|
||||
val weatherByCity = lines
|
||||
.flatMap(parseLine)
|
||||
.filter(line => cities.contains(line.city))
|
||||
.filter(line => userQuery.cities.contains(line.city))
|
||||
.groupBy(_.city)
|
||||
|
||||
def aggregateValue(
|
||||
selector: MeteoData => Option[Double],
|
||||
aggregate: List[Double] => Double,
|
||||
constructor: Double => AggregateValue
|
||||
): Map[String, AggregateValue] = {
|
||||
weatherByCity.map { case (city, weatherData) =>
|
||||
val values = weatherData.flatMap(wd => selector(wd.meteo))
|
||||
city -> constructor(aggregate(values))
|
||||
weatherByCity.map { case (city, weatherStationData) =>
|
||||
userQuery.field match {
|
||||
case "phenomena" => {
|
||||
val phenomenaList = weatherStationData.map(_.weather.phenomena)
|
||||
(city -> aggregatePhenomenaValues(userQuery.key, phenomenaList))
|
||||
}
|
||||
case field => {
|
||||
val doubleList = extractDoubleFieldValues(field, weatherStationData.map(_.weather))
|
||||
(city -> aggregateDoubleValues(userQuery.key, doubleList))
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
def aggregateList(
|
||||
selector: WeatherStationData => Option[(LocalDateTime, Double)],
|
||||
constructor: Map[LocalDateTime, Double] => AggregateValue
|
||||
): Map[String, AggregateValue] = {
|
||||
weatherByCity.map { case (city, weatherData) =>
|
||||
val values = weatherData.flatMap(selector).toMap
|
||||
city -> constructor(values)
|
||||
}
|
||||
}
|
||||
|
||||
val AggKey = AggregateKey
|
||||
val AggVal = AggregateValue
|
||||
|
||||
aggregator match {
|
||||
case AggKey.tempMax => aggregateValue(_.tempMax, _.max, AggVal.tempMax)
|
||||
case AggKey.tempMaxList => aggregateList(wd => wd.meteo.tempMax.map(wd.timestamp -> _), AggVal.tempMaxList)
|
||||
|
||||
case AggKey.tempMin => aggregateValue(_.tempMin, _.min, AggVal.tempMin)
|
||||
case AggKey.tempMinList => aggregateList(wd => wd.meteo.tempMin.map(wd.timestamp -> _), AggVal.tempMinList)
|
||||
|
||||
case AggKey.tempAvg => aggregateValue(_.tempAvg, values => values.sum / values.length, AggVal.tempAvg)
|
||||
case AggKey.tempAvgList => aggregateList(wd => wd.meteo.tempAvg.map(wd.timestamp -> _), AggVal.tempAvgList)
|
||||
|
||||
case AggKey.precipitationSum => aggregateValue(_.precipitation, _.sum, AggVal.precipitationSum)
|
||||
case AggKey.precipitationList => aggregateList(wd => wd.meteo.precipitation.map(wd.timestamp -> _), AggVal.precipitationList)
|
||||
|
||||
case AggKey.windAvg => aggregateValue(_.windSpeedAvg, values => values.sum / values.length, AggVal.windAvg)
|
||||
case AggKey.windAvgList => aggregateList(wd => wd.meteo.windSpeedAvg.map(wd.timestamp -> _), AggVal.windAvgList)
|
||||
|
||||
case AggKey.windMax => aggregateValue(_.windGustMax, _.max, AggVal.windMax)
|
||||
case AggKey.windMaxList => aggregateList(wd => wd.meteo.windGustMax.map(wd.timestamp -> _), AggVal.windMaxList)
|
||||
|
||||
case AggKey.visibilityMin => aggregateValue(_.visibilityMin, _.min, AggVal.visibilityMin)
|
||||
case AggKey.visibilityMinList => aggregateList(wd => wd.meteo.visibilityMin.map(wd.timestamp -> _), AggVal.visibilityMinList)
|
||||
|
||||
case AggKey.visibilityAvg => aggregateValue(_.visibilityAvg, values => values.sum / values.length, AggVal.visibilityAvg)
|
||||
case AggKey.visibilityAvgList => aggregateList(wd => wd.meteo.visibilityAvg.map(wd.timestamp -> _), AggVal.visibilityAvgList)
|
||||
|
||||
case AggKey.snowAvgList => aggregateList(wd => wd.meteo.snowThicknessAvg.map(wd.timestamp -> _), AggVal.snowAvgList)
|
||||
case AggKey.snowAvg => aggregateValue(_.snowThicknessAvg, values => values.sum / values.length, AggVal.snowAvg)
|
||||
case AggKey.snowMax => aggregateValue(_.snowThicknessAvg, _.max, AggVal.snowMax)
|
||||
|
||||
case AggKey.atmPressureList => aggregateList(wd => wd.meteo.atmPressure.map(wd.timestamp -> _), AggVal.atmPressureList)
|
||||
case AggKey.atmPressureMin => aggregateValue(_.atmPressure, _.min, AggVal.atmPressureMin)
|
||||
case AggKey.atmPressureMax => aggregateValue(_.atmPressure, _.max, AggVal.atmPressureMax)
|
||||
case AggKey.atmPressureAvg => aggregateValue(_.atmPressure, values => values.sum / values.length, AggVal.atmPressureAvg)
|
||||
|
||||
case AggKey.dewList => aggregateList(wd => wd.meteo.dewPoint.map(wd.timestamp -> _), AggVal.dewList)
|
||||
case AggKey.dewMin => aggregateValue(_.dewPoint, _.min, AggVal.dewMin)
|
||||
case AggKey.dewMax => aggregateValue(_.dewPoint, _.max, AggVal.dewMax)
|
||||
case AggKey.dewAvg => aggregateValue(_.dewPoint, values => values.sum / values.length, AggVal.dewAvg)
|
||||
|
||||
case AggKey.humidityAvg => aggregateValue(_.airHumidity, values => values.sum / values.length, AggVal.humidityAvg)
|
||||
case AggKey.humidityAvgList => aggregateList(wd => wd.meteo.airHumidity.map(wd.timestamp -> _), AggVal.humidityAvgList)
|
||||
|
||||
case AggKey.sunDurationList => aggregateList(wd => wd.meteo.sunshineDuration.map(wd.timestamp -> _), AggVal.sunDurationList)
|
||||
case AggKey.sunDurationSum => aggregateValue(_.sunshineDuration, _.sum, AggVal.sunDurationSum)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -7,65 +7,70 @@ import scala.reflect.runtime.universe._
|
||||
case class WeatherStationData(
|
||||
city: String,
|
||||
timestamp: LocalDateTime,
|
||||
meteo: MeteoData,
|
||||
phenomena: List[String],
|
||||
weather: WeatherData,
|
||||
)
|
||||
|
||||
case class MeteoData(
|
||||
case class WeatherData(
|
||||
tempMax: Option[Double],
|
||||
tempMin: Option[Double],
|
||||
tempAvg: Option[Double],
|
||||
precipitation: Option[Double],
|
||||
windSpeedAvg: Option[Double],
|
||||
windGustMax: Option[Double],
|
||||
windAvg: Option[Double],
|
||||
windMax: Option[Double],
|
||||
visibilityMin: Option[Double],
|
||||
visibilityAvg: Option[Double],
|
||||
snowThicknessAvg: Option[Double],
|
||||
snowAvg: Option[Double],
|
||||
atmPressure: Option[Double],
|
||||
dewPoint: Option[Double],
|
||||
airHumidity: Option[Double],
|
||||
sunshineDuration: Option[Double],
|
||||
humidity: Option[Double],
|
||||
sunDuration: Option[Double],
|
||||
phenomena: List[String],
|
||||
)
|
||||
|
||||
object MeteoData {
|
||||
def fromDoubles(data: List[Option[Double]]): Option[MeteoData] = data match {
|
||||
object WeatherData {
|
||||
def fromDoubles(data: List[Option[Double]], phenomena: List[String]): Option[WeatherData] = data match {
|
||||
case List(
|
||||
tempMax,
|
||||
tempMin,
|
||||
tempAvg,
|
||||
precipitation,
|
||||
windSpeedAvg,
|
||||
windGustMax,
|
||||
windAvg,
|
||||
windMax,
|
||||
visibilityMin,
|
||||
visibilityAvg,
|
||||
snowThicknessAvg,
|
||||
snowAvg,
|
||||
atmPressure,
|
||||
dewPoint,
|
||||
airHumidity,
|
||||
sunshineDuration
|
||||
) => Some(MeteoData(
|
||||
humidity,
|
||||
sunDuration
|
||||
) => Some(WeatherData(
|
||||
tempMax,
|
||||
tempMin,
|
||||
tempAvg,
|
||||
precipitation,
|
||||
windSpeedAvg,
|
||||
windGustMax,
|
||||
windAvg,
|
||||
windMax,
|
||||
visibilityMin,
|
||||
visibilityAvg,
|
||||
snowThicknessAvg,
|
||||
snowAvg,
|
||||
atmPressure,
|
||||
dewPoint,
|
||||
airHumidity,
|
||||
sunshineDuration,
|
||||
humidity,
|
||||
sunDuration,
|
||||
phenomena,
|
||||
))
|
||||
case _ => None
|
||||
}
|
||||
|
||||
def getCount: Int = {
|
||||
val constructor = typeOf[MeteoData].decl(termNames.CONSTRUCTOR).asMethod
|
||||
def getParamCount: Int = {
|
||||
val constructor = typeOf[WeatherData].decl(termNames.CONSTRUCTOR).asMethod
|
||||
val paramCount = constructor.paramLists.flatten.size
|
||||
paramCount
|
||||
}
|
||||
|
||||
def getDoubleParamCount: Int = {
|
||||
getParamCount - 1 // minus one because 'phenomena' List[String] not Double
|
||||
}
|
||||
}
|
||||
|
||||
//val cityList: List[String] = List("Ainaži", "Alūksne", "Bauska", "Dagda", "Daugavgrīva", "Daugavpils", "Dobele", "Gulbene", "Jelgava", "Kalnciems", "Kolka", "Kuldīga", "Lielpēči", "Liepāja", "Madona", "Mērsrags", "Pāvilosta", "Piedruja", "Priekuļi", "Rēzekne", "Rīga", "Rucava", "Rūjiena", "Saldus", "Sigulda", "Sīļi", "Skrīveri", "Skulte", "Stende", "Ventspils", "Vičaki", "Zīlāni", "Zosēni", "Ainaži", "Alūksne", "Bauska", "Dagda", "Daugavgrīva", "Daugavpils", "Dobele", "Gulbene", "Jelgava", "Kalnciems", "Kolka", "Kuldīga", "Lielpēči", "Liepāja", "Madona", "Mērsrags", "Pāvilosta", "Piedruja", "Priekuļi", "Rēzekne", "Rīga", "Rucava", "Rūjiena", "Saldus", "Sigulda", "Sīļi", "Skrīveri", "Skulte", "Stende", "Ventspils", "Vičaki", "Zīlāni", "Zosēni")
|
||||
@@ -76,14 +81,14 @@ object MeteoData {
|
||||
// "tempMin" -> "Stundas minimālā temperatūra",
|
||||
// "tempAvg" -> "Stundas vidējā temperatūra",
|
||||
// "precipitation" -> "Stundas nokrišņu daudzums",
|
||||
// "windSpeedAvg" -> "Vidējais vēja ātrums novērojumu termiņā (10 min. vidējais)",
|
||||
// "windGustMax" -> "Stundas maksimālās vēja brāzmas",
|
||||
// "windAvg" -> "Vidējais vēja ātrums novērojumu termiņā (10 min. vidējais)",
|
||||
// "windMax" -> "Stundas maksimālās vēja brāzmas",
|
||||
// "visibilityMin" -> "Stundas minimālā redzamība",
|
||||
// "visibilityAvg" -> "Stundas vidējā redzamība",
|
||||
// "snowThicknessAvg" -> "Stundas vidējais sniega segas biezums",
|
||||
// "snowAvg" -> "Stundas vidējais sniega segas biezums",
|
||||
// "atmPressure" -> "Atmosfēras spiediens jūras līmenī novērojuma termiņā (milibāros)",
|
||||
// "dewPoint" -> "Rasas punkta temperatūra novērojuma termiņā",
|
||||
// "airHumidity" -> "Relatīvais gaisa mitrums novērojumu termiņā",
|
||||
// "sunshineDuration" -> "Saules spīdēšanas ilgums",
|
||||
//// "weatherPhenomena" -> "Laika parādības",
|
||||
// "humidity" -> "Relatīvais gaisa mitrums novērojumu termiņā",
|
||||
// "sunDuration" -> "Saules spīdēšanas ilgums",
|
||||
//// "phenomena" -> "Laika parādības",
|
||||
//)
|
||||
@@ -4,14 +4,16 @@ import cats.effect._
|
||||
import cats.implicits.toTraverseOps
|
||||
import db.DBService
|
||||
import fetch.FetchService
|
||||
import parse.{AggregateKey, AggregateValue, Parser}
|
||||
import server.ValidateRoutes.{Aggregate, CityList, DateTimeRange, ValidDate}
|
||||
import parse.Parser
|
||||
import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, ValidDate}
|
||||
import io.circe.{Encoder, Json, Printer}
|
||||
import org.http4s._
|
||||
import org.http4s.dsl.io._
|
||||
import org.http4s.server.Router
|
||||
import org.http4s.server.blaze.BlazeServerBuilder
|
||||
import io.circe.syntax._
|
||||
import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder
|
||||
import parse.Aggregate.{AggregateKey, UserQuery}
|
||||
|
||||
|
||||
object Server extends IOApp {
|
||||
@@ -26,12 +28,11 @@ object Server extends IOApp {
|
||||
|
||||
private val appRoutes = HttpRoutes.of[IO] {
|
||||
|
||||
// http://localhost:3000/query/20230414_2200-20230501_1230/Liepāja,Rēzekne/tempAvg
|
||||
case GET -> Root / "query" / DateTimeRange(from, to) / CityList(cities) / Aggregate(aggregate) =>
|
||||
// http://localhost:3000/query/20230414_2200-20230501_1230/Liepāja,Rēzekne/tempMax/max
|
||||
case GET -> Root / "query" / DateTimeRange(from, to) / CityList(cities) / field / AggKey(key) =>
|
||||
DBService.getInRange(from, to)
|
||||
.map(Parser.queryData(_, cities, aggregate))
|
||||
.map(Parser.queryData(UserQuery(cities, field, key), _))
|
||||
.flatMap(result => Ok(result.asJson.pretty))
|
||||
// .flatMap(result => Ok("make json encoder"))
|
||||
|
||||
// http://localhost:3000/fetch/date/20230423
|
||||
case GET -> Root / "fetch" / "date" / ValidDate(date) =>
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
package server
|
||||
|
||||
import parse.AggregateKey
|
||||
import parse.Aggregate.AggregateKey
|
||||
|
||||
import java.time.{LocalDate, LocalDateTime}
|
||||
import java.time.format.DateTimeFormatter
|
||||
@@ -35,9 +35,9 @@ object ValidateRoutes {
|
||||
}
|
||||
}
|
||||
|
||||
object Aggregate {
|
||||
object AggKey {
|
||||
def unapply(str: String): Option[AggregateKey] = {
|
||||
AggregateKey.stringToAggregateParam(str)
|
||||
AggregateKey.fromString(str)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user