1698 lines
42 KiB
R
1698 lines
42 KiB
R
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", "weekly", "monthly")) {
|
|
aggregate <- match.arg(aggregate)
|
|
|
|
if (aggregate != "raw") {
|
|
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)"
|
|
)
|
|
}
|
|
|
|
|
|
get_time_bucket_expression <- function(aggregate = c("daily", "weekly", "monthly")) {
|
|
aggregate <- match.arg(aggregate)
|
|
|
|
switch(
|
|
aggregate,
|
|
daily = "observed_day",
|
|
weekly = paste(
|
|
"date(",
|
|
"observed_day,",
|
|
"'-' || ((CAST(strftime('%w', observed_day) AS integer) + 6) % 7) || ' days'",
|
|
")"
|
|
),
|
|
"date(observed_day, 'start of month')"
|
|
)
|
|
}
|
|
|
|
|
|
query_cached_metric <- function(
|
|
location_id,
|
|
metric_id,
|
|
start_date,
|
|
end_date = Sys.Date(),
|
|
aggregate = c("raw", "daily", "weekly", "monthly"),
|
|
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 != "raw") {
|
|
period_expression <- get_time_bucket_expression(aggregate)
|
|
sprintf(
|
|
paste(
|
|
"SELECT station_id, station_name, %s AS 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, %s, metric_id, metric_label, unit",
|
|
"ORDER BY observed_day, station_name;"
|
|
),
|
|
period_expression,
|
|
get_daily_aggregate_expression(metric_id),
|
|
sql_string(location_id),
|
|
sql_string(metric$metric_id),
|
|
sql_string(start_at),
|
|
sql_string(end_at),
|
|
period_expression
|
|
)
|
|
} 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", "weekly", "monthly"),
|
|
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)
|
|
requested_start_date <- end_date - as.integer(initial_backfill_days) + 1L
|
|
|
|
latest_data <- read_sqlite_query(
|
|
sprintf(
|
|
paste(
|
|
"SELECT MIN(observed_at) AS earliest_observed_at,",
|
|
"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
|
|
)
|
|
|
|
earliest_observed_at <- if (nrow(latest_data)) latest_data$earliest_observed_at[1] else ""
|
|
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(requested_start_date)
|
|
}
|
|
|
|
earliest_observed_day <- as.Date(earliest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ")
|
|
latest_observed_day <- as.Date(latest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ")
|
|
if (!is.na(earliest_observed_day) && earliest_observed_day > requested_start_date) {
|
|
return(requested_start_date)
|
|
}
|
|
if (latest_observed_day > end_date) {
|
|
return(end_date + 1L)
|
|
}
|
|
|
|
latest_observed_day - 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 = "",
|
|
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)
|
|
}
|
|
|
|
|
|
get_metric_plot_y_label <- function(
|
|
metric_id,
|
|
view = c("raw", "daily", "weekly", "monthly")
|
|
) {
|
|
view <- match.arg(view)
|
|
metric <- get_weather_metric(metric_id)
|
|
|
|
if (!identical(metric_id, "rain_6m") || identical(view, "raw")) {
|
|
return(sprintf("%s (%s)", metric$label, metric$unit))
|
|
}
|
|
|
|
switch(
|
|
view,
|
|
daily = "Daily rain total (mm)",
|
|
weekly = "Weekly rain total (mm)",
|
|
monthly = "Monthly rain total (mm)"
|
|
)
|
|
}
|
|
|
|
|
|
build_metric_plot_time_axis <- function(
|
|
values,
|
|
view = c("raw", "daily", "weekly", "monthly")
|
|
) {
|
|
view <- match.arg(view)
|
|
|
|
if (identical(view, "raw")) {
|
|
values <- as.POSIXct(values, tz = "UTC")
|
|
values <- values[!is.na(values)]
|
|
|
|
if (!length(values)) {
|
|
return(list(at = values, labels = character(), las = 1, cex.axis = 0.8))
|
|
}
|
|
|
|
span_days <- max(1, as.numeric(difftime(max(values), min(values), units = "days")))
|
|
tick_count <- min(14L, max(6L, ceiling(span_days * 2)))
|
|
axis_at <- as.POSIXct(
|
|
pretty(as.numeric(values), n = tick_count),
|
|
origin = "1970-01-01",
|
|
tz = "UTC"
|
|
)
|
|
axis_at <- axis_at[axis_at >= min(values) & axis_at <= max(values)]
|
|
if (length(axis_at) < min(6L, tick_count)) {
|
|
axis_at <- seq(
|
|
from = min(values),
|
|
to = max(values),
|
|
length.out = tick_count
|
|
)
|
|
}
|
|
|
|
return(list(
|
|
at = axis_at,
|
|
labels = format(axis_at, "%d %b\n%H:%M", tz = "UTC"),
|
|
las = 1,
|
|
cex.axis = 0.8
|
|
))
|
|
}
|
|
|
|
values <- as.Date(values)
|
|
values <- values[!is.na(values)]
|
|
|
|
if (!length(values)) {
|
|
return(list(at = values, labels = character(), las = 1, cex.axis = 0.85))
|
|
}
|
|
|
|
span_days <- max(1, as.integer(max(values) - min(values)))
|
|
tick_count <- switch(
|
|
view,
|
|
daily = min(14L, max(6L, span_days + 1L)),
|
|
weekly = min(12L, max(6L, ceiling(span_days / 7))),
|
|
monthly = min(12L, max(6L, ceiling(span_days / 30))),
|
|
8L
|
|
)
|
|
label_format <- switch(
|
|
view,
|
|
daily = "%d %b",
|
|
weekly = "%d %b",
|
|
monthly = "%b\n%Y"
|
|
)
|
|
|
|
axis_at <- as.Date(
|
|
pretty(as.numeric(values), n = tick_count),
|
|
origin = "1970-01-01"
|
|
)
|
|
axis_at <- sort(unique(axis_at[axis_at >= min(values) & axis_at <= max(values)]))
|
|
if (length(axis_at) < min(6L, tick_count)) {
|
|
axis_at <- sort(unique(as.Date(seq(
|
|
from = min(values),
|
|
to = max(values),
|
|
length.out = tick_count
|
|
))))
|
|
}
|
|
|
|
list(
|
|
at = axis_at,
|
|
labels = format(axis_at, label_format),
|
|
las = 1,
|
|
cex.axis = 0.85
|
|
)
|
|
}
|
|
|
|
|
|
draw_metric_plot_time_axis <- function(
|
|
values,
|
|
view = c("raw", "daily", "weekly", "monthly")
|
|
) {
|
|
view <- match.arg(view)
|
|
axis_spec <- build_metric_plot_time_axis(values = values, view = view)
|
|
|
|
if (!length(axis_spec$at)) {
|
|
return(invisible(NULL))
|
|
}
|
|
|
|
if (identical(view, "raw")) {
|
|
axis.POSIXct(
|
|
side = 1,
|
|
at = axis_spec$at,
|
|
labels = axis_spec$labels,
|
|
las = axis_spec$las,
|
|
cex.axis = axis_spec$cex.axis
|
|
)
|
|
} else {
|
|
axis.Date(
|
|
side = 1,
|
|
at = axis_spec$at,
|
|
labels = axis_spec$labels,
|
|
las = axis_spec$las,
|
|
cex.axis = axis_spec$cex.axis
|
|
)
|
|
}
|
|
|
|
invisible(axis_spec)
|
|
}
|
|
|
|
|
|
plot_cached_metric <- function(
|
|
metric_data,
|
|
metric_id,
|
|
view = c("raw", "daily", "weekly", "monthly"),
|
|
main = NULL
|
|
) {
|
|
view <- match.arg(view)
|
|
metric <- get_weather_metric(metric_id)
|
|
old_par <- graphics::par(no.readonly = TRUE)
|
|
on.exit(graphics::par(old_par), add = TRUE)
|
|
graphics::par(mar = c(6.5, 4.5, 4, 1) + 0.1)
|
|
|
|
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 <- get_metric_plot_y_label(metric_id = metric_id, view = view)
|
|
|
|
if (view != "raw") {
|
|
metric_data$observed_day <- as.Date(metric_data$observed_day)
|
|
|
|
plot(
|
|
range(metric_data$observed_day, na.rm = TRUE),
|
|
y_limits,
|
|
type = "n",
|
|
xaxt = "n",
|
|
xaxs = "i",
|
|
xlab = "",
|
|
ylab = y_label,
|
|
main = main
|
|
)
|
|
draw_metric_plot_time_axis(metric_data$observed_day, view = view)
|
|
|
|
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",
|
|
xaxt = "n",
|
|
xaxs = "i",
|
|
xlab = "",
|
|
ylab = y_label,
|
|
main = main
|
|
)
|
|
draw_metric_plot_time_axis(metric_data$observed_at, view = view)
|
|
|
|
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", "weekly", "monthly"),
|
|
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
|
|
)
|
|
}
|