diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..35f0d6d --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +*.swp +*.swo +*.swj +*.swk +*.swl +*.swm +*.swn +data/*.sqlite +data/*.sqlite-shm +data/*.sqlite-wal diff --git a/DESCRIPTION b/DESCRIPTION index 62bdfd6..c7f5c27 100644 --- a/DESCRIPTION +++ b/DESCRIPTION @@ -1,15 +1,22 @@ Package: beaumontmeteo -Title: What the Package Does (One Line, Title Case) +Title: Cache Local Weather Observations for Beaumont Version: 0.0.0.9000 Authors@R: person("First", "Last", , "first.last@example.com", role = c("aut", "cre"), comment = c(ORCID = "YOUR-ORCID-ID")) -Description: What the package does (one paragraph). +Description: Shiny app and helper functions to cache Meteo France rainfall + and station observation metrics locally and browse them without + re-querying the API every time. License: `use_mit_license()`, `use_gpl3_license()` or friends to pick a license Encoding: UTF-8 +Imports: + httr, + shiny, + shinycssloaders Roxygen: list(markdown = TRUE) RoxygenNote: 7.3.2 Suggests: testthat (>= 3.0.0) +SystemRequirements: sqlite3 Config/testthat/edition: 3 diff --git a/R/getStationData.R b/R/getStationData.R index 7f256a5..cd89a27 100644 --- a/R/getStationData.R +++ b/R/getStationData.R @@ -1,12 +1,23 @@ -getStationData <- function(start_date, end_date, station_id,headers ,dpclim = "public/DPClim/v1/",timesleep=5,base="https://public-api.meteofrance.fr") { +get_dpclim_period_bounds <- function(start_date, end_date) { start_date <- as.Date(start_date) end_date <- as.Date(end_date) - url <- httr::modify_url(base,path = paste0(dpclim,"commande-station/infrahoraire-6m")) - # Format dates in ISO8601 format - formatted_start_date <- format(start_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") - formatted_end_date <- format(end_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") - parameters <- list("id-station" = station_id, "date-deb-periode" = formatted_start_date, "date-fin-periode" = formatted_end_date) + list( + start = format(start_date, "%Y-%m-%dT00:00:00Z", tz = "GMT"), + end = format(end_date + 1L, "%Y-%m-%dT00:00:00Z", tz = "GMT") + ) +} + + +getStationData <- function(start_date, end_date, station_id,headers ,dpclim = "public/DPClim/v1/",timesleep=5,base="https://public-api.meteofrance.fr") { + url <- httr::modify_url(base,path = paste0(dpclim,"commande-station/infrahoraire-6m")) + period_bounds <- get_dpclim_period_bounds(start_date, end_date) + + parameters <- list( + "id-station" = station_id, + "date-deb-periode" = period_bounds$start, + "date-fin-periode" = period_bounds$end + ) # Create the URL with parameters to ask for the csv file url_query_file <- httr::modify_url(url, query = parameters) @@ -45,14 +56,14 @@ getStationData <- function(start_date, end_date, station_id,headers ,dpclim = "p getStationDataDPObs <- function(start_date, end_date, station_id,headers ,dpclim = "public/DPClim/v1/",timesleep=5,base="https://public-api.meteofrance.fr") { - start_date <- as.Date(start_date) - end_date <- as.Date(end_date) url <- httr::modify_url(base,path = paste0(dpclim,"commande-station/infrahoraire-6m")) - # Format dates in ISO8601 format - formatted_start_date <- format(start_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") - formatted_end_date <- format(end_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") + period_bounds <- get_dpclim_period_bounds(start_date, end_date) - parameters <- list("id-station" = station_id, "date-deb-periode" = formatted_start_date, "date-fin-periode" = formatted_end_date) + parameters <- list( + "id-station" = station_id, + "date-deb-periode" = period_bounds$start, + "date-fin-periode" = period_bounds$end + ) # Create the URL with parameters to ask for the csv file url_query_file <- httr::modify_url(url, query = parameters) @@ -90,14 +101,14 @@ getStationDataDPObs <- function(start_date, end_date, station_id,headers ,dpclim } getStationDataTemp <- function(start_date, end_date, station_id,headers ,dpclim = "public/DPClim/v1/",timesleep=5,base="https://public-api.meteofrance.fr") { - start_date <- as.Date(start_date) - end_date <- as.Date(end_date) url <- httr::modify_url(base,path = paste0(dpclim,"commande-station/infrahoraire-6m")) - # Format dates in ISO8601 format - formatted_start_date <- format(start_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") - formatted_end_date <- format(end_date, "%Y-%m-%dT00:00:00Z", tz = "GMT") + period_bounds <- get_dpclim_period_bounds(start_date, end_date) - parameters <- list("id-station" = station_id, "date-deb-periode" = formatted_start_date, "date-fin-periode" = formatted_end_date) + parameters <- list( + "id-station" = station_id, + "date-deb-periode" = period_bounds$start, + "date-fin-periode" = period_bounds$end + ) # Create the URL with parameters to ask for the csv file url_query_file <- httr::modify_url(url, query = parameters) diff --git a/R/rain_db.R b/R/rain_db.R new file mode 100644 index 0000000..ee5efd3 --- /dev/null +++ b/R/rain_db.R @@ -0,0 +1,1522 @@ +default_rain_db_path <- function() { + path <- Sys.getenv( + "BEAUMONT_RAIN_DB", + unset = file.path("data", "rain_history.sqlite") + ) + + normalizePath(path, winslash = "/", mustWork = FALSE) +} + + +get_rain_locations <- function() { + data.frame( + location_id = c("vignasses", "la_mure", "la_couchonnier"), + label = c("Les Vignasses", "La Mure", "La Couchonnier"), + latitude = c(44.8550665, 44.9167, 45.3722971), + longitude = c(5.8441789, 5.8000, 5.6387118), + station_count = c(3L, 3L, 3L), + stringsAsFactors = FALSE + ) +} + + +get_rain_location <- function(location_id, locations = get_rain_locations()) { + location <- locations[locations$location_id == location_id, , drop = FALSE] + + if (!nrow(location)) { + stop(sprintf("Unknown location_id '%s'.", location_id)) + } + + location[1, , drop = FALSE] +} + + +load_station_catalog <- function(path = file.path("data", "allstations.csv")) { + if (!file.exists(path)) { + stop(sprintf("Station catalog not found at '%s'.", path)) + } + + read.csv(path, stringsAsFactors = FALSE) +} + + +source_secret_env <- function(secret_path = file.path("data", "secrets")) { + if (!file.exists(secret_path)) { + return(NULL) + } + + secret_env <- new.env(parent = baseenv()) + sys.source(secret_path, envir = secret_env) + secret_env +} + + +load_api_headers <- function( + secret_path = file.path("data", "secrets"), + token_name = "token4" +) { + secret_env <- source_secret_env(secret_path = secret_path) + if (is.null(secret_env)) { + return(NULL) + } + + api_key <- secret_env[[token_name]] + if (is.null(api_key) || !nzchar(api_key)) { + return(NULL) + } + + httr::add_headers( + accept = "*/*", + apikey = api_key + ) +} + + +ensure_sqlite_cli <- function() { + sqlite_bin <- Sys.which("sqlite3") + + if (!nzchar(sqlite_bin)) { + stop("sqlite3 is required but was not found on PATH.") + } + + sqlite_bin +} + + +run_sqlite <- function(args, stdin = "", timeout = 0) { + sqlite_bin <- ensure_sqlite_cli() + + output <- system2( + sqlite_bin, + args = args, + stdout = TRUE, + stderr = TRUE, + stdin = stdin, + timeout = timeout + ) + + status <- attr(output, "status") + if (!is.null(status) && status != 0) { + stop(paste(output, collapse = "\n")) + } + + output +} + + +sql_string <- function(value) { + if (is.null(value) || is.na(value)) { + return("NULL") + } + + paste0("'", gsub("'", "''", as.character(value), fixed = TRUE), "'") +} + + +parse_station_timestamp <- function(x) { + if (inherits(x, "POSIXt")) { + return(as.POSIXct(x, tz = "UTC")) + } + + if (is.numeric(x)) { + x <- format(x, scientific = FALSE, trim = TRUE) + } + + x <- trimws(as.character(x)) + x <- sub("\\.0+$", "", x) + + if (!length(x)) { + return(as.POSIXct(character(), tz = "UTC")) + } + + if (grepl("T", x[1], fixed = TRUE)) { + return(as.POSIXct(x, format = "%Y-%m-%dT%H:%M:%SZ", tz = "UTC")) + } + + as.POSIXct(strptime(x, format = "%Y%m%d%H%M", tz = "UTC")) +} + + +format_utc_timestamp <- function(x) { + format(as.POSIXct(x, tz = "UTC"), "%Y-%m-%dT%H:%M:%SZ", tz = "UTC") +} + + +round_metric_values <- function(values, digits = 2L) { + values <- suppressWarnings(as.numeric(values)) + round(values, digits = as.integer(digits)) +} + + +celsius_from_kelvin <- function(values) { + values <- suppressWarnings(as.numeric(values)) + values - 273.15 +} + + +hpa_from_pa <- function(values) { + values <- suppressWarnings(as.numeric(values)) + values / 100 +} + + +identity_numeric <- function(values) { + suppressWarnings(as.numeric(values)) +} + + +get_weather_metric_definitions <- function() { + list( + rain_6m = list( + label = "Rain", + unit = "mm / 6 min", + daily_agg = "sum", + digits = 2L, + source_name = "DPClim" + ), + air_temperature = list( + label = "Temperature", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "t", + transform = celsius_from_kelvin + ), + dew_point = list( + label = "Dew point", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "td", + transform = celsius_from_kelvin + ), + humidity = list( + label = "Humidity", + unit = "%", + daily_agg = "mean", + digits = 1L, + source_name = "DPPaquetObs", + obs_column = "u", + transform = identity_numeric + ), + wind_direction = list( + label = "Wind direction", + unit = "deg", + daily_agg = "mean", + digits = 0L, + source_name = "DPPaquetObs", + obs_column = "dd", + transform = identity_numeric + ), + wind_speed = list( + label = "Wind speed", + unit = "m/s", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "ff", + transform = identity_numeric + ), + gust_direction = list( + label = "Gust direction", + unit = "deg", + daily_agg = "mean", + digits = 0L, + source_name = "DPPaquetObs", + obs_column = "dxi10", + transform = identity_numeric + ), + wind_gust = list( + label = "Wind gust", + unit = "m/s", + daily_agg = "max", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "fxi10", + transform = identity_numeric + ), + soil_temperature_10cm = list( + label = "Soil temp 10 cm", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "t_10", + transform = celsius_from_kelvin + ), + soil_temperature_20cm = list( + label = "Soil temp 20 cm", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "t_20", + transform = celsius_from_kelvin + ), + soil_temperature_50cm = list( + label = "Soil temp 50 cm", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "t_50", + transform = celsius_from_kelvin + ), + soil_temperature_100cm = list( + label = "Soil temp 100 cm", + unit = "C", + daily_agg = "mean", + digits = 2L, + source_name = "DPPaquetObs", + obs_column = "t_100", + transform = celsius_from_kelvin + ), + visibility = list( + label = "Visibility", + unit = "m", + daily_agg = "mean", + digits = 0L, + source_name = "DPPaquetObs", + obs_column = "vv", + transform = identity_numeric + ), + station_pressure = list( + label = "Station pressure", + unit = "hPa", + daily_agg = "mean", + digits = 1L, + source_name = "DPPaquetObs", + obs_column = "pres", + transform = hpa_from_pa + ), + sea_level_pressure = list( + label = "Sea-level pressure", + unit = "hPa", + daily_agg = "mean", + digits = 1L, + source_name = "DPPaquetObs", + obs_column = "pmer", + transform = hpa_from_pa + ) + ) +} + + +get_weather_metric <- function(metric_id) { + definitions <- get_weather_metric_definitions() + definition <- definitions[[metric_id]] + + if (is.null(definition)) { + stop(sprintf("Unknown metric_id '%s'.", metric_id)) + } + + definition$metric_id <- metric_id + definition +} + + +get_weather_metric_catalog <- function(include_rain = TRUE) { + definitions <- get_weather_metric_definitions() + metric_ids <- names(definitions) + if (!include_rain) { + metric_ids <- setdiff(metric_ids, "rain_6m") + } + + data.frame( + metric_id = metric_ids, + label = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$label, character(1)), + unit = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$unit, character(1)), + daily_agg = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$daily_agg, character(1)), + source_name = vapply(metric_ids, function(metric_id) definitions[[metric_id]]$source_name, character(1)), + obs_column = vapply( + metric_ids, + function(metric_id) { + column_name <- definitions[[metric_id]]$obs_column + if (is.null(column_name)) { + "" + } else { + column_name + } + }, + character(1) + ), + stringsAsFactors = FALSE + ) +} + + +empty_weather_cache <- function() { + data.frame( + location_id = character(), + station_id = character(), + station_name = character(), + observed_at = character(), + observed_day = character(), + metric_id = character(), + metric_label = character(), + unit = character(), + source_name = character(), + value_num = numeric(), + fetched_at = character(), + stringsAsFactors = FALSE + ) +} + + +empty_metric_query <- function(aggregate = c("raw", "daily")) { + aggregate <- match.arg(aggregate) + + if (aggregate == "daily") { + return(data.frame( + station_id = character(), + station_name = character(), + observed_day = character(), + metric_id = character(), + metric_label = character(), + unit = character(), + value_num = numeric(), + stringsAsFactors = FALSE + )) + } + + data.frame( + location_id = character(), + station_id = character(), + station_name = character(), + observed_at = character(), + observed_day = character(), + metric_id = character(), + metric_label = character(), + unit = character(), + value_num = numeric(), + stringsAsFactors = FALSE + ) +} + + +normalise_rainfall_data <- function( + raw_data, + location_id, + fetched_at = Sys.time() +) { + if (is.null(raw_data) || !nrow(raw_data)) { + return(empty_weather_cache()) + } + + required_columns <- c("POSTE", "DATE", "RR6", "Nom_usuel") + missing_columns <- setdiff(required_columns, names(raw_data)) + + if (length(missing_columns)) { + stop( + sprintf( + "Rainfall payload is missing required columns: %s", + paste(missing_columns, collapse = ", ") + ) + ) + } + + metric <- get_weather_metric("rain_6m") + observed_at <- parse_station_timestamp(raw_data$DATE) + station_id <- as.character(raw_data$POSTE) + station_name <- trimws(as.character(raw_data$Nom_usuel)) + station_name[!nzchar(station_name)] <- station_id[!nzchar(station_name)] + fetched_at <- format_utc_timestamp(fetched_at) + + data.frame( + location_id = location_id, + station_id = station_id, + station_name = station_name, + observed_at = format_utc_timestamp(observed_at), + observed_day = format(observed_at, "%Y-%m-%d"), + metric_id = metric$metric_id, + metric_label = metric$label, + unit = metric$unit, + source_name = metric$source_name, + value_num = round_metric_values(raw_data$RR6, metric$digits), + fetched_at = fetched_at, + stringsAsFactors = FALSE + ) +} + + +normalise_observation_package_data <- function( + raw_data, + location_id, + station_name = NULL, + fetched_at = Sys.time() +) { + if (is.null(raw_data) || !nrow(raw_data)) { + return(empty_weather_cache()) + } + + required_columns <- c("geo_id_insee", "validity_time") + missing_columns <- setdiff(required_columns, names(raw_data)) + if (length(missing_columns)) { + stop( + sprintf( + "Observation payload is missing required columns: %s", + paste(missing_columns, collapse = ", ") + ) + ) + } + + observed_at <- parse_station_timestamp(raw_data$validity_time) + observed_day <- format(observed_at, "%Y-%m-%d") + station_id <- as.character(raw_data$geo_id_insee) + if (is.null(station_name)) { + station_name <- station_id + } + station_name <- rep(as.character(station_name)[1], length.out = nrow(raw_data)) + fetched_at <- format_utc_timestamp(fetched_at) + + definitions <- get_weather_metric_definitions() + metric_ids <- setdiff(names(definitions), "rain_6m") + + measurement_rows <- lapply(metric_ids, function(metric_id) { + definition <- definitions[[metric_id]] + column_name <- definition$obs_column + + if (is.null(column_name) || !column_name %in% names(raw_data)) { + return(NULL) + } + + values <- definition$transform(raw_data[[column_name]]) + values <- round_metric_values(values, definition$digits) + keep <- !is.na(values) + if (!any(keep)) { + return(NULL) + } + + data.frame( + location_id = location_id, + station_id = station_id[keep], + station_name = station_name[keep], + observed_at = format_utc_timestamp(observed_at[keep]), + observed_day = observed_day[keep], + metric_id = metric_id, + metric_label = definition$label, + unit = definition$unit, + source_name = definition$source_name, + value_num = values[keep], + fetched_at = fetched_at, + stringsAsFactors = FALSE + ) + }) + + measurement_rows <- Filter(Negate(is.null), measurement_rows) + if (!length(measurement_rows)) { + return(empty_weather_cache()) + } + + do.call(rbind, measurement_rows) +} + + +sqlite_scalar_query <- function(sql, db_path) { + script_file <- tempfile(fileext = ".sql") + on.exit(unlink(script_file), add = TRUE) + + writeLines(c(".bail on", sql), script_file) + output <- run_sqlite(args = c(db_path), stdin = script_file) + + if (!length(output)) { + return("") + } + + utils::tail(output, 1) +} + + +sqlite_table_exists <- function(db_path, table_name) { + query <- sprintf( + paste( + "SELECT COUNT(*)", + "FROM sqlite_master", + "WHERE type = 'table'", + " AND name = %s;" + ), + sql_string(table_name) + ) + + suppressWarnings(as.integer(sqlite_scalar_query(query, db_path))) > 0 +} + + +migrate_legacy_rainfall_table <- function(db_path) { + if (!sqlite_table_exists(db_path, "rainfall_observations")) { + return(invisible(FALSE)) + } + + script_file <- tempfile(fileext = ".sql") + on.exit(unlink(script_file), add = TRUE) + + writeLines( + c( + ".bail on", + "BEGIN;", + paste( + "INSERT OR IGNORE INTO weather_measurements", + "(location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, source_name, value_num, fetched_at)", + "SELECT location_id, station_id, station_name, observed_at, observed_day,", + "'rain_6m', 'Rain', 'mm / 6 min', 'DPClim', rain_mm, fetched_at", + "FROM rainfall_observations;" + ), + "COMMIT;" + ), + script_file + ) + + invisible(run_sqlite(args = c(db_path), stdin = script_file)) + invisible(TRUE) +} + + +ensure_weather_db <- function(db_path = default_rain_db_path()) { + dir.create(dirname(db_path), recursive = TRUE, showWarnings = FALSE) + + script_file <- tempfile(fileext = ".sql") + on.exit(unlink(script_file), add = TRUE) + + writeLines( + c( + ".bail on", + "PRAGMA journal_mode = WAL;", + "CREATE TABLE IF NOT EXISTS weather_measurements (", + " location_id TEXT NOT NULL,", + " station_id TEXT NOT NULL,", + " station_name TEXT NOT NULL,", + " observed_at TEXT NOT NULL,", + " observed_day TEXT NOT NULL,", + " metric_id TEXT NOT NULL,", + " metric_label TEXT NOT NULL,", + " unit TEXT NOT NULL,", + " source_name TEXT NOT NULL,", + " value_num REAL,", + " fetched_at TEXT NOT NULL,", + " PRIMARY KEY (location_id, station_id, observed_at, metric_id)", + ");", + "CREATE INDEX IF NOT EXISTS idx_weather_location_time", + " ON weather_measurements(location_id, observed_at);", + "CREATE INDEX IF NOT EXISTS idx_weather_location_metric_time", + " ON weather_measurements(location_id, metric_id, observed_at);", + "CREATE INDEX IF NOT EXISTS idx_weather_location_metric_day", + " ON weather_measurements(location_id, metric_id, observed_day);" + ), + script_file + ) + + invisible(run_sqlite(args = c(db_path), stdin = script_file)) + migrate_legacy_rainfall_table(db_path) + invisible(db_path) +} + + +ensure_rain_db <- function(db_path = default_rain_db_path()) { + ensure_weather_db(db_path = db_path) +} + + +upsert_weather_measurements <- function( + weather_data, + db_path = default_rain_db_path() +) { + ensure_weather_db(db_path) + + if (is.null(weather_data) || !nrow(weather_data)) { + return(0L) + } + + columns <- c( + "location_id", + "station_id", + "station_name", + "observed_at", + "observed_day", + "metric_id", + "metric_label", + "unit", + "source_name", + "value_num", + "fetched_at" + ) + + weather_data <- weather_data[, columns, drop = FALSE] + + csv_file <- tempfile(fileext = ".csv") + script_file <- tempfile(fileext = ".sql") + on.exit(unlink(c(csv_file, script_file)), add = TRUE) + + utils::write.csv( + weather_data, + file = csv_file, + row.names = FALSE, + quote = TRUE, + na = "" + ) + + writeLines( + c( + ".bail on", + "BEGIN;", + "CREATE TEMP TABLE weather_import (", + " location_id TEXT,", + " station_id TEXT,", + " station_name TEXT,", + " observed_at TEXT,", + " observed_day TEXT,", + " metric_id TEXT,", + " metric_label TEXT,", + " unit TEXT,", + " source_name TEXT,", + " value_num TEXT,", + " fetched_at TEXT", + ");", + ".mode csv", + sprintf(".import --skip 1 %s weather_import", shQuote(csv_file)), + paste( + "INSERT OR REPLACE INTO weather_measurements", + "(location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, source_name, value_num, fetched_at)", + "SELECT location_id, station_id, station_name, observed_at, observed_day,", + "metric_id, metric_label, unit, source_name, CAST(NULLIF(value_num, '') AS REAL), fetched_at", + "FROM weather_import;" + ), + "SELECT changes();", + "DROP TABLE weather_import;", + "COMMIT;" + ), + script_file + ) + + output <- run_sqlite(args = c(db_path), stdin = script_file) + rows_changed <- suppressWarnings(as.integer(utils::tail(output, 1))) + + if (is.na(rows_changed)) { + return(nrow(weather_data)) + } + + rows_changed +} + + +upsert_rainfall_observations <- function( + rain_data, + db_path = default_rain_db_path() +) { + upsert_weather_measurements( + weather_data = rain_data, + db_path = db_path + ) +} + + +read_sqlite_query <- function(sql, db_path = default_rain_db_path()) { + ensure_weather_db(db_path) + + script_file <- tempfile(fileext = ".sql") + on.exit(unlink(script_file), add = TRUE) + + writeLines( + c( + ".headers on", + ".mode csv", + sql + ), + script_file + ) + + output <- run_sqlite(args = c(db_path), stdin = script_file) + if (!length(output)) { + return(data.frame()) + } + + utils::read.csv( + text = paste(output, collapse = "\n"), + stringsAsFactors = FALSE + ) +} + + +get_location_cache_status <- function( + location_id, + db_path = default_rain_db_path() +) { + status <- read_sqlite_query( + sprintf( + paste( + "SELECT COUNT(*) AS row_count,", + "COUNT(DISTINCT station_id) AS station_count,", + "COUNT(DISTINCT metric_id) AS metric_count,", + "MIN(observed_at) AS earliest_observed_at,", + "MAX(observed_at) AS latest_observed_at", + "FROM weather_measurements", + "WHERE location_id = %s;" + ), + sql_string(location_id) + ), + db_path = db_path + ) + + if (!nrow(status)) { + status <- data.frame( + row_count = 0, + station_count = 0, + metric_count = 0, + earliest_observed_at = "", + latest_observed_at = "", + stringsAsFactors = FALSE + ) + } + + status$row_count <- as.integer(status$row_count) + status$station_count <- as.integer(status$station_count) + status$metric_count <- as.integer(status$metric_count) + status +} + + +describe_cache_status <- function( + location_id, + db_path = default_rain_db_path(), + locations = get_rain_locations() +) { + location <- get_rain_location(location_id, locations = locations) + status <- get_location_cache_status(location_id, db_path = db_path) + + if (!status$row_count[1]) { + return( + paste( + sprintf("Location: %s", location$label[1]), + sprintf("Database: %s", db_path), + "No cached weather measurements yet.", + sep = "\n" + ) + ) + } + + paste( + sprintf("Location: %s", location$label[1]), + sprintf("Database: %s", db_path), + sprintf("Cached rows: %s", format(status$row_count[1], big.mark = ",")), + sprintf("Stations: %s", status$station_count[1]), + sprintf("Metrics: %s", status$metric_count[1]), + sprintf("Latest observation: %s", status$latest_observed_at[1]), + sep = "\n" + ) +} + + +get_daily_aggregate_expression <- function(metric_id) { + metric <- get_weather_metric(metric_id) + + switch( + metric$daily_agg, + sum = "ROUND(SUM(COALESCE(value_num, 0)), 2)", + max = "ROUND(MAX(value_num), 2)", + min = "ROUND(MIN(value_num), 2)", + "ROUND(AVG(value_num), 2)" + ) +} + + +query_cached_metric <- function( + location_id, + metric_id, + start_date, + end_date = Sys.Date(), + aggregate = c("raw", "daily"), + db_path = default_rain_db_path() +) { + aggregate <- match.arg(aggregate) + metric <- get_weather_metric(metric_id) + + start_date <- as.Date(start_date) + end_date <- as.Date(end_date) + if (start_date > end_date) { + return(empty_metric_query(aggregate = aggregate)) + } + + start_at <- format_utc_timestamp(as.POSIXct(start_date, tz = "UTC")) + end_at <- format_utc_timestamp(as.POSIXct(end_date + 1, tz = "UTC")) + + sql <- if (aggregate == "daily") { + sprintf( + paste( + "SELECT station_id, station_name, observed_day, metric_id, metric_label, unit,", + "%s AS value_num", + "FROM weather_measurements", + "WHERE location_id = %s", + " AND metric_id = %s", + " AND observed_at >= %s", + " AND observed_at < %s", + "GROUP BY station_id, station_name, observed_day, metric_id, metric_label, unit", + "ORDER BY observed_day, station_name;" + ), + get_daily_aggregate_expression(metric_id), + sql_string(location_id), + sql_string(metric$metric_id), + sql_string(start_at), + sql_string(end_at) + ) + } else { + sprintf( + paste( + "SELECT location_id, station_id, station_name, observed_at, observed_day, metric_id, metric_label, unit, value_num", + "FROM weather_measurements", + "WHERE location_id = %s", + " AND metric_id = %s", + " AND observed_at >= %s", + " AND observed_at < %s", + "ORDER BY observed_at, station_name;" + ), + sql_string(location_id), + sql_string(metric$metric_id), + sql_string(start_at), + sql_string(end_at) + ) + } + + result <- read_sqlite_query(sql, db_path = db_path) + if (!nrow(result)) { + return(empty_metric_query(aggregate = aggregate)) + } + + result +} + + +query_cached_rainfall <- function( + location_id, + start_date, + end_date = Sys.Date(), + aggregate = c("raw", "daily"), + db_path = default_rain_db_path() +) { + aggregate <- match.arg(aggregate) + result <- query_cached_metric( + location_id = location_id, + metric_id = "rain_6m", + start_date = start_date, + end_date = end_date, + aggregate = aggregate, + db_path = db_path + ) + + if ("value_num" %in% names(result)) { + names(result)[names(result) == "value_num"] <- "rain_mm" + } + + result +} + + +query_latest_metric_values <- function( + location_id, + metric_id, + db_path = default_rain_db_path() +) { + metric <- get_weather_metric(metric_id) + + sql <- sprintf( + paste( + "SELECT m.station_id, m.station_name, m.observed_at, m.metric_id, m.metric_label, m.unit, m.value_num", + "FROM weather_measurements AS m", + "INNER JOIN (", + " SELECT station_id, MAX(observed_at) AS observed_at", + " FROM weather_measurements", + " WHERE location_id = %s", + " AND metric_id = %s", + " GROUP BY station_id", + ") AS latest", + "ON latest.station_id = m.station_id", + "AND latest.observed_at = m.observed_at", + "WHERE m.location_id = %s", + " AND m.metric_id = %s", + "ORDER BY m.station_name;" + ), + sql_string(location_id), + sql_string(metric$metric_id), + sql_string(location_id), + sql_string(metric$metric_id) + ) + + result <- read_sqlite_query(sql, db_path = db_path) + if (!nrow(result)) { + return(data.frame( + station_id = character(), + station_name = character(), + observed_at = character(), + metric_id = character(), + metric_label = character(), + unit = character(), + value_num = numeric(), + stringsAsFactors = FALSE + )) + } + + result +} + + +split_sync_ranges <- function(start_date, end_date, chunk_days = 120L) { + start_date <- as.Date(start_date) + end_date <- as.Date(end_date) + chunk_days <- as.integer(chunk_days) + + if (start_date > end_date) { + return(data.frame( + start_date = as.Date(character()), + end_date = as.Date(character()) + )) + } + + ranges <- vector("list", length = 0L) + current_start <- start_date + + while (current_start <= end_date) { + current_end <- min(current_start + chunk_days - 1L, end_date) + ranges[[length(ranges) + 1L]] <- data.frame( + start_date = current_start, + end_date = current_end + ) + current_start <- current_end + 1L + } + + do.call(rbind, ranges) +} + + +get_sync_start_date <- function( + location_id, + db_path = default_rain_db_path(), + end_date = Sys.Date(), + initial_backfill_days = 21L, + overlap_days = 1L, + metric_id = "rain_6m" +) { + metric <- get_weather_metric(metric_id) + end_date <- as.Date(end_date) + + latest_data <- read_sqlite_query( + sprintf( + paste( + "SELECT MAX(observed_at) AS latest_observed_at", + "FROM weather_measurements", + "WHERE location_id = %s", + " AND metric_id = %s;" + ), + sql_string(location_id), + sql_string(metric$metric_id) + ), + db_path = db_path + ) + + latest_observed_at <- if (nrow(latest_data)) latest_data$latest_observed_at[1] else "" + if (is.na(latest_observed_at) || !nzchar(latest_observed_at)) { + return(end_date - as.integer(initial_backfill_days) + 1L) + } + + as.Date(latest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ") - as.integer(overlap_days) +} + + +get_nearby_stations_for_location <- function( + location_id, + allstations = load_station_catalog(), + locations = get_rain_locations() +) { + location <- get_rain_location(location_id, locations = locations) + + getIdFromCoords( + coords = c(location$latitude[1], location$longitude[1]), + stations = allstations, + N = location$station_count[1] + ) +} + + +fetch_rainfall_from_api <- function( + location_id, + start_date, + end_date, + headers, + allstations = load_station_catalog(), + locations = get_rain_locations(), + btw_station_sleep = 0.25, + within_station_sleep = 1 +) { + if (is.null(headers)) { + stop("API credentials are not available.") + } + + location <- get_rain_location(location_id, locations = locations) + + getAllFromCoord( + coord = c(location$latitude[1], location$longitude[1]), + start_date = format(as.Date(start_date), "%Y-%m-%d"), + end_date = format(as.Date(end_date), "%Y-%m-%d"), + allstations = allstations, + N = location$station_count[1], + headers = headers, + btw_station_sleep = btw_station_sleep, + within_statio_sleep = within_station_sleep + ) +} + + +fetch_observations_from_api <- function( + location_id, + headers, + allstations = load_station_catalog(), + locations = get_rain_locations(), + btw_station_sleep = 0.25 +) { + if (is.null(headers)) { + stop("API credentials are not available.") + } + + nearby_stations <- get_nearby_stations_for_location( + location_id = location_id, + allstations = allstations, + locations = locations + ) + + measurement_rows <- lapply(seq_len(nrow(nearby_stations)), function(row_index) { + station_id <- as.character(nearby_stations$Id_station[row_index]) + station_name <- as.character(nearby_stations$Nom_usuel[row_index]) + + raw_data <- tryCatch( + getStationPaquet(id_station = station_id, headers = headers), + error = function(error) { + message(conditionMessage(error)) + NULL + } + ) + + Sys.sleep(btw_station_sleep) + + normalise_observation_package_data( + raw_data = raw_data, + location_id = location_id, + station_name = station_name + ) + }) + + measurement_rows <- Filter(function(x) !is.null(x) && nrow(x), measurement_rows) + if (!length(measurement_rows)) { + return(empty_weather_cache()) + } + + do.call(rbind, measurement_rows) +} + + +sync_location_rainfall <- function( + location_id, + db_path = default_rain_db_path(), + headers = load_api_headers(), + allstations = load_station_catalog(), + locations = get_rain_locations(), + start_date = NULL, + end_date = Sys.Date(), + initial_backfill_days = 21L, + overlap_days = 1L, + chunk_days = 120L, + btw_station_sleep = 0.25, + within_station_sleep = 1 +) { + ensure_weather_db(db_path) + + if (is.null(headers)) { + stop("API credentials are not available. Add token4 to data/secrets.") + } + + end_date <- as.Date(end_date) + if (is.null(start_date)) { + start_date <- get_sync_start_date( + location_id = location_id, + db_path = db_path, + end_date = end_date, + initial_backfill_days = initial_backfill_days, + overlap_days = overlap_days, + metric_id = "rain_6m" + ) + } + + start_date <- as.Date(start_date) + if (start_date > end_date) { + return(data.frame( + location_id = location_id, + start_date = as.character(start_date), + end_date = as.character(end_date), + rows_fetched = 0L, + rows_written = 0L, + status = "up_to_date", + message = "", + stringsAsFactors = FALSE + )) + } + + ranges <- split_sync_ranges( + start_date = start_date, + end_date = end_date, + chunk_days = chunk_days + ) + + rows_fetched <- 0L + rows_written <- 0L + + for (range_index in seq_len(nrow(ranges))) { + raw_data <- fetch_rainfall_from_api( + location_id = location_id, + start_date = ranges$start_date[range_index], + end_date = ranges$end_date[range_index], + headers = headers, + allstations = allstations, + locations = locations, + btw_station_sleep = btw_station_sleep, + within_station_sleep = within_station_sleep + ) + + weather_data <- normalise_rainfall_data( + raw_data = raw_data, + location_id = location_id + ) + + rows_fetched <- rows_fetched + nrow(weather_data) + rows_written <- rows_written + upsert_weather_measurements( + weather_data = weather_data, + db_path = db_path + ) + } + + data.frame( + location_id = location_id, + start_date = as.character(start_date), + end_date = as.character(end_date), + rows_fetched = rows_fetched, + rows_written = rows_written, + status = "ok", + message = "", + stringsAsFactors = FALSE + ) +} + + +sync_location_observations <- function( + location_id, + db_path = default_rain_db_path(), + headers = load_api_headers(), + allstations = load_station_catalog(), + locations = get_rain_locations(), + btw_station_sleep = 0.25 +) { + ensure_weather_db(db_path) + + if (is.null(headers)) { + stop("API credentials are not available. Add token4 to data/secrets.") + } + + weather_data <- fetch_observations_from_api( + location_id = location_id, + headers = headers, + allstations = allstations, + locations = locations, + btw_station_sleep = btw_station_sleep + ) + + rows_written <- upsert_weather_measurements( + weather_data = weather_data, + db_path = db_path + ) + + data.frame( + location_id = location_id, + rows_fetched = nrow(weather_data), + rows_written = rows_written, + status = "ok", + message = "", + stringsAsFactors = FALSE + ) +} + + +sync_location_weather <- function( + location_id, + db_path = default_rain_db_path(), + headers = load_api_headers(), + allstations = load_station_catalog(), + locations = get_rain_locations(), + start_date = NULL, + end_date = Sys.Date(), + initial_backfill_days = 21L, + overlap_days = 1L, + chunk_days = 120L, + btw_station_sleep = 0.25, + within_station_sleep = 1 +) { + rain_result <- sync_location_rainfall( + location_id = location_id, + db_path = db_path, + headers = headers, + allstations = allstations, + locations = locations, + start_date = start_date, + end_date = end_date, + initial_backfill_days = initial_backfill_days, + overlap_days = overlap_days, + chunk_days = chunk_days, + btw_station_sleep = btw_station_sleep, + within_station_sleep = within_station_sleep + ) + + observation_result <- sync_location_observations( + location_id = location_id, + db_path = db_path, + headers = headers, + allstations = allstations, + locations = locations, + btw_station_sleep = btw_station_sleep + ) + + status <- get_location_cache_status(location_id, db_path = db_path) + + data.frame( + location_id = location_id, + start_date = rain_result$start_date[1], + end_date = rain_result$end_date[1], + rain_rows_fetched = rain_result$rows_fetched[1], + rain_rows_written = rain_result$rows_written[1], + observation_rows_fetched = observation_result$rows_fetched[1], + observation_rows_written = observation_result$rows_written[1], + rows_written = rain_result$rows_written[1] + observation_result$rows_written[1], + latest_observed_at = status$latest_observed_at[1], + status = if (identical(rain_result$status[1], "error") || identical(observation_result$status[1], "error")) "error" else "ok", + message = paste( + c(rain_result$message[1], observation_result$message[1]), + collapse = " " + ), + stringsAsFactors = FALSE + ) +} + + +sync_all_weather_locations <- function( + location_ids = get_rain_locations()$location_id, + db_path = default_rain_db_path(), + headers = load_api_headers(), + allstations = load_station_catalog(), + locations = get_rain_locations(), + start_date = NULL, + end_date = Sys.Date(), + initial_backfill_days = 21L, + overlap_days = 1L, + chunk_days = 120L, + btw_station_sleep = 0.25, + within_station_sleep = 1 +) { + results <- lapply(location_ids, function(location_id) { + tryCatch( + sync_location_weather( + location_id = location_id, + db_path = db_path, + headers = headers, + allstations = allstations, + locations = locations, + start_date = start_date, + end_date = end_date, + initial_backfill_days = initial_backfill_days, + overlap_days = overlap_days, + chunk_days = chunk_days, + btw_station_sleep = btw_station_sleep, + within_station_sleep = within_station_sleep + ), + error = function(error) { + data.frame( + location_id = location_id, + start_date = if (is.null(start_date)) "" else as.character(as.Date(start_date)), + end_date = as.character(as.Date(end_date)), + rain_rows_fetched = 0L, + rain_rows_written = 0L, + observation_rows_fetched = 0L, + observation_rows_written = 0L, + rows_written = 0L, + latest_observed_at = "", + status = "error", + message = conditionMessage(error), + stringsAsFactors = FALSE + ) + } + ) + }) + + do.call(rbind, results) +} + + +sync_all_rain_locations <- function( + location_ids = get_rain_locations()$location_id, + db_path = default_rain_db_path(), + headers = load_api_headers(), + allstations = load_station_catalog(), + locations = get_rain_locations(), + start_date = NULL, + end_date = Sys.Date(), + initial_backfill_days = 21L, + overlap_days = 1L, + chunk_days = 120L, + btw_station_sleep = 0.25, + within_station_sleep = 1 +) { + sync_all_weather_locations( + location_ids = location_ids, + db_path = db_path, + headers = headers, + allstations = allstations, + locations = locations, + start_date = start_date, + end_date = end_date, + initial_backfill_days = initial_backfill_days, + overlap_days = overlap_days, + chunk_days = chunk_days, + btw_station_sleep = btw_station_sleep, + within_station_sleep = within_station_sleep + ) +} + + +compute_plot_limits <- function(values) { + values <- suppressWarnings(as.numeric(values)) + values <- values[is.finite(values)] + + if (!length(values)) { + return(c(0, 1)) + } + + lower_limit <- min(values) + upper_limit <- max(values) + + if (identical(lower_limit, upper_limit)) { + padding <- if (upper_limit == 0) 1 else abs(upper_limit) * 0.15 + return(c(lower_limit - padding, upper_limit + padding)) + } + + padding <- (upper_limit - lower_limit) * 0.1 + c(lower_limit - padding, upper_limit + padding) +} + + +plot_cached_metric <- function( + metric_data, + metric_id, + view = c("raw", "daily"), + main = NULL +) { + view <- match.arg(view) + metric <- get_weather_metric(metric_id) + + if (is.null(metric_data) || !nrow(metric_data)) { + stop(sprintf("No cached %s data available.", tolower(metric$label))) + } + + station_names <- unique(as.character(metric_data$station_name)) + colours <- grDevices::hcl.colors(length(station_names), "Dark 3") + names(colours) <- station_names + + y_limits <- compute_plot_limits(metric_data$value_num) + y_label <- sprintf("%s (%s)", metric$label, metric$unit) + + if (view == "daily") { + metric_data$observed_day <- as.Date(metric_data$observed_day) + + plot( + range(metric_data$observed_day, na.rm = TRUE), + y_limits, + type = "n", + xlab = "", + ylab = y_label, + main = main + ) + + for (station_name in station_names) { + station_data <- metric_data[metric_data$station_name == station_name, , drop = FALSE] + lines( + station_data$observed_day, + station_data$value_num, + type = "o", + col = colours[[station_name]], + lwd = 2, + pch = 19 + ) + } + } else { + metric_data$observed_at <- as.POSIXct(metric_data$observed_at, tz = "UTC") + + plot( + range(metric_data$observed_at, na.rm = TRUE), + y_limits, + type = "n", + xlab = "", + ylab = y_label, + main = main + ) + + for (station_name in station_names) { + station_data <- metric_data[metric_data$station_name == station_name, , drop = FALSE] + lines( + station_data$observed_at, + station_data$value_num, + col = colours[[station_name]], + lwd = 1.5 + ) + points( + station_data$observed_at, + station_data$value_num, + pch = 19, + col = grDevices::adjustcolor(colours[[station_name]], 0.55), + cex = 0.85 + ) + } + } + + legend( + "topleft", + legend = station_names, + col = colours, + lwd = 2, + pch = 19, + bty = "n" + ) +} + + +plot_cached_rainfall <- function( + rain_data, + view = c("raw", "daily"), + hide_zero = TRUE, + main = NULL +) { + view <- match.arg(view) + + if (is.null(rain_data) || !nrow(rain_data)) { + stop("No rainfall data available.") + } + + if (hide_zero) { + rain_data <- rain_data[is.na(rain_data$rain_mm) | rain_data$rain_mm > 0, , drop = FALSE] + } + + if (!nrow(rain_data)) { + stop("No rainfall data left after removing zero values.") + } + + rain_data$value_num <- rain_data$rain_mm + + plot_cached_metric( + metric_data = rain_data, + metric_id = "rain_6m", + view = view, + main = main + ) +} diff --git a/README.md b/README.md index bd876e9..aad066d 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,89 @@ # Beaumontmeteo -Quick and dirty R interface for meteofrance API. +Shiny app and helper scripts to cache Meteo France station data around Beaumont in a local SQLite database. -project perso to monitor rain in the Beaumont and elsewhere. +## What is stored -you need api keys and stuff and I am not sure how it all works but yeah, one day I'll clean all. +The cache now uses one unified dataset called `weather_measurements` inside `data/rain_history.sqlite`. +Each row stores: + +- location, +- station, +- observation timestamp, +- metric id, +- numeric value, +- source metadata. + +That means rain, temperature, humidity, wind and pressure can all be compared on the same time axis later. + +## App + +Run from the project root: + +```r +shiny::runApp() +``` + +The UI now has two panels: + +- a rain panel backed by historical `DPClim` rainfall, +- a second panel backed by the same SQLite dataset, with a selector for temperature and other station metrics. + +## Syncing the cache + +Put your Meteo France API key in `data/secrets` as `token4`, then run: + +```bash +Rscript scripts/update_rain_db.R +``` + +For a slow historical rainfall backfill into the same SQLite dataset: + +```bash +Rscript scripts/backfill_two_years.R +``` + +Useful options: + +- `update_rain_db.R --location=vignasses` +- `update_rain_db.R --location=vignasses,la_mure` +- `update_rain_db.R --days=60` +- `update_rain_db.R --db-path=/somewhere/else/rain.sqlite` +- `backfill_two_years.R --location=vignasses` +- `backfill_two_years.R --db-path=/somewhere/else/rain.sqlite` +- `backfill_two_years.R --chunk-days=30` +- `backfill_two_years.R --resume=false` +- `backfill_two_years.R --between-chunk-sleep=20` + +## Important note about frequency + +Historical rain can be backfilled from the climatology endpoint, so a weekly sync is fine there. + +Temperature, humidity, wind and pressure come from the rolling station observation feed. To keep those metrics continuous in the database, run the sync daily rather than weekly. + +## Cron examples + +Daily full weather sync: + +```bash +0 7 * * * cd /path/to/beaumontmeteo && Rscript scripts/update_rain_db.R >> /tmp/beaumontmeteo-sync.log 2>&1 +``` + +If you only care about rain history, the same script can still be run weekly because rainfall is backfilled on each sync. + +## Why SQLite + +SQLite is a good fit here because this is a local, read-heavy app with a small periodic write job. + +- No database server to install or operate. +- One file to move, back up and inspect. +- Proper indexed queries and deduplication, unlike CSV or RDS files. +- The app still works offline once data has been cached. + +If this ever becomes a shared multi-user service with concurrent writers, Postgres would be the next upgrade. + +## Notes + +- The database path can be overridden with `BEAUMONT_RAIN_DB`. +- The repo ignores `data/*.sqlite` so the cache file stays local. diff --git a/app.R b/app.R index 42601d1..abdb0ab 100644 --- a/app.R +++ b/app.R @@ -1,96 +1,493 @@ # Load necessary libraries library(shiny) -library(jsonlite) library(httr) -library(shinycssloaders) # For the spinner +library(shinycssloaders) -# Assuming you have your custom functions in a file called 'your_functions.R' +invisible(lapply( + list.files(path = "R", pattern = "\\.R$", full.names = TRUE), + source +)) -# List all R files in the directory and source them +locations <- get_rain_locations() +secondary_metrics <- get_weather_metric_catalog(include_rain = FALSE) +db_path <- default_rain_db_path() +ensure_weather_db(db_path) + +get_window_slider_config <- function(unit = "days") { + switch( + unit, + days = list(label = "Past days", max = 60L, value = 10L), + months = list(label = "Past months", max = 24L, value = 6L), + years = list(label = "Past years", max = 10L, value = 1L), + list(label = "Past days", max = 60L, value = 10L) + ) +} -lapply(list.files(path = "R", pattern = "\\.R$", full.names = T), source) +last_day_of_month <- function(year, month) { + next_month_year <- year + if (month == 12L) 1L else 0L + next_month <- if (month == 12L) 1L else month + 1L + next_month_start <- as.Date(sprintf("%04d-%02d-01", next_month_year, next_month)) + as.integer(format(next_month_start - 1L, "%d")) +} -source("data/secrets") -# Create a Bearer token header -headers.default <- add_headers( - accept = "*/*", - apikey = token4 -) +subtract_calendar_period <- function(date, amount, unit) { + date <- as.Date(date) + amount <- as.integer(amount) -# Load station data + if (amount <= 0L) { + return(date) + } + + date_lt <- as.POSIXlt(date, tz = "UTC") + year <- date_lt$year + 1900L + month <- date_lt$mon + 1L + day <- date_lt$mday + + if (identical(unit, "months")) { + total_months <- year * 12L + (month - 1L) - amount + target_year <- total_months %/% 12L + target_month <- total_months %% 12L + 1L + target_day <- min(day, last_day_of_month(target_year, target_month)) + + return(as.Date(sprintf("%04d-%02d-%02d", target_year, target_month, target_day))) + } + + if (identical(unit, "years")) { + target_year <- year - amount + target_day <- min(day, last_day_of_month(target_year, month)) + + return(as.Date(sprintf("%04d-%02d-%02d", target_year, month, target_day))) + } + + date - amount +} + + +window_start_date <- function(end_date, amount, unit) { + end_date <- as.Date(end_date) + amount <- max(1L, as.integer(amount)) + + if (identical(unit, "days")) { + return(end_date - amount + 1L) + } + + subtract_calendar_period(end_date, amount, unit) + 1L +} + + +format_window_label <- function(amount, unit) { + amount <- max(1L, as.integer(amount)) + unit_label <- if (amount == 1L) sub("s$", "", unit) else unit + + sprintf("last %s %s", amount, unit_label) +} -# Define UI for application ui <- fluidPage( tags$head( tags$style(HTML(" - #main { - height: 100vh; /* Full height of the viewport */ - overflow-y: auto; /* Scroll if content overflows */ - } .container-fluid { - height: 100vh; /* Full height of the viewport */ - display: flex; - flex-direction: column; + max-width: 1480px; } - .row { - flex-grow: 1; - display: flex; + .status-block { + margin-top: 16px; + padding: 12px; + background: #f6f8f9; + border: 1px solid #d9e1e5; + border-radius: 6px; } - .col-sm-8 { - flex-grow: 1; /* Allow the main panel to grow */ + .help-block { + margin-top: 12px; + } + .panel-card { + min-height: 100%; + padding: 16px; + background: #fbfcfd; + border: 1px solid #d9e1e5; + border-radius: 8px; + } + .panel-title { + margin-top: 0; } ")) ), - - # Application title - titlePanel("Beaumont aka SECMONT"), - - # Sidebar for user inputs + + titlePanel("Beaumont Weather Cache"), + sidebarLayout( sidebarPanel( - width=2, - numericInput(inputId = "daysBefore", label = "Nb jour à check:", value=10, min=1) + width = 3, + selectInput( + inputId = "location_id", + label = "Location", + choices = stats::setNames(locations$location_id, locations$label), + selected = locations$location_id[1] + ), + radioButtons( + inputId = "window_unit", + label = "Time window", + choices = c( + "Days" = "days", + "Months" = "months", + "Years" = "years" + ), + selected = "days", + inline = TRUE + ), + sliderInput( + inputId = "window_amount", + label = "Past days", + min = 1, + max = 60, + value = 10, + step = 1 + ), + radioButtons( + inputId = "rain_view_mode", + label = "Rain display", + choices = c( + "6-minute rain" = "raw", + "Daily total" = "daily" + ), + selected = "raw" + ), + checkboxInput( + inputId = "hide_zero", + label = "Hide zero rainfall", + value = TRUE + ), + selectInput( + inputId = "metric_id", + label = "Second panel", + choices = stats::setNames(secondary_metrics$metric_id, secondary_metrics$label), + selected = "air_temperature" + ), + radioButtons( + inputId = "metric_view_mode", + label = "Second panel display", + choices = c( + "Raw observations" = "raw", + "Daily aggregate" = "daily" + ), + selected = "raw" + ), + uiOutput("syncControls"), + div( + class = "status-block", + verbatimTextOutput("cacheStatus") + ) ), - - # Main panel for displaying plot + mainPanel( - width = 6, - withSpinner( plotOutput("dataPlot")) + width = 9, + fluidRow( + column( + width = 6, + div( + class = "panel-card", + h3(class = "panel-title", "Rain"), + withSpinner(plotOutput("rainPlot", height = "420px")), + h4("Daily rain totals"), + tableOutput("dailySummary") + ) + ), + column( + width = 6, + div( + class = "panel-card", + h3(class = "panel-title", textOutput("metricTitle", container = span)), + withSpinner(plotOutput("metricPlot", height = "420px")), + h4("Latest values by station"), + tableOutput("metricLatest") + ) + ) + ) ) ) ) -# Define server logic required to generate and plot data -server <- function(input, output) { - # Reactive expression to fetch data based on input and display spinner while loading - - - # Render the plot - output$dataPlot <- renderPlot({ - startdate <- format(Sys.Date() - input$daysBefore, "%Y-%m-%d") - vignass.coor=c(44.8550665,5.8441789) - allstations <- read.csv("data/allstations.csv") - test1 <- getAllFromCoord(vignass.coor, start_date = startdate, end_date = format(Sys.Date(), "%Y-%m-%d"), allstations, headers = headers.default,btw_station_sleep = .5,within_statio_sleep = 1) - - # Prepare colors - cols <- palette.colors()[1:length(unique(test1$Nom_usuel))] - names(cols) <- unique(test1$Nom_usuel) - - # Filter data - testsep <- test1 - testsep[testsep[,3] == 0 & !is.na(testsep[,3]), c(3,4)] <- NA - - plot(getDate(testsep[,2]), testsep[,3], pch = 20, col = adjustcolor(cols[testsep$Nom_usuel], .4), cex = 1.3, ylim = c(0, 8)) - - # Add legend - legend("topleft", col = cols, legend = names(cols), pch = 20, cex = 1) - - # Highlight specific date - abline(v = as.numeric(as.POSIXlt("2024-08-07", format = "%Y-%m-%d")), lwd = 3, col = "red") + +server <- function(input, output, session) { + api_headers <- load_api_headers() + data_version <- reactiveVal(0L) + last_sync_message <- reactiveVal("") + + selected_location <- reactive({ + get_rain_location(input$location_id, locations = locations) }) + + selected_metric <- reactive({ + get_weather_metric(input$metric_id) + }) + + observeEvent(input$window_unit, { + settings <- get_window_slider_config(input$window_unit) + current_value <- if (is.null(input$window_amount)) { + settings$value + } else { + as.integer(input$window_amount) + } + + updateSliderInput( + session = session, + inputId = "window_amount", + label = settings$label, + min = 1, + max = settings$max, + value = min(max(current_value, 1L), settings$max), + step = 1 + ) + }, ignoreInit = TRUE) + + selected_start_date <- reactive({ + window_start_date( + end_date = Sys.Date(), + amount = input$window_amount, + unit = input$window_unit + ) + }) + + selected_window_label <- reactive({ + format_window_label( + amount = input$window_amount, + unit = input$window_unit + ) + }) + + selected_window_days <- reactive({ + as.integer(Sys.Date() - selected_start_date()) + 1L + }) + + cached_rain <- reactive({ + data_version() + + query_cached_rainfall( + location_id = input$location_id, + start_date = selected_start_date(), + end_date = Sys.Date(), + aggregate = input$rain_view_mode, + db_path = db_path + ) + }) + + cached_metric <- reactive({ + data_version() + + query_cached_metric( + location_id = input$location_id, + metric_id = input$metric_id, + start_date = selected_start_date(), + end_date = Sys.Date(), + aggregate = input$metric_view_mode, + db_path = db_path + ) + }) + + daily_summary <- reactive({ + data_version() + + query_cached_rainfall( + location_id = input$location_id, + start_date = selected_start_date(), + end_date = Sys.Date(), + aggregate = "daily", + db_path = db_path + ) + }) + + metric_latest <- reactive({ + data_version() + + query_latest_metric_values( + location_id = input$location_id, + metric_id = input$metric_id, + db_path = db_path + ) + }) + + output$metricTitle <- renderText({ + selected_metric()$label + }) + + output$syncControls <- renderUI({ + if (is.null(api_headers)) { + return( + div( + class = "help-block", + helpText( + "This app is reading the local SQLite cache only. Add token4 to data/secrets and run", + "`Rscript scripts/update_rain_db.R`", + "to refresh rain plus the second-panel weather metrics." + ), + helpText( + "Rain can be backfilled. Temperature, humidity, wind and pressure come from the rolling observation feed, so sync that script daily if you want a continuous history." + ) + ) + ) + } + + tagList( + actionButton("sync_now", "Sync location from API"), + div( + class = "help-block", + helpText( + "This refresh pulls historical rain plus the latest station observations into the same SQLite dataset." + ) + ) + ) + }) + + observeEvent(input$sync_now, { + req(!is.null(api_headers)) + + location <- selected_location() + result <- tryCatch( + withProgress( + message = sprintf("Syncing %s into SQLite", location$label[1]), + value = 0.3, + { + sync_location_weather( + location_id = input$location_id, + db_path = db_path, + headers = api_headers, + initial_backfill_days = max(21L, selected_window_days() + 7L) + ) + } + ), + error = function(error) { + last_sync_message( + sprintf("Last sync failed: %s", conditionMessage(error)) + ) + NULL + } + ) + + if (is.null(result)) { + return() + } + + latest_observed_at <- result$latest_observed_at[1] + if (is.na(latest_observed_at) || !nzchar(latest_observed_at)) { + latest_observed_at <- "no data returned" + } + + last_sync_message( + sprintf( + paste( + "Last sync: %s rows written", + "(rain %s, observations %s).", + "Cache now reaches %s." + ), + format(result$rows_written[1], big.mark = ","), + format(result$rain_rows_written[1], big.mark = ","), + format(result$observation_rows_written[1], big.mark = ","), + latest_observed_at + ) + ) + data_version(data_version() + 1L) + }) + + output$cacheStatus <- renderText({ + data_version() + + status_text <- describe_cache_status( + location_id = input$location_id, + db_path = db_path, + locations = locations + ) + + if (!nzchar(last_sync_message())) { + return(status_text) + } + + paste(status_text, last_sync_message(), sep = "\n\n") + }) + + output$rainPlot <- renderPlot({ + plot_data <- cached_rain() + + shiny::validate( + shiny::need( + nrow(plot_data) > 0, + "No cached rainfall for this period yet. Run the sync script or use the API sync button if credentials are configured." + ) + ) + + if (input$hide_zero) { + plot_data <- plot_data[ + is.na(plot_data$rain_mm) | plot_data$rain_mm > 0, + , + drop = FALSE + ] + } + + shiny::validate( + shiny::need( + nrow(plot_data) > 0, + "No non-zero rainfall in this window. Untick 'Hide zero rainfall' if you want to inspect dry periods too." + ) + ) + + plot_cached_rainfall( + rain_data = plot_data, + view = input$rain_view_mode, + hide_zero = FALSE, + main = sprintf( + "%s - %s", + selected_location()$label[1], + selected_window_label() + ) + ) + }) + + output$metricPlot <- renderPlot({ + plot_data <- cached_metric() + metric <- selected_metric() + + shiny::validate( + shiny::need( + nrow(plot_data) > 0, + sprintf( + "No cached %s data for this window yet. Run the sync script more frequently if you want a continuous history for that metric.", + tolower(metric$label) + ) + ) + ) + + plot_cached_metric( + metric_data = plot_data, + metric_id = input$metric_id, + view = input$metric_view_mode, + main = sprintf( + "%s - %s - %s", + selected_location()$label[1], + metric$label, + selected_window_label() + ) + ) + }) + + output$dailySummary <- renderTable({ + summary_data <- daily_summary() + if (!nrow(summary_data)) { + return(NULL) + } + + names(summary_data) <- c("Station ID", "Station", "Day", "Metric", "Label", "Unit", "Rain") + summary_data[, c("Station", "Day", "Rain")] + }, striped = TRUE, spacing = "s", digits = 2) + + output$metricLatest <- renderTable({ + latest_data <- metric_latest() + if (!nrow(latest_data)) { + return(NULL) + } + + names(latest_data) <- c("Station ID", "Station", "Observed At", "Metric", "Label", "Unit", "Value") + latest_data[, c("Station", "Observed At", "Value", "Unit")] + }, striped = TRUE, spacing = "s", digits = 2) } -# Run the application + shinyApp(ui = ui, server = server) diff --git a/scripts/backfill_two_years.R b/scripts/backfill_two_years.R new file mode 100644 index 0000000..5a92807 --- /dev/null +++ b/scripts/backfill_two_years.R @@ -0,0 +1,299 @@ +command_line_args <- commandArgs(trailingOnly = TRUE) + +parse_cli_args <- function(args) { + parsed <- list() + + for (arg in args) { + if (!grepl("^--", arg)) { + next + } + + parts <- strsplit(sub("^--", "", arg), "=", fixed = TRUE)[[1]] + key <- parts[1] + value <- if (length(parts) > 1) paste(parts[-1], collapse = "=") else TRUE + parsed[[key]] <- value + } + + parsed +} + + +parse_flag <- function(value, default = FALSE) { + if (is.null(value)) { + return(default) + } + + normalized <- tolower(as.character(value)) + if (normalized %in% c("true", "t", "1", "yes", "y")) { + return(TRUE) + } + if (normalized %in% c("false", "f", "0", "no", "n")) { + return(FALSE) + } + + default +} + + +sleep_with_message <- function(seconds, reason) { + seconds <- as.numeric(seconds) + + if (!is.finite(seconds) || seconds <= 0) { + return(invisible(NULL)) + } + + cat(sprintf("%s Sleeping %.1f seconds.\n", reason, seconds)) + Sys.sleep(seconds) +} + + +script_file_arg <- grep("^--file=", commandArgs(FALSE), value = TRUE) +script_path <- normalizePath(sub("^--file=", "", script_file_arg[1]), winslash = "/") +project_root <- normalizePath(file.path(dirname(script_path), ".."), winslash = "/") + +setwd(project_root) +invisible(lapply( + list.files(path = "R", pattern = "\\.R$", full.names = TRUE), + source +)) + +args <- parse_cli_args(command_line_args) +db_path <- if (!is.null(args[["db-path"]])) { + normalizePath(args[["db-path"]], winslash = "/", mustWork = FALSE) +} else { + default_rain_db_path() +} + +location_ids <- if (!is.null(args[["location"]])) { + strsplit(args[["location"]], ",", fixed = TRUE)[[1]] +} else { + get_rain_locations()$location_id +} + +days_back <- if (!is.null(args[["days"]])) { + as.integer(args[["days"]]) +} else { + 730L +} + +chunk_days <- if (!is.null(args[["chunk-days"]])) { + as.integer(args[["chunk-days"]]) +} else { + 30L +} + +btw_station_sleep <- if (!is.null(args[["btw-station-sleep"]])) { + as.numeric(args[["btw-station-sleep"]]) +} else { + 4 +} + +within_station_sleep <- if (!is.null(args[["within-station-sleep"]])) { + as.numeric(args[["within-station-sleep"]]) +} else { + 10 +} + +between_chunk_sleep <- if (!is.null(args[["between-chunk-sleep"]])) { + as.numeric(args[["between-chunk-sleep"]]) +} else { + 20 +} + +between_location_sleep <- if (!is.null(args[["between-location-sleep"]])) { + as.numeric(args[["between-location-sleep"]]) +} else { + 60 +} + +resume <- parse_flag(args[["resume"]], default = TRUE) +end_date <- if (!is.null(args[["end-date"]])) { + as.Date(args[["end-date"]]) +} else { + Sys.Date() - 1L +} + +headers <- load_api_headers() +if (is.null(headers)) { + stop("No API credentials found. Add token4 to data/secrets before running the backfill.") +} + +allstations <- load_station_catalog() +locations <- get_rain_locations() +ensure_weather_db(db_path) + +cat("Starting slow historical backfill.\n") +cat(sprintf("Database: %s\n", db_path)) +cat(sprintf("Locations: %s\n", paste(location_ids, collapse = ", "))) +cat(sprintf("Rainfall window: last %s days ending on %s\n", days_back, as.character(end_date))) +cat(sprintf("Chunk size: %s days\n", chunk_days)) +cat(sprintf("Sleep between stations: %.1f seconds\n", btw_station_sleep)) +cat(sprintf("Sleep while waiting for station files: %.1f seconds\n", within_station_sleep)) +cat(sprintf("Sleep between chunks: %.1f seconds\n", between_chunk_sleep)) +cat(sprintf("Sleep between locations: %.1f seconds\n", between_location_sleep)) +cat(sprintf("Resume mode: %s\n\n", if (resume) "on" else "off")) + +overall_start_date <- end_date - days_back + 1L +results <- vector("list", length(location_ids)) + +for (location_index in seq_along(location_ids)) { + location_id <- location_ids[location_index] + location_label <- get_rain_location(location_id, locations = locations)$label[1] + + location_start_date <- overall_start_date + if (resume) { + resume_start_date <- get_sync_start_date( + location_id = location_id, + db_path = db_path, + end_date = end_date, + initial_backfill_days = days_back, + overlap_days = 1L, + metric_id = "rain_6m" + ) + location_start_date <- as.Date(max(location_start_date, resume_start_date)) + } + + if (location_start_date > end_date) { + cat(sprintf("[%s] %s already has data up to %s. Skipping.\n", location_id, location_label, as.character(end_date))) + status <- get_location_cache_status(location_id, db_path = db_path) + results[[location_index]] <- data.frame( + location_id = location_id, + start_date = as.character(location_start_date), + end_date = as.character(end_date), + chunk_days = chunk_days, + rows_fetched = 0L, + rows_written = 0L, + latest_observed_at = status$latest_observed_at[1], + status = "up_to_date", + message = "", + stringsAsFactors = FALSE + ) + next + } + + ranges <- split_sync_ranges( + start_date = location_start_date, + end_date = end_date, + chunk_days = chunk_days + ) + + cat(sprintf("[%s] Backfilling %s from %s to %s in %s chunk(s).\n", + location_id, + location_label, + as.character(location_start_date), + as.character(end_date), + nrow(ranges) + )) + + location_rows_fetched <- 0L + location_rows_written <- 0L + location_status <- "ok" + location_message <- "" + + for (range_index in seq_len(nrow(ranges))) { + chunk_start <- ranges$start_date[range_index] + chunk_end <- ranges$end_date[range_index] + + cat(sprintf( + "[%s] Chunk %s/%s: %s -> %s\n", + location_id, + range_index, + nrow(ranges), + as.character(chunk_start), + as.character(chunk_end) + )) + + chunk_result <- tryCatch( + { + raw_data <- fetch_rainfall_from_api( + location_id = location_id, + start_date = chunk_start, + end_date = chunk_end, + headers = headers, + allstations = allstations, + locations = locations, + btw_station_sleep = btw_station_sleep, + within_station_sleep = within_station_sleep + ) + + weather_data <- normalise_rainfall_data( + raw_data = raw_data, + location_id = location_id + ) + + rows_written <- upsert_weather_measurements( + weather_data = weather_data, + db_path = db_path + ) + + list( + rows_fetched = nrow(weather_data), + rows_written = rows_written, + status = "ok", + message = "" + ) + }, + error = function(error) { + list( + rows_fetched = 0L, + rows_written = 0L, + status = "error", + message = conditionMessage(error) + ) + } + ) + + location_rows_fetched <- location_rows_fetched + chunk_result$rows_fetched + location_rows_written <- location_rows_written + chunk_result$rows_written + + cat(sprintf( + "[%s] Chunk result: fetched %s rows, wrote %s rows.\n", + location_id, + format(chunk_result$rows_fetched, big.mark = ","), + format(chunk_result$rows_written, big.mark = ",") + )) + + if (!identical(chunk_result$status, "ok")) { + location_status <- "error" + location_message <- chunk_result$message + cat(sprintf("[%s] Stopping because of error: %s\n", location_id, location_message)) + break + } + + if (range_index < nrow(ranges)) { + sleep_with_message( + seconds = between_chunk_sleep, + reason = sprintf("[%s] Chunk complete.", location_id) + ) + } + } + + status <- get_location_cache_status(location_id, db_path = db_path) + results[[location_index]] <- data.frame( + location_id = location_id, + start_date = as.character(location_start_date), + end_date = as.character(end_date), + chunk_days = chunk_days, + rows_fetched = location_rows_fetched, + rows_written = location_rows_written, + latest_observed_at = status$latest_observed_at[1], + status = location_status, + message = location_message, + stringsAsFactors = FALSE + ) + + if (location_index < length(location_ids)) { + sleep_with_message( + seconds = between_location_sleep, + reason = sprintf("[%s] Location complete.", location_id) + ) + } +} + +results <- do.call(rbind, results) +cat("\nBackfill summary:\n") +print(results, row.names = FALSE) + +if (any(results$status == "error")) { + stop("At least one location failed during the historical backfill.") +} diff --git a/scripts/update_rain_db.R b/scripts/update_rain_db.R new file mode 100644 index 0000000..ce5f80e --- /dev/null +++ b/scripts/update_rain_db.R @@ -0,0 +1,66 @@ +command_line_args <- commandArgs(trailingOnly = TRUE) + +parse_cli_args <- function(args) { + parsed <- list() + + for (arg in args) { + if (!grepl("^--", arg)) { + next + } + + parts <- strsplit(sub("^--", "", arg), "=", fixed = TRUE)[[1]] + key <- parts[1] + value <- if (length(parts) > 1) paste(parts[-1], collapse = "=") else TRUE + parsed[[key]] <- value + } + + parsed +} + + +script_file_arg <- grep("^--file=", commandArgs(FALSE), value = TRUE) +script_path <- normalizePath(sub("^--file=", "", script_file_arg[1]), winslash = "/") +project_root <- normalizePath(file.path(dirname(script_path), ".."), winslash = "/") + +setwd(project_root) +invisible(lapply( + list.files(path = "R", pattern = "\\.R$", full.names = TRUE), + source +)) + +args <- parse_cli_args(command_line_args) +db_path <- if (!is.null(args[["db-path"]])) { + normalizePath(args[["db-path"]], winslash = "/", mustWork = FALSE) +} else { + default_rain_db_path() +} + +location_ids <- if (!is.null(args[["location"]])) { + strsplit(args[["location"]], ",", fixed = TRUE)[[1]] +} else { + get_rain_locations()$location_id +} + +initial_backfill_days <- if (!is.null(args[["days"]])) { + as.integer(args[["days"]]) +} else { + 21L +} + +headers <- load_api_headers() +if (is.null(headers)) { + stop("No API credentials found. Add token4 to data/secrets before running the sync.") +} + +results <- sync_all_weather_locations( + location_ids = location_ids, + db_path = db_path, + headers = headers, + initial_backfill_days = initial_backfill_days +) + +print(results, row.names = FALSE) + +if (any(results$status == "error")) { + stop("At least one location failed to sync.") +} diff --git a/tests/testthat/test-getStationData.R b/tests/testthat/test-getStationData.R index 008cf54..ddf2445 100644 --- a/tests/testthat/test-getStationData.R +++ b/tests/testthat/test-getStationData.R @@ -1,19 +1,37 @@ -library(jsonlite) -library(httr) - - -source(file.path(here::here(),"secrets")) -# Create a Bearer token header -headers.default <- add_headers( - accept = "*/*", - apikey = token2 +testthat::skip_if_not( + identical(tolower(Sys.getenv("BEAUMONT_RUN_LIVE_API_TESTS", "false")), "true"), + "Set BEAUMONT_RUN_LIVE_API_TESTS=true to run live Meteo France API tests." ) -test_that("get station return something right", -{ - test=getStationData(start_date="2024-06-24",end_date="2024-08-14",station_id="38269004",headers=headers.default) - if(!is.null(test)){testthat::expect_length(test,3)} - else testthat::expect_null(test) +testthat::skip_if_not( + file.exists(file.path("data", "secrets")), + "Live API tests require data/secrets." +) + +library(httr) + +source(file.path("R", "request.R")) +source(file.path("R", "getStationData.R")) + +secret_env <- new.env(parent = baseenv()) +sys.source(file.path("data", "secrets"), envir = secret_env) + +headers.default <- add_headers( + accept = "*/*", + apikey = secret_env$token2 +) + +test_that("get station return something right", { + test <- getStationData( + start_date = "2024-06-24", + end_date = "2024-08-14", + station_id = "38269004", + headers = headers.default + ) + + if (!is.null(test)) { + testthat::expect_length(test, 3) + } else { + testthat::expect_null(test) + } }) - - diff --git a/tests/testthat/test-rain-db.R b/tests/testthat/test-rain-db.R new file mode 100644 index 0000000..9f5e890 --- /dev/null +++ b/tests/testthat/test-rain-db.R @@ -0,0 +1,180 @@ +source(testthat::test_path("..", "..", "R", "rain_db.R")) + +test_that("normalise_rainfall_data converts API payload into unified metric rows", { + raw_data <- data.frame( + POSTE = c("1001", "1001"), + DATE = c(202401020000, 202401020006), + RR6 = c(0, 1.5), + Nom_usuel = c("Station A", "Station A"), + stringsAsFactors = FALSE + ) + + cache_rows <- normalise_rainfall_data( + raw_data = raw_data, + location_id = "vignasses", + fetched_at = as.POSIXct("2026-04-08 09:00:00", tz = "UTC") + ) + + expect_identical( + names(cache_rows), + c( + "location_id", + "station_id", + "station_name", + "observed_at", + "observed_day", + "metric_id", + "metric_label", + "unit", + "source_name", + "value_num", + "fetched_at" + ) + ) + expect_equal(cache_rows$metric_id[1], "rain_6m") + expect_equal(cache_rows$observed_at[2], "2024-01-02T00:06:00Z") + expect_equal(cache_rows$fetched_at[1], "2026-04-08T09:00:00Z") +}) + + +test_that("normalise_observation_package_data extracts temperature and other metrics", { + raw_data <- data.frame( + geo_id_insee = c("1001", "1001"), + validity_time = c("2026-04-08T09:24:00Z", "2026-04-08T09:30:00Z"), + t = c(293.15, 294.15), + td = c(289.15, 290.15), + u = c(40, 42), + ff = c(1.5, 2.0), + fxi10 = c(3.5, 4.0), + pres = c(101325, 101425), + pmer = c(101525, 101625), + stringsAsFactors = FALSE + ) + + cache_rows <- normalise_observation_package_data( + raw_data = raw_data, + location_id = "vignasses", + station_name = "Station A", + fetched_at = as.POSIXct("2026-04-08 10:00:00", tz = "UTC") + ) + + expect_true(all(c( + "air_temperature", + "dew_point", + "humidity", + "wind_speed", + "wind_gust", + "station_pressure", + "sea_level_pressure" + ) %in% unique(cache_rows$metric_id))) + expect_equal(cache_rows$value_num[cache_rows$metric_id == "air_temperature"][1], 20) + expect_equal(cache_rows$value_num[cache_rows$metric_id == "station_pressure"][1], 1013.2) +}) + + +test_that("SQLite cache upserts and queries multiple metrics from one dataset", { + skip_if_not(nzchar(Sys.which("sqlite3")), "sqlite3 is required for cache tests.") + + db_path <- tempfile(fileext = ".sqlite") + on.exit(unlink(c(db_path, paste0(db_path, c("-shm", "-wal")))), add = TRUE) + + ensure_weather_db(db_path) + + rain_rows <- normalise_rainfall_data( + raw_data = data.frame( + POSTE = c("1001", "1001"), + DATE = c(202401020000, 202401020006), + RR6 = c(1.25, 0.5), + Nom_usuel = c("Station A", "Station A"), + stringsAsFactors = FALSE + ), + location_id = "vignasses", + fetched_at = as.POSIXct("2026-04-08 09:00:00", tz = "UTC") + ) + + obs_rows <- normalise_observation_package_data( + raw_data = data.frame( + geo_id_insee = c("1001", "1001"), + validity_time = c("2024-01-02T00:00:00Z", "2024-01-02T06:00:00Z"), + t = c(293.15, 295.15), + u = c(40, 44), + ff = c(1.5, 2.5), + pres = c(101325, 101225), + stringsAsFactors = FALSE + ), + location_id = "vignasses", + station_name = "Station A", + fetched_at = as.POSIXct("2026-04-08 09:00:00", tz = "UTC") + ) + + expect_equal(upsert_weather_measurements(rain_rows, db_path), 2) + expect_true(upsert_weather_measurements(obs_rows, db_path) >= 6) + + updated_rain <- rain_rows[1, , drop = FALSE] + updated_rain$value_num <- 2 + expect_true(upsert_rainfall_observations(updated_rain, db_path) >= 1) + + rain_raw <- query_cached_rainfall( + location_id = "vignasses", + start_date = "2024-01-02", + end_date = "2024-01-02", + aggregate = "raw", + db_path = db_path + ) + + rain_daily <- query_cached_rainfall( + location_id = "vignasses", + start_date = "2024-01-02", + end_date = "2024-01-02", + aggregate = "daily", + db_path = db_path + ) + + temperature_raw <- query_cached_metric( + location_id = "vignasses", + metric_id = "air_temperature", + start_date = "2024-01-02", + end_date = "2024-01-02", + aggregate = "raw", + db_path = db_path + ) + + temperature_daily <- query_cached_metric( + location_id = "vignasses", + metric_id = "air_temperature", + start_date = "2024-01-02", + end_date = "2024-01-02", + aggregate = "daily", + db_path = db_path + ) + + latest_temperature <- query_latest_metric_values( + location_id = "vignasses", + metric_id = "air_temperature", + db_path = db_path + ) + + expect_equal(nrow(rain_raw), 2) + expect_equal(rain_raw$rain_mm[1], 2) + expect_equal(rain_daily$rain_mm[1], 2.5) + expect_equal(nrow(temperature_raw), 2) + expect_equal(temperature_daily$value_num[1], 21) + expect_equal(latest_temperature$value_num[1], 22) + expect_equal( + as.character(get_sync_start_date("vignasses", db_path, end_date = as.Date("2024-01-10"))), + "2024-01-01" + ) +}) + + +test_that("sync ranges are chunked predictably", { + ranges <- split_sync_ranges( + start_date = as.Date("2024-01-01"), + end_date = as.Date("2024-05-15"), + chunk_days = 60L + ) + + expect_equal(nrow(ranges), 3) + expect_equal(as.character(ranges$start_date[1]), "2024-01-01") + expect_equal(as.character(ranges$end_date[3]), "2024-05-15") +}) diff --git a/tests/testthat/test-script-helpers.R b/tests/testthat/test-script-helpers.R index e56e559..e8a49a7 100644 --- a/tests/testthat/test-script-helpers.R +++ b/tests/testthat/test-script-helpers.R @@ -16,6 +16,20 @@ test_that("sourced getStationData handles request failure without attached httr" }) +test_that("DPClim period bounds include the full end day", { + helper_env <- new.env(parent = globalenv()) + sys.source(testthat::test_path("..", "..", "R", "getStationData.R"), envir = helper_env) + + bounds <- helper_env$get_dpclim_period_bounds( + start_date = "2024-08-09", + end_date = "2024-09-07" + ) + + expect_identical(bounds$start, "2024-08-09T00:00:00Z") + expect_identical(bounds$end, "2024-09-08T00:00:00Z") +}) + + test_that("getAllFromCoord returns NULL when every station fetch fails", { helper_env <- new.env(parent = globalenv()) sys.source(testthat::test_path("..", "..", "R", "getAllFromCoord.R"), envir = helper_env)