From c5b84727a68cd43fb254382f77ef1278862a5644 Mon Sep 17 00:00:00 2001 From: Guntis Smaukstelis Date: Mon, 3 Jul 2023 23:17:43 +0300 Subject: [PATCH] Add granularity for aggregation, group by hour/day/month/year --- src/main/scala/parse/Aggregate.scala | 82 ++++++++++++++++------ src/main/scala/parse/Main.scala | 63 +++++++++++++++++ src/main/scala/parse/Parser.scala | 7 +- src/main/scala/server/Server.scala | 7 +- src/main/scala/server/ValidateRoutes.scala | 13 ++++ src/test/scala/parse/ParserSpec.scala | 13 ++-- web/src/aggregator/Aggregator.tsx | 4 ++ web/src/aggregator/CityChart.tsx | 16 ++++- web/src/aggregator/Result.tsx | 3 +- web/src/aggregator/SelectGranularity.tsx | 22 ++++++ web/src/consts.ts | 2 + 11 files changed, 198 insertions(+), 34 deletions(-) create mode 100644 src/main/scala/parse/Main.scala create mode 100644 web/src/aggregator/SelectGranularity.tsx diff --git a/src/main/scala/parse/Aggregate.scala b/src/main/scala/parse/Aggregate.scala index 5cb536c..40a688a 100644 --- a/src/main/scala/parse/Aggregate.scala +++ b/src/main/scala/parse/Aggregate.scala @@ -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 } } diff --git a/src/main/scala/parse/Main.scala b/src/main/scala/parse/Main.scala new file mode 100644 index 0000000..b1897f1 --- /dev/null +++ b/src/main/scala/parse/Main.scala @@ -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) + + } +} diff --git a/src/main/scala/parse/Parser.scala b/src/main/scala/parse/Parser.scala index 2aaf3cd..5140bf8 100644 --- a/src/main/scala/parse/Parser.scala +++ b/src/main/scala/parse/Parser.scala @@ -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)) } } diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index 2174176..6b769bc 100644 --- a/src/main/scala/server/Server.scala +++ b/src/main/scala/server/Server.scala @@ -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 diff --git a/src/main/scala/server/ValidateRoutes.scala b/src/main/scala/server/ValidateRoutes.scala index 1bb3d08..010fb1e 100644 --- a/src/main/scala/server/ValidateRoutes.scala +++ b/src/main/scala/server/ValidateRoutes.scala @@ -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 + } + } + } } diff --git a/src/test/scala/parse/ParserSpec.scala b/src/test/scala/parse/ParserSpec.scala index f11614a..c42c6ec 100644 --- a/src/test/scala/parse/ParserSpec.scala +++ b/src/test/scala/parse/ParserSpec.scala @@ -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)), ))) ) } diff --git a/web/src/aggregator/Aggregator.tsx b/web/src/aggregator/Aggregator.tsx index ecb0d3f..0176b5a 100644 --- a/web/src/aggregator/Aggregator.tsx +++ b/web/src/aggregator/Aggregator.tsx @@ -4,6 +4,7 @@ import { createSignal } from "solid-js"; import { Result } from "./Result"; import { SelectCity } from "./SelectCity"; import { SelectField } from "./SelectField"; +import { SelectGranularity } from "./SelectGranularity"; import { SelectKey } from "./SelectKey"; import { SelectTimeRange } from "./SelectTimeRange"; @@ -16,6 +17,7 @@ export function Aggregator() { const [getEnd, setEnd] = createSignal(nowRounded); const [getField, setField] = createSignal("tempMax"); const [getKey, setKey] = createSignal("max"); + const [getGranularity, setGranularity] = createSignal("hour"); return (
@@ -36,6 +38,7 @@ export function Aggregator() {
+
@@ -45,6 +48,7 @@ export function Aggregator() { getEnd={getEnd} getField={getField} getKey={getKey} + getGranularity={getGranularity} />
diff --git a/web/src/aggregator/CityChart.tsx b/web/src/aggregator/CityChart.tsx index afb270c..aae74b5 100644 --- a/web/src/aggregator/CityChart.tsx +++ b/web/src/aggregator/CityChart.tsx @@ -5,11 +5,25 @@ import moment from "moment"; // Register the controllers, elements, scales, and plugins we'll be using Chart.register(LineController, LinearScale, PointElement, LineElement, Title, CategoryScale); +function formatDateString(str: string): string { + const date = moment(new Date(str)); + switch (str.length) { + // year: 2023 + case 4: return date.format("yyyy"); + // year-month: 2023-06 + case 7: return date.format("yyyy-MM"); + // year-month-day: 2023-06-23 + case 10: return date.format("yyyy-MM-DD"); + // year-month-day hour:minute 2023-06-23T23:59 + default: return date.format("HH:mm"); + } +} + export const CityChart: Component<{ data: [string, number | null][] }> = (props) => { const [canvas, setCanvas] = createSignal(); let chart: Chart; const data: [string, number | null][] = props.data.map(([dateStr, value]) => [ - moment(new Date(dateStr)).format("HH:mm"), + formatDateString(dateStr), value, ]); diff --git a/web/src/aggregator/Result.tsx b/web/src/aggregator/Result.tsx index 185fbcd..14de578 100644 --- a/web/src/aggregator/Result.tsx +++ b/web/src/aggregator/Result.tsx @@ -12,6 +12,7 @@ export const Result: Component<{ getEnd: Accessor, getField: Accessor, getKey: Accessor, + getGranularity: Accessor, }> = (props) => { const cities = () => [...props.getCities()].join(","); const queryStart = () => moment(props.getStart()).format("YYYYMMDD_HHmm"); @@ -21,7 +22,7 @@ export const Result: Component<{ const fetchQuery = async (timestamp: number) => { if (cities() === "") return new Error("ERROR: Select cities!"); await new Promise(resolve => setTimeout(resolve, 500)) - const response = await fetch(`${apiHost}/api/query/${queryStart()}-${queryEnd()}/${cities()}/${props.getField()}/${props.getKey()}`); + const response = await fetch(`${apiHost}/api/query/${queryStart()}-${queryEnd()}/${props.getGranularity()}/${cities()}/${props.getField()}/${props.getKey()}`); const json = await response.json(); return json; } diff --git a/web/src/aggregator/SelectGranularity.tsx b/web/src/aggregator/SelectGranularity.tsx new file mode 100644 index 0000000..d04ab4d --- /dev/null +++ b/web/src/aggregator/SelectGranularity.tsx @@ -0,0 +1,22 @@ +import { Accessor, Component, Setter } from "solid-js"; + +import { aggregateGranularity } from "../consts"; + +export const SelectGranularity: Component<{ + getGranularity: Accessor, + setGranularity: Setter, +}> = ({ getGranularity, setGranularity }) => { + return ( +
+

Select granularity

+ +
+ ); +} \ No newline at end of file diff --git a/web/src/consts.ts b/web/src/consts.ts index 4d717c4..bb3a60d 100644 --- a/web/src/consts.ts +++ b/web/src/consts.ts @@ -6,6 +6,8 @@ export const weatherField = ["tempMax", "tempMin", "tempAvg", "precipitation", " export const aggregateKey = ["min", "max", "avg", "sum", "list", "distinct"]; +export const aggregateGranularity = ["hour", "day", "month", "year"]; + export type ResultKeyVal = [key: string, value: number | Array]; export const resultOrder = {