fix: normalize pipeline inputs and validate keys

NEWS.md

@@ -33,6 +33,9 @@
 - Added the experimental `snake_case` workflow: `new_alert_inputs()`,
   `fetch_alert_inputs()`, `fetch_alert_parameters()`, `run_alert_pipeline()`,
   `as_alert_history()`, and `write_alert_results()`.
+- Standardized new pipeline inputs on `municipio_geocodigo` and validate case,
+  climate, and parameter keys before joins; legacy input aliases are normalized
+  at the API boundary.
 - Added focused readers and computational names including `fetch_cases()`,
   `fetch_case_records()`, `fetch_climate()`, `estimate_rt()`,
   `nowcast_cases()`, `define_alert_rules()`, and `classify_alerts()`.

R/api.R

@@ -79,16 +79,17 @@
     stop("`population` must contain a geocode and population column.", call. = FALSE)
   }
   normalized <- data.frame(
-    cidade = as.numeric(population[[code[[1]]]]),
+    municipio_geocodigo = as.numeric(population[[code[[1]]]]),
     pop = as.numeric(population[[value[[1]]]])
   )
   if ("year" %in% names(population)) {
     normalized$year <- as.integer(population$year)
-    normalized <- normalized[order(normalized$cidade, normalized$year), , drop = FALSE]
-    normalized <- normalized[!duplicated(normalized$cidade, fromLast = TRUE), , drop = FALSE]
+    normalized <- normalized[order(normalized$municipio_geocodigo, normalized$year), , drop = FALSE]
+    normalized <- normalized[!duplicated(normalized$municipio_geocodigo, fromLast = TRUE), , drop = FALSE]
     normalized$year <- NULL
   }
