bot/WeatherTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add granularity for aggregation, group by hour/day/month/year

Browse Source
This commit is contained in:
Guntis Smaukstelis
2023-07-03 23:17:43 +03:00
parent 4d93fe68dc
commit c5b84727a6
11 changed files with 198 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/parse/Aggregate.scala
+62 -20
View File
@@ -4,14 +4,18 @@ import cats.implicits.{catsSyntaxOptionId, toFoldableOps}
import io.circe.generic.semiauto.deriveEncoder
import io.circe.syntax.EncoderOps
import io.circe.{Encoder, Json}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
object Aggregate {
final case class UserQuery(
cities: List[String],
field: String,
key: AggregateKey
key: AggregateKey,
granularity: ChronoUnit,
)
sealed trait AggregateKey
@@ -46,7 +50,7 @@ object Aggregate {
sealed trait AggregateValue
final case class DoubleValue(value: Double) extends AggregateValue
final case class TimeDoubleList(list: List[(LocalDateTime, Option[Double])]) extends AggregateValue
final case class TimeDoubleList(list: List[(String, Option[Double])]) extends AggregateValue
final case class StringListList(list: List[List[String]]) extends AggregateValue
final case class DistinctStringList(list: List[String]) extends AggregateValue
object AggregateValue {
@@ -80,21 +84,46 @@ object Aggregate {
implicit val distinctStringListEncoder: Encoder[DistinctStringList] = deriveEncoder[DistinctStringList]
}
def extractDoubleFieldValues(field: String, weatherData: List[WeatherData]): List[Option[Double]] =
private def convertDateTime(granularity: ChronoUnit)(data: (LocalDateTime, _)): String = {
val dateTime = data._1
granularity match {
case ChronoUnit.DAYS => dateTime.toLocalDate.toString
case ChronoUnit.MONTHS => dateTime.format(DateTimeFormatter.ofPattern("yyyy-MM"))
case ChronoUnit.YEARS => dateTime.format(DateTimeFormatter.ofPattern("yyyy"))
case _ => dateTime.withMinute(0).toString
}
}
private def groupByGranularity(
list: List[(LocalDateTime, Option[Double])],
granularity: ChronoUnit,
aggregateKey: AggregateKey,
): List[(String, Option[Double])] = {
list
.groupBy(convertDateTime(granularity))
.toList
.map { case(dateTime, groupedList) => (dateTime, flattenField(aggregateKey, groupedList)) }
}
def aggregateByField(
field: String,
granularity: ChronoUnit,
data: List[WeatherStationData],
): List[(String, 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 "tempMax" => groupByGranularity(data.map(d => (d.timestamp, d.weather.tempMax)), granularity, AggregateKey.Max)
case "tempMin" => groupByGranularity(data.map(d => (d.timestamp, d.weather.tempMin)), granularity, AggregateKey.Min)
case "tempAvg" => groupByGranularity(data.map(d => (d.timestamp, d.weather.tempAvg)), granularity, AggregateKey.Avg)
case "precipitation" => groupByGranularity(data.map(d => (d.timestamp, d.weather.precipitation)), granularity, AggregateKey.Sum)
case "windAvg" => groupByGranularity(data.map(d => (d.timestamp, d.weather.windAvg)), granularity, AggregateKey.Avg)
case "windMax" => groupByGranularity(data.map(d => (d.timestamp, d.weather.windMax)), granularity, AggregateKey.Max)
case "visibilityMin" => groupByGranularity(data.map(d => (d.timestamp, d.weather.visibilityMin)), granularity, AggregateKey.Min)
case "visibilityAvg" => groupByGranularity(data.map(d => (d.timestamp, d.weather.visibilityAvg)), granularity, AggregateKey.Avg)
case "snowAvg" => groupByGranularity(data.map(d => (d.timestamp, d.weather.snowAvg)), granularity, AggregateKey.Avg)
case "atmPressure" => groupByGranularity(data.map(d => (d.timestamp, d.weather.atmPressure)), granularity, AggregateKey.Avg)
case "dewPoint" => groupByGranularity(data.map(d => (d.timestamp, d.weather.dewPoint)), granularity, AggregateKey.Avg)
case "humidity" => groupByGranularity(data.map(d => (d.timestamp, d.weather.humidity)), granularity, AggregateKey.Avg)
case "sunDuration" => groupByGranularity(data.map(d => (d.timestamp, d.weather.sunDuration)), granularity, AggregateKey.Sum)
case _ => List.empty
}
@@ -102,18 +131,31 @@ object Aggregate {
def roundDecimal: Double = BigDecimal(d).setScale(1, BigDecimal.RoundingMode.HALF_UP).toDouble
}
private def flattenField(
AggKey: AggregateKey,
values: List[(LocalDateTime, Option[Double])],
): Option[Double] = {
val flatValues = values.flatMap(_._2)
AggKey match {
case AggregateKey.Min => flatValues.minimumOption
case AggregateKey.Max => flatValues.maximumOption
case AggregateKey.Avg => flatValues.reduceOption(_ + _).map(_ / flatValues.length)
case AggregateKey.Sum => flatValues.sum.some
case _ => None
}
}
def aggregateDoubleValues(
AggKey: AggregateKey,
values: List[Option[Double]],
timestamps: List[LocalDateTime],
values: List[(String, Option[Double])],
): Option[AggregateValue] = {
val flatValues = values.flatten
val flatValues = values.flatMap(_._2)
AggKey match {
case AggregateKey.Min => flatValues.minimumOption.map(value => DoubleValue(value.roundDecimal))
case AggregateKey.Max => flatValues.maximumOption.map(value => DoubleValue(value.roundDecimal))
case AggregateKey.Avg => flatValues.reduceOption(_ + _).map(sum => DoubleValue((sum / flatValues.length).roundDecimal))
case AggregateKey.Sum => flatValues.sum.some.map(value => DoubleValue(value.roundDecimal))
case AggregateKey.List => TimeDoubleList(timestamps.zip(values).sorted).some
case AggregateKey.List => TimeDoubleList(values.sorted).some
case AggregateKey.Distinct => None
}
}
src/main/scala/parse/Main.scala
+63
View File
@@ -0,0 +1,63 @@
package parse
import cats.effect.unsafe.implicits.global
import db.DBService
import io.circe.syntax.EncoderOps
import parse.Aggregate.{AggregateKey, UserQuery}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import parse.Aggregate.AggregateValueImplicits.aggregateValueEncoder
// TODO remake this as a test with different granularities and especially check avg value calculations
object Main {
def main(args: Array[String]): Unit = {
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
def groupByGranularity(granularity: ChronoUnit) = {
granularity match {
case ChronoUnit.DAYS => (dateTime: LocalDateTime) => dateTime.toLocalDate
case ChronoUnit.MONTHS => (dateTime: LocalDateTime) => dateTime.toLocalDate.withDayOfMonth(1)
case ChronoUnit.YEARS => (dateTime: LocalDateTime) => dateTime.toLocalDate.withDayOfYear(1)
case _ => (dateTime: LocalDateTime) => dateTime.withMinute(0).withSecond(0).withNano(0)
}
}
val list = List(
(LocalDateTime.parse("20230515_0900", formatter), Some(20d)),
(LocalDateTime.parse("20230515_1000", formatter), None),
(LocalDateTime.parse("20230515_1100", formatter), Some(6d)),
(LocalDateTime.parse("20230516_1100", formatter), Some(10d)),
(LocalDateTime.parse("20230416_1100", formatter), Some(10d)),
)
val grouped = list.groupBy{ case (dateTime, _) => groupByGranularity(ChronoUnit.HOURS)(dateTime) }
// println(grouped)
// grouped.foreach { case (key, value) =>
// println(s"$key -> $value")
// }
val from = LocalDateTime.parse("20230627_0000", formatter)
val to = LocalDateTime.parse("20230627_2359", formatter)
val dbService = DBService.of.unsafeRunSync()
val lines = dbService.getInRange(from, to).unsafeRunSync()
val query = UserQuery(List("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.HOURS)
val parsed = Parser.queryData(query, lines)
println(parsed.asJson)
// val query2 = UserQuery(List("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.DAYS)
// val parsed2 = Parser.queryData(query2, lines)
// println(parsed2.asJson)
// val query3 = UserQuery(List("Ainaži"), "tempAvg", AggregateKey.Avg, ChronoUnit.MONTHS)
// val parsed3 = Parser.queryData(query3, lines)
// println(parsed3.asJson)
}
}
src/main/scala/parse/Parser.scala
+4 -3
View File
@@ -1,9 +1,10 @@
package parse
import parse.Aggregate.{AggregateValue, UserQuery, aggregateDoubleValues, aggregatePhenomenaValues, extractDoubleFieldValues}
import parse.Aggregate.{AggregateValue, UserQuery, aggregateDoubleValues, aggregatePhenomenaValues, aggregateByField}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import scala.util.Try
object Parser {
@@ -48,8 +49,8 @@ object Parser {
(city -> aggregatePhenomenaValues(userQuery.key, phenomenaList))
}
case field => {
val doubleList = extractDoubleFieldValues(field, weatherStationData.map(_.weather))
(city -> aggregateDoubleValues(userQuery.key, doubleList, weatherStationData.map(_.timestamp)))
val doubleList = aggregateByField(field, userQuery.granularity, weatherStationData)
(city -> aggregateDoubleValues(userQuery.key, doubleList))
}
}
src/main/scala/server/Server.scala
+4 -3
View File
@@ -6,7 +6,7 @@ import com.comcast.ip4s.IpLiteralSyntax
import db.DataService
import fetch.FetchService
import parse.{Parser, WeatherData}
import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, ValidDate}
import server.ValidateRoutes.{AggKey, CityList, DateTimeRange, Granularity, ValidDate}
import io.circe.{Json, Printer}
import org.http4s._
import org.http4s.dsl.io._
@@ -23,6 +23,7 @@ import org.http4s.circe.jsonEncoder
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
import java.time.temporal.ChronoUnit
import scala.concurrent.duration.DurationInt
@@ -47,9 +48,9 @@ class Server(dataService: DataService, fetch: FetchService, log: Logger[IO]) {
private val apiRoutes = HttpRoutes.of[IO] {
// http://0.0.0.0:8080/api/query/20230414_2200-20230501_1230/Liepāja,Rēzekne/tempMax/max
case GET -> Root / "query" / DateTimeRange(from, to) / CityList(cities) / field / AggKey(key) =>
case GET -> Root / "query" / DateTimeRange(from, to) / Granularity(granularity) / CityList(cities) / field / AggKey(key) =>
dataService.getInRange(from, to)
.map(Parser.queryData(UserQuery(cities, field, key), _))
.map(Parser.queryData(UserQuery(cities, field, key, granularity), _))
.flatMap(result => Ok(result.asJson.pretty))
// http://0.0.0.0:8080/api/fetch/date/20230514
src/main/scala/server/ValidateRoutes.scala
+13
View File
@@ -4,6 +4,7 @@ import parse.Aggregate.AggregateKey
import java.time.{LocalDate, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import scala.util.Try
object ValidateRoutes {
@@ -40,4 +41,16 @@ object ValidateRoutes {
AggregateKey.fromString(str)
}
}
object Granularity {
def unapply(str: String): Option[ChronoUnit] = {
str match {
case "hour" => Some(ChronoUnit.HOURS)
case "day" => Some(ChronoUnit.DAYS)
case "month" => Some(ChronoUnit.MONTHS)
case "year" => Some(ChronoUnit.YEARS)
case _ => None
}
}
}
}
src/test/scala/parse/ParserSpec.scala
+7 -6
View File
@@ -8,6 +8,7 @@ import parse.Aggregate.{AggregateKey, DoubleValue, TimeDoubleList, UserQuery}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import scala.collection.immutable.HashMap
class ParserSpec extends AnyFunSuite with Matchers {
@@ -16,7 +17,7 @@ class ParserSpec extends AnyFunSuite with Matchers {
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
val from = LocalDateTime.parse("20230515_0905", formatter)
val to = LocalDateTime.parse("20230516_0942", formatter)
val userQuery = UserQuery(List("Bauska", "Dagda", "Daugavgrīva", "Rīga"), "precipitation", AggregateKey.Sum)
val userQuery = UserQuery(List("Bauska", "Dagda", "Daugavgrīva", "Rīga"), "precipitation", AggregateKey.Sum, ChronoUnit.HOURS)
val dbService = DBService.of.unsafeRunSync()
@@ -35,7 +36,7 @@ class ParserSpec extends AnyFunSuite with Matchers {
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
val from = LocalDateTime.parse("20230516_0400", formatter)
val to = LocalDateTime.parse("20230516_0800", formatter)
val userQuery = UserQuery(List("Rīga"), "precipitation", AggregateKey.List)
val userQuery = UserQuery(List("Rīga"), "precipitation", AggregateKey.List, ChronoUnit.HOURS)
val dbService = DBService.of.unsafeRunSync()
@@ -45,10 +46,10 @@ class ParserSpec extends AnyFunSuite with Matchers {
parsed shouldBe HashMap(
"Rīga" ->
Some(TimeDoubleList(List(
(LocalDateTime.parse("2023-05-16T04:00"), Some(1.9)),
(LocalDateTime.parse("2023-05-16T05:00"), Some(4.5)),
(LocalDateTime.parse("2023-05-16T06:00"), Some(1.5)),
(LocalDateTime.parse("2023-05-16T07:00"), Some(0.0)),
("2023-05-16T04:00", Some(1.9)),
("2023-05-16T05:00", Some(4.5)),
("2023-05-16T06:00", Some(1.5)),
("2023-05-16T07:00", Some(0.0)),
)))
)
}