| import os |
|
|
| filepath = "/Users/alexandrudumitrescu/Documents/clima/2026/ssod/server.R" |
|
|
| with open(filepath, "r") as f: |
| lines = f.readlines() |
|
|
| start_idx = -1 |
| end_idx = -1 |
|
|
| for i, line in enumerate(lines): |
| if "# Stage 3: Init" in line: |
| start_idx = i |
| if "output$station_ready <- reactive({" in line: |
| end_idx = i |
| break |
|
|
| if start_idx != -1 and end_idx != -1: |
| new_fetch_logic = """ # --- Async Fetch Machine --- |
| |
| # State variables for year-by-year fetching |
| years_to_fetch <- reactiveVal(numeric(0)) |
| fetched_data <- reactiveVal(list()) |
| current_year_idx <- reactiveVal(1) |
| |
| # Stage 3: Init Fetching |
| observe({ |
| req(fetch_stage() == 3) |
| station_id <- current_station_id() |
| if (is.null(station_id)) { |
| reset_fetch() |
| return() |
| } |
| |
| current_start <- isolate(input$date_range[1]) |
| current_end <- isolate(input$date_range[2]) |
| |
| if (is.null(current_start) || is.null(current_end)) { |
| reset_fetch("Invalid date range.") |
| return() |
| } |
| |
| start_yr <- as.integer(format(as.Date(current_start), "%Y")) |
| end_yr <- as.integer(format(as.Date(current_end), "%Y")) |
| years <- seq(start_yr, end_yr) |
| |
| years_to_fetch(years) |
| fetched_data(list()) |
| current_year_idx(1) |
| fetch_retry_count(0) |
| |
| fetch_message(paste("Checking availability for", length(years), "years...")) |
| |
| this_token <- current_fetch_token() |
| later::later( |
| function() { |
| if (identical(isolate(current_fetch_token()), this_token)) { |
| fetch_stage(4) |
| } |
| }, |
| 0.1 |
| ) |
| fetch_stage(-1) |
| }) |
| |
| # Stage 4: Fetch Loop |
| observe({ |
| req(fetch_stage() == 4) |
| station_id <- current_station_id() |
| if (is.null(station_id)) { |
| reset_fetch() |
| return() |
| } |
| |
| idx <- current_year_idx() |
| years <- years_to_fetch() |
| |
| if (idx > length(years)) { |
| # Transition to finalize stage |
| fetch_stage(6) |
| return() |
| } |
| |
| year <- years[idx] |
| fetch_message(paste0("Fetching data for ", year, " (", idx, "/", length(years), ")...")) |
| session$sendCustomMessage("freezeUI", list(text = paste0("Fetching data for ", year, " (", idx, "/", length(years), ")..."), station = current_station_label())) |
| |
| res <- fetch_ssod_year_mem(station_id, year) |
| |
| if (!is.null(res$error)) { |
| if (res$error == "transient") { |
| if (fetch_retry_count() < 3) { |
| fetch_retry_count(fetch_retry_count() + 1) |
| fetch_message(paste("Network drop. Retrying year", year, "- Attempt", fetch_retry_count())) |
| this_token <- current_fetch_token() |
| later::later(function() { |
| if (identical(isolate(current_fetch_token()), this_token)) { |
| fetch_stage(4) |
| } |
| }, 1.0) |
| fetch_stage(-1) |
| return() |
| } else { |
| reset_fetch(paste("Connection failed after retries for year", year)) |
| return() |
| } |
| } else { |
| reset_fetch(paste("Fatal error fetching year", year, ":", res$message)) |
| return() |
| } |
| } |
| |
| if (res$status == 200) { |
| # Parse data |
| tmp <- tempfile(fileext=".csv") |
| writeBin(res$content, tmp) |
| |
| tryCatch({ |
| df <- parse_ssod_data(tmp) |
| if (!is.null(df) && nrow(df) > 0) { |
| cur_data <- fetched_data() |
| cur_data[[as.character(year)]] <- df |
| fetched_data(cur_data) |
| } |
| }, error = function(e) { |
| # Ignore parse errors for a single year and continue |
| }) |
| unlink(tmp) |
| } |
| |
| # Success or 404, move to next year |
| fetch_retry_count(0) |
| current_year_idx(idx + 1) |
| |
| this_token <- current_fetch_token() |
| later::later(function() { |
| if (identical(isolate(current_fetch_token()), this_token)) { |
| fetch_stage(4) |
| } |
| }, 0.1) |
| fetch_stage(-1) |
| }) |
| |
| # Stage 6: Combine & Finalize |
| observe({ |
| req(fetch_stage() == 6) |
| dfs <- fetched_data() |
| |
| if (length(dfs) > 0) { |
| df_combined <- bind_rows(dfs) %>% arrange(datetime) |
| full_station_data(df_combined) |
| |
| # Auto-adjust date range if current range has no data |
| current_start <- isolate(input$date_range[1]) |
| current_end <- isolate(input$date_range[2]) |
| if (!is.null(current_start) && !is.null(current_end)) { |
| actual_max_date <- as.Date(max(df_combined$datetime, na.rm = TRUE)) |
| actual_min_date <- as.Date(min(df_combined$datetime, na.rm = TRUE)) |
| |
| if (as.Date(current_start) > actual_max_date || as.Date(current_end) < actual_min_date) { |
| new_end <- actual_max_date |
| new_start <- max(actual_min_date, new_end - 731) |
| updateDateRangeInput(session, "date_range", start = new_start, end = new_end) |
| prev_start_date(new_start) |
| prev_end_date(new_end) |
| } |
| } |
| |
| mem_size <- format(object.size(df_combined), units = "Mb") |
| final_msg <- paste0("Processing Complete!<br><span style='font-size: 0.9em; color: #555;'>In-Memory Data: <b>", mem_size, "</b></span>") |
| fetch_message(final_msg) |
| |
| loading_station(FALSE) |
| fetch_stage(0) |
| |
| this_token <- current_fetch_token() |
| session$sendCustomMessage("freezeUI", list(text = "Rendering plots...", station = current_station_label())) |
| updateNavbarPage(session, "main_nav", selected = "Dashboard") |
| |
| session$sendCustomMessage("freezeUI", list(text = "Rendering plots...", allowCancel = TRUE)) |
| session$onFlushed(function() { |
| if (identical(isolate(current_fetch_token()), this_token)) { |
| session$sendCustomMessage("waitForPlots", list()) |
| } |
| }, once = TRUE) |
| |
| } else { |
| reset_fetch("No valid records found in the selected date range.") |
| } |
| fetch_stage(-1) |
| }) |
| |
| """ |
| new_lines = lines[:start_idx] + [new_fetch_logic] + lines[end_idx:] |
| with open(filepath, "w") as f: |
| f.writelines(new_lines) |
| print("Successfully replaced fetch logic!") |
| else: |
| print(f"Could not find start/end indices. Start: {start_idx}, End: {end_idx}") |
|
|
