diff --git a/src/main/scala/Main.scala b/src/main/scala/Main.scala index c14c2b6..103ce66 100644 --- a/src/main/scala/Main.scala +++ b/src/main/scala/Main.scala @@ -26,6 +26,7 @@ object Main extends IOApp { dataService <- DataService.of fetchLvgmcService <- lvgmc.FetchService.of warningService <- WarningService.of + waterTemperatureService <- lvgmc.WaterTemperatureService.of fetchWeatherStations = for { @@ -52,7 +53,7 @@ object Main extends IOApp { fetchGrib <- dmi.FetchService.of(dataService) fetchGribTask = scheduler.scheduleTask("Fetch Grib", List(43), fetchGrib.fetchRecentForecasts()).compile.drain - server <- Server.of(postgresService, dataService, fetchLvgmcService, warningService) + server <- Server.of(postgresService, dataService, fetchLvgmcService, warningService, waterTemperatureService) serverTask = server.run // Split from the legacy provider jobs below: parMapN cancels and fails diff --git a/src/main/scala/fetch/lvgmc/WaterTemperatureService.scala b/src/main/scala/fetch/lvgmc/WaterTemperatureService.scala new file mode 100644 index 0000000..339a093 --- /dev/null +++ b/src/main/scala/fetch/lvgmc/WaterTemperatureService.scala @@ -0,0 +1,120 @@ +package fetch.lvgmc + +import cats.effect.{IO, Ref} +import cats.syntax.all._ +import io.circe.Json +import io.circe.parser +import org.http4s.Uri +import org.http4s.ember.client.EmberClientBuilder +import org.typelevel.log4cats.Logger +import org.typelevel.log4cats.slf4j.Slf4jLogger + +import java.time.{Duration, Instant} + +final case class WaterTemperatureReading(area: String, observedAt: String, value: Option[Double]) +final case class WaterTemperatureResponse(source: String, readings: Vector[WaterTemperatureReading]) + +object WaterTemperatureService { + def of: IO[WaterTemperatureService] = + for { + logger <- Slf4jLogger.create[IO] + cache <- Ref.of[IO, Option[(Instant, WaterTemperatureResponse)]](None) + } yield new WaterTemperatureService(logger, cache) +} + +// Real-time only, by design: the Ūdens workspace has never persisted history +// (its ranges are manual editorial inputs), and this follows the same +// pattern — an in-memory cache, no database table, mirroring +// fetch.warnings.WarningService rather than the station-observation path. +final class WaterTemperatureService private (logger: Logger[IO], cache: Ref[IO, Option[(Instant, WaterTemperatureResponse)]]) { + private val cacheLifetime = Duration.ofMinutes(15) + + private val observationsUrl = sys.env.getOrElse( + "LVGMC_OPENDATA_HYDRO_URL", + "https://data.gov.lv/dati/api/3/action/datastore_search?resource_id=de5f06e9-6f44-497d-8ec2-72a2483608e8" + ) + + // One representative real station per named water area on the Ūdens + // production template, verified 2026-08-23 against LVĢMC's open + // hydrological data. Jūra/Līcis are coastal sea-temperature stations + // (SEDUT); the four historical regions use inland river/lake stations + // (WTEMD). Approximate by design — see docs/UPDATE_ROADMAP.md. + private val areaStations: List[(String, String, String)] = List( + ("Jūra", "RIVE99PA", "SEDUT"), + ("Līcis", "RISE99MS", "SEDUT"), + ("Kurzeme", "HD073613", "WTEMD"), + ("Zemgale", "HD073801", "WTEMD"), + ("Vidzeme", "HD073062", "WTEMD"), + ("Latgale", "HD073141", "WTEMD"), + ) + + def fetchWaterTemperatures: IO[WaterTemperatureResponse] = + for { + now <- IO.delay(Instant.now) + cached <- cache.get + response <- cached match { + case Some((storedAt, value)) if Duration.between(storedAt, now).compareTo(cacheLifetime) < 0 => + IO.pure(value) + case stale => + fetchFresh.attempt.flatMap { + case Right(value) => cache.set(Some(now -> value)).as(value) + case Left(error) => stale match { + // Re-stamp with `now` even on failure so a dead upstream is + // retried at most once per cache lifetime, not on every request + // (see the identical fix in WarningService). + case Some((_, value)) => + logger.warn(error)("Using stale water temperature data; will not retry until the cache lifetime elapses") *> + cache.set(Some(now -> value)).as(value) + case None => IO.raiseError(error) + } + } + } + } yield response + + private def fetchFresh: IO[WaterTemperatureResponse] = + EmberClientBuilder.default[IO].build.use { client => + for { + baseUri <- Uri.fromString(observationsUrl).liftTo[IO] + stationIds = areaStations.map(_._2).distinct + abbreviations = areaStations.map(_._3).distinct + filters = Json.obj( + "STATION_ID" -> Json.arr(stationIds.map(Json.fromString): _*), + "ABBREVIATION" -> Json.arr(abbreviations.map(Json.fromString): _*), + ).noSpaces + // The underlying resource is LVĢMC's rolling 48h operational window, + // so this narrow a filter (6 stations, 2 parameters) never needs + // pagination the way the larger station/warning fetches do. + uri = baseUri.withQueryParam("limit", "2000").withQueryParam("filters", filters) + body <- client.expect[String](uri) + rows <- parseRecords(body) + readings = areaStations.map { case (area, stationId, abbreviation) => + latestFor(rows, stationId, abbreviation) match { + case Some((observedAt, value)) => WaterTemperatureReading(area, observedAt, Some(value)) + case None => WaterTemperatureReading(area, "", None) + } + } + _ <- logger.info(s"Loaded water temperatures for ${readings.count(_.value.isDefined)} of ${readings.size} areas") + } yield WaterTemperatureResponse("LVĢMC", readings.toVector) + } + + private def latestFor(rows: Vector[Json], stationId: String, abbreviation: String): Option[(String, Double)] = + rows.flatMap { row => + val cursor = row.hcursor + for { + sid <- cursor.downField("STATION_ID").as[String].toOption if sid == stationId + ab <- cursor.downField("ABBREVIATION").as[String].toOption if ab == abbreviation + datetime <- cursor.downField("DATETIME").as[String].toOption + value <- cursor.downField("VALUE").as[Double].toOption + } yield datetime -> value + }.sortBy(_._1)(Ordering[String].reverse).headOption + + private def parseRecords(body: String): IO[Vector[Json]] = + IO.fromEither( + parser.parse(body) + .leftMap(error => new RuntimeException(s"Unable to parse water temperature data: ${error.message}")) + .flatMap { json => + json.hcursor.downField("result").downField("records").as[Vector[Json]] + .leftMap(error => new RuntimeException(s"Invalid water temperature response: ${error.message}")) + } + ) +} diff --git a/src/main/scala/server/Server.scala b/src/main/scala/server/Server.scala index 8e04433..177e36d 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 data.DataService import db.PostgresService import fetch.csv.FileNameService -import fetch.lvgmc.FetchService +import fetch.lvgmc.{FetchService, WaterTemperatureService} import fetch.warnings.WarningService import fs2.io.file.{Files, Path} import server.ValidateRoutes.{AggFieldList, AggKey, CityList, DateTimeRange, Granularity, ValidateDate, ValidateDateTime, ValidateInt, ValidateMonths, ValidateZonedDateTime} @@ -34,14 +34,14 @@ import scala.concurrent.duration.DurationInt object Server { - def of(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService): IO[Server] = { + def of(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService, waterTemperatures: WaterTemperatureService): IO[Server] = { Slf4jLogger.create[IO].map { - new Server(postgresService, dataService, fetch, warnings, _) + new Server(postgresService, dataService, fetch, warnings, waterTemperatures, _) } } } -class Server(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService, log: Logger[IO]) { +class Server(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService, waterTemperatures: WaterTemperatureService, log: Logger[IO]) { private case class ResponseWrapper(result: Map[String, Option[Aggregate.AggregateValue]], query: UserQuery) private case class LatestTemperature(city: String, observedAt: String, value: Option[Double]) @@ -54,6 +54,14 @@ class Server(postgresService: PostgresService, dataService: DataService, fetch: ServiceUnavailable("LVĢMC brīdinājumu dati pašlaik nav pieejami") } + case GET -> Root / "water-temperatures" => + waterTemperatures.fetchWaterTemperatures + .flatMap(result => Ok(result.asJson)) + .handleErrorWith { error => + log.error(error)("Unable to load LVĢMC water temperatures") *> + ServiceUnavailable("Ūdens temperatūras dati pašlaik nav pieejami") + } + // http://0.0.0.0:8080/api/show/lvgmc-forecast/Latvija_LTV_pilsetas_tekosa_dn.csv case GET -> Root / "show" / "lvgmc-forecast" / fileName => fetch.fetchFile(fileName).flatMap(bytes =>