Files
WeatherTool/src/main/scala/db/PostgresService.scala
T

173 lines
6.8 KiB
Scala
Raw Normal View History

2023-10-21 21:12:58 +03:00
package db
2023-10-22 21:19:11 +03:00
import cats.data.NonEmptyList
2023-10-21 21:12:58 +03:00
import cats.effect._
import cats.implicits.{catsSyntaxParallelTraverse1, toFoldableOps}
2023-10-21 21:12:58 +03:00
import doobie._
import doobie.implicits._
import java.time.format.DateTimeFormatter
2023-12-12 15:36:30 +02:00
import java.time.{LocalDate, LocalDateTime, OffsetDateTime, ZoneId, ZonedDateTime}
2023-10-21 21:12:58 +03:00
import doobie.postgres.implicits._
2023-10-22 23:51:59 +03:00
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
import parse.{Parser, WeatherStationData}
2023-10-21 21:12:58 +03:00
2023-10-22 23:51:59 +03:00
object PostgresService {
def of(transactor: Transactor[IO]): IO[PostgresService] = {
Slf4jLogger.create[IO].map(logger => new PostgresService(transactor, logger))
}
}
2023-10-21 21:12:58 +03:00
class PostgresService(transactor: Transactor[IO], log: Logger[IO]) extends DataServiceTrait {
private val rigaZone = ZoneId.of("Europe/Riga")
// in pgAdmin run: ```SET TIMEZONE = 'Europe/Riga';```
def save(fileName: String, content: String): IO[String] = {
val strLines = content.split(System.lineSeparator()).toList
val weatherStationData = strLines.flatMap(Parser.parseLine)
insertInWeatherTable(weatherStationData)
.attempt
.flatMap {
case Left(error) => log.error(s"Write db '$fileName' failed with error: ${error.getMessage}") *> IO.raiseError(error)
case Right(rowCount) => log.info(s"write rows: $rowCount file: $fileName").as(fileName)
}
}
def getDatesByMonths(monthList: List[LocalDate]): IO[List[LocalDate]] = {
val monthYearPairs = monthList.map { localDate =>
val month = localDate.atStartOfDay.atZone(rigaZone).getMonthValue
val year = localDate.atStartOfDay.atZone(rigaZone).getYear
(month, year)
}
val whereClauses = monthYearPairs.map {
case (month, year) => fr"(EXTRACT(MONTH FROM dateTime) = $month AND EXTRACT(YEAR FROM dateTime) = $year)"
}
val combinedWhereClause = whereClauses.intercalate(fr" OR ")
(
fr"""
SELECT DISTINCT DATE(dateTime)
FROM weather
WHERE """ ++ combinedWhereClause
)
.query[LocalDate]
.to[List]
.transact(transactor)
}
def readFile(fileName: String): IO[List[String]] = ???
def getInRange(from: LocalDateTime, to: LocalDateTime): IO[List[String]] = ???
def getDates: IO[List[LocalDate]] = ???
2023-12-12 15:36:30 +02:00
def getDateTimeEntries(dateTime: LocalDateTime): IO[List[String]] = {
2023-12-12 16:15:36 +02:00
val columnNames = List("City; TempMax; TempMin; TempAvg; Precipitation; WindAvg; WindMax; VisibilityMin; VisibilityAvg; SnowAvg; AtmPressure; DewPoint; Humidity; SunDuration; Phenomena")
val result = fr"""
SELECT city || ';' || COALESCE(tempmax::text, '' ) || ';' || COALESCE(tempmin::text, '' ) || ';' || COALESCE(tempavg::text, '' ) || ';' || COALESCE(precipitation::text, '' ) || ';' || COALESCE(windavg::text, '' ) || ';' || COALESCE(windmax::text, '' ) || ';' || COALESCE(tempmax::text, '' ) || ';' || COALESCE(visibilitymin::text, '' ) || ';' || COALESCE(visibilityavg::text, '' ) || ';' || COALESCE(snowavg::text, '' ) || ';' || COALESCE(atmpressure::text, '' ) || ';' || COALESCE(dewpoint::text, '' ) || ';' || COALESCE(humidity::text, '' ) || ';' || COALESCE(sunduration::text, '' ) || ';' || array_to_string(phenomena, ', ')
2023-12-12 15:36:30 +02:00
FROM weather
WHERE dateTime = $dateTime
ORDER BY city
"""
.query[String]
.to[List]
.transact(transactor)
2023-12-12 16:15:36 +02:00
result.map(columnNames ++ _)
2023-12-12 15:36:30 +02:00
}
def getDateFileNames(date: LocalDate): IO[List[String]] = {
val year = date.atStartOfDay.atZone(rigaZone).getYear
val month = date.atStartOfDay.atZone(rigaZone).getMonthValue
val day = date.atStartOfDay.atZone(rigaZone).getDayOfMonth
val whereClauses = fr"(EXTRACT(DAY FROM dateTime) = $day AND EXTRACT(MONTH FROM dateTime) = $month AND EXTRACT(YEAR FROM dateTime) = $year)"
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HH00")
(
fr"""
SELECT DISTINCT dateTime
FROM weather
WHERE """ ++ whereClauses ++
fr"ORDER BY dateTime"
)
.query[OffsetDateTime]
.to[List]
.transact(transactor)
2023-12-12 16:15:36 +02:00
.map { dateTimes =>
dateTimes.map(dateTime =>
dateTime.atZoneSameInstant(rigaZone).format(formatter)
)
2023-12-12 15:36:30 +02:00
}
}
2023-12-12 16:15:36 +02:00
2023-10-21 21:12:58 +03:00
def getResourceContent(path: String): IO[String] = {
val streamResource = Resource.make(IO(getClass.getResourceAsStream(path))) { stream =>
IO(stream.close()).handleErrorWith(_ => IO.unit)
}
streamResource.use { stream =>
IO(scala.io.Source.fromInputStream(stream).mkString)
}
}
def createWeatherTable: IO[Int] = {
2023-10-21 21:12:58 +03:00
for {
createTableSql <- getResourceContent("/db/create_weather_table.sql")
2023-10-22 23:51:59 +03:00
result <- Update0(createTableSql, None).run.transact(transactor)
2023-10-21 21:12:58 +03:00
} yield result
}
implicit val doubleOptionMeta: Meta[Option[Double]] = Meta[Double].imap(Option(_))(_.getOrElse(Double.NaN))
def insertInWeatherTable(data: List[WeatherStationData]): IO[Int] = {
2023-12-11 00:12:52 +02:00
getResourceContent("/db/insert_weather_table.sql").flatMap { insertTableSql =>
val insertData = data.map(line => {
val zonedTime: ZonedDateTime = line.timestamp.atZone(rigaZone)
val w = line.weather
(zonedTime, line.city, w.tempMax, w.tempMin, w.tempAvg, w.precipitation, w.windAvg, w.windMax, w.visibilityMin, w.visibilityAvg, w.snowAvg, w.atmPressure, w.dewPoint, w.humidity, w.sunDuration, w.phenomena)
})
Update[(ZonedDateTime, String, Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], Option[Double], List[String])](insertTableSql)
.updateMany(insertData)
.transact(transactor)
2023-12-11 00:12:52 +02:00
}
2023-10-21 21:12:58 +03:00
}
2023-10-22 23:51:59 +03:00
def selectWeatherTable(): IO[List[(String, Option[Double])]] = {
2023-10-22 21:19:11 +03:00
val formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmm")
val from = LocalDateTime.parse("20230516_0100", formatter).atZone(rigaZone)
val to = LocalDateTime.parse("20230516_1500", formatter).atZone(rigaZone)
val citiesNel = NonEmptyList.of("Rīga", "Rēzekne")
val granularity = "HOUR" // "HOUR", "DAY", "MONTH", "YEAR"
val columnName = "tempMin"
val baseQuery =
fr"""
SELECT
city,
2023-10-22 23:51:59 +03:00
MIN(""" ++ Fragment.const(columnName) ++ fr""") AS tempMin
2023-10-22 21:19:11 +03:00
FROM weather
WHERE
""" ++ Fragments.in(fr"city", citiesNel) ++ fr"""
AND dateTime BETWEEN $from AND $to
AND """ ++ Fragment.const(columnName) ++fr""" IS NOT NULL
GROUP BY city, EXTRACT(""" ++ Fragment.const(granularity) ++ fr""" FROM dateTime)
"""
baseQuery
.query[(String, Option[Double])]
.to[List]
2023-10-22 23:51:59 +03:00
.transact(transactor)
2023-10-22 21:19:11 +03:00
}
2023-10-22 23:51:59 +03:00
def dropWeatherTable(): IO[Int] = {
2023-10-21 21:12:58 +03:00
for {
dropTableSql <- getResourceContent("/db/drop_weather_table.sql")
2023-10-22 23:51:59 +03:00
result <- Update0(dropTableSql, None).run.transact(transactor)
2023-10-21 21:12:58 +03:00
} yield result
}
}