---
title: "Ingest CCE-LTER Euphausiids"
calcofi:
target_name: ingest_cce_lter_euphausiids
workflow_type: ingest
dependency:
- ingest_swfsc_ichthyo
output: data/parquet/cce-lter_euphausiids/manifest.json
provider: cce-lter
dataset: euphausiids
workflow_url: https://calcofi.io/workflows/ingest_cce-lter_euphausiids.html
questions_file: metadata/cce-lter/euphausiids/questions.csv
dataset_meta:
dataset_name: CCE-LTER Euphausiid Abundance
# 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: Euphausiids (Krill)
category: Euphausiids (Krill)
color: "#b197fc"
description: >
Species- and life-stage-resolved euphausiid (krill) abundance from
CalCOFI / CCE-LTER net tows (BTEDB export), 1951-present, one row per
tow x species x life stage. Supersedes the prior single-Abundance-
column ingest, which had no taxonomic scope (Q02, now resolved) β
units remain provisional pending provider confirmation (Q01).
# source: https://cite.edirepository.org/cite/knb-lter-cce.313.1?style=ESIP, checked 2026-09-03
# (revision 1 is the newest: DataONE has no object at .../313/2, /3, /4)
citation_main: "Ohman, M.D. 2022. California Current Ecosystem Euphausiid data, Brinton and Townsend Euphausiid Database (BTEDB) ver 1. Environmental Data Initiative. https://doi.org/10.6073/pasta/4a92a0044bcd1523a4f994ece874a57d"
# source: https://cite.edirepository.org/cite/knb-lter-cce.313.1?style=ESIP, checked 2026-09-03
doi: 10.6073/pasta/4a92a0044bcd1523a4f994ece874a57d
link_calcofi_org: ""
# the portal URL the data is actually fetched from β libs/download_euphausiids.R
# pulls EDI package knb-lter-cce.313.1. This field held the prose "BTEDB
# (Bongo Tow Euphausiid Database) export", which named the upstream database
# rather than a link; that provenance is in `description` where it belongs.
link_data_source: https://portal.edirepository.org/nis/mapbrowse?scope=knb-lter-cce&identifier=313
link_others: []
# NOT CC-BY-4.0: the EML <intellectualRights> (DataONE mirror, since the EDI
# portal blocks automated fetches with a Cloudflare bot check) is bespoke,
# non-CC text requiring the acknowledgement below and deposit of reprints
# with the Pelagic Invertebrate Collection and the CCE-LTER program office.
# source: https://cn.dataone.org/cn/v2/object/https%3A%2F%2Fpasta.lternet.edu%2Fpackage%2Fmetadata%2Feml%2Fknb-lter-cce%2F313%2F1, checked 2026-09-03
license: custom
license_url: https://portal.edirepository.org/nis/metadataviewer?packageid=knb-lter-cce.313.1
# source: EML <intellectualRights> (same object as license, above), checked 2026-09-03 β
# the required acknowledgement text verbatim
acknowledgement: "Data originate from the Brinton and Townsend Euphausiid Database of the Pelagic Invertebrate Collection, Scripps Institution of Oceanography. Supported in recent years by NSF grants to M.D. Ohman and the CCE-LTER site."
pi_names: Rasmus Swalethorp; Linsey Sala
# publishes the consolidated core: the source shape is wrangled in the notebook
# and projected into the shared core tables.
tables_owned:
- {table: sample, shared: true, note: "core event dimension (tow grain)"}
- {table: obs, shared: true, note: "core occurrence headline (bio, species x life stage)"}
- {table: taxon, shared: true, note: "shared taxa reference (37 BTEDB species)"}
- {table: dataset_taxon, shared: true, note: "BTEDB taxon_id -> taxon_key crosswalk"}
- {table: measurement_type, shared: true, note: "shared registry across datasets"}
erd:
color: "#d2f0bb"
editor_options:
chunk_output_type: console
---
## Overview
**Source**: BTEDB (Bongo Tow Euphausiid Database) export, `data.csv` β one row
per net tow, 237 columns: 12 tow/position/time columns + 225
`{Genus}_{species}_{life_stage}_Abundance` columns spanning 37 species across
8 genera (*Euphausia*, *Nematobrachion*, *Nematoscelis*, *Nyctiphanes*,
*Stylocheiron*, *Tessarabrachion*, *Thysanoessa*, *Thysanopoda*) and 16
life-stage tokens (adult, juvenile, calyptopis + C1βC3, furcilia + F1βF7,
larvae, metanauplius, damaged).
- **Provider**: `cce-lter`
- **Grain**: one row per net tow x species x life stage (long format)
- **Strategy**: load the tow-level position columns into `euphausiids_tow`
exactly as before; build a small `euphausiids_taxon` reference table from
the distinct species named in the abundance columns; **pivot the 225 wide
columns into long format** (species + life_stage + value per tow) into
`euphausiids_measurement`; summarize into `euphausiids_summary`.
**What changed from the prior ingest**: the previous version read a single
pre-aggregated `Abundance` column with no species dimension at all β that
was the entire content of open question Q02 ("no species column"). This
version reads the species-resolved BTEDB export instead, which **resolves
Q02** but does not change Q01 (units) or the ship/cruise/coordinate
questions, which are unchanged from before and still open.
```{mermaid}
graph LR
A[data.csv<br/>7,482 tows x 225 species/stage cols] --> B[euphausiids_tow<br/>position + keys]
A --> T[euphausiids_taxon<br/>37 species]
A --> C[euphausiids_measurement<br/>long: tow x taxon x life_stage]
C --> D[euphausiids_summary<br/>avg/stddev]
B -.ship/cruise/grid.-> E[(shared refs)]
```
## 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))
# common ingest settings (overwrite, dir_data)
source(here("libs/ingest.R"))
# provider/dataset/metadata from this file's authoritative YAML block
cc <- read_calcofi_meta(here("ingest_cce-lter_euphausiids.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")
# load unified measurement_type reference
meas_type_csv <- here("metadata/measurement_type.csv")
d_meas_type <- read_measurement_type(meas_type_csv)
```
## Read Source Data
The source CSV is fetched reproducibly from EDI by `libs/download_euphausiids.R`
(package `knb-lter-cce.313.1`, entity *Brinton and Townsend Euphausiid Abundance
Data*), pinned to revision 1 with its md5 asserted, so a republished package
fails loudly instead of silently re-shaping the ingest. It lands at
`{dir_data}/cce-lter/euphausiids/data.csv` β the conventional
`{provider}/{dataset}` path, replacing the hand-staged `{dir_data}/euphausiids/`
extract the prior ingest read (12 columns, one undifferentiated `Abundance`).
Source columns are renamed to canonical names per
`metadata/cce-lter/euphausiids/flds_redefine.csv`.
```{r}
#| label: read-source
source(here("libs/download_euphausiids.R"))
euph_csv <- download_euphausiids(
path_expand(glue("{dir_data}/{provider}/{dataset}")),
overwrite = overwrite_all)
stopifnot("euphausiids data.csv not found" = file_exists(euph_csv))
# archive source to GCS for provenance
sync_to_gcs(
local_dir = path_dir(euph_csv),
gcs_prefix = glue("archive/{provider}/{dataset}"),
bucket = "calcofi-files-public",
exclude = c(".DS_Store", "*.tmp", "*.gdoc")) # .gdoc is a Drive stub
d_raw <- read_csv(euph_csv)
abund_cols <- names(d_raw) |> str_subset("_Abundance$")
cat(glue("Read {format(nrow(d_raw), big.mark=',')} rows, ",
"{ncol(d_raw)} columns from {basename(euph_csv)} ",
"({length(abund_cols)} species/stage abundance columns)"), "\n")
```
## Clean, Type-Cast, and Correct
Tow-level cleaning, applying the decisions recorded in `questions.csv`. This
section only touches the 12 non-abundance columns. Canonical field names follow
`metadata/field_dictionary.csv`.
**Q07 (timezone) β the prior ingest was wrong.** It read `TowBegin`/`TowEnd` as
already-UTC. The EDI package metadata states these are *local* time ("Local time
of beginning/end of plankton tow"), so they are converted from
`America/Los_Angeles`, which resolves each historical date's PST/PDT offset
rather than applying a fixed -8. This shifts every tow by 7-8 h, and
`datetime_start_utc` is the cross-dataset match key, so it also changes which
cruise/cast a tow lines up with.
**Q06 (TowEnd year 2371)** is a single-digit transcription error on `tow_id`
7359, whose month/day match its own `TowBegin` β corrected to 2015 rather than
nulled. The out-of-range guard is kept as a net for anything else.
**Q05 (coordinates)**: verified against the source file, exactly one sign-error
longitude (`tow_id` 638, +121.433) and one unrecoverable point (`tow_id` 7364,
lat 87.25 / lon -34.454, outside the EDI bounding box with no confident
transposition fix) β flipped and nulled respectively by the general rules below.
```{r}
#| label: clean-data
# Q06: correct the known transcription error before any timestamp parsing, so
# the fix is applied to the source value rather than to a nulled-out NA
d_raw <- d_raw |>
mutate(TowEnd = if_else(
as.integer(RowNumber) == 7359L & year(as_datetime(TowEnd)) == 2371L,
`year<-`(as_datetime(TowEnd), 2015L), as_datetime(TowEnd)))
# Q07: source timestamps are LOCAL Pacific per EDI metadata, not UTC
local_to_utc <- function(x)
force_tz(as_datetime(x), "America/Los_Angeles") |> with_tz("UTC")
d_clean <- d_raw |>
transmute(
tow_id = as.integer(RowNumber),
cruise_orig = Cruise,
ship_name = Ship,
date = as.Date(Date),
line = suppressWarnings(as.numeric(Line)),
station = suppressWarnings(as.numeric(Station)),
region = Region,
datetime_start_utc = local_to_utc(TowBegin), # Q07 resolved
datetime_end_utc = local_to_utc(TowEnd),
latitude = as.numeric(Latitude),
longitude = as.numeric(Longitude)) |>
mutate(
# Q05: sign error -> flip; unrecoverable point -> null both coordinates
longitude = if_else(longitude > 0, -longitude, longitude),
bad_coord = latitude > 51 | latitude < 20 | longitude < -135 | longitude > -105,
latitude = if_else(bad_coord, NA_real_, latitude),
longitude = if_else(bad_coord, NA_real_, longitude),
# residual out-of-range tow-end guard (Q06's known case is fixed above)
datetime_end_utc = if_else(
year(datetime_end_utc) > 2026 | year(datetime_end_utc) < 1949,
as_datetime(NA), datetime_end_utc),
site_key = if_else(
is.na(line) | is.na(station), NA_character_,
sprintf("%05.1f %05.1f", line, station)))
# assert the two named Q05/Q06 corrections actually landed, so a future source
# revision that renumbers rows fails here instead of silently skipping the fix
stopifnot(
"Q06: tow_id 7359 tow-end should be corrected to 2015, not nulled" =
year(d_clean$datetime_end_utc[d_clean$tow_id == 7359L]) == 2015L,
"Q05: tow_id 638 longitude should be flipped negative" =
d_clean$longitude[d_clean$tow_id == 638L] < 0,
"Q05: tow_id 7364 coordinates should be nulled" =
is.na(d_clean$latitude[d_clean$tow_id == 7364L]))
cat(glue("Cleaned {nrow(d_clean)} tows; ",
"{sum(d_clean$bad_coord, na.rm=TRUE)} coordinate(s) nulled (Q05), ",
"{sum(is.na(d_clean$datetime_end_utc))} tow-end null; ",
"timestamps converted America/Los_Angeles -> UTC (Q07)"), "\n")
```
## Build Taxon Reference Table
Parse the 225 abundance column names into genus/species/life_stage. Column
pattern is `{Genus}_{species}_{life_stage}_Abundance`; life_stage itself may
contain an underscore (e.g. `calyptopis_C1`, `furcilia_F6`), so the regex
captures everything between the species token and the trailing `_Abundance`
rather than assuming a fixed number of segments.
```{r}
#| label: parse-columns
col_parts <- tibble(col = abund_cols) |>
mutate(
stem = str_remove(col, "_Abundance$"),
genus = str_extract(stem, "^[A-Za-z]+"),
rest = str_remove(stem, "^[A-Za-z]+_"),
species = str_extract(rest, "^[a-z]+"),
life_stage = str_remove(rest, "^[a-z]+_") |> str_replace_all("_", " "))
# a column with no life-stage token would silently yield the species epithet as
# its life_stage, so require every column to parse into all three parts
bad_parse <- col_parts |>
filter(is.na(genus) | is.na(species) | life_stage == species | life_stage == "")
stopifnot(
"every *_Abundance column must parse into genus + species + life_stage" =
nrow(bad_parse) == 0)
n_species <- n_distinct(col_parts$genus, col_parts$species)
cat(glue("Parsed {nrow(col_parts)} columns -> ",
"{n_species} distinct species x {n_distinct(col_parts$life_stage)} life-stage tokens"), "\n")
# Question 09 (NEW): two species present in the canonical DB taxon list are absent
# from this BTEDB export entirely (Bentheuphausia amblyops, Thysanopoda
# cristata) β nothing to ingest for these until the provider confirms
# whether BTEDB tracks them at all. Tracked below, not silently dropped.
col_parts <- col_parts |>
mutate(scientific_name = as.character(glue("{genus} {species}")))
euph_taxon <- col_parts |>
distinct(scientific_name, genus, species) |>
arrange(scientific_name) |>
mutate(taxon_id = row_number(), .before = 1)
cat(glue("euphausiids_taxon: {nrow(euph_taxon)} species"), "\n")
```
### Resolve to WoRMS
`taxon_id` is a local key; the shared `taxon` / `dataset_taxon` model keys on
`worms:<AphiaID>`, so resolve each name against WoRMS with
`calcofi4db::standardize_species()` and carry `worms_id` on
`euphausiids_taxon`. *Emit Core Tables* then stages that column with
`calcofi4db::append_dataset_taxon()` and lets `resolve_dataset_taxon()` fill
`taxon_key`, which is what lets `obs.taxon_key` be populated for euphausiids.
This is also what answers **Q10**: if BTEDB's `Nematoscelis` and the DB's older
`Hansarsia` are synonyms, WoRMS returns the same accepted AphiaID for both, so
the check is an assertion rather than a provider question.
```{r}
#| label: taxon-worms
dbWriteTable(con, "euphausiids_taxon", euph_taxon, overwrite = TRUE)
taxon_std <- standardize_species(
con, species_tbl = "euphausiids_taxon", id_col = "taxon_id",
sci_name_col = "scientific_name", update_in_place = TRUE, include_gbif = FALSE)
# standardize_species() adds a gbif_id column for every run; with include_gbif =
# FALSE it is entirely NULL, so drop it rather than publish an all-empty column
if ("gbif_id" %in% dbListFields(con, "euphausiids_taxon")) {
n_gbif <- dbGetQuery(con,
"SELECT COUNT(gbif_id) AS n FROM euphausiids_taxon")$n
if (n_gbif == 0)
dbExecute(con, "ALTER TABLE euphausiids_taxon DROP COLUMN gbif_id")
}
euph_taxon <- dbGetQuery(con, "SELECT * FROM euphausiids_taxon ORDER BY taxon_id")
n_worms <- sum(!is.na(euph_taxon$worms_id))
cat(glue("WoRMS resolved: {n_worms}/{nrow(euph_taxon)} species ",
"({round(100*n_worms/nrow(euph_taxon),1)}%)"), "\n")
# Q10: Nematoscelis (BTEDB) vs Hansarsia (older DB name) β same AphiaID or not
q10 <- taxon_std |>
filter(str_detect(scientific_name, "^Nematoscelis ")) |>
select(scientific_name, worms_id, accepted_name, taxonomic_status)
q10 |> datatable(
caption = "Q10: Nematoscelis WoRMS resolution (accepted_name reveals any Hansarsia synonymy)",
rownames = FALSE)
```
## Pivot to Long Format
Reshape the 225 wide columns into one row per tow x species x life_stage,
dropping true zeros (a tow where a species/stage genuinely wasn't observed)
from the fact table β same convention as `euphausiids_measurement` not
storing NULLs before.
```{r}
#| label: pivot-long
d_wide <- d_raw |>
transmute(tow_id = as.integer(RowNumber), across(all_of(abund_cols)))
d_long <- d_wide |>
pivot_longer(-tow_id, names_to = "col", values_to = "abundance") |>
left_join(col_parts, by = "col") |>
left_join(euph_taxon |> select(taxon_id, scientific_name), by = "scientific_name") |>
filter(!is.na(abundance), abundance > 0) |>
select(tow_id, taxon_id, life_stage, abundance)
cat(glue("Pivoted to {format(nrow(d_long), big.mark=',')} non-zero tow x species x stage rows ",
"(from {format(nrow(d_wide) * length(abund_cols), big.mark=',')} wide cells)"), "\n")
```
## Resolve Ship and Cruise Keys
**Q04 (vessel names) resolved**: the raw `Ship` column holds 37 distinct values
that collapse to 31 canonical vessels once case, the `R/V` prefix, internal
whitespace and trailing periods are normalized (e.g. *New Horizon* / *NEW
HORIZON*, *R/V BLACK DOUGLAS* / *BLACK DOUGLAS*, *Paolina T* / *Paolina T.*).
One merge is not derivable by string rules and is applied explicitly:
`SHIMADA` = `BELL M. SHIMADA` (same NOAA vessel). Unresolved names are reported
below rather than silently dropped β `ship_key` is a secondary provenance field
here, so a miss is non-blocking (Q03/Q04 both downgraded to low priority).
```{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)
# shared normalization so both sides collapse the same way
norm_ship <- function(x)
x |>
str_replace("^R/?V\\.?\\s+", "") |>
str_to_upper() |>
str_replace_all("\\.", "") |>
str_squish()
# vessel aliases that no string rule recovers (Q04)
SHIP_ALIASES <- c("SHIMADA" = "BELL M SHIMADA")
d_ship <- dbGetQuery(con, "SELECT ship_key, ship_name, ship_nodc FROM ship") |>
mutate(ship_name_norm = norm_ship(ship_name))
d_keys <- d_clean |>
mutate(
ship_name_norm = norm_ship(ship_name),
ship_name_norm = coalesce(SHIP_ALIASES[ship_name_norm], ship_name_norm)) |>
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) β the old
# month rule put a tow in the first days of a month on the NEXT cruise
dbWriteTable(con, "zk", d_keys |> select(tow_id, ship_key, datetime = date), overwrite = TRUE)
resolve_cruise_key(con, "zk", datetime_col = "datetime", require_in_cruise = TRUE) |>
datatable(caption = "cruise_key resolution method")
d_keys <- d_keys |>
left_join(dbGetQuery(con, "SELECT tow_id, cruise_key FROM zk"), by = "tow_id")
dbExecute(con, "DROP TABLE zk")
n_raw_ships <- n_distinct(d_clean$ship_name)
n_norm_ships <- n_distinct(d_keys$ship_name_norm)
cat(glue("Q04: {n_raw_ships} raw vessel names -> {n_norm_ships} canonical"), "\n")
n_ship <- sum(!is.na(d_keys$ship_key))
n_cruise <- sum(!is.na(d_keys$cruise_key))
cat(glue(
"Ship match: {n_ship}/{nrow(d_keys)} ({round(100*n_ship/nrow(d_keys),1)}%); ",
"cruise_key match: {n_cruise}/{nrow(d_keys)} ",
"({round(100*n_cruise/nrow(d_keys),1)}%)"), "\n")
# report vessel names that still don't resolve, for the Q04 follow-up
d_keys |>
filter(is.na(ship_key)) |>
count(ship_name, ship_name_norm, sort = TRUE) |>
datatable(caption = "Unresolved vessel names (Q04)", rownames = FALSE)
```
## Load Tidy Tow Table
```{r}
#| label: load-tow
d_tow <- d_keys |>
transmute(
tow_id, cruise_key, ship_key, ship_name, cruise_orig,
site_key, line, station, region,
datetime_start_utc, datetime_end_utc,
latitude, longitude)
dbWriteTable(con, "euphausiids_tow", d_tow, overwrite = TRUE)
# euphausiids_taxon was written (and WoRMS-enriched) in the taxon step above
cat(glue("euphausiids_tow: {dbGetQuery(con,'SELECT COUNT(*) FROM euphausiids_tow')[[1]]} rows"), "\n")
```
## Add Spatial
```{r}
#| label: spatial
# Recover a position from CalCOFI line/station where the source carries none.
#
# The station plan IS a coordinate system β PROJ ships it as `+proj=calcofi` β so
# this is a PROJECTION, not a lookup against `grid`. That matters here: a lookup
# resolves only stations present in the grid table, while the transform resolves
# any line/station pair, including the historical inshore stations and the Gulf
# of California / Baja lines the modern pattern dropped.
#
# One tow qualifies today (2015-04-32NM, line 86.7 station 33 -> -118.49, 33.89,
# in the Southern California Bight). Small, but it is a real position we were
# holding and discarding, and the rule protects future rows for free. It runs
# BEFORE add_point_geom()/assign_grid_key() so a recovered row gets its geometry,
# grid_key and hex_id like any other β recovering the coordinate is only half the
# job if the row still lands ungridded.
#
# Must also precede add_point_geom() for a second reason: DuckDB fails an UPDATE
# on a table carrying a CRS-tagged GEOMETRY column, and `geom` does not exist yet.
d_nopos <- dbGetQuery(con, "
SELECT tow_id, line, station FROM euphausiids_tow
WHERE (latitude IS NULL OR isnan(latitude)
OR longitude IS NULL OR isnan(longitude))
AND line IS NOT NULL AND station IS NOT NULL")
if (nrow(d_nopos) > 0) {
ll <- cc_calcofi_to_lonlat(d_nopos$line, d_nopos$station)
d_nopos$latitude_new <- ll$latitude
d_nopos$longitude_new <- ll$longitude
dbWriteTable(con, "_pos_recover", d_nopos, overwrite = TRUE)
n_fix <- dbExecute(con, "
UPDATE euphausiids_tow t
SET latitude = r.latitude_new, longitude = r.longitude_new
FROM _pos_recover r
WHERE r.tow_id = t.tow_id AND r.latitude_new IS NOT NULL")
dbExecute(con, "DROP TABLE _pos_recover")
cat(glue("recovered {n_fix} position(s) from CalCOFI line/station ",
"via +proj=calcofi"), "\n")
} else {
cat("no positions to recover from line/station\n")
}
add_point_geom(con, "euphausiids_tow", lon_col = "longitude", lat_col = "latitude")
grid_stats <- assign_grid_key(con, "euphausiids_tow")
grid_stats |> datatable(caption = "Grid assignment")
```
## Load Measurement Table
One measurement type, `euphausiid_abundance`, now dimensioned by
`taxon_id` + `life_stage` instead of being a single undifferentiated value
per tow β this is what resolves Q02.
```{r}
#| label: load-measurement
dbWriteTable(con, "euph_long_staged", d_long, overwrite = TRUE)
dbExecute(con,
"CREATE OR REPLACE TABLE euphausiids_measurement AS
SELECT ROW_NUMBER() OVER (ORDER BY tow_id, taxon_id, life_stage) AS euphausiids_measurement_id,
tow_id, taxon_id, life_stage,
'euphausiid_abundance' AS measurement_type,
CAST(abundance AS DOUBLE) AS measurement_value,
NULL::VARCHAR AS measurement_qual
FROM euph_long_staged
WHERE abundance IS NOT NULL
AND NOT isnan(CAST(abundance AS DOUBLE))
AND isfinite(CAST(abundance AS DOUBLE))")
dbExecute(con, "DROP TABLE euph_long_staged")
n_meas <- dbGetQuery(con, "SELECT COUNT(*) FROM euphausiids_measurement")[[1]]
cat(glue("euphausiids_measurement: {format(n_meas, big.mark=',')} rows"), "\n")
```
## Summarize Replicate Measurements
Aggregate replicate tows at each unique position x species x life_stage
into mean and standard deviation β same pattern as before, now with the
taxon/stage dimension carried through.
```{r}
#| label: measurement-summary
dbExecute(con,
"CREATE OR REPLACE TABLE euphausiids_summary AS
SELECT
t.site_key, t.datetime_start_utc, t.latitude, t.longitude,
m.taxon_id, m.life_stage, m.measurement_type,
AVG(m.measurement_value) AS avg,
CASE WHEN COUNT(*) = 1 THEN 0
ELSE COALESCE(STDDEV_SAMP(m.measurement_value), 0) END AS stddev,
COUNT(*) AS n_obs
FROM euphausiids_measurement m
JOIN euphausiids_tow t USING (tow_id)
WHERE NOT isnan(m.measurement_value) AND isfinite(m.measurement_value)
GROUP BY t.site_key, t.datetime_start_utc, t.latitude, t.longitude,
m.taxon_id, m.life_stage, m.measurement_type")
cat(glue("euphausiids_summary: {dbGetQuery(con,'SELECT COUNT(*) FROM euphausiids_summary')[[1]]} rows"), "\n")
```
## Add Measurement Type
**Q01 (units) resolved**: the EDI package metadata (abstract + per-column unit
definitions) gives abundance as the vertically integrated number of individuals
beneath 1 mΒ² of sea surface β `numberPerMeterSquared`, an *areal* quantity, not
the volumetric `count/1000m3` the prior ingest guessed. Since `units` is what
every downstream consumer reads off `measurement_type`, the row is **upserted**
rather than skipped-if-present, so a registry left behind by an earlier run with
the provisional units is corrected in place.
```{r}
#| label: add-measurement-type
euph_types <- tibble(
measurement_type = "euphausiid_abundance",
description = paste(
"Euphausiid (krill) abundance per net tow, by species and life stage;",
"vertically integrated individuals beneath 1 m2 of sea surface",
"(EDI knb-lter-cce.313.1)."),
units = "numberPerMeterSquared", # Q01 resolved
is_canonical = TRUE,
`_source_column` = "*_Abundance (pivoted; see euphausiids_taxon + life_stage)",
`_source_table` = "euphausiids_measurement",
`_source_datasets` = "cce-lter_euphausiids",
`_qual_column` = NA_character_,
`_prec_column` = NA_character_,
grain = "obs")
# upsert, not delete-and-replace: the literal below carries no valid_min/
# valid_max, and the naive `filter(!= x) |> bind_rows()` destroyed those
# curated bounds on every re-run (it silently un-declared euphausiid_abundance
# and the picoplankton types mid-release, failing the bounds gate).
d_meas_type <- upsert_measurement_types(d_meas_type, euph_types)
write_csv(d_meas_type, meas_type_csv, na = "")
cat(glue("Registered euphausiid_abundance (units = {euph_types$units})"), "\n")
dbWriteTable(con, "measurement_type", d_meas_type, overwrite = TRUE)
```
## 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
euph_rels <- list(
primary_keys = list(
euphausiids_tow = "tow_id",
euphausiids_taxon = "taxon_id",
euphausiids_measurement = "euphausiids_measurement_id",
measurement_type = "measurement_type"),
foreign_keys = list(
list(table = "euphausiids_measurement", column = "tow_id",
ref_table = "euphausiids_tow", ref_column = "tow_id"),
list(table = "euphausiids_measurement", column = "taxon_id",
ref_table = "euphausiids_taxon", ref_column = "taxon_id"),
list(table = "euphausiids_measurement", column = "measurement_type",
ref_table = "measurement_type", ref_column = "measurement_type")))
euph_tables <- c(
"euphausiids_tow", "euphausiids_taxon", "euphausiids_measurement",
"euphausiids_summary", "measurement_type", "dataset")
cc_erd(
con, tables = euph_tables, rels = euph_rels,
colors = list(
lightblue = c("euphausiids_tow", "euphausiids_measurement", "euphausiids_summary"),
lightgreen = "euphausiids_taxon",
lightyellow = "measurement_type",
white = "dataset"))
build_relationships_json(
rels = euph_rels, output_dir = dir_parquet,
provider = provider, dataset = dataset)
```
## 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")
if (length(results$warnings) > 0)
cat("Warnings:\n", paste("-", results$warnings, collapse = "\n"), "\n")
# NOTE: `Validation: FAILED` here is expected and non-blocking (strict = FALSE).
# NULLs in the cross-dataset FK keys (cruise_key, site_key, grid_key) are the
# accepted unmatched remainder of partial matching β the same behavior as
# calcofi_dic's nullable cast_id/bottle_id (issue #47) β tracked in questions
# Q03 (cruise) and Q04 (ship), not a hard failure. `itis_id` NULLs are species
# WoRMS resolves but ITIS does not. The "missing expected tables" warning lists
# ichthyo tables this dataset legitimately does not own.
cat(glue(
"\nMatch coverage (non-NULL): ",
"ship_key {round(100*dbGetQuery(con,\"SELECT AVG(CASE WHEN ship_key IS NOT NULL THEN 1 ELSE 0 END) FROM euphausiids_tow\")[[1]],1)}%, ",
"cruise_key {round(100*dbGetQuery(con,\"SELECT AVG(CASE WHEN cruise_key IS NOT NULL THEN 1 ELSE 0 END) FROM euphausiids_tow\")[[1]],1)}%, ",
"grid_key {round(100*dbGetQuery(con,\"SELECT AVG(CASE WHEN grid_key IS NOT NULL THEN 1 ELSE 0 END) FROM euphausiids_tow\")[[1]],1)}%"), "\n")
n_dup <- dbGetQuery(con,
"SELECT COUNT(*) FROM (
SELECT tow_id, COUNT(*) n FROM euphausiids_tow GROUP BY tow_id HAVING COUNT(*)>1)")[[1]]
cat(glue("euphausiids_tow tow_id duplicates: {n_dup}"), "\n")
```
## Data Preview
```{r}
#| label: preview-taxon
euph_taxon |> datatable(caption = "euphausiids_taxon β all species", rownames = FALSE)
```
```{r}
#| label: preview-measurement
dbGetQuery(con, "
SELECT m.tow_id, t.scientific_name, m.life_stage, m.measurement_value
FROM euphausiids_measurement m
JOIN euphausiids_taxon t USING (taxon_id)
LIMIT 100") |>
datatable(caption = "euphausiids_measurement β first 100 rows (joined for readability)", rownames = FALSE)
```
## Emit Core Tables
Project this dataset into the shared consolidated core model
(`design_env-bio-consolidation.md`). The projection below is owned by this
notebook β the
single source of truth for the per-dataset projection into `sample` / `obs` /
`obs_attribute` / `sample_measurement`, also used by `release_database.qmd` to
assemble the authoritative cross-dataset release.
With the species-resolved export, the euphausiid `obs` arm now carries
`life_stage` and resolves `taxon_key` through `dataset_taxon` (built centrally
at release time, so it reads NULL here) instead of emitting one undifferentiated
row per tow.
```{r}
#| label: emit_core
ds_key <- "cce-lter_euphausiids"
# 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.
# the vocabulary is DECLARED here and RESOLVED by the package (taxon plan D1,
# calcofi4db >= 3.29.0): append_dataset_taxon() stages one `dataset_taxon` row
# per BTEDB species with `taxon_key` empty and the AphiaID standardize_species()
# resolved above carried along as `ds_source_json`; resolve_dataset_taxon() fills
# the key from the authorities. There is no euphausiids arm inside calcofi4db any
# more -- a column renamed or dropped here now fails at this call rather than
# becoming NA at release.
d_vocab <- dbGetQuery(con, "
SELECT CAST(taxon_id AS VARCHAR) AS ds_taxa_code,
scientific_name AS ds_scientific_name,
CAST(worms_id AS INTEGER) AS worms_id
FROM euphausiids_taxon ORDER BY taxon_id")
n_staged <- append_dataset_taxon(con, ds_key, d_vocab)
# 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 <- resolve_dataset_taxon(con, mt_taxon, tx_over)
# the ingest asserts its own crosswalk (D6): every taxon_id the measurements use
# is in the vocabulary and every staged row keys an authority id. Halts on any
# finding.
obs_codes <- dbGetQuery(con,
"SELECT DISTINCT CAST(taxon_id AS VARCHAR) AS c FROM euphausiids_measurement
WHERE taxon_id IS NOT NULL")$c
chk_taxon <- check_dataset_taxon(con, ds_key, codes = obs_codes)
cat(glue("vocabulary: {n_staged} species staged; taxon {n_taxon}, ",
"dataset_taxon {n_ds_taxon}"), "\n")
# NOTE: no `_measurement_taxon` staging. The crosswalk's euphausiid rows describe
# the OLD single-`Abundance` export, where the taxon was baked into the type name.
# The BTEDB export is species- AND life-stage-resolved, so taxon_key comes from
# dataset_taxon and life_stage rides the headline β decomposing through the
# crosswalk here is exactly the bug that collapsed all 37 species to
# worms:110513 (Euphausiidae) and nulled life_stage in the old release arm.
append_sample(con, sample_arm_self(
ds_key, "euphausiids_tow", "tow_id", "tow", site_expr = "site_key",
depth_min = "NULL::DOUBLE", depth_max = "NULL::DOUBLE"))
append_obs(con, glue("
SELECT 'bio', '{ds_key}', {ns_key(ds_key, 'tow', 'tw.tow_id')},
tw.grid_key, tw.cruise_key, tw.latitude, tw.longitude,
CAST(tw.datetime_start_utc AS TIMESTAMP), NULL::DOUBLE, NULL::DOUBLE,
dt.taxon_key, m.life_stage, m.measurement_type, m.measurement_value,
m.measurement_qual, NULL::DOUBLE
FROM euphausiids_measurement m JOIN euphausiids_tow tw USING (tow_id)
LEFT JOIN dataset_taxon dt ON dt.dataset_key = '{ds_key}'
AND dt.ds_taxa_code = CAST(m.taxon_id AS VARCHAR)"))
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} ",
"obs_attribute={core$obs_attribute %||% 0} ",
"sample_measurement={core$sample_measurement %||% 0}\n"))
# the obs headline must carry the new taxon x life-stage grain, not collapse
# back to one row per tow β assert rather than eyeball
n_obs <- dbGetQuery(con, "SELECT COUNT(*) FROM obs")[[1]]
n_meas <- dbGetQuery(con,
"SELECT COUNT(*) FROM euphausiids_measurement m JOIN euphausiids_tow t
USING (tow_id)")[[1]]
n_stage <- dbGetQuery(con, "SELECT COUNT(DISTINCT life_stage) FROM obs")[[1]]
stopifnot(
"obs must be one row per measurement (tow x taxon x life_stage)" = n_obs == n_meas,
"obs.life_stage must carry the BTEDB life-stage dimension" = n_stage > 1)
cat(glue("obs parity: {format(n_obs, big.mark=',')} rows across ",
"{n_stage} life stages"), "\n")
# the species x life-stage grain must survive into taxon_key, not collapse to
# family Euphausiidae (which is what decomposing via measurement_taxon would do)
stopifnot(
"euphausiid obs must resolve species-level taxon_key" =
dbGetQuery(con, "SELECT COUNT(DISTINCT taxon_key) FROM obs")[[1]] > 1)
cat(glue("taxa resolved: ",
"{dbGetQuery(con, 'SELECT COUNT(DISTINCT taxon_key) FROM obs')[[1]]} distinct"), "\n")
```
## Write Parquet Outputs
```{r}
#| label: write-parquet
dir_create(dir_parquet)
mismatches <- list(
measurement_types = collect_measurement_type_mismatches(
con, here("metadata/measurement_type.csv")),
cruise_keys = collect_cruise_key_mismatches(con, "euphausiids_tow"))
tbls_out <- core_output_tables(con, extra = c("measurement_type", "dataset"))
parquet_stats <- write_parquet_outputs(
con = con,
output_dir = dir_parquet,
tables = tbls_out,
sort_by = list(obs = c("grid_key", "measurement_type")),
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/cce-lter/euphausiids/tbls_redefine.csv"))
d_flds_rd <- read_csv(here("metadata/cce-lter/euphausiids/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 = here("metadata/core_dictionary.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 CCE-LTER (Rasmus Swalethorp, Linsey Sala), **ranked by
importance**. Q02 (taxonomic scope) is now settled by this ingest β kept
below with status `answered` for record-keeping, not because it's still
open. Questions 09/10 are new, arising directly from cross-checking this
export against the DB's canonical euphausiid species list.
```{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 CCE-LTER euphausiid data providers (ranked)")
```
## Cleanup
```{r}
#| label: cleanup
close_duckdb(con)
cat(glue("Parquet outputs written to: {dir_parquet}"), "\n")
```
::: {.callout-caution collapse="true"}
## Session Info
```{r session_info}
devtools::session_info()
```
:::
