Ingest Underway CUFES Fish Eggs

Published

2026-08-14

1 Overview

Source: NOAA CoastWatch ERDDAP erdCalCOFIcufes — Continuous Underway Fish Egg Sampler egg counts (sardine/anchovy/jack mackerel/ hake/squid/other) + underway environment. Provider swfsc (Noelle Bowlin), 1996-present.

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, 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_swfsc_cufes.qmd"))
provider <- cc$provider; dataset <- cc$dataset; 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 && file_exists(db_path)) file_delete(db_path)
dir_create(dirname(db_path)); con <- get_duckdb_con(db_path); load_duckdb_extension(con, "spatial")
meas_type_csv <- here("metadata/measurement_type.csv"); d_meas_type <- read_measurement_type(meas_type_csv)

3 Download from ERDDAP

Code

dir_dl  <- here(glue("data/cache/{dir_label}")); dir_create(dir_dl)
cufes_csv <- file.path(dir_dl, "erdCalCOFIcufes.csv")
erddap_url <- "https://coastwatch.pfeg.noaa.gov/erddap/tabledap/erdCalCOFIcufes.csv"
if (overwrite_all || !file_exists(cufes_csv))
  download.file(erddap_url, cufes_csv, quiet = TRUE)
# ERDDAP CSV: row 1 = column names, row 2 = units, rows 3+ = data
hdr   <- read_csv(cufes_csv, n_max = 0) |> names()
d_raw <- read_csv(cufes_csv, skip = 2, col_names = hdr)
cat(glue("Downloaded {format(nrow(d_raw), big.mark=',')} CUFES samples"), "\n")

Downloaded 49,572 CUFES samples

Code

sync_to_gcs(local_dir = dir_dl, gcs_prefix = glue("archive/{provider}/{dataset}"),
            bucket = "calcofi-files-public", exclude = c(".DS_Store"))

# A tibble: 0 × 4
# ℹ 4 variables: file <chr>, action <chr>, size <dbl>, reason <chr>

4 Build Sample + Measurement Tables

Code

d <- d_raw |>
  mutate(sample_id = row_number(), .before = 1) |>
  mutate(
    datetime_start_utc = as_datetime(time),
    datetime_end_utc   = suppressWarnings(as_datetime(stop_time)),
    across(c(latitude, longitude, stop_latitude, stop_longitude,
             start_temperature, start_salinity, start_wind_speed, start_wind_direction, start_pump_speed,
             stop_temperature, stop_salinity, stop_wind_speed, stop_wind_direction, stop_pump_speed),
           ~ suppressWarnings(as.numeric(.x))))

# A CUFES sample is a SEGMENT, not a point: the pump runs while the ship steams,
# and the source records where the sample started and where it stopped. Those
# ends are a median 8.71 km apart, so `latitude`/`longitude` — which every
# downstream step treats as *the* position, and from which `grid_key` and
# `hex_id` are derived — was reporting a point ~4.4 km from the sample's centre.
# On a grid whose cells are of that order, that is enough to place a sample in
# the wrong one.
#
# So the position is the MIDPOINT of the segment, with the ends preserved as
# `latitude_start`/`longitude_start` and `latitude_stop`/`longitude_stop` so the
# segment itself is not lost and a consumer can still draw it.
#
# The midpoint also recovers 20 samples whose START coordinate is NaN while the
# STOP is good: taking whichever end exists is strictly better than discarding a
# position we hold. NaN is tested explicitly because it passes IS NOT NULL and
# would otherwise poison the mean (see calcofi4db 3.13.1).
ok <- function(x) !is.na(x) & !is.nan(x) & is.finite(x)

