299 lines
8 KiB
R
299 lines
8 KiB
R
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 = "",
|
|
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.")
|
|
}
|