diff --git a/R/rain_db.R b/R/rain_db.R index ee5efd3..63fd049 100644 --- a/R/rain_db.R +++ b/R/rain_db.R @@ -365,10 +365,10 @@ empty_weather_cache <- function() { } -empty_metric_query <- function(aggregate = c("raw", "daily")) { +empty_metric_query <- function(aggregate = c("raw", "daily", "weekly", "monthly")) { aggregate <- match.arg(aggregate) - if (aggregate == "daily") { + if (aggregate != "raw") { return(data.frame( station_id = character(), station_name = character(), @@ -820,12 +820,29 @@ get_daily_aggregate_expression <- function(metric_id) { } +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"), + aggregate = c("raw", "daily", "weekly", "monthly"), db_path = default_rain_db_path() ) { aggregate <- match.arg(aggregate) @@ -840,24 +857,27 @@ query_cached_metric <- function( 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") { + sql <- if (aggregate != "raw") { + period_expression <- get_time_bucket_expression(aggregate) sprintf( paste( - "SELECT station_id, station_name, observed_day, metric_id, metric_label, unit,", + "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, observed_day, metric_id, metric_label, unit", + "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) + sql_string(end_at), + period_expression ) } else { sprintf( @@ -890,7 +910,7 @@ query_cached_rainfall <- function( location_id, start_date, end_date = Sys.Date(), - aggregate = c("raw", "daily"), + aggregate = c("raw", "daily", "weekly", "monthly"), db_path = default_rain_db_path() ) { aggregate <- match.arg(aggregate) @@ -997,11 +1017,13 @@ get_sync_start_date <- function( ) { 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 MAX(observed_at) AS latest_observed_at", + "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;" @@ -1012,12 +1034,22 @@ get_sync_start_date <- function( 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(end_date - as.integer(initial_backfill_days) + 1L) + return(requested_start_date) } - as.Date(latest_observed_at, format = "%Y-%m-%dT%H:%M:%SZ") - as.integer(overlap_days) + 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) } @@ -1148,7 +1180,7 @@ sync_location_rainfall <- function( if (start_date > end_date) { return(data.frame( location_id = location_id, - start_date = as.character(start_date), + start_date = "", end_date = as.character(end_date), rows_fetched = 0L, rows_written = 0L, @@ -1410,7 +1442,7 @@ compute_plot_limits <- function(values) { plot_cached_metric <- function( metric_data, metric_id, - view = c("raw", "daily"), + view = c("raw", "daily", "weekly", "monthly"), main = NULL ) { view <- match.arg(view) @@ -1427,7 +1459,7 @@ plot_cached_metric <- function( y_limits <- compute_plot_limits(metric_data$value_num) y_label <- sprintf("%s (%s)", metric$label, metric$unit) - if (view == "daily") { + if (view != "raw") { metric_data$observed_day <- as.Date(metric_data$observed_day) plot(