Ingest SIO PIC Zooplankton Net Tows

Published

2026-08-14

1 Overview

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).

  • Provider: pic
  • Grain: one row per zooplankton net tow (sample metadata only)
  • Scope decision: ingest CalCOFI-program tows only, selected by Expedition_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.
  • Biovolume is NOT in this file — only the tow/sample registry is built here as zooplankton_tow; the biovolume measurement table is pending from the provider (blocker Q01) and will be added as zooplankton_measurement.
Code
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]

2 Setup

Code
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")

3 Read Source Data

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.

Code
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>
Code
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 

4 Clean, Type-Cast, and Filter to CalCOFI Cruises

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.

Code
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 
Code
# 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)

5 Resolve Ship and Cruise Keys

Code
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   
Code
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") |>
  mutate(cruise_key = if_else(
    is.na(ship_nodc) | is.na(date), NA_character_,
    as.character(glue("{format(date, '%Y-%m')}-{ship_nodc}"))))

valid_ck <- dbGetQuery(con, "SELECT DISTINCT cruise_key FROM cruise")$cruise_key
zt <- zt |> mutate(cruise_key = if_else(cruise_key %in% valid_ck, cruise_key, NA_character_))

# write resolved keys back
dbWriteTable(con, "zt_keys", zt |> select(tow_id, ship_key, cruise_key), overwrite = TRUE)
dbExecute(con, "ALTER TABLE zooplankton_tow ADD COLUMN IF NOT EXISTS ship_key VARCHAR")
[1] 0
Code
dbExecute(con, "ALTER TABLE zooplankton_tow ADD COLUMN IF NOT EXISTS cruise_key VARCHAR")
[1] 0
Code
dbExecute(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] 82343
Code
dbExecute(con, "DROP TABLE zt_keys")
[1] 0
Code
cat(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: 67592/82343 (82.1%) 

6 Add Spatial

Code
add_point_geom(con, "zooplankton_tow", lon_col = "longitude", lat_col = "latitude")
assign_grid_key(con, "zooplankton_tow") |> datatable(caption = "Grid assignment")

7 Load Dataset Metadata

Code
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 

8 Schema Documentation

Code
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"))

Code
# the SOURCE shape above documents the wrangling; relationships.json is written
# with the parquet outputs below, from core_relationships().

9 Validate

Code
results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")
Validation: FAILED 
Code
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 14751 NULL values in required column 'cruise_key'
- Table 'zooplankton_tow' has 409 NULL values in required column 'grid_key' 
Code
# 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 

10 Data Preview

Code
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")

11 Emit Core Tables

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.

Code
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 
Code
# 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 

12 Write Parquet Outputs

Code
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"
Code
parquet_stats |> mutate(file = basename(path)) |> select(-path) |>
  datatable(caption = "Parquet export statistics")

13 Write Metadata

Code
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, 34 columns) — these render blank in cc_describe_table() / cc_db_catalog():
tables with no description_md: 1    dataset
columns with no description_md: 17    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 (+5 more)
measurement columns with no units: 14    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 (+2 more)
  backfill via metadata/{provider}/{dataset}/flds_redefine.csv, then re-run

14 Upload to GCS

Code
sync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet, gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")
# A tibble: 4 × 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 rsync
4 <rsync> uploaded    NA parallel rsync

15 Questions for Data Providers

Follow-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.

Code
# 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)")

16 Cleanup

Code
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