workflows
Source: SIOPIC_DB_PNTtable_allRecords_9Feb2026.csv (~19 MB, 148,129 net tows), the parent-net-tow (PNT) registry of the SIO Pelagic Invertebrate Collection (PI: Rasmus Swalethorp, Ed Weber, Linsey Sala).
picExpedition_Type_pnt == "CalCOFI" (Q02). This replaced a CalCOFI-region bounding box, which was a proxy for the program and was wrong in both directions ā see below.zooplankton_tow; the biovolume measurement table is pending from the provider (blocker Q01) and will be added as zooplankton_measurement.graph LR
A[PNTtable.csv<br/>148,129 tows] --> B[filter Expedition_Type = CalCOFI]
B --> C[zooplankton_tow<br/>position + gear + keys]
C -.ship/cruise/grid.-> D[(shared refs)]
C -.biovolume pending.-> E[zooplankton_measurement]graph LR A[PNTtable.csv<br/>148,129 tows] --> B[filter Expedition_Type = CalCOFI] B --> C[zooplankton_tow<br/>position + gear + keys] C -.ship/cruise/grid.-> D[(shared refs)] C -.biovolume pending.-> E[zooplankton_measurement]
devtools::load_all(here::here("../calcofi4db"))
devtools::load_all(here::here("../calcofi4r"))
librarian::shelf(
CalCOFI/calcofi4db, CalCOFI/calcofi4r,
DBI, dplyr, DT, fs, glue, here, htmltools, janitor, jsonlite, knitr,
lubridate, purrr, readr, sf, stringr, tibble, tidyr, units,
quiet = T)
options(readr.show_col_types = F)
options(DT.options = list(scrollX = TRUE))
source(here("libs/ingest.R"))
cc <- read_calcofi_meta(here("ingest_sio_pic-zooplankton.qmd"))
provider <- cc$provider
dataset <- cc$dataset
dataset_name <- cc$dataset_meta$dataset_name
tables_owned <- cc$tables_owned
dir_label <- glue("{provider}_{dataset}")
dir_parquet <- here(glue("data/parquet/{dir_label}"))
dir_stage <- cc_stage_path("parquet", dir_label, create = TRUE)
db_path <- here(glue("data/wrangling/{dir_label}.duckdb"))
if (overwrite) {
if (file_exists(db_path)) file_delete(db_path)
if (file_exists(paste0(db_path, ".wal"))) file_delete(paste0(db_path, ".wal"))
if (dir_exists(paste0(db_path, ".tmp"))) dir_delete(paste0(db_path, ".tmp"))
}
dir_create(dirname(db_path))
con <- get_duckdb_con(db_path)
load_duckdb_extension(con, "spatial")Read directly from {dir_data}/calcofi/zooplankton/ (the file is not under a pic/ subdirectory). Columns renamed per metadata/sio/pic-zooplankton/flds_redefine.csv.
zoo_csv <- path_expand(glue(
"{dir_data}/calcofi/zooplankton/SIOPIC_DB_PNTtable_allRecords_9Feb2026.csv"))
stopifnot("zooplankton CSV not found" = file_exists(zoo_csv))
sync_to_gcs(
local_dir = path_dir(zoo_csv),
gcs_prefix = glue("archive/{provider}/{dataset}"),
bucket = "calcofi-files-public",
exclude = c(".DS_Store", "*.tmp", "*.gdoc"))# A tibble: 0 Ć 4
# ā¹ 4 variables: file <chr>, action <chr>, size <dbl>, reason <chr>d_raw <- read_csv(zoo_csv, col_types = cols(.default = "c")) # read all as text, cast below
cat(glue("Read {format(nrow(d_raw), big.mark=',')} tows, {ncol(d_raw)} columns"), "\n")Read 148,129 tows, 20 columns Canonical names from metadata/field_dictionary.csv. Datetimes are built from date (MM/DD/YYYY) + HHMM time (treated as given pending the timezone answer, Q03).
The scope filter is the program, Expedition_Type_pnt == "CalCOFI", not a CalCOFI-region bounding box. The flag is trustworthy: it agrees with "CALCOFI" appearing in EXPEDITION_pnt on all 148,129 source rows, with zero disagreement either way.
The bounding box it replaces was a proxy for the program, and it was wrong in both directions:
| tows | resolve to a CalCOFI cruise | |
|---|---|---|
| in bbox, not CalCOFI ā now dropped | 32,719 | 35.5% |
| CalCOFI, outside bbox ā now restored | 15,532 | 83.2% |
| CalCOFI and in bbox ā kept either way | 66,811 | 81.8% |
The 15,532 restored tows are not coordinate errors ā the CalCOFI subset contains zero missing, impossible or NaN coordinates, so Q04ās antimeridian problem lives entirely in the non-CalCOFI holdings. They are the historical grid: 1950sā80s inshore stations, the Gulf of California, and the Baja lines down to 19.9Ā°N. That they resolve to a CalCOFI cruise more often than the average tow (83.2%) is the evidence they belong.
The dropped tows (CWPA Squid Survey, RACER, SOSO, Huntington Beach, CCE-LTER process cruises ā¦) matter more than their count suggests: 11,613 of them carried a cruise_key, but that key is derived from ship + year-month, so a non-CalCOFI cruise sharing a vessel and month with a real CalCOFI cruise was being stamped with it. Dropping them removes false attributions, not just out-of-scope rows.
hhmm_to_dt <- function(date, hhmm) {
t <- str_pad(na_if(str_trim(hhmm), ""), 4, pad = "0")
out <- suppressWarnings(as_datetime(paste0(
date, " ", substr(t, 1, 2), ":", substr(t, 3, 4), ":00")))
out[is.na(t) & !is.na(date)] <- as_datetime(as.character(date[is.na(t) & !is.na(date)]))
out
}
d_clean <- d_raw |>
transmute(
expedition = EXPEDITION_pnt,
expedition_code = EXPED_CODE_pnt,
expedition_type = Expedition_Type_pnt,
ship_name = SHIP_pnt,
order_occ = suppressWarnings(as.integer(SWFOrder_Occ)),
line = suppressWarnings(as.numeric(STATION_LINE_pnt)),
station = suppressWarnings(as.numeric(STATION_NUMBER_pnt)),
latitude = suppressWarnings(as.numeric(LAT_DECIMAL_pnt)),
longitude = suppressWarnings(as.numeric(LONG_DECIMAL_pnt)),
date = suppressWarnings(as.Date(SAMPLE_DATE_pnt, format = "%m/%d/%Y")),
datetime_start_utc = hhmm_to_dt(date, START_TIME_pnt), # Q03: assumed UTC
datetime_end_utc = hhmm_to_dt(date, END_TIME_pnt),
depth_min_m = suppressWarnings(as.numeric(DEPTH_MIN_pnt)),
depth_max_m = suppressWarnings(as.numeric(DEPTH_MAX_pnt)),
max_wire_out_m = suppressWarnings(as.numeric(MAX_MWO_pnt)),
net_type = NET_TYPE_pnt,
mesh_size_mm = suppressWarnings(as.numeric(MESH_SIZE_pnt)),
tow_type = TOW_TYPE_pnt,
fixative = FIXATIVE_pnt,
preservative = PRESERVATIVE_pnt)
n_all <- nrow(d_clean)
# scope filter: the CalCOFI program (Q02), not a region. the coordinate guard is
# a validity check only -- it rejects impossible/NaN positions (Q04) rather than
# clipping the historical grid to the modern one.
d_cc <- d_clean |>
filter(expedition_type == "CalCOFI") |>
filter(!is.na(latitude), !is.na(longitude),
!is.nan(latitude), !is.nan(longitude),
abs(latitude) <= 90, abs(longitude) <= 180) |>
mutate(
site_key = if_else(
is.na(line) | is.na(station), NA_character_,
sprintf("%05.1f %05.1f", line, station)),
tow_id = row_number())
n_cc <- sum(d_clean$expedition_type == "CalCOFI", na.rm = TRUE)
n_badpos <- n_cc - nrow(d_cc)
cat(glue(
"CalCOFI-program filter: {format(n_cc, big.mark=',')}/",
"{format(n_all, big.mark=',')} tows are CalCOFI ",
"({round(100*n_cc/n_all, 1)}%); ",
"{n_badpos} dropped for an invalid position; ",
"{format(nrow(d_cc), big.mark=',')} retained"), "\n")CalCOFI-program filter: 82,343/148,129 tows are CalCOFI (55.6%); 0 dropped for an invalid position; 82,343 retained # the program flag must agree with the expedition name, or the scope is not what
# it claims -- this is the assumption the whole filter rests on
disagree <- d_clean |>
filter(xor(coalesce(expedition_type, "") == "CalCOFI",
str_detect(coalesce(expedition, ""), regex("calcofi", ignore_case = TRUE))))
stopifnot(
"Expedition_Type must agree with CALCOFI in the expedition name" = nrow(disagree) == 0,
"CalCOFI subset must have no invalid positions (Q04 is non-CalCOFI only)" = n_badpos == 0)
dbWriteTable(con, "zooplankton_tow", d_cc, overwrite = TRUE)load_prior_tables(
con, parquet_dir = cc_stage_path("parquet", "swfsc_ichthyo"),
tables = c("ship", "cruise", "grid"), geom_tables = c("grid"), as_view = TRUE)# A tibble: 3 Ć 3
table rows has_geom
<chr> <dbl> <lgl>
1 cruise 691 FALSE
2 grid 218 TRUE
3 ship 48 FALSE d_ship <- dbGetQuery(con, "SELECT ship_key, ship_name, ship_nodc FROM ship") |>
mutate(ship_name_norm = ship_name |> str_to_upper() |> str_squish())
zt <- dbGetQuery(con, "SELECT tow_id, ship_name, date FROM zooplankton_tow") |>
mutate(ship_name_norm = ship_name |>
str_replace("^R/?V\\.?\\s+", "") |> str_to_upper() |> str_squish()) |>
left_join(d_ship |> select(ship_name_norm, ship_key, ship_nodc), by = "ship_name_norm")
# cruise_key: the reference cruise whose date span holds the tow, else its month,
# keys not in `cruise` left NULL (calcofi4db::resolve_cruise_key). A tow in the
# first days of a month belongs to the cruise that began the month before ā the
# old month rule put it on the NEXT cruise of the same ship (30 of this dataset's
# consecutive-month cruise pairs ran into each other within 3 days).
dbWriteTable(con, "zt_keys", zt |> select(tow_id, ship_key, datetime = date), overwrite = TRUE)
resolve_cruise_key(con, "zt_keys", datetime_col = "datetime", require_in_cruise = TRUE) |>
datatable(caption = "cruise_key resolution method")zt <- zt |> left_join(dbGetQuery(con, "SELECT tow_id, cruise_key FROM zt_keys"), by = "tow_id")
dbExecute(con, "ALTER TABLE zooplankton_tow ADD COLUMN IF NOT EXISTS ship_key VARCHAR")[1] 0dbExecute(con, "ALTER TABLE zooplankton_tow ADD COLUMN IF NOT EXISTS cruise_key VARCHAR")[1] 0dbExecute(con,
"UPDATE zooplankton_tow t SET ship_key = k.ship_key, cruise_key = k.cruise_key
FROM zt_keys k WHERE t.tow_id = k.tow_id")[1] 82343dbExecute(con, "DROP TABLE zt_keys")[1] 0cat(glue(
"ship match: {sum(!is.na(zt$ship_key))}/{nrow(zt)} ",
"({round(100*mean(!is.na(zt$ship_key)),1)}%); ",
"cruise_key match: {sum(!is.na(zt$cruise_key))}/{nrow(zt)} ",
"({round(100*mean(!is.na(zt$cruise_key)),1)}%)"), "\n")ship match: 81033/82343 (98.4%); cruise_key match: 77256/82343 (93.8%) add_point_geom(con, "zooplankton_tow", lon_col = "longitude", lat_col = "latitude")
assign_grid_key(con, "zooplankton_tow") |> datatable(caption = "Grid assignment")d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here()))
dbWriteTable(con, "dataset", d_dataset, overwrite = TRUE)
cat(glue("dataset: {nrow(d_dataset)} datasets registered"), "\n")dataset: 16 datasets registered zoo_rels <- list(
primary_keys = list(zooplankton_tow = "tow_id"),
foreign_keys = list()) # cross-dataset FKs (cruise/ship/grid) in relationships_cross.csv
cc_erd(
con, tables = c("zooplankton_tow", "dataset"), rels = zoo_rels,
colors = list(lightblue = "zooplankton_tow", white = "dataset"))
# the SOURCE shape above documents the wrangling; relationships.json is written
# with the parquet outputs below, from core_relationships().results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")Validation: FAILED if (length(results$errors) > 0)
cat("Errors:\n", paste("-", results$errors, collapse = "\n"), "\n")Errors:
- Table 'zooplankton_tow' has 428 NULL values in required column 'site_key'
- Table 'zooplankton_tow' has 1310 NULL values in required column 'ship_key'
- Table 'zooplankton_tow' has 5087 NULL values in required column 'cruise_key'
- Table 'zooplankton_tow' has 409 NULL values in required column 'grid_key' # NULL cruise_key/ship_key/site_key are the expected unmatched remainder (Q03/Q05),
# the same accepted behavior as calcofi_dic; not a hard failure.
n_dup <- dbGetQuery(con,
"SELECT COUNT(*) FROM (SELECT tow_id, COUNT(*) n FROM zooplankton_tow GROUP BY tow_id HAVING COUNT(*)>1)")[[1]]
cat(glue("zooplankton_tow tow_id duplicates: {n_dup}"), "\n")zooplankton_tow tow_id duplicates: 0 cols <- dbGetQuery(con,
"SELECT column_name FROM information_schema.columns
WHERE table_name='zooplankton_tow' AND data_type NOT LIKE 'GEOMETRY%'")$column_name
dbGetQuery(con, glue("SELECT {paste(cols, collapse=', ')} FROM zooplankton_tow LIMIT 100")) |>
datatable(caption = "zooplankton_tow ā first 100 rows", rownames = FALSE, filter = "top")Project this dataset into the shared consolidated core model (design_env-bio-consolidation.md). These core tables are this ingestās output: release_database.qmd concatenates the per-dataset shards rather than re-deriving the core from per-dataset tables, so there is exactly one projection to keep correct.
This source carries tow effort and position only ā no measurements ā so it contributes sample and nothing else.
ds_key <- "sio_pic-zooplankton"
# This projection lives here, in the notebook that owns the dataset, not in a
# switch(dataset_key, ...) arm inside calcofi4db. The reusable SHAPE stays in the
# package (sample_arm_self), so this is a declaration.
append_sample(con, sample_arm_self(
ds_key, "zooplankton_tow", "tow_id", "tow",
site_expr = "site_key", ord_expr = "CAST(order_occ AS INTEGER)",
depth_min = "depth_min_m", depth_max = "depth_max_m"))
core <- list(sample = dbGetQuery(con, "SELECT COUNT(*) FROM sample")[[1]])
cat(glue("core projection ā sample={core$sample %||% 0}"), "\n")core projection ā sample=82343 # one sample row per source tow, and the cross-dataset columns must survive the
# projection (site_key/order_occ are how `site` is rebuilt from the core).
# Assert the no-measurement case explicitly, rather than let a silent zero-row
# obs look like a projection bug.
n_tow <- dbGetQuery(con, "SELECT COUNT(*) FROM zooplankton_tow")[[1]]
n_site <- dbGetQuery(con, "SELECT COUNT(*) FROM sample WHERE site_key IS NOT NULL")[[1]]
n_ord <- dbGetQuery(con, "SELECT COUNT(*) FROM sample WHERE order_occ IS NOT NULL")[[1]]
stopifnot(
"sample must be one row per source tow" = core$sample == n_tow,
"sio_pic-zooplankton contributes sample rows" = core$sample > 0,
"sio_pic-zooplankton has no measurements to project" = !"obs" %in% dbListTables(con),
"sample must carry site_key for most tows" = n_site > 0,
"sample must carry order_occ where the source has it" = n_ord > 0)
cat(glue("sample parity: {format(core$sample, big.mark = ',')} tows; ",
"{format(n_site, big.mark = ',')} with site_key, ",
"{format(n_ord, big.mark = ',')} with order_occ"), "\n")sample parity: 82,343 tows; 81,915 with site_key, 1,457 with order_occ dir_create(dir_parquet)
mismatches <- list(cruise_keys = collect_cruise_key_mismatches(con, "zooplankton_tow"))
tbls_out <- core_output_tables(con, extra = "dataset")
parquet_stats <- write_parquet_outputs(
con = con, output_dir = dir_parquet,
tables = tbls_out,
strip_provenance = FALSE, mismatches = mismatches)
build_relationships_json(
rels = core_relationships(tbls_out), output_dir = dir_parquet,
provider = provider, dataset = dataset)[1] "/Users/bbest/Github/CalCOFI/workflows/data/parquet/sio_pic-zooplankton/relationships.json"parquet_stats |> mutate(file = basename(path)) |> select(-path) |>
datatable(caption = "Parquet export statistics")d_tbls_rd <- read_csv(here("metadata/sio/pic-zooplankton/tbls_redefine.csv"))
d_flds_rd <- read_csv(here("metadata/sio/pic-zooplankton/flds_redefine.csv"))
metadata_path <- build_metadata_json(
con = con, d_tbls_rd = d_tbls_rd, d_flds_rd = d_flds_rd,
metadata_derived_csv = c(here("metadata/core_dictionary.csv"),
here("metadata/sio/pic-zooplankton/metadata_derived.csv")),
output_dir = dir_parquet, tables = tbls_out,
set_comments = TRUE, provider = provider, dataset = dataset,
workflow_url = cc$workflow_url, tables_owned = tables_owned)metadata.json documentation gaps (2 tables, 39 columns) ā these render blank in cc_describe_table() / cc_db_catalog():
tables with no description_md: 1 dataset
columns with no description_md: 22 dataset.provider, dataset.dataset, dataset.dataset_name, dataset.dataset_name_short, dataset.category, dataset.color, dataset.description, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others (+10 more)
measurement columns with no units: 17 dataset.dataset_name_short, dataset.category, dataset.color, dataset.citation_main, dataset.citation_others, dataset.link_calcofi_org, dataset.link_data_source, dataset.link_others, dataset.tables, dataset.coverage_temporal, dataset.coverage_spatial, dataset.license (+5 more)
backfill via metadata/{provider}/{dataset}/flds_redefine.csv, then re-runsync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet, gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")# A tibble: 3 Ć 4
file action size reason
<chr> <chr> <dbl> <chr>
1 <rsync> uploaded NA parallel rsync
2 <rsync> uploaded NA parallel rsync
3 <rsync> uploaded NA parallel rsyncFollow-up questions for SIO PIC (Rasmus Swalethorp, Ed Weber, Linsey Sala), ranked by importance. The headline blocker (Q01) is that biovolume is absent from the provided file, so only the tow registry is ingested here. Tracked in metadata/sio/pic-zooplankton/questions.csv; aggregated by questions_email.qmd.
# one validated read + render for every ingest: the vocabulary and the column
# order live in calcofi4db, not in 16 hand-written factor() calls
questions_datatable(
here(cc$questions_file),
caption = "Questions for the SIO PIC zooplankton data providers (ranked)")close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")Parquet outputs written to: /Users/bbest/Github/CalCOFI/workflows/data/parquet/sio_pic-zooplankton