# A position is a PAIR. Resolving latitude and longitude with independent rules
# lets a row keep a latitude while its longitude resolves to nothing — 11 samples
# here publish `latitude 42` with a NaN longitude at BOTH ends, and taking them
# separately faithfully reproduces that half-position (66 obs rows with a
# latitude and no hex_id). Worse, independent rules could in principle pair a
# latitude from one end with a longitude from the other, inventing a place the
# ship never was.
#
# So choose the SOURCE first — midpoint, else start, else stop — and require both
# components of that source to be real. Otherwise the position is NULL, which is
# the honest answer and what check_ungridded_obs()'s n_no_position counts.
d_sample <- d |>
  mutate(
    .use_mid   = ok(latitude) & ok(longitude) & ok(stop_latitude) & ok(stop_longitude),
    .use_start = !.use_mid & ok(latitude) & ok(longitude),
    .use_stop  = !.use_mid & !.use_start & ok(stop_latitude) & ok(stop_longitude)) |>
  transmute(
    sample_id, cruise_orig = cruise, ship_name = ship, ship_code,
    sample_number = suppressWarnings(as.integer(sample_number)),
    datetime_start_utc, datetime_end_utc,
    latitude_start = latitude, longitude_start = longitude,
    latitude_stop = stop_latitude, longitude_stop = stop_longitude,
    latitude = case_when(
      .use_mid   ~ (latitude + stop_latitude) / 2,
      .use_start ~ latitude,
      .use_stop  ~ stop_latitude,
      .default   = NA_real_),
    longitude = case_when(
      .use_mid   ~ (longitude + stop_longitude) / 2,
      .use_start ~ longitude,
      .use_stop  ~ stop_longitude,
      .default   = NA_real_),
    start_temperature, start_salinity, start_wind_speed, start_wind_direction, start_pump_speed,
    stop_temperature, stop_salinity, stop_wind_speed, stop_wind_direction, stop_pump_speed)

cat(glue(
  "position: {sum(ok(d_sample$latitude) & ok(d_sample$longitude))} of ",
  "{nrow(d_sample)} samples positioned; ",
  "{sum(!ok(d_sample$latitude) | !ok(d_sample$longitude))} have none ",
  "(a half-position is not a position)"), "\n")

position: 48009 of 49572 samples positioned; 1563 have none (a half-position is not a position)

Code

# no row may keep one coordinate without the other
stopifnot(
  "latitude and longitude must be present together or not at all" =
    sum(ok(d_sample$latitude) != ok(d_sample$longitude)) == 0)

dbWriteTable(con, "cufes_sample", d_sample, overwrite = TRUE)

# pivot egg counts -> long measurement
egg_cols <- c("sardine_eggs","anchovy_eggs","jack_mackerel_eggs","hake_eggs","squid_eggs","other_fish_eggs")
d_meas <- d |>
  select(sample_id, all_of(egg_cols)) |>
  pivot_longer(all_of(egg_cols), names_to = "measurement_type", values_to = "measurement_value") |>
  filter(!is.na(measurement_value)) |>
  mutate(measurement_value = as.double(measurement_value), measurement_qual = NA_character_) |>
  mutate(cufes_measurement_id = row_number(), .before = 1)
dbWriteTable(con, "cufes_measurement", d_meas, overwrite = TRUE)
cat(glue("cufes_sample {nrow(d_sample)}, cufes_measurement {nrow(d_meas)}"), "\n")

cufes_sample 49572, cufes_measurement 284097

