Fix grouped rainfall plot views

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(
@@ -1493,7 +1525,7 @@ plot_cached_metric <- function(
 
 plot_cached_rainfall <- function(
   rain_data,
-  view = c("raw", "daily"),
+  view = c("raw", "daily", "weekly", "monthly"),
   hide_zero = TRUE,
   main = NULL
 ) {

app.R

@@ -84,6 +84,99 @@ format_window_label <- function(amount, unit) {
   sprintf("last %s %s", amount, unit_label)
 }
 
+
+get_rain_view_choices <- function(window_unit = "days") {
+  switch(
+    window_unit,
+    days = c(
+      "6-minute rain" = "raw",
+      "Daily total" = "daily"
+    ),
+    months = c(
+      "Daily total" = "daily",
+      "Weekly total" = "weekly"
+    ),
+    years = c(
+      "Weekly total" = "weekly",
+      "Monthly total" = "monthly"
+    ),
+    c(
+      "6-minute rain" = "raw",
+      "Daily total" = "daily"
+    )
+  )
+}
+
+
+get_metric_view_choices <- function(window_unit = "days") {
+  switch(
+    window_unit,
+    days = c(
+      "Raw observations" = "raw",
+      "Daily aggregate" = "daily"
+    ),
+    months = c(
+      "Daily aggregate" = "daily",
+      "Weekly aggregate" = "weekly"
+    ),
+    years = c(
+      "Weekly aggregate" = "weekly",
+      "Monthly aggregate" = "monthly"
+    ),
+    c(
+      "Raw observations" = "raw",
+      "Daily aggregate" = "daily"
+    )
+  )
+}
+
+
+get_preferred_window_view <- function(window_unit = "days") {
+  switch(
+    window_unit,
+    days = "raw",
+    months = "weekly",
+    years = "monthly",
+    "raw"
+  )
+}
+
+
+get_summary_aggregate <- function(window_unit = "days") {
+  switch(
+    window_unit,
+    days = "daily",
+    months = "weekly",
+    years = "monthly",
+    "daily"
+  )
+}
+
+
+format_aggregate_label <- function(aggregate = c("raw", "daily", "weekly", "monthly")) {
+  aggregate <- match.arg(aggregate)
+
+  switch(
+    aggregate,
+    raw = "Raw",
+    daily = "Daily",
+    weekly = "Weekly",
+    monthly = "Monthly"
+  )
+}
+
+
+format_aggregate_period_label <- function(aggregate = c("daily", "weekly", "monthly")) {
+  aggregate <- match.arg(aggregate)
+
+  switch(
+    aggregate,
+    daily = "Day",
+    weekly = "Week",
+    monthly = "Month"
+  )
+}
+
 ui <- fluidPage(
   tags$head(
     tags$style(HTML("
@@ -188,7 +281,7 @@ ui <- fluidPage(
             class = "panel-card",
             h3(class = "panel-title", "Rain"),
             withSpinner(plotOutput("rainPlot", height = "420px")),
-            h4("Daily rain totals"),
+            h4(textOutput("summaryTitle", container = span)),
             tableOutput("dailySummary")
           )
         ),
@@ -223,11 +316,24 @@ server <- function(input, output, session) {
 
   observeEvent(input$window_unit, {
     settings <- get_window_slider_config(input$window_unit)
+    preferred_view <- get_preferred_window_view(input$window_unit)
+    rain_choices <- get_rain_view_choices(input$window_unit)
+    metric_choices <- get_metric_view_choices(input$window_unit)
     current_value <- if (is.null(input$window_amount)) {
       settings$value
     } else {
       as.integer(input$window_amount)
     }
+    current_rain_view <- if (!is.null(input$rain_view_mode) && input$rain_view_mode %in% rain_choices) {
+      input$rain_view_mode
+    } else {
+      preferred_view
+    }
+    current_metric_view <- if (!is.null(input$metric_view_mode) && input$metric_view_mode %in% metric_choices) {
+      input$metric_view_mode
+    } else {
+      preferred_view
+    }
 
     updateSliderInput(
       session = session,
@@ -238,6 +344,20 @@ server <- function(input, output, session) {
       value = min(max(current_value, 1L), settings$max),
       step = 1
     )
+
+    updateRadioButtons(
+      session = session,
+      inputId = "rain_view_mode",
+      choices = rain_choices,
+      selected = current_rain_view
+    )
+
+    updateRadioButtons(
+      session = session,
+      inputId = "metric_view_mode",
+      choices = metric_choices,
+      selected = current_metric_view
+    )
   }, ignoreInit = TRUE)
 
   selected_start_date <- reactive({
@@ -259,6 +379,10 @@ server <- function(input, output, session) {
     as.integer(Sys.Date() - selected_start_date()) + 1L
   })
 
+  selected_summary_aggregate <- reactive({
+    get_summary_aggregate(input$window_unit)
+  })
+
   cached_rain <- reactive({
     data_version()
 
@@ -291,7 +415,7 @@ server <- function(input, output, session) {
       location_id = input$location_id,
       start_date = selected_start_date(),
       end_date = Sys.Date(),
-      aggregate = "daily",
+      aggregate = selected_summary_aggregate(),
       db_path = db_path
     )
   })
@@ -310,6 +434,13 @@ server <- function(input, output, session) {
     selected_metric()$label
   })
 
+  output$summaryTitle <- renderText({
+    sprintf(
+      "%s rain totals",
+      format_aggregate_label(selected_summary_aggregate())
+    )
+  })
+
   output$syncControls <- renderUI({
     if (is.null(api_headers)) {
       return(
@@ -474,8 +605,9 @@ server <- function(input, output, session) {
       return(NULL)
     }
 
-    names(summary_data) <- c("Station ID", "Station", "Day", "Metric", "Label", "Unit", "Rain")
-    summary_data[, c("Station", "Day", "Rain")]
+    period_label <- format_aggregate_period_label(selected_summary_aggregate())
+    names(summary_data) <- c("Station ID", "Station", period_label, "Metric", "Label", "Unit", "Rain")
+    summary_data[, c("Station", period_label, "Rain")]
   }, striped = TRUE, spacing = "s", digits = 2)
 
   output$metricLatest <- renderTable({

getCloseStat.R

@@ -9,40 +9,32 @@ headers.default <- add_headers(
   apikey = token4
 )
 
-getAllFromCoord <- function(coord,start_date,end_date,allstations,N=3,headers,base="https://public-api.meteofrance.fr"){
-    three_station=getIdFromCoords(coord,allstations,N=N)
-    alldata=lapply(three_station$Id_station,function(statid){
-                       print(paste("recuperer station",statid))
-                       allstat=tryCatch(getStationData(start_date=start_date,end_date=end_date,station_id=statid,headers=headers,base=base),
-                                error=function(e){print(e);NULL})
-                       print(paste("done, sleep 5 sec"))
-                       print(dim(allstat))
-                       Sys.sleep(5)
-                       return(allstat)
-    })
-   alldata= do.call("rbind.data.frame",alldata)
-   ids=three_station$Nom_usuel
-   names(ids)=three_station$Id_station
-   cbind.data.frame(alldata,Nom_usuel=ids[as.character(alldata[,1])])
-}
-
 
 allstations=read.csv("allstations.csv") #get all station
 lacouch.coor <- c(45.3722971,5.6387118)
-lamure.coor <- c(44.9167, 5.8000)
 vignass.coor=c(44.8550665,5.8441789)
 foreve=list()
-for(y in 0:3){
-    enddate=format(Sys.Date()-y*365, "%Y-%m-%d")
-    startdate=format(Sys.Date()-(y*365+364), "%Y-%m-%d")
-    foreve=tryCatch(getAllFromCoord(vignass.coor,startdate,enddate,allstations,headers=headers.default,base="https://public-api.meteofrance.fr"),error=function(e)e)
+n_periods = 10
+for (i in 0:(n_periods-1)) {
+    enddate = format(Sys.Date() - i * 182, "%Y-%m-%d") # Approximately 6 months, adjust if you need more precision
+    # Calculate start date for the period, 182 days before the end date
+    startdate = format(Sys.Date() - (i * 182 + 181), "%Y-%m-%d") # Approximately 6 months
+    foreve[[paste0("per",i)]]=tryCatch(getAllFromCoord(lacouch.coor,startdate,enddate,allstations,headers=headers.default,base="https://public-api.meteofrance.fr"),error=function(e)e)
 }
 
-test1=getAllFromCoord(vignass.coor,start_date=startdate,end_date=enddate,allstations,headers=headers.default)
+startdate = format(Sys.Date() - 5, "%Y-%m-%d")
+test1=getAllFromCoord(vignass.coor,start_date=startdate,end_date= format(Sys.Date() , "%Y-%m-%d"),allstations,headers=headers.default)
+#test1=do.call("rbind.data.frame",foreve)
 
+#test1=read.csv("allvignass.csv")[,-1]
 cols=palette.colors()[1:length(unique(test1$Nom_usuel))]
 names(cols)=unique(test1$Nom_usuel)
-plot(getDate(test1[,2]),test1[,3],pch=20,col=cols[test1$Nom_usuel],cex=2)
+testsep=test1#[test1[,3]>0,]
+testsep[testsep[,3]==0,c(3,4)]=NA
+testsep=testsep[getDate(testsep[,2])>a,]
+plot(getDate(testsep[,2]),testsep[,3],pch=20,col=adjustcolor(cols[testsep$Nom_usuel],.4),cex=1.3,ylim=c(0,8))
 legend("topleft",col=cols,legend=names(cols),pch=20,cex=2)
 
+
 abline(v=as.numeric(as.POSIXlt("2024-08-07",format="%Y-%m-%d")),lwd=3,col="red")
+#write.csv(file="allvignass.csv",test1)

meteofranceapi.R

@@ -8,6 +8,10 @@ headersLIM <- add_headers(
   accept = "*/*",
   apikey = token2
 )
+headersLIM2 <- add_headers(
+  accept = "*/*",
+  apikey = token
+)
 headersPaquet <- add_headers(
   accept = "*/*",
   apikey = yearTokenPaquer
@@ -21,19 +25,55 @@ start_date <- as.Date("2024-03-24")
 end_date <- as.Date("2024-08-14")
 station_id <- "38269004"
 
-allstations=getStations(headersLIM)
+#allstations=getStations(headersLIM)
 allstations=read.csv("allstations.csv")
 #write.csv(file="allstations.csv",allstation,row.names=F)
-lacouch=c(45.3722971,5.6387118)
+lacouch.coor <- c(45.3722971,5.6387118)
+lamure.coor <- c(44.9167, 5.8000)
 
-alldist=dist(rbind(lacouch,cbind(allstations$Latitude,allstations$Longitude)))
-staupre=allstation$Id_station[which.min(as.matrix(alldist)[1,-1])]
-lamure=38269004
+vignass.coor=c(44.8550665,5.8441789)
 
-allmure=getStationData(start_date="2024-06-24",end_date="2024-08-14",station_id=lamure,headers=headersLIM)
+lamure.id=38269004
+lavaldens.id=38269004
+lacouch.id=38269004
+lacouch.stats=getIdFromCoords(lacouch.coor,allstations,N=3)
+lamure.stats=getIdFromCoords(lamure.coor,allstations,N=3)
+vignass.stats=getIdFromCoords(vignass.coor,allstations,N=3)
+
+allvignass=getStationData(start_date="2024-06-24",end_date="2024-08-14",station_id=lamure.id,headers=headersLIM)
 alllavaldens=getStationData(start_date="2024-06-24",end_date="2024-08-14",station_id=38207001,headers=headersLIM)
+allcouch=getStationData(start_date="2024-06-24",end_date="2024-08-16",station_id=lacouch.id,headers=headersLIM)
+
+today=format(Sys.Date(), "%Y-%m-%d")
+monthan=format(Sys.Date()-45, "%Y-%m-%d")
+
+allvignass=lapply(vignass.stats$Id_station,function(statid){print(paste("recuperer station",statid));Sys.sleep(5);getStationData(start_date=monthan,end_date=today,station_id=statid,headers=headersLIM);paste("done, sleep 5 sec");Sys.sleep(5)})
+
+alllacouch=lapply(lacouch.stats$Id_station,function(statid){print(paste("recuperer station",statid));statdat=tryCatch(getStationData(start_date=monthan,end_date=today,station_id=statid,headers=headersLIM),error=function(e)NULL);paste("done, sleep 5 sec");Sys.sleep(5);statdat})
+
+getStationData(start_date="2024-07-04",end_date="2024-08-16",station_id=vignass.stats$Id_station[3],headers=headersLIM)
+
+allvignass.df=do.call("rbind.data.frame",allvignass)
+allvignass.df=allvignass.df[allvignass.df[,3]>0,]
+ids=vignass.stats$Nom_usuel
+names(ids)=vignass.stats$Id_station
+cols=palette.colors()[1:nrow(vignass.stats)]
+names(cols)=vignass.stats$Id_station
+plot(getDate(allvignass.df[,2]),allvignass.df[,3],pch=20,col=cols[as.character(allvignass.df[,1])],cex=2)
+legend("topleft",col=cols,legend=ids[names(cols)],pch=20,cex=2)
+
+abline(v=as.numeric(as.POSIXlt("2024-08-07",format="%Y-%m-%d")),lwd=3,col="red")
+
+alllacouch=do.call("rbind.data.frame",alllacouch)
+alllacouch=alllacouch[alllacouch[,3]>0,]
+ids=lacouch.stats$Nom_usuel
+names(ids)=lacouch.stats$Id_station
+cols=palette.colors()[1:nrow(lacouch.stats)]
+names(cols)=lacouch.stats$Id_station
+plot(getDate(alllacouch[,2]),alllacouch[,3],pch=20,col=cols[as.character(alllacouch[,1])],cex=2)
+legend("topleft",col=cols,legend=ids[names(cols)],pch=20,cex=2)
+
 paquetLamure=getStationPaquet(id_station=lamure)
-allcouch=getStationData(start_date="2024-06-24",end_date="2024-08-14",station_id=staupre,headers=headersLIM)
 allal=getStationData(start_date="2024-01-24",end_date="2024-08-13",station_id=staupre,headers=headersLIM)
 
 plot(getDate(alllavaldens[,2]),alllavaldens[,3],lwd=3,type="l",col="red")

scripts/backfill_two_years.R

@@ -158,7 +158,7 @@ for (location_index in seq_along(location_ids)) {
     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),
+      start_date = "",
       end_date = as.character(end_date),
       chunk_days = chunk_days,
       rows_fetched = 0L,

tests/testthat/test-rain-db.R

@@ -162,7 +162,11 @@ test_that("SQLite cache upserts and queries multiple metrics from one dataset",
   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"
+    "2023-12-21"
+  )
+  expect_equal(
+    as.character(get_sync_start_date("vignasses", db_path, end_date = as.Date("2024-01-01"))),
+    "2023-12-12"
   )
 })
 
@@ -178,3 +182,48 @@ test_that("sync ranges are chunked predictably", {
   expect_equal(as.character(ranges$start_date[1]), "2024-01-01")
   expect_equal(as.character(ranges$end_date[3]), "2024-05-15")
 })
+
+
+test_that("weekly and monthly rainfall aggregates collapse long windows", {
+  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", "1001"),
+      DATE = c(202401020000, 202401090000, 202402010000),
+      RR6 = c(1, 2, 3),
+      Nom_usuel = c("Station A", "Station A", "Station A"),
+      stringsAsFactors = FALSE
+    ),
+    location_id = "vignasses",
+    fetched_at = as.POSIXct("2026-04-08 09:00:00", tz = "UTC")
+  )
+
+  expect_equal(upsert_weather_measurements(rain_rows, db_path), 3)
+
+  weekly_rain <- query_cached_rainfall(
+    location_id = "vignasses",
+    start_date = "2024-01-01",
+    end_date = "2024-02-01",
+    aggregate = "weekly",
+    db_path = db_path
+  )
+
+  monthly_rain <- query_cached_rainfall(
+    location_id = "vignasses",
+    start_date = "2024-01-01",
+    end_date = "2024-02-01",
+    aggregate = "monthly",
+    db_path = db_path
+  )
+
+  expect_equal(as.character(weekly_rain$observed_day), c("2024-01-01", "2024-01-08", "2024-01-29"))
+  expect_equal(weekly_rain$rain_mm, c(1, 2, 3))
+  expect_equal(as.character(monthly_rain$observed_day), c("2024-01-01", "2024-02-01"))
+  expect_equal(monthly_rain$rain_mm, c(3, 3))
+})