---
title: "Build the model + dataset registry (`dataset`, `model`, `taxon_model`, `listing`)"
msens:
target_name: build_registry
workflow_type: schema
dependency: [merge_taxon] # reads model_{ds}.csv (ingests, via merge chain) + merge.duckdb taxon
output: data/manifests/build_registry.json
editor_options:
chunk_output_type: console
---
Assemble the **`dataset`** and **`model`** registry tables that the v8 ingests define
*implicitly* — dataset properties in each `ingest_*.qmd` `msens:` frontmatter block, model
properties in each `{dist}/model_{ds}.csv`. v7 wrote these tables inline in every ingest
(`INSERT INTO dataset/model`); the v8 Parquet rewrite dropped those inserts, so the registry
was never consolidated. Building it **once here** (rather than per-ingest) makes it a
first-class target that **fails loudly if an ingested dataset is missing its metadata** — the
single source of truth for STAC, the marine-atlas release, apps and the API.
Also carries across the two **relational** tables the merge stage already built into
`merge.duckdb` but which nothing forwarded — `taxon_model` (taxon ↔ contributing models) and
`listing` (the federal MMPA/MBTA/ESA lookup). v3–v7 published both; without this v8 was the
only release whose published tables could not answer *which models fed this taxon*.
Those edges also settle **which datasets actually fed the scores**: `dataset.is_scored` is
introspected from `taxon_model`, not declared. `gm` + `nc` are ingested to `dist/` but their
density (#/km²) is not yet on the [0,100] suitability scale, so they are excluded from the
merge and contribute to no score — without the flag the registry reads as 11 input datasets
where 8 produced the numbers.
Written to `sdm.duckdb` (`dataset`, `model`, `taxon_model`, `listing`) and staged as Parquet
for the release.
## Design
```{mermaid}
%%| label: fig-design
%%| fig-cap: "Registry consolidation: dataset props from front-matter, model props from per-ingest CSVs + merged taxa, relations from the merge stage"
flowchart LR
fm["ingest_*.qmd<br/>msens: dataset: blocks"] --> ds["dataset table"]
csv["dist/model_{ds}.csv<br/>(per-ingest)"] --> mdl["model table"]
tax["merge.duckdb taxon<br/>(ms_merge models)"] --> mdl
rel["merge.duckdb<br/>taxon_model + listing"] --> rl["taxon_model + listing"]
ds --> db[("sdm.duckdb + registry/*.parquet<br/>V2/zstd")]
mdl --> db
rl --> db
db --> mf["hash_query() of all four<br/>→ content-addressed manifest"]
```
## Setup
```{r}
#| label: setup
librarian::shelf(DBI, dplyr, duckdb, fs, glue, here, jsonlite, logger, purrr, readr,
stringr, tibble, yaml, quiet = T)
source(here("libs/paths.R"))
options(readr.show_col_types = F)
if (!exists("%||%")) `%||%` <- function(a, b) if (is.null(a)) b else a
dir_atlas <- glue("{dir_big_v}/marine-atlas")
dir_dist <- glue("{dir_atlas}/dist")
merge_db <- glue("{dir_atlas}/merge.duckdb")
manifest <- here("data/manifests/build_registry.json"); dir_create(path_dir(manifest))
stopifnot("run ingests first" = dir_exists(dir_dist), "run build_cell_grid first" = file_exists(sdm_db))
# canonical dataset order (merged first, then contributing datasets)
ds_order <- c("ms_merge","ax","am","ca_nmfs","ch_nmfs","ch_fws","rng_fws","bl","rng_iucn",
"rng_turtle_swot_dps","gm","nc")
```
## `dataset` — from the `msens:` frontmatter (+ v7 metadata where keys match)
```{r}
#| label: dataset
# one row per ingest that declares a `msens: dataset:` block (the authoritative v8 props)
fm_dataset <- function(f) {
txt <- readLines(f, warn = FALSE)
d <- which(txt == "---")
if (length(d) < 2) return(NULL)
m <- tryCatch(yaml.load(paste(txt[(d[1] + 1):(d[2] - 1)], collapse = "\n")), error = \(e) NULL)
ds <- m$msens$dataset
if (is.null(ds$ds_key)) return(NULL)
# the rich metadata (name/description/citation/links/value_info/...) is OPTIONAL in the block:
# a dataset that v7 also had inherits it below; a NEW dataset (ax) has nowhere else to declare
# it, and without it the docs would print a glue'd fallback label and no citation
opt <- function(k) { v <- ds[[k]]; if (is.null(v)) NA else v }
tibble(ds_key = ds$ds_key, response_type = ds$response_type %||% NA,
source_broad = ds$source_authority %||% NA, temporal_res = ds$temporal_interval %||% "static",
native_format = ds$native_format %||% NA, qmd = path_file(f),
fm_name_short = opt("name_short"), fm_name_display = opt("name_display"),
fm_description = opt("description"), fm_citation = opt("citation"),
fm_link_info = opt("link_info"), fm_link_download = opt("link_download"),
fm_link_metadata = opt("link_metadata"), fm_value_info = opt("value_info"),
fm_regions = opt("regions"), fm_taxa_groups = opt("taxa_groups"),
fm_is_mask = as.logical(opt("is_mask")), fm_env_start = as.character(opt("env_start")),
fm_env_end = as.character(opt("env_end")),
# the native raster IS the analysis grid (AquaX): its two representations are "as
# delivered" vs "as ingested", not original vs interpolated -- the app labels them so
fm_on_grid = as.logical(opt("on_grid")))
}
ds_fm <- map_dfr(dir_ls(".", glob = "ingest_*.qmd"), fm_dataset)
# the merged product is its own "dataset"
ds_fm <- bind_rows(ds_fm, tibble(
ds_key = "ms_merge", response_type = "mixed", source_broad = "MarineSensitivity",
temporal_res = "static", native_format = "parquet", qmd = "merge_models.qmd"))
# inherit rich metadata (name_short/description/citation/links/env dates) from v7 for
# matching keys; v8 frontmatter props override. am was `am_0.05` in v7.
v7 <- dbConnect(duckdb(glue("{dir_big}/{ver_prev}/sdm.duckdb"), read_only = TRUE))
d_v7 <- dbGetQuery(v7, "SELECT * FROM dataset") |>
mutate(ds_key = if_else(ds_key == "am_0.05", "am", ds_key))
dbDisconnect(v7, shutdown = TRUE)
# front-matter values win over inherited ones where declared; a column the inherited table
# lacks is created from the front-matter alone
take <- function(d, col, fm) {
have <- if (col %in% names(d)) d[[col]] else rep(NA, nrow(d))
v <- coalesce(as.character(d[[fm]]), as.character(have))
if (is.logical(have) || col %in% c("is_mask", "on_grid")) as.logical(v) else v
}
# drop from the inherited table every column the front-matter / this notebook supplies -- v8's
# dataset table carries native_format / spatial_res_deg / is_scored / on_grid, which v7's did not,
# and a left_join would otherwise suffix both copies (.x/.y) and lose the column
dataset <- ds_fm |>
left_join(d_v7 |> select(-any_of(c("response_type", "source_broad", "temporal_res", "sort_order",
"native_format", "spatial_res_deg", "is_scored", "on_grid", "qmd"))),
by = "ds_key")
for (col in c("name_short","name_display","description","citation","link_info","link_download",
"link_metadata","value_info","regions","taxa_groups","is_mask","env_start","env_end","on_grid"))
dataset[[col]] <- take(dataset, col, paste0("fm_", col))
dataset <- dataset |>
select(-starts_with("fm_")) |>
mutate(
name_short = coalesce(name_short, glue("{source_broad} ({response_type})")),
description = coalesce(description, name_short),
is_mask = coalesce(is_mask, FALSE),
on_grid = coalesce(as.logical(on_grid), FALSE),
spatial_res_deg = 0.05,
sort_order = match(ds_key, ds_order)) |>
arrange(sort_order) |>
relocate(ds_key, name_short, response_type, source_broad, temporal_res, native_format, sort_order)
# datasets that were actually ingested (have a dist partition) MUST have a dataset row
ingested <- dir_ls(dir_dist, type = "directory", glob = "*dataset=*") |> path_file() |> str_remove("dataset=")
missing_meta <- setdiff(ingested, dataset$ds_key)
if (length(missing_meta))
stop("ingested datasets missing a `msens: dataset:` block: ", paste(missing_meta, collapse = ", "))
not_ingested <- setdiff(setdiff(dataset$ds_key, "ms_merge"), ingested)
if (length(not_ingested))
log_warn("dataset(s) declared but NOT ingested (no dist partition): {paste(not_ingested, collapse=', ')}")
knitr::kable(dataset |> select(ds_key, name_short, response_type, source_broad, temporal_res, native_format, sort_order))
```
## `model` — union of every `model_{ds}.csv` (+ merged `ms_merge` models)
```{r}
#| label: model
# each ingest writes model_{ds}.csv with dataset-native columns; map to a common schema
# keyed by the stable mdl_key.
read_model_csv <- function(f) {
ds <- str_match(path_file(f), "^model_(.+)\\.csv$")[, 2]
d <- read_csv(f) |> rename_with(tolower)
pick <- function(cands) { hit <- intersect(tolower(cands), names(d)); if (length(hit)) d[[hit[1]]] else NA }
tibble(
mdl_key = d$mdl_key,
ds_key = ds,
sp_id = as.character(pick(c("sp_key","sisid","sp_id","spcode","id_no","code"))),
sci_name = as.character(pick(c("taxa","sci_name","sciname","sciename","scientific_name"))),
common_name = as.character(pick(c("common","comname","common_name"))),
er_score = suppressWarnings(as.numeric(pick(c("er_score")))))
}
model_raw <- map_dfr(dir_ls(dir_dist, glob = "*/model_*.csv") |> c(dir_ls(dir_dist, glob = "model_*.csv")),
read_model_csv) |>
# a crosswalk may list species WITHOUT a surface (model_ax.csv keeps the 2,742 "absent in US"
# species with mdl_key = NA so they stay reviewable); a model row must have a key, else the
# registry publishes an NA id that poisons assign_mdl_id()'s max and the stability guard
filter(!is.na(mdl_key), nzchar(mdl_key)) |>
distinct(mdl_key, .keep_all = TRUE)
# merged per-taxon models (mdl_key = ms_merge_key) from the merge taxon
con <- dbConnect(duckdb(merge_db, read_only = TRUE))
model_merged <- dbGetQuery(con, "SELECT ms_merge_key AS mdl_key, scientific_name AS sci_name,
er_score, sp_cat FROM taxon") |>
transmute(mdl_key, ds_key = "ms_merge", sp_id = mdl_key, sci_name, common_name = NA_character_, er_score, sp_cat)
dbDisconnect(con, shutdown = TRUE)
model <- bind_rows(model_raw |> mutate(sp_cat = NA_character_), model_merged) |>
mutate(ds_key = factor(ds_key, levels = ds_order)) |> arrange(ds_key, mdl_key) |>
mutate(ds_key = as.character(ds_key))
# stable integer id per mdl_key: the compact partition key for the serving surface
# (serve/model_cell/mdl_id=*, so a titiler tile is a partition-pruned point read) and
# the mdl_key->mdl_id map the app/tile-URL builder joins on.
#
# This was `dense_rank(mdl_key)`, which makes the id a function of the model SET — so
# adding a model renumbers every model sorted after it. Ingesting gm + nc into v8's
# dist/ (registered, deliberately NOT merged) moved 45,499 of 80,261 ids; nothing would
# have failed, the registry and the published partitions would just disagree and titiler
# would serve the wrong species past ch_nmfs. So ids are assigned AGAINST the published
# registry when this version has already shipped: published keys keep their id, new keys
# append above the max (msens::assign_mdl_id()).
#
# Reachability is asserted, not tolerated: for a version listed in versions.json the
# published registry MUST be readable, else a transient network failure silently
# downgrades to renumber-everything. Only a version that has never shipped starts fresh.
is_published <- ver %in% msens::atlas_versions()$ver
published <- if (is_published) {
cx <- dbConnect(duckdb()); on.exit(dbDisconnect(cx, shutdown = TRUE), add = TRUE)
dbExecute(cx, "INSTALL httpfs; LOAD httpfs; SET s3_url_style='path'; SET s3_region='us-east-1';")
dbGetQuery(cx, glue(
"SELECT mdl_key, mdl_id FROM read_parquet('{msens::atlas_base_url()}/{ver}/tables/model.parquet')"))
} else NULL
model <- model |>
mutate(mdl_id = msens::assign_mdl_id(mdl_key, published)) |>
relocate(mdl_id, .after = mdl_key)
if (!is.null(published)) {
kept <- inner_join(published, select(model, mdl_key, new_id = mdl_id), by = "mdl_key")
stopifnot("published mdl_id renumbered — serve/model_cell partitions would be wrong" =
all(kept$mdl_id == kept$new_id))
log_info("mdl_id: {nrow(kept)} published ids preserved, ",
"{nrow(model) - nrow(kept)} new models appended above {max(published$mdl_id, na.rm = TRUE)}")
} else log_info("mdl_id: {nrow(model)} assigned fresh ({ver} not yet in versions.json)")
log_info("model registry: {nrow(model)} models ({n_distinct(model_raw$ds_key)} raw datasets + ms_merge)")
knitr::kable(model |> count(ds_key, name = "n_models"))
```
## `taxon_model` + `listing` — the relational tables the merge stage already built
```{r}
#| label: relational
# v3–v7 published `taxon_model` (which models fed a taxon) and `listing` (the federal
# MMPA/MBTA/ESA lookup). v8 BUILT both — merge_models_prep writes them to merge.duckdb —
# but nothing carried them into sdm.duckdb, so the release staged neither and the NEWEST
# release became the only one unable to answer "which models fed this taxon". They belong
# here rather than in the release notebook because `manifest_build()` introspects
# sdm.duckdb: a table staged straight to `tables/` is published but undiscoverable (the
# same trap the v1/v2 taxon_model reconstruction hit).
con <- dbConnect(duckdb(merge_db, read_only = TRUE))
taxon_model <- dbGetQuery(con, "SELECT mdl_key, taxon_authority, taxon_id, ms_merge_key FROM taxon_model")
listing <- dbGetQuery(con, "SELECT * FROM listing")
dbDisconnect(con, shutdown = TRUE)
# ds_key is derived from the key grammar, never re-split inline — mdl_key_parse() is the
# single source of truth for what a `|`-separated mdl_key means.
taxon_model <- taxon_model |>
mutate(ds_key = msens::mdl_key_parse(mdl_key)$dataset_key) |>
relocate(ds_key, .after = mdl_key) |>
arrange(ms_merge_key, ds_key, mdl_key)
# every edge must point at a model in the registry, else the relation is decorative
orphan <- setdiff(taxon_model$mdl_key, model$mdl_key)
stopifnot("taxon_model references mdl_keys absent from the model registry" = !length(orphan))
log_info("taxon_model: {nrow(taxon_model)} edges over {n_distinct(taxon_model$ms_merge_key)} taxa; ",
"listing: {nrow(listing)} rows")
# `is_scored` — registered is NOT used. gm + nc are ingested to dist/ but their density
# (#/km2) is not yet mapped onto the [0,100] suitability scale, so they are excluded from
# the merge and contribute to no score. Without this flag the docs (and anyone reading the
# registry) count 11 input datasets where 8 fed the numbers. Introspected from the edges
# rather than declared in front-matter, so it cannot drift from the merge.
dataset <- dataset |>
mutate(is_scored = msens::dataset_is_scored(ds_key, taxon_model$ds_key)) |>
relocate(is_scored, .after = is_mask)
log_info("datasets scored: {paste(dataset$ds_key[dataset$is_scored], collapse=', ')}; ",
"registered but unscored: {paste(dataset$ds_key[!dataset$is_scored], collapse=', ')}")
knitr::kable(taxon_model |> count(ds_key, name = "n_edges"))
```
## Write registry → `sdm.duckdb` + release Parquet
```{r}
#| label: write
con <- dbConnect(duckdb(sdm_db))
dbWriteTable(con, "dataset", dataset, overwrite = TRUE)
dbWriteTable(con, "model", model, overwrite = TRUE)
dbWriteTable(con, "taxon_model", taxon_model, overwrite = TRUE)
dbWriteTable(con, "listing", listing, overwrite = TRUE)
# stage as Parquet (V2/zstd/80MB row groups) alongside the other release tables
reg_dir <- glue("{dir_atlas}/registry"); dir_create(reg_dir)
for (t in c("dataset", "model", "taxon_model", "listing"))
msens::copy_atlas_parquet(con, t, glue("{reg_dir}/{t}.parquet"))
smry <- dbGetQuery(con, "SELECT (SELECT count(*) FROM dataset) n_datasets, (SELECT count(*) FROM model) n_models,
(SELECT count(*) FROM taxon_model) n_taxon_model, (SELECT count(*) FROM listing) n_listing")
# content fingerprints of all four registry tables (fold into one manifest hash)
reg_hash <- paste0(msens::hash_query(con, "dataset"), msens::hash_query(con, "model"),
msens::hash_query(con, "taxon_model"), msens::hash_query(con, "listing"))
dbDisconnect(con, shutdown = TRUE)
msens::report_table(smry, caption = "registry counts")
```
## Manifest
```{r}
#| label: manifest
# content-addressed: fingerprint of dataset+model tables (sorted lists for determinism)
msens::write_manifest(
manifest, target = "build_registry", content_hash = reg_hash,
stats = list(ver = ver, n_datasets = smry$n_datasets, n_models = smry$n_models,
n_taxon_model = smry$n_taxon_model, n_listing = smry$n_listing,
ingested = paste(sort(ingested), collapse = ","),
not_ingested = if (length(not_ingested)) paste(sort(not_ingested), collapse = ",") else ""),
force = msens::force_target("build_registry"))
```