bot/WeatherTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add LVGMC warning map workflow

Browse Source
This commit is contained in:
b0txec
2026-08-22 20:11:20 +03:00
parent 218ca2fa85
commit bc5bee171c
14 changed files with 538 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/Main.scala
+3 -1
View File
@@ -6,6 +6,7 @@ import db.{DBConnection, PostgresService}
import fetch.csv.{FileNameService}
import fetch.dmi
import fetch.lvgmc
import fetch.warnings.WarningService
import scheduler.Scheduler
import server.Server
@@ -24,6 +25,7 @@ object Main extends IOApp {
scheduler <- Scheduler.of
dataService <- DataService.of
fetchLvgmcService <- lvgmc.FetchService.of
warningService <- WarningService.of
fetchWeatherStations = for {
@@ -40,7 +42,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)
server <- Server.of(postgresService, dataService, fetchLvgmcService, warningService)
serverTask = server.run
scheduledTasks =
src/main/scala/fetch/warnings/WarningService.scala
+194
View File
@@ -0,0 +1,194 @@
package fetch.warnings
import cats.effect.{IO, Ref}
import cats.syntax.all._
import io.circe.{Json, parser}
import org.http4s.Uri
import org.http4s.client.Client
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 WarningPoint(lat: Double, lon: Double)
final case class WarningPolygon(id: String, points: Vector[WarningPoint])
final case class WeatherWarning(
id: String,
number: String,
intensity: String,
phenomenon: String,
regions: String,
validFrom: String,
validUntil: String,
description: String,
risks: String,
polygons: Vector[WarningPolygon]
)
final case class WarningResponse(
source: String,
fetchedAt: String,
warnings: Vector[WeatherWarning]
)
final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(Instant, WarningResponse)]]) {
private val cacheLifetime = Duration.ofMinutes(5)
private val polygonPageSize = 32000
private val metadataUrl = sys.env.getOrElse(
"LVGMC_WARNINGS_METADATA_URL",
"https://data.gov.lv/dati/api/3/action/datastore_search?resource_id=59c111fb-8c9a-4a63-8284-0a64a2920681&limit=1000"
)
private val polygonsUrl = sys.env.getOrElse(
"LVGMC_WARNINGS_POLYGONS_URL",
"https://data.gov.lv/dati/api/3/action/datastore_search?resource_id=01dc7d3c-34e5-4cc3-8f1a-aaf022872a02"
)
def fetchWarnings: IO[WarningResponse] =
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 {
case Some((_, value)) => logger.warn(error)("Using stale LVĢMC warning data") *> IO.pure(value)
case None => IO.raiseError(error)
}
}
}
} yield response
private def fetchFresh: IO[WarningResponse] =
EmberClientBuilder.default[IO].build.use { client =>
for {
metadataUri <- Uri.fromString(metadataUrl).liftTo[IO]
polygonsBaseUri <- Uri.fromString(polygonsUrl).liftTo[IO]
metadataBody <- client.expect[String](metadataUri)
metadata <- parseRecords(metadataBody, "warning metadata")
eventIds = metadata.flatMap(record => stringValue(record, "WEATHER_WARNING_EV_ID")).distinct
polygonRows <- eventIds.traverse(eventId => fetchPolygonRows(client, polygonsBaseUri, eventId)).map(_.flatten)
polygons = buildPolygons(polygonRows)
warnings = metadata.flatMap(record => buildWarning(record, polygons))
.filter(_.polygons.nonEmpty)
_ <- logger.info(s"Loaded ${warnings.size} LVĢMC warnings")
} yield WarningResponse("LVĢMC", Instant.now.toString, warnings)
}
private def parseRecords(body: String, label: String): IO[Vector[Json]] =
parsePage(body, label).map(_._2)
private def parsePage(body: String, label: String): IO[(Int, Vector[Json])] =
IO.fromEither(
parser
.parse(body)
.leftMap(error => new RuntimeException(s"Unable to parse $label: ${error.message}"))
.flatMap { json =>
val result = json.hcursor.downField("result")
(for {
total <- result.downField("total").as[Int]
records <- result.downField("records").as[Vector[Json]]
} yield total -> records)
.leftMap(error => new RuntimeException(s"Invalid $label response: ${error.message}"))
}
)
private def fetchPolygonRows(
client: Client[IO],
baseUri: Uri,
eventId: String,
offset: Int = 0,
accumulated: Vector[Json] = Vector.empty
): IO[Vector[Json]] = {
val filterValue = eventId.toLongOption.map(Json.fromLong).getOrElse(Json.fromString(eventId))
val filters = Json.obj("WEATHER_WARNING_EV_ID" -> filterValue).noSpaces
val uri = baseUri
.withQueryParam("limit", polygonPageSize.toString)
.withQueryParam("offset", offset.toString)
.withQueryParam("sort", "_id asc")
.withQueryParam("filters", filters)
client.expect[String](uri).flatMap(parsePage(_, s"warning polygons for $eventId")).flatMap {
case (total, rows) =>
val allRows = accumulated ++ rows
if (allRows.size >= total || rows.isEmpty) IO.pure(allRows)
else fetchPolygonRows(client, baseUri, eventId, offset + rows.size, allRows)
}
}
private def buildPolygons(
rows: Vector[Json]
): Map[String, Vector[WarningPolygon]] =
rows
.flatMap { row =>
for {
eventId <- stringValue(row, "WEATHER_WARNING_EV_ID")
polygonId <- stringValue(row, "POLIGON_ID")
lat <- doubleValue(row, "LAT")
lon <- doubleValue(row, "LON")
} yield ((eventId, polygonId), intValue(row, "NPK").getOrElse(0), WarningPoint(lat, lon))
}
.groupBy(_._1)
.toVector
.groupBy(_._1._1)
.view
.mapValues { polygonGroups =>
polygonGroups
.map { case ((_, polygonId), points) =>
WarningPolygon(polygonId, points.sortBy(_._2).map(_._3))
}
.sortBy(_.id)
.toVector
}
.toMap
private def buildWarning(
record: Json,
polygons: Map[String, Vector[WarningPolygon]]
): Option[WeatherWarning] =
stringValue(record, "WEATHER_WARNING_EV_ID").map { eventId =>
WeatherWarning(
id = eventId,
number = stringValue(record, "WARNING_NO").getOrElse(""),
intensity = stringValue(record, "INTENSITY_LV")
.orElse(stringValue(record, "INTENSITY_EN"))
.getOrElse(""),
phenomenon = stringValue(record, "PARADIBA").getOrElse(""),
regions = stringValue(record, "REGIONS").getOrElse(""),
validFrom = stringValue(record, "TIME_FROM").getOrElse(""),
validUntil = stringValue(record, "TIME_TILL").getOrElse(""),
description = stringValue(record, "TEKSTS_LV").getOrElse(""),
risks = stringValue(record, "RISKS_LV").getOrElse(""),
polygons = polygons.getOrElse(eventId, Vector.empty)
)
}
private def field(json: Json, name: String): Option[Json] =
json.asObject.flatMap(_(name)).filterNot(_.isNull)
private def stringValue(json: Json, name: String): Option[String] =
field(json, name).flatMap(value =>
value.asString.orElse(value.asNumber.map(_.toString)).map(_.trim).filter(_.nonEmpty)
)
private def doubleValue(json: Json, name: String): Option[Double] =
field(json, name).flatMap(value =>
value.asNumber.map(_.toDouble).orElse(value.asString.flatMap(_.trim.toDoubleOption))
)
private def intValue(json: Json, name: String): Option[Int] =
field(json, name).flatMap(value =>
value.asNumber.flatMap(_.toInt).orElse(value.asString.flatMap(_.trim.toIntOption))
)
}
object WarningService {
def of: IO[WarningService] =
for {
logger <- Slf4jLogger.create[IO]
cache <- Ref.of[IO, Option[(Instant, WarningResponse)]](None)
} yield new WarningService(logger, cache)
}
src/main/scala/server/Server.scala
+13 -3
View File
@@ -8,6 +8,7 @@ import db.PostgresService
import fetch.csv.FileNameService
//import fetch.csv.FetchService
import fetch.lvgmc.FetchService
import fetch.warnings.WarningService
import fs2.io.file.{Files, Path}
import server.ValidateRoutes.{AggFieldList, AggKey, CityList, DateTimeRange, Granularity, ValidateDate, ValidateDateTime, ValidateInt, ValidateMonths, ValidateZonedDateTime}
import org.http4s._
@@ -34,18 +35,26 @@ import scala.concurrent.duration.DurationInt
object Server {
def of(postgresService: PostgresService, dataService: DataService, fetch: FetchService): IO[Server] = {
def of(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService): IO[Server] = {
Slf4jLogger.create[IO].map {
new Server(postgresService, dataService, fetch, _)
new Server(postgresService, dataService, fetch, warnings, _)
}
}
}
class Server(postgresService: PostgresService, dataService: DataService, fetch: FetchService, log: Logger[IO]) {
class Server(postgresService: PostgresService, dataService: DataService, fetch: FetchService, warnings: WarningService, 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])
private val apiRoutes = HttpRoutes.of[IO] {
case GET -> Root / "warnings" =>
warnings.fetchWarnings
.flatMap(result => Ok(result.asJson))
.handleErrorWith { error =>
log.error(error)("Unable to load LVĢMC warnings") *>
ServiceUnavailable("LVĢMC brīdinājumu 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 =>
@@ -198,6 +207,7 @@ class Server(postgresService: PostgresService, dataService: DataService, fetch:
"/database" -> staticcontent.fileService[IO](FileService.Config("./web/dist")),
"/harmonie" -> staticcontent.fileService[IO](FileService.Config("./web/dist")),
"/lvgmc-forecast" -> staticcontent.fileService[IO](FileService.Config("./web/dist")),
"/bridinajumi" -> staticcontent.fileService[IO](FileService.Config("./web/dist")),
).orNotFound
def run: IO[ExitCode] =