b0txec
3315f00fb1
Same root cause as the station-observation fix (this uses the same open-data portal, same UTC DATETIME field): the raw UTC string was passed straight through to the frontend as a display label, which JS's Date parser then reads as local time for a string with no timezone suffix — silently showing observation times 2-3h behind the newsroom's actual clock. Internal recency filtering (isRecent) was already self-consistent either way; this only affects display.
239 lines
10 KiB
Scala
239 lines
10 KiB
Scala
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, LocalDateTime, ZoneId, ZoneOffset}
|
|
import scala.util.Try
|
|
|
|
final case class WaterTemperatureReading(area: String, observedAt: String, min: Option[Double], max: 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)
|
|
|
|
// Marine (SEDUT) stations report less often than inland gauges — 12h
|
|
// comfortably covers that gap while still dropping genuinely stuck
|
|
// sensors (one inland station was found reporting a value 14h stale
|
|
// while its neighbors were all current).
|
|
private val recencyWindow = Duration.ofHours(12)
|
|
|
|
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"
|
|
)
|
|
|
|
private val areaOrder: Vector[String] = Vector("Jūra", "Līcis", "Kurzeme", "Zemgale", "Vidzeme", "Latgale")
|
|
|
|
// This dataset's DATETIME field is UTC (same source/portal as the station
|
|
// observations feed, same verified offset); converted to local so the
|
|
// observedAt label shown on Ūdens matches the newsroom's own clock instead
|
|
// of reading 2-3h behind.
|
|
private val rigaZone = ZoneId.of("Europe/Riga")
|
|
|
|
private def toLocal(utcDatetime: String): Option[String] =
|
|
Try(LocalDateTime.parse(utcDatetime).atZone(ZoneOffset.UTC).withZoneSameInstant(rigaZone).toLocalDateTime.toString).toOption
|
|
|
|
// Every LVĢMC hydrological station that reports water temperature (WTEMD
|
|
// inland/river stations, SEDUT coastal stations), grouped into the six
|
|
// named Ūdens zones. Boundaries follow Latvia's historical regions
|
|
// approximately, not administrative novadi — the open-data portal has no
|
|
// region field to key off. A handful of stations near region borders
|
|
// (Jēkabpils, Carnikava, Kolka, the cluster just west of Rīga) were
|
|
// judgment calls; verified against the live station/metadata list on
|
|
// 2026-08-23 — see docs/UPDATE_ROADMAP.md.
|
|
private val stationZones: Map[String, String] = Map(
|
|
// Jūra — open Baltic coast, west side of Kurzeme
|
|
"RILP99PA" -> "Jūra", // Liepāja
|
|
"RIVE99PA" -> "Jūra", // Ventspils
|
|
|
|
// Līcis — Gulf of Riga coast
|
|
"RIKO99PA" -> "Līcis", // Kolka
|
|
"RIME99MS" -> "Līcis", // Mērsrags
|
|
"SERO99MS" -> "Līcis", // Roja
|
|
"RISE99MS" -> "Līcis", // Skulte
|
|
"SALACGRI" -> "Līcis", // Salacgrīva
|
|
"DAUGAVGR" -> "Līcis", // Daugavgrīva
|
|
"SEJU99MS" -> "Līcis", // Lielupes grīva
|
|
"HD073813" -> "Līcis", // Carnikava — inland gauge, but right at the Gulf coast
|
|
|
|
// Kurzeme
|
|
"HD073580" -> "Kurzeme", // Rojupe
|
|
"HD073605" -> "Kurzeme", // Vārdava
|
|
"HD073613" -> "Kurzeme", // Kuldīga
|
|
"HD073616" -> "Kurzeme", // Vendzava
|
|
"HD073641" -> "Kurzeme", // Pakuļu HES
|
|
"HD073659" -> "Kurzeme", // Tērande
|
|
"HD073682" -> "Kurzeme", // Dūkupji
|
|
"HD073709" -> "Kurzeme", // Cīrava
|
|
"HD073726" -> "Kurzeme", // Alši
|
|
"HD073982" -> "Kurzeme", // Usma
|
|
"HD073529" -> "Kurzeme", // Lielveisi
|
|
|
|
// Zemgale
|
|
"HD073151" -> "Zemgale", // Jēkabpils
|
|
"HD073411" -> "Zemgale", // Zaķi
|
|
"HD073418" -> "Zemgale", // Stariņi
|
|
"HD073422" -> "Zemgale", // Mežotne
|
|
"HD073424" -> "Zemgale", // Staļģene
|
|
"HD073461" -> "Zemgale", // Sudrabkalni
|
|
"HD073482" -> "Zemgale", // Bauska
|
|
"HD073548" -> "Zemgale", // Bramberģe
|
|
"HD073552" -> "Zemgale", // Baloži
|
|
"HD073801" -> "Zemgale", // Jelgava
|
|
"HD073802" -> "Zemgale", // Kalnciems
|
|
|
|
// Vidzeme
|
|
"HD073003" -> "Vidzeme", // Mazsalaca
|
|
"HD073009" -> "Vidzeme", // Lagaste
|
|
"HD073025" -> "Vidzeme", // Oleri
|
|
"HD073044" -> "Vidzeme", // Velēna
|
|
"HD073052" -> "Vidzeme", // Valmiera
|
|
"HD073062" -> "Vidzeme", // Sigulda
|
|
"HD073071" -> "Vidzeme", // Zosēni
|
|
"HD073075" -> "Vidzeme", // Taurene
|
|
"HD073079" -> "Vidzeme", // Lejasciems
|
|
"HD073081" -> "Vidzeme", // Vilkzemnieki
|
|
"HD073098" -> "Vidzeme", // Melturi
|
|
"HD073345" -> "Vidzeme", // Lubāna
|
|
"HD073352" -> "Vidzeme", // Aiviekstes HES
|
|
"HD073366" -> "Vidzeme", // Litene
|
|
"HD073400" -> "Vidzeme", // Lielpeči
|
|
"HD073410" -> "Vidzeme", // Alderi
|
|
"HD073727" -> "Vidzeme", // Meņģele
|
|
"HD073810" -> "Vidzeme", // Rīga
|
|
"HD073902" -> "Vidzeme", // Zeļķi
|
|
"HD073904" -> "Vidzeme", // Pļaviņas
|
|
"HD073970" -> "Vidzeme", // Alūksne
|
|
|
|
// Latgale
|
|
"HD072694" -> "Latgale", // Lozdova
|
|
"HD072971" -> "Latgale", // Ludza
|
|
"HD073135" -> "Latgale", // Piedruja
|
|
"HD073137" -> "Latgale", // Krāslava
|
|
"HD073144" -> "Latgale", // Vaikuļāni
|
|
"HD073146" -> "Latgale", // Jersika
|
|
"HD073334" -> "Latgale", // Kūlenieki
|
|
"HD073380" -> "Latgale", // Griškāni
|
|
"HD073720" -> "Latgale", // Brūnuļi
|
|
"HD073721" -> "Latgale", // Lenderņa
|
|
"HD073953" -> "Latgale", // Spīdoles
|
|
"HD073972" -> "Latgale", // Kaunata
|
|
)
|
|
|
|
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]
|
|
filters = Json.obj(
|
|
"ABBREVIATION" -> Json.arr(Json.fromString("WTEMD"), Json.fromString("SEDUT")),
|
|
).noSpaces
|
|
// The combined WTEMD+SEDUT rolling window currently holds ~3,000
|
|
// rows; 5000 keeps headroom without needing pagination.
|
|
uri = baseUri.withQueryParam("limit", "5000").withQueryParam("filters", filters)
|
|
body <- client.expect[String](uri)
|
|
rows <- parseRecords(body)
|
|
readings = buildReadings(rows)
|
|
_ <- logger.info(s"Loaded water temperatures for ${readings.count(_.min.isDefined)} of ${readings.size} zones from ${rows.size} station readings")
|
|
} yield WaterTemperatureResponse("LVĢMC", readings)
|
|
}
|
|
|
|
private case class StationReading(stationId: String, datetime: String, value: Double)
|
|
|
|
private def buildReadings(rows: Vector[Json]): Vector[WaterTemperatureReading] = {
|
|
val parsed = rows.flatMap { row =>
|
|
val cursor = row.hcursor
|
|
for {
|
|
stationId <- cursor.downField("STATION_ID").as[String].toOption
|
|
rawDatetime <- cursor.downField("DATETIME").as[String].toOption
|
|
datetime <- toLocal(rawDatetime)
|
|
value <- cursor.downField("VALUE").as[Double].toOption
|
|
} yield StationReading(stationId, datetime, value)
|
|
}
|
|
|
|
if (parsed.isEmpty) {
|
|
areaOrder.map(WaterTemperatureReading(_, "", None, None))
|
|
} else {
|
|
// "Recent" is relative to the freshest timestamp this fetch actually
|
|
// returned, not wall-clock time — sidesteps any timezone ambiguity
|
|
// between the API and the app, since both readings come from the
|
|
// same source clock.
|
|
val latestTimestamp = parsed.map(_.datetime).max
|
|
val latestPerStation = parsed.groupBy(_.stationId).values.map(_.maxBy(_.datetime))
|
|
val fresh = latestPerStation.filter(reading => isRecent(reading.datetime, latestTimestamp))
|
|
|
|
val byZone = fresh
|
|
.flatMap(reading => stationZones.get(reading.stationId).map(_ -> reading))
|
|
.groupBy(_._1)
|
|
.view.mapValues(_.map(_._2))
|
|
.toMap
|
|
|
|
areaOrder.map { area =>
|
|
byZone.getOrElse(area, Iterable.empty).toVector match {
|
|
case values if values.nonEmpty =>
|
|
WaterTemperatureReading(area, values.map(_.datetime).max, Some(values.map(_.value).min), Some(values.map(_.value).max))
|
|
case _ =>
|
|
WaterTemperatureReading(area, "", None, None)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private def isRecent(datetime: String, referenceTimestamp: String): Boolean =
|
|
(for {
|
|
observed <- Try(LocalDateTime.parse(datetime)).toOption
|
|
reference <- Try(LocalDateTime.parse(referenceTimestamp)).toOption
|
|
} yield Duration.between(observed, reference).compareTo(recencyWindow) <= 0).getOrElse(false)
|
|
|
|
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}"))
|
|
}
|
|
)
|
|
}
|