-  if (anyNA(normalized) || any(normalized$pop <= 0) || anyDuplicated(normalized$cidade)) {
+  if (anyNA(normalized) || any(normalized$pop <= 0) ||
+      anyDuplicated(normalized$municipio_geocodigo)) {
     stop("`population` must have one positive value per geocode.", call. = FALSE)
   }
   normalized
@@ -97,10 +98,13 @@
 #' Construct validated inputs for the alert pipeline
 #'
 #' `cases` has one row per municipality, disease, and epidemiological week and
-#' must contain `SE`, `cidade`, `CID10`, `casos`, `cas_prov`, `cas_lab`,
+#' must contain `SE`, `municipio_geocodigo`, `CID10`, `casos`, `cas_prov`, `cas_lab`,
 #' `localidade`, and `nome`. It must also contain `pop`, unless `population` is
 #' supplied. `climate` has one row per municipality and week, keyed by
-#' `geocodigo` and `SE`. Rows are ordered by municipality and week.
+#' `municipio_geocodigo` and `SE`. The legacy input aliases `cidade` and
+#' `geocodigo` are accepted and normalized to `municipio_geocodigo`. Duplicate
+#' case and climate keys are rejected before computation. Rows are ordered by
+#' municipality, disease, and week.
 #'
 #' @param cases Weekly case data frame.
 #' @param climate Weekly climate data frame.
@@ -116,18 +120,16 @@
 #' @export
 new_alert_inputs <- function(cases, climate, population = NULL,
                              case_records = NULL, metadata = list()) {
+  canonical <- .canonicalize_pipeline_geocodes(cases, climate)
+  cases <- canonical$cases
+  climate <- canonical$climate
   .validate_columns(
     cases,
-    c("SE", "cidade", "CID10", "casos", "cas_prov", "cas_lab", "localidade", "nome"),
+    c("SE", "municipio_geocodigo", "CID10", "casos", "cas_prov", "cas_lab",
+      "localidade", "nome"),
     "cases"
   )
-  .validate_columns(climate, c("SE", "geocodigo"), "climate")
-  if (!is.numeric(cases$SE) || !is.numeric(cases$cidade) ||
-      !is.numeric(climate$SE) || !is.numeric(climate$geocodigo) ||
-      !all(vapply(cases[c("casos", "cas_prov", "cas_lab")], is.numeric, logical(1)))) {
-    stop("Case and climate keys and counts must be numeric.", call. = FALSE)
-  }
-  .validate_epiweek(c(cases$SE, climate$SE))
+  .validate_pipeline_inputs(cases, climate, require_population = FALSE)
   if (!is.null(case_records) && !is.data.frame(case_records)) {
     stop("`case_records` must be NULL or a data frame.", call. = FALSE)
   }
@@ -138,21 +140,13 @@ new_alert_inputs <- function(cases, climate, population = NULL,
   population <- .normalize_population(population)
   if (!"pop" %in% names(cases)) cases$pop <- NA_real_
   if (!is.null(population)) {
-    matched <- match(cases$cidade, population$cidade)
+    matched <- match(cases$municipio_geocodigo, population$municipio_geocodigo)
     replace <- is.na(cases$pop)
     cases$pop[replace] <- population$pop[matched[replace]]
   }
-  if (anyNA(cases$pop) || any(cases$pop <= 0)) {
-    stop("Every case row must have a positive population.", call. = FALSE)
-  }
-  if (anyDuplicated(cases[c("cidade", "CID10", "SE")])) {
-    stop("`cases` keys (`cidade`, `CID10`, `SE`) must be unique.", call. = FALSE)
-  }
-  if (anyDuplicated(climate[c("geocodigo", "SE")])) {
-    stop("`climate` keys (`geocodigo`, `SE`) must be unique.", call. = FALSE)
-  }
-  cases <- cases[order(cases$cidade, cases$CID10, cases$SE), , drop = FALSE]
-  climate <- climate[order(climate$geocodigo, climate$SE), , drop = FALSE]
+  .validate_pipeline_inputs(cases, climate)
+  cases <- cases[order(cases$municipio_geocodigo, cases$CID10, cases$SE), , drop = FALSE]
+  climate <- climate[order(climate$municipio_geocodigo, climate$SE), , drop = FALSE]
   rownames(cases) <- NULL
   rownames(climate) <- NULL
 
@@ -168,7 +162,7 @@ print.alerttools_inputs <- function(x, ...) {
   cat("<alerttools_inputs>\n")
   cat("  case rows:", nrow(x$cases), "\n")
   cat("  climate rows:", nrow(x$climate), "\n")
-  cat("  municipalities:", length(unique(x$cases$cidade)), "\n")
+  cat("  municipalities:", length(unique(x$cases$municipio_geocodigo)), "\n")
   invisible(x)
 }
 

R/core_pipeline.R

@@ -43,6 +43,91 @@
   invisible(data)
 }
 
+.canonicalize_municipality_key <- function(data, aliases, name) {
+  .validate_columns(data, character(), name)
+  if (anyDuplicated(names(data))) {
+    stop("`", name, "` must not contain duplicate column names.", call. = FALSE)
+  }
+  canonical <- "municipio_geocodigo"
+  present_aliases <- intersect(aliases, names(data))
+  if (canonical %in% names(data)) {
+    for (alias in present_aliases) {
+      if (!identical(as.character(data[[canonical]]), as.character(data[[alias]]))) {
+        stop("`", name, "` contains conflicting municipality identifiers.",
+             call. = FALSE)
+      }
+    }
+    data[present_aliases] <- NULL
+  } else {
+    if (length(present_aliases) != 1L) {
+      stop("`", name, "` must contain `municipio_geocodigo`.", call. = FALSE)
+    }
+    data[[canonical]] <- data[[present_aliases[[1L]]]]
+    data[present_aliases[[1L]]] <- NULL
+  }
+  data
+}
+
+.canonicalize_pipeline_geocodes <- function(cases, climate) {
+  list(
+    cases = .canonicalize_municipality_key(cases, "cidade", "cases"),
+    climate = .canonicalize_municipality_key(climate, "geocodigo", "climate")
+  )
+}
+
+.validate_unique_keys <- function(data, keys, name) {
+  if (anyNA(data[keys])) {
+    stop("`", name, "` keys must not contain missing values.", call. = FALSE)
+  }
+  if (anyDuplicated(data[keys])) {
+    stop("`", name, "` keys (", paste(keys, collapse = ", "),
+         ") must be unique.", call. = FALSE)
+  }
+  invisible(data)
+}
+
+.validate_pipeline_inputs <- function(cases, climate, parameters = NULL,
+                                      require_population = TRUE) {
+  case_columns <- c("SE", "municipio_geocodigo", "CID10", "casos", "cas_prov",
+                    "cas_lab", "localidade", "nome")
+  if (require_population) case_columns <- c(case_columns, "pop")
+  .validate_columns(cases, case_columns, "cases")
+  .validate_columns(climate, c("SE", "municipio_geocodigo"), "climate")
+  numeric_case_columns <- c("SE", "municipio_geocodigo", "casos", "cas_prov", "cas_lab")
+  if (!all(vapply(cases[numeric_case_columns], is.numeric, logical(1))) ||
+      !is.numeric(climate$SE) || !is.numeric(climate$municipio_geocodigo) ||
+      !is.character(cases$CID10) || anyNA(cases$CID10)) {
+    stop("Case and climate keys and counts must use valid numeric and disease types.",
+         call. = FALSE)
+  }
+  if (require_population &&
+      (!is.numeric(cases$pop) || anyNA(cases$pop) || any(cases$pop <= 0))) {
+    stop("Every case row must have a positive population.", call. = FALSE)
+  }
+  if (any(cases$casos < 0) || any(cases$cas_prov < 0) || any(cases$cas_lab < 0)) {
+    stop("Case counts must not be negative.", call. = FALSE)
+  }
+  .validate_epiweek(c(cases$SE, climate$SE))
+  .validate_unique_keys(cases, c("municipio_geocodigo", "CID10", "SE"), "cases")
+  .validate_unique_keys(climate, c("municipio_geocodigo", "SE"), "climate")
+
+  if (!is.null(parameters)) {
+    .validate_columns(
+      parameters,
+      c("municipio_geocodigo", "limiar_preseason", "limiar_epidemico",
+        "varcli", "clicrit", "cid10", "codmodelo"),
+      "parameters"
+    )
+    if (anyDuplicated(names(parameters)) || !is.numeric(parameters$municipio_geocodigo) ||
+        !is.character(parameters$cid10) || anyNA(parameters$cid10)) {
+      stop("`parameters` must have valid municipality and disease identifiers.",
+           call. = FALSE)
+    }
+    .validate_unique_keys(parameters, c("municipio_geocodigo", "cid10"), "parameters")
+  }
+  invisible(list(cases = cases, climate = climate, parameters = parameters))
+}
+
 #' Calculate incidence from in-memory vectors
 #'
 #' @param count Numeric vector of case counts.
@@ -105,8 +190,12 @@ new_alerttools_result <- function(data, alerts, parameters,
 #' This function performs no database or file access and does not depend on the
 #' current working directory.
 #'
-#' @param cases Weekly case data using the `getCases()` domain schema.
-#' @param climate Weekly municipal climate data with `SE` and `geocodigo`.
+#' @param cases Weekly case data keyed by `municipio_geocodigo`, disease, and
+#'   epidemiological week. The legacy `cidade` alias is accepted at the input
+#'   boundary.
+#' @param climate Weekly municipal climate data keyed by `SE` and
+#'   `municipio_geocodigo`. The legacy `geocodigo` alias is accepted at the input
+#'   boundary.
 #' @param parameters Alert parameters, one row per municipality and disease.
 #' @param case_records Optional individual notification records for Bayesian
 #'   nowcasting.
@@ -126,19 +215,16 @@ alerttools_pipeline <- function(cases, climate, parameters, case_records = NULL,
   if (!is.logical(verbose) || length(verbose) != 1L || is.na(verbose)) {
     stop("`verbose` must be TRUE or FALSE.", call. = FALSE)
   }
-  .validate_columns(
-    cases,
-    c("SE", "cidade", "CID10", "casos", "cas_prov", "cas_lab",
-      "localidade", "nome", "pop"),
-    "cases"
-  )
-  .validate_columns(climate, c("SE", "geocodigo"), "climate")
-  .validate_columns(
-    parameters,
-    c("municipio_geocodigo", "limiar_preseason", "limiar_epidemico",
-      "varcli", "clicrit", "cid10", "codmodelo"),
-    "parameters"
-  )
+  canonical <- .canonicalize_pipeline_geocodes(cases, climate)
+  cases <- canonical$cases
+  climate <- canonical$climate
+  .validate_pipeline_inputs(cases, climate, parameters)
+  # Keep the established calculation and result schema stable while accepting
+  # canonical identifiers at the public API boundary.
+  cases$cidade <- cases$municipio_geocodigo
+  cases$municipio_geocodigo <- NULL
+  climate$geocodigo <- climate$municipio_geocodigo
+  climate$municipio_geocodigo <- NULL
   if (length(report_week) != 1L || !is.numeric(report_week) || is.na(report_week)) {
     stop("`report_week` must be one epidemiological week.", call. = FALSE)
   }

docs/refactoring/api-migration.md

@@ -62,13 +62,18 @@ O argumento `disease` aceita nomes de domínio ou os códigos equivalentes:
 
 Campos: `cases`, `climate`, `population`, `case_records` e `metadata`.
 
-- `cases`: chave única `cidade`, `CID10`, `SE`; contagens `casos`, `cas_prov`,
+- `cases`: chave única `municipio_geocodigo`, `CID10`, `SE`; contagens `casos`, `cas_prov`,
   `cas_lab`; população positiva em `pop`; ordenação por cidade, doença e semana;
-- `climate`: chave única `geocodigo`, `SE`; ordenação por município e semana;
-- `population`: opcional, normalizada para `cidade`, `pop`;
+- `climate`: chave única `municipio_geocodigo`, `SE`; ordenação por município e semana;
+- `population`: opcional, normalizada para `municipio_geocodigo`, `pop`;
 - `case_records`: registros individuais opcionais para nowcasting;
 - `metadata`: lista nomeada com proveniência e período.
 
+Os aliases legados `cidade` em casos e `geocodigo` em clima são aceitos apenas na
+fronteira de entrada e normalizados para `municipio_geocodigo`. Os pontos de
+entrada públicos validam tipos, chaves duplicadas e cardinalidade antes dos
+joins.
+
 ### `alerttools_result`
 
 Campos: `data`, `alerts`, `parameters`, `metadata` e `diagnostics`. `data` e

tests/testthat/test-api.R

@@ -12,6 +12,10 @@ test_that("new_alert_inputs validates, normalizes, and orders public schemas", {
 
   expect_s3_class(inputs, "alerttools_inputs")
   expect_named(inputs, c("cases", "climate", "population", "case_records", "metadata"))
+  expect_true("municipio_geocodigo" %in% names(inputs$cases))
+  expect_false("cidade" %in% names(inputs$cases))
+  expect_true("municipio_geocodigo" %in% names(inputs$climate))
+  expect_false("geocodigo" %in% names(inputs$climate))
   expect_equal(inputs$cases$SE, sort(inputs$cases$SE))
   expect_equal(inputs$climate$SE, sort(inputs$climate$SE))
   expect_true(all(inputs$cases$pop == 10000))
@@ -22,6 +26,8 @@ test_that("new_alert_inputs validates, normalizes, and orders public schemas", {
   duplicate$pop <- 10000
   expect_error(new_alert_inputs(duplicate, climate), "keys.*unique")
   expect_error(new_alert_inputs(cases, climate), "positive population")
+  duplicate_climate <- rbind(climate, climate[1, ])
+  expect_error(new_alert_inputs(cases, duplicate_climate), "climate.*keys.*unique")
 })
 
 test_that("run_alert_pipeline is equivalent to the approved pure pipeline", {
@@ -41,6 +47,33 @@ test_that("run_alert_pipeline is equivalent to the approved pure pipeline", {
   expect_error(run_alert_pipeline(list(), parameters), "new_alert_inputs")
 })
 
+test_that("public computational entry validates canonical keys before joining", {
+  cases <- read_fixture("cases.csv")
+  climate <- read_fixture("climate.csv")
+  parameters <- read_fixture("parameters.csv", stringsAsFactors = FALSE)
+  names(cases)[names(cases) == "cidade"] <- "municipio_geocodigo"
+  names(climate)[names(climate) == "geocodigo"] <- "municipio_geocodigo"
+
+  canonical <- alerttools_pipeline(cases, climate, parameters, report_week = 202106)
+  expect_s3_class(canonical, "alerttools_result")
+
+  duplicate_cases <- rbind(cases, cases[1, ])
+  expect_error(
+    alerttools_pipeline(duplicate_cases, climate, parameters, report_week = 202106),
+    "cases.*keys.*unique"
+  )
+  duplicate_climate <- rbind(climate, climate[1, ])
+  expect_error(
+    alerttools_pipeline(cases, duplicate_climate, parameters, report_week = 202106),
+    "climate.*keys.*unique"
+  )
+  duplicate_parameters <- rbind(parameters, parameters[1, ])
+  expect_error(
+    alerttools_pipeline(cases, climate, duplicate_parameters, report_week = 202106),
+    "parameters.*keys.*unique"
+  )
+})
+
 test_that("database API fetches a complete, explicitly bounded input bundle", {
   connection <- sqlite_fixture_connection(environment())
   inputs <- fetch_alert_inputs(