5 Resolve Keys + Spatial

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 sample_id, ship_name, datetime_start_utc FROM cufes_sample") |>
  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(datetime_start_utc), NA_character_,
                              as.character(glue("{format(datetime_start_utc,'%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_))
dbWriteTable(con, "zk", zt |> select(sample_id, ship_key, cruise_key), overwrite = TRUE)
dbExecute(con, "ALTER TABLE cufes_sample ADD COLUMN IF NOT EXISTS ship_key VARCHAR")

[1] 0

Code

dbExecute(con, "ALTER TABLE cufes_sample ADD COLUMN IF NOT EXISTS cruise_key VARCHAR")

[1] 0

Code

dbExecute(con, "UPDATE cufes_sample s SET ship_key=z.ship_key, cruise_key=z.cruise_key FROM zk z WHERE s.sample_id=z.sample_id")

[1] 49572

Code

dbExecute(con, "DROP TABLE zk")

[1] 0

Code

cat(glue("ship match {sum(!is.na(zt$ship_key))}/{nrow(zt)}; cruise_key {sum(!is.na(zt$cruise_key))}/{nrow(zt)}"), "\n")

ship match 49572/49572; cruise_key 38859/49572

Code

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

6 Add Measurement Types

Code

cufes_types <- tibble(
  measurement_type = egg_cols,
  description = c("Sardine egg count","Northern anchovy egg count","Jack mackerel egg count",
                  "Pacific hake egg count","Squid egg count","Other fish egg count"),
  units = "count", `_source_column` = egg_cols, `_source_table` = "cufes_measurement",
  `_source_datasets` = "swfsc_cufes", `_qual_column` = NA_character_, `_prec_column` = NA_character_)
new_types <- cufes_types |> filter(!measurement_type %in% d_meas_type$measurement_type)
if (nrow(new_types) > 0) { d_meas_type <- bind_rows(d_meas_type, new_types); write_csv(d_meas_type, meas_type_csv, na = "") }
dbWriteTable(con, "measurement_type", d_meas_type, overwrite = TRUE)
cat(glue("added {nrow(new_types)} measurement types"), "\n")

added 0 measurement types

7 Metadata, Schema, Validate, Outputs

8 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, so there is exactly one projection to keep correct.

CUFES bakes the taxon into the measurement type name (sardine_eggs, anchovy_eggs, …). metadata/measurement_taxon.csv decomposes each raw type into a real taxon_key, the canonical measurement_type, and life_stage, so the egg counts land in obs as ordinary taxon-resolved occurrences.

Code

ds_key <- "swfsc_cufes"
# filter the crosswalk to THIS dataset -- an unfiltered read leaks other datasets'
# taxa into this shard (the retired emit_core_tables() wrapper filtered internally)
mt_taxon <- read_csv(here("metadata/measurement_taxon.csv"),
                     col_types = cols(worms_id = "i", itis_id = "i",
                                      bin_value = "d", .default = "c")) |>
  filter(dataset_key == ds_key)
tx_over  <- read_csv(here("metadata/taxon_override.csv"), show_col_types = FALSE)

# This projection lives here, in the notebook that owns the dataset, not in a
# switch(dataset_key, ...) arm inside calcofi4db. The reusable SHAPES stay in the
# package (sample_arm_self / compat_measurement_sql), so this is a declaration.

# cross-reference: resolve each taxon against BOTH authorities (cached in
# metadata/taxon_xref.csv, so a re-run costs no API calls). This fills the
# `worms_id` COLUMN on itis:-keyed taxa without touching their key — a consumer
# joining on worms_id used to match ZERO rows for every seabird and marine
# mammal — backfills `itis_id` the other way, replaces an id its authority has
# deprecated so the key is always an accepted id, and fetches the real
# `taxonomic_status` with the date it was checked. Must precede the lineage
# fetch, which should ask about the accepted id, not the deprecated one.
ensure_taxon_xref(con, mt_taxon, tx_over,
                  cache_csv = here("metadata/taxon_xref.csv"))

# lineage: fetch each taxon's WoRMS/ITIS classification (cached in
# metadata/taxon_lineage.csv, so a re-run costs no API calls) and stage it as the
# `taxon` hierarchy build_taxon_reference() reads. Without it a crosswalk- or
# vocabulary-resolved taxon reaches the release with a key and a name and NOTHING
# else — no rank, no parent_taxon_key, no classification — so hierarchy rollups
# ("all Decapoda") silently match nothing and no error is raised anywhere.
ensure_taxon_lineage(con, mt_taxon, tx_over,
                     cache_csv = here("metadata/taxon_lineage.csv"))
n_taxon    <- build_taxon_reference(con, mt_taxon, tx_over)
n_ds_taxon <- build_dataset_taxon(con,   mt_taxon, tx_over)
# stage the crosswalk WITH its derived taxon_key. Do NOT dbWriteTable() the raw
# CSV: it has no taxon_key column at all, so `mx.taxon_key` below is a binder
# error — and a 'worms:' || worms_id string built inline would mis-key any
# ITIS-resolved taxon.
ensure_measurement_taxon(con, mt_taxon, dataset_key = ds_key)

append_sample(con, sample_arm_self(ds_key, "cufes_sample", "sample_id", "underway"))

append_obs(con, glue("
  SELECT 'bio', '{ds_key}', {ns_key(ds_key, 'underway', 'c.sample_id')},
         c.grid_key, c.cruise_key, c.latitude, c.longitude,
         CAST(c.datetime_start_utc AS TIMESTAMP), 0::DOUBLE, 0::DOUBLE,
         mx.taxon_key, mx.life_stage, mx.measurement_type, m.measurement_value,
         m.measurement_qual, NULL::DOUBLE
  FROM cufes_measurement m JOIN cufes_sample c USING (sample_id)
  JOIN _measurement_taxon mx ON mx.dataset_key = '{ds_key}'
                            AND mx.raw_measurement_type = m.measurement_type
                            AND mx.target = 'obs'"))

core <- list(
  sample        = dbGetQuery(con, "SELECT COUNT(*) FROM sample")[[1]],
  obs           = dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]],
  taxon         = n_taxon,
  dataset_taxon = n_ds_taxon)
cat(glue(
  "core projection — sample={core$sample %||% 0} obs={core$obs %||% 0} ",
  "taxon={core$taxon %||% 0} dataset_taxon={core$dataset_taxon %||% 0}\n"))

core projection — sample=49572 obs=284097 taxon=33 dataset_taxon=6

Code

# every measurement whose raw type is registered must reach obs,
# with a real taxon_key — an unregistered raw type is dropped by the INNER join,
# so assert the registry actually covers the vocabulary in the data
n_obs <- dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]]
n_exp <- dbGetQuery(con, "
  SELECT COUNT(*) FROM cufes_measurement m JOIN cufes_sample c USING (sample_id)
  JOIN _measurement_taxon mx ON mx.raw_measurement_type = m.measurement_type
                            AND mx.target = 'obs'")[[1]]
n_unreg <- dbGetQuery(con, "
  SELECT COUNT(DISTINCT m.measurement_type) FROM cufes_measurement m
  WHERE m.measurement_type NOT IN (
    SELECT raw_measurement_type FROM _measurement_taxon)")[[1]]
stopifnot(
  "obs must be one row per registered measurement" = n_obs == n_exp,
  "every cufes measurement_type must be in measurement_taxon.csv" = n_unreg == 0,
  "cufes obs must all carry a taxon_key" =
    dbGetQuery(con, "SELECT COUNT(*) FROM obs WHERE taxon_key IS NULL")[[1]] == 0)
cat(glue("obs parity: {format(n_obs, big.mark = ',')} rows, all taxon-resolved"), "\n")

obs parity: 284,097 rows, all taxon-resolved

Code

# serve the retired per-dataset table names as VIEWs over the core, so
# in-notebook consumers and ad-hoc queries keep working against the old names
# (exact for every column the core models, lossy for the rest)
invisible(dbExecute(con, "DROP TABLE IF EXISTS cufes_measurement_src"))
invisible(dbExecute(con, "ALTER TABLE cufes_measurement RENAME TO cufes_measurement_src"))
invisible(dbExecute(con, glue(
  "CREATE OR REPLACE VIEW cufes_measurement AS
   {compat_measurement_sql(ds_key, 'underway', 'sample_id', 'cufes_measurement_id')}")))
cat(glue("compat view cufes_measurement over obs: ",
         "{dbGetQuery(con, 'SELECT COUNT(*) FROM cufes_measurement')[[1]]} rows"), "\n")

compat view cufes_measurement over obs: 284097 rows

Code

d_dataset <- ingest_yaml_to_dataset_df(read_ingest_yaml(here()))
dbWriteTable(con, "dataset", d_dataset, overwrite = TRUE)

cufes_rels <- list(
  primary_keys = list(cufes_sample = "sample_id", cufes_measurement = "cufes_measurement_id",
                      measurement_type = "measurement_type"),
  foreign_keys = list(
    list(table="cufes_measurement", column="sample_id", ref_table="cufes_sample", ref_column="sample_id"),
    list(table="cufes_measurement", column="measurement_type", ref_table="measurement_type", ref_column="measurement_type")))
# the SOURCE shape, documenting the wrangling above; the published tables are
# the consolidated core, so relationships.json comes from core_relationships()
cc_erd(con, tables = c("cufes_sample","cufes_measurement","measurement_type","dataset"), rels = cufes_rels,
       colors = list(lightblue = c("cufes_sample","cufes_measurement"), lightyellow = "measurement_type", white = "dataset"))

Code

results <- validate_for_release(con, checks = "all", strict = FALSE)
cat("Validation:", ifelse(results$passed, "PASSED", "FAILED"), "\n")

Validation: FAILED

Code

dir_create(dir_parquet)
tbls_out <- core_output_tables(con, extra = c("measurement_type", "dataset"))
write_parquet_outputs(con = con, output_dir = dir_parquet,
  tables = tbls_out,
  sort_by = list(obs = c("grid_key", "measurement_type")), strip_provenance = FALSE)

# A tibble: 6 × 5
  table              rows file_size path                             partitioned
  <chr>             <dbl>     <dbl> <chr>                            <lgl>      
1 sample            49572   1315341 /Users/bbest/_big/calcofi/parqu… FALSE      
2 obs              284097   4498936 /Users/bbest/_big/calcofi/parqu… FALSE      
3 taxon                33      4846 /Users/bbest/_big/calcofi/parqu… FALSE      
4 dataset_taxon         6      1643 /Users/bbest/_big/calcofi/parqu… FALSE      
5 measurement_type    200     12063 measurement_type.parquet         FALSE      
6 dataset              16     11099 dataset.parquet                  FALSE

Code

build_relationships_json(rels = core_relationships(tbls_out), output_dir = dir_parquet,
  provider = provider, dataset = dataset)

[1] "/Users/bbest/Github/CalCOFI/workflows/data/parquet/swfsc_cufes/relationships.json"

Code

d_tbls_rd <- read_csv(here("metadata/swfsc/cufes/tbls_redefine.csv"))
d_flds_rd <- read_csv(here("metadata/swfsc/cufes/flds_redefine.csv"))
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/swfsc/cufes/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 (6 tables, 92 columns) — these render blank in cc_describe_table() / cc_db_catalog():
tables with no description_md: 2    measurement_type, dataset
columns with no description_md: 32    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 (+20 more)
measurement columns with no units: 25    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 (+13 more)
  backfill via metadata/{provider}/{dataset}/flds_redefine.csv, then re-run

[1] "/Users/bbest/Github/CalCOFI/workflows/data/parquet/swfsc_cufes/metadata.json"

Code

sync_to_gcs(local_dir = dir_stage, sidecar_dir = dir_parquet, gcs_prefix = glue("ingest/{dir_label}"), bucket = "calcofi-db")

# A tibble: 5 × 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
5 <rsync> uploaded    NA parallel rsync

9 Questions for Data Providers

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 SWFSC CUFES data providers (ranked)")

Code

close_duckdb(con); cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")

Parquet outputs written to: /Users/bbest/Github/CalCOFI/workflows/data/parquet/swfsc_cufes