Add real-time water temperature backend (Ūdens), fetch-on-demand like warnings

2026-08-23 11:15:33 +03:00
parent ea1c563ff3
commit 4036d2477f
3 changed files with 134 additions and 5 deletions
@@ -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}"))
+        }
+    )
+}