add finer grouping selection

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(