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

137 lines
4.9 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
import java.time.{LocalDate, LocalDateTime, 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]] = ???
def getDateFileNames(date: LocalDate): IO[List[String]] = ???
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
// IO.blocking {
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
}
// }.flatten
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
}
}