---
title: "Ingest SIO PIC Zooplankton Net Tows"
calcofi:
target_name: ingest_sio_pic_zooplankton
workflow_type: ingest
dependency:
- ingest_swfsc_ichthyo
output: data/parquet/sio_pic-zooplankton/manifest.json
provider: sio
dataset: pic-zooplankton
workflow_url: https://calcofi.io/workflows/ingest_sio_pic-zooplankton.html
questions_file: metadata/sio/pic-zooplankton/questions.csv
dataset_meta:
dataset_name: SIO PIC Zooplankton Net Tows
# display trio, read by the release `dataset` table and the
# consumer apps (calcofi4db >= 3.15.0) — see NEWS for why these
# left the apps' own hardcoded maps
dataset_name_short: Zooplankton Net Tows
category: Zooplankton
color: "#69db7c"
description: >
Zooplankton net-tow sample registry from the SIO Pelagic Invertebrate
Collection, scoped to CalCOFI-program expeditions. Position, time, depth,
and gear metadata per tow. Includes the historical grid (1950s-80s inshore,
Gulf of California and Baja lines), not just the modern pattern. Biovolume
measurements are not in the source file and are pending from the provider
(see Questions for Data Providers).
citation_main: ""
# the source is a CSV export from the PIC's internal collection database, so
# there is no data-portal URL to record — leave it empty rather than putting
# the prose "SIO Pelagic Invertebrate Collection DB (CSV export)" in a link
# field (that provenance is in `description`). The CalCOFI zooplankton page
# is the landing page a reader should be sent to.
link_calcofi_org: https://calcofi.org/data/marine-ecosystem-data/zooplankton/
link_data_source: ""
link_others: []
license: ""
pi_names: Rasmus Swalethorp; Ed Weber; Linsey Sala
tables_owned:
- {table: sample, shared: true, note: "core event dimension (tow grain; effort/position only)"}
erd:
color: "#bbe0f0"
editor_options:
chunk_output_type: console
---
## 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`.
```{mermaid}
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]
```
## Setup
```{r}
#| label: setup
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 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`.
```{r}
#| label: read-source
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"))
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")
```
## 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.
```{r}
#| label: clean-filter
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")
# 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)
```
## Resolve Ship and Cruise Keys
```{r}
#| label: resolve-keys
load_prior_tables(
con, parquet_dir = cc_stage_path("parquet", "swfsc_ichthyo"),
tables = c("ship", "cruise", "grid"), geom_tables = c("grid"), as_view = TRUE)
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")
dbExecute(con, "ALTER TABLE zooplankton_tow ADD COLUMN IF NOT EXISTS cruise_key VARCHAR")
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")
dbExecute(con, "DROP TABLE zt_keys")
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")
```
## Add Spatial
```{r}
#| label: spatial
add_point_geom(con, "zooplankton_tow", lon_col = "longitude", lat_col = "latitude")
assign_grid_key(con, "zooplankton_tow") |> datatable(caption = "Grid assignment")
```
## Load Dataset Metadata
```{r}
#| label: load-dataset-metadata
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")
```
## Schema Documentation
```{r}
#| label: schema
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().
```
## Validate
```{r}
#| label: validate
results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")
if (length(results$errors) > 0)
cat("Errors:\n", paste("-", results$errors, collapse = "\n"), "\n")
# 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")
```
## Data Preview
```{r}
#| label: preview
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")
```
## 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.
```{r}
#| label: emit_core
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")
# 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")
```
## Write Parquet Outputs
```{r}
#| label: write-parquet
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)
parquet_stats |> mutate(file = basename(path)) |> select(-path) |>
datatable(caption = "Parquet export statistics")
```
## Write Metadata
```{r}
#| label: write-metadata
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)
```
## Upload to GCS
```{r}
#| label: upload-gcs
sync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet, gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")
```
## 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`.
```{r}
#| label: provider-questions
# 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)")
```
## Cleanup
```{r}
#| label: cleanup
close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")
```