---
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).
citation_main: ""
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: []
license: ""
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`. `calcofi4db::build_dataset_taxon()` reads that column to
crosswalk this dataset's vocabulary into the global `taxon` table, which is what
lets `obs.taxon_key` be populated for euphausiids (see *Emit Core Tables*).
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") |>
mutate(
cruise_key = if_else(
is.na(ship_nodc), NA_character_,
glue("{format(date, '%Y-%m')}-{ship_nodc}") |> as.character()))
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))
valid_ck <- dbGetQuery(con, "SELECT DISTINCT cruise_key FROM cruise")$cruise_key
d_keys <- d_keys |>
mutate(cruise_key = if_else(cruise_key %in% valid_ck, cruise_key, NA_character_))
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.
# 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)
# 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()
```
:::
