Show real regional water-temperature ranges, not one station duplicated as both bounds
Classify all 65 LVĢMC stations reporting water temperature (56 inland WTEMD + 9 coastal SEDUT) into the six Ūdens zones by geography, and report the real min/max across each zone's currently-reporting stations instead of one hand-picked station's single value shown twice. Drops readings older than 12h so a stuck sensor can't skew a zone's range.
This commit is contained in:
b0txec
@@ -9,9 +9,10 @@ import org.http4s.ember.client.EmberClientBuilder
|
||||
import org.typelevel.log4cats.Logger
|
||||
import org.typelevel.log4cats.slf4j.Slf4jLogger
|
||||
|
||||
import java.time.{Duration, Instant}
|
||||
import java.time.{Duration, Instant, LocalDateTime}
|
||||
import scala.util.Try
|
||||
|
||||
final case class WaterTemperatureReading(area: String, observedAt: String, value: Option[Double])
|
||||
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 {
|
||||
@@ -29,23 +30,104 @@ object WaterTemperatureService {
|
||||
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"
|
||||
)
|
||||
|
||||
// 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", "HD072971", "WTEMD"), // Ludza — Daugavpils's station (HD073141) only reports water level (LIMEN), not temperature
|
||||
private val areaOrder: Vector[String] = Vector("Jūra", "Līcis", "Kurzeme", "Zemgale", "Vidzeme", "Latgale")
|
||||
|
||||
// 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] =
|
||||
@@ -75,38 +157,64 @@ final class WaterTemperatureService private (logger: Logger[IO], cache: Ref[IO,
|
||||
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): _*),
|
||||
"ABBREVIATION" -> Json.arr(Json.fromString("WTEMD"), Json.fromString("SEDUT")),
|
||||
).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)
|
||||
// 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 = 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)
|
||||
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 def latestFor(rows: Vector[Json], stationId: String, abbreviation: String): Option[(String, Double)] =
|
||||
rows.flatMap { row =>
|
||||
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 {
|
||||
sid <- cursor.downField("STATION_ID").as[String].toOption if sid == stationId
|
||||
ab <- cursor.downField("ABBREVIATION").as[String].toOption if ab == abbreviation
|
||||
stationId <- cursor.downField("STATION_ID").as[String].toOption
|
||||
datetime <- cursor.downField("DATETIME").as[String].toOption
|
||||
value <- cursor.downField("VALUE").as[Double].toOption
|
||||
} yield datetime -> value
|
||||
}.sortBy(_._1)(Ordering[String].reverse).headOption
|
||||
} 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(
|
||||
|
||||
Reference in New Issue
Block a user