---
title: "Clean a CTD cruise × variable"
subtitle: "propose flags in PostgreSQL, curators review, clean 1 m bins fall out — originals untouched"
format:
html:
toc: true
code-fold: false
code-tools: true
jupyter: python3
---
## What this is
A **parameterized QA/QC notebook** for one cruise × one variable, written in Python for
the CTD team to copy and tweak. It runs against the multi-user PostgreSQL database
(`calcofi`, schema `ctd`) where the **entire cast-file archive lives verbatim and
immutable** — every one of the 82 source columns, every `-99`, every superseded
preliminary file. Cleaning here never edits a value: you **propose flags** into the
`ctd.flag` ledger, curators (Rasmus, Ben G, Kelsey, Erin) accept or reject them, and
derived products like the 1 m bins are **recomputed from originals + accepted flags** —
so any product can be regenerated, and any decision can be revisited.
@fig-flow is the whole loop: raw files land once in immutable tables; rules propose,
humans decide, and every output is derived — each arrow is a section below with the code
that runs it.
::: {.column-page}
```{mermaid}
%%| label: fig-flow
%%| echo: false
%%| fig-width: 10
%%| fig-cap: "The QA/QC loop. **Originals** (navy — `ctd.file` / `ctd.scan` / `ctd.scan_issue`, all 82 source columns, immutable by trigger) are loaded once from the calcofi.org cast-file archive by `load_pg_ctd.qmd`. Automated **QC rules** (§4: `cc_qc_spike`, `cc_qc_sensor_pair`, `cc_qc_range`) produce evidence and triage tables (§5), which `cc_propose_flags()` writes idempotently into the **ledger** (yellow — `ctd.flag`, every change audited). **Curators** accept or reject (§6, one UPDATE in pgAdmin or SQL); the **clean views** (`ctd.v_scan_qc` / `ctd.v_scan_clean`) apply only accepted verdicts, so every product — like the 1 m bins `cc_bin_1m()` writes to `work.clean_1m_*` (§7) — can be regenerated at any time. Each night the accepted flags are published to `gs://calcofi-db/qc/ctd/flag_accepted.parquet`, where the public database release applies them as `measurement_qual`."
flowchart TB
A["calcofi.org archive<br/>409 files, 1993 → now"]
A -->|"load once, verbatim"| B
subgraph PG["PostgreSQL — schema ctd"]
B[("ORIGINALS<br/>10.8M scans<br/>immutable")]
F[("the LEDGER<br/>ctd.flag")]
V["clean QC views"]
end
B -->|"QC rules (§4)"| P["flag candidates (§5)"]
P -->|"propose (§6)"| F
F -->|"curators review (§6)"| V
B --> V
V -->|"1 m bins (§7)"| W[("work.clean_1m_*")]
F -->|"nightly export"| G["flag_accepted.parquet<br/>(public GCS)"]
G -.->|"measurement_qual"| R["public DB release"]
style PG fill:#f4f8fc,stroke:#2A4F7C,color:#193E6D
classDef orig fill:#193E6D,color:#fff,stroke:#2A4F7C
classDef ledger fill:#F4D530,color:#193E6D,stroke:#c9ad13
classDef derived fill:#dbe7f3,color:#193E6D,stroke:#2A4F7C
classDef ext fill:#f4f4f4,color:#333,stroke:#bbb
class B orig
class F ledger
class V,W derived
class A,P,G,R ext
```
:::
The QC checks are the same SQL rules the [ctd-qaqc app](https://app.calcofi.io/ctd-qaqc/)
runs ([`metadata/qc_rules/`](https://github.com/CalCOFI/workflows/tree/main/metadata/qc_rules)),
ported into documented [calcofi4py](https://calcofi.io/calcofi4py/) helpers — SQL is SQL,
so they run identically from Python, R, or psql.
**To run it yourself** you need an account ([Server Access](https://calcofi.io/docs/server-access.html)
walks through the SSH key, tunnel and `~/.pgpass`), then:
```bash
pip install "calcofi4py[viz] @ git+https://github.com/CalCOFI/calcofi4py" jupyter
quarto render clean_ctd_cruise-var.qmd -P study:2607SH -P variable:tempave
```
**Rendering as a cleaning archive.** The rendered HTML is a complete, self-contained
record of one cleaning pass — parameters, evidence, what was proposed, and (in the
[Reproducibility](#sec-repro) accordion at the end) the exact software and rule versions
that produced it. Name the output per cruise × variable and keep it:
```bash
quarto render clean_ctd_cruise-var.qmd \
-P study:2607SH -P variable:tempave \
--output clean_ctd_2607SH_tempave.html # lands in _output/ next to the default name
```
Parameters (the cell below; override with `-P`):
```{python}
#| tags: [parameters]
study = "2607SH" # source cruise id — the most recent, Jul 2026, R/V Bell M. Shimada
variable = "tempave" # a ctd.scan measurement column (see ctd.scan_column)
units = "degC"
sensor_pair = ("temp1", "temp2") # the redundant pair behind the averaged variable
```
## Connect
`cc_pg_connect(tunnel=True)` opens `ssh -N calcofi` for you and authenticates from
`~/.pgpass` — no password appears in this notebook, ever.
```{python}
import pandas as pd
import calcofi4py as cc
con = cc.cc_pg_connect(tunnel=True)
pg_version = con.execute("SHOW server_version").fetchone()[0]
pd.options.display.max_rows = 12
```
## The cruise at a glance
Every station occupation is a numbered **`cast_seq`** (extracted from `cast_id`:
`"2607_001d"` → 1) with a down- and an upcast at the same position — that number is the
key to cross-reference between the map (@fig-map), the profiles (@fig-profiles), the
section (@fig-section) and the flag triage (@fig-triage).
```{python}
casts = cc.cc_ctd_casts(con, study)
occ = casts.groupby("cast_seq").size()
print(f"{study} (cruise_key {casts.cruise_key.iloc[0]}): "
f"{casts.cast_seq.nunique()} station occupations (cast_seq "
f"{int(casts.cast_seq.min())}-{int(casts.cast_seq.max())}), "
f"{len(casts)} casts ({(casts.cast_dir == 'D').sum()} down / {(casts.cast_dir == 'U').sum()} up; "
f"{(occ == 2).sum()} complete down+up pairs), "
f"{casts.datetime_utc.min():%Y-%m-%d} to {casts.datetime_utc.max():%Y-%m-%d}, "
f"max depth {casts.depth_max.max():.0f} m, "
f"{casts.n_scans.sum():,} scans total")
casts.head(4)
```
```{python}
#| label: fig-map
#| fig-cap: "Station occupations, one labeled marker per `cast_seq` (the down/up pair collapses to one point; hover for station, time, both directions' scan counts). Occupations whose source files carry `-99` positions are absent here but present in every table."
cc.cc_station_map(casts, title=f"{study} — station occupations by cast_seq")
```
```{python}
scans = cc.cc_ctd_scans(con, study, columns=[variable, *sensor_pair])
print(f"{len(scans):,} scans; accepted-QC columns come along: "
f"{[c for c in scans.columns if c.endswith(('_qc', '_fix'))]}")
```
The source files use `-99` as a missing-position sentinel, kept verbatim in `ctd.scan`
(only the derived `ctd.cast` positions NULL it). @tbl-nopos is the accounting: any
occupation missing from @fig-map appears here with how many of its scans carry the
sentinel.
```{python}
#| label: tbl-nopos
#| tbl-cap: "Occupations whose source files carry `-99` positions (scan-level counts; `on_map` = whether a usable position remains for @fig-map)."
pos = (scans.assign(no_pos=(scans.lat_dec == -99) | (scans.lon_dec == -99))
.groupby("cast_seq")
.agg(n_scans=("scan_id", "count"), n_scans_neg99=("no_pos", "sum")))
mapped = casts.dropna(subset=["lat", "lon"]).cast_seq.unique()
nopos = (pos[pos.n_scans_neg99 > 0]
.assign(pct=lambda d: (100 * d.n_scans_neg99 / d.n_scans).round(1),
on_map=lambda d: d.index.isin(mapped))
.reset_index())
if len(nopos):
display(nopos)
else:
print(f"none — all {pos.index.nunique()} occupations carry real positions on every scan")
```
```{python}
#| label: fig-section
#| fig-cap: "Quick-look section: `cast_seq` × depth, colored by value (each vertical stripe is one downcast's scans — not an interpolated product). The x axis is the same `cast_seq` as @fig-map."
cc.cc_section_plot(scans, casts, variable, units=units,
title=f"{study} — {variable} section")
```
## Run the QC checks
Three portable rules, straight from the registry. Each returns a DataFrame of suspect
scans — evidence first, flags second.
**Spikes** — a scan that leaps away from the midpoint of its neighbours *while the
neighbours agree with each other*. That second clause is the whole trick: without it,
every steep-but-smooth thermocline gradient fires (on the reference cruise: 92 naive
hits, 19 with neighbour agreement — the other 73 were real gradients).
```{python}
spikes = cc.cc_qc_spike(con, study, variable, spike_threshold=0.5, neighbour_tol=0.5)
print(f"{len(spikes)} spike candidates on {spikes.cast_id.nunique()} casts "
f"({(spikes.cast_dir == 'U').mean():.0%} on upcasts)")
spikes.sort_values("excursion", ascending=False).head(6)
```
**Sensor pair** — the two temperature sensors disagreeing beyond calibration tolerance.
The source's own `*Q` codes 1/2 ("use primary"/"use secondary") exist precisely because
one of a pair misbehaves; a persistent gap is how that is caught early.
```{python}
pair = cc.cc_qc_sensor_pair(con, study, *sensor_pair, threshold=0.05)
print(f"{len(pair)} scans with |{sensor_pair[0]} - {sensor_pair[1]}| > 0.05 {units}")
pair.head(4)
```
**Declared bounds** — the impossible, not the unusual (an unconverted sentinel, a
scaling error). Bounds are generous on purpose.
```{python}
oob = cc.cc_qc_range(con, study, variable, valid_min=-3, valid_max=40)
print(f"{len(oob)} values outside [-3, 40] {units}")
```
## Triage: which casts most need a human
Flags rolled up per `cast_seq` — depth span, value range, share of the cast's scans.
Casts absent from the table have no flags.
```{python}
#| label: tbl-triage
#| tbl-cap: "Proposed + accepted flags per cast_seq (sorted by count) — the inspection worklist."
candidates = pd.concat([
spikes[["scan_id", "depth"]].assign(rule_key="ctd_spike_v1"),
oob[["scan_id", "depth"]].assign(rule_key="ctd_value_out_of_range_v1"),
]).assign(flag_id=lambda d: range(len(d)))
triage = cc.cc_flag_summary(candidates, scans, variable)
worst = int(triage.cast_seq.iloc[0])
triage
```
```{python}
#| label: fig-triage
#| fig-cap: "The same triage visually: flag candidates per `cast_seq`, stacked by rule. A tall bar is a cast to open in @fig-profiles; compare against its neighbours in @fig-section."
import plotly.express as px
bar = (candidates.merge(scans[["scan_id", "cast_seq"]], on="scan_id")
.groupby(["cast_seq", "rule_key"]).size().rename("n").reset_index())
fig = px.bar(bar, x="cast_seq", y="n", color="rule_key", height=340,
color_discrete_map={"ctd_spike_v1": "#2A4F7C",
"ctd_value_out_of_range_v1": "#F4D530"},
labels={"cast_seq": "cast_seq", "n": "flag candidates"},
title=f"{study} — {variable} flag candidates by cast")
fig.update_xaxes(dtick=5)
fig
```
Cast `{python} worst` dominates — open it (@fig-profiles selects it by default; use the
dropdown for any other `cast_seq`). Up- and downcast of the same occupation should
roughly agree: a one-sided excursion is instrument, not ocean.
```{python}
#| label: fig-profiles
#| fig-cap: "Depth profiles with a dropdown per `cast_seq` (default: the worst cast from @tbl-triage; 'all casts' shows the cruise envelope). Down = blue, up = orange, flag candidates = violet ×."
cc.cc_profile_explorer(scans, variable, flags=candidates, units=units, default=worst,
title=f"{study} — {variable} profiles")
```
## Propose flags into the ledger
`cc_propose_flags()` writes one `ctd.flag` row per scan × variable with IODE code **4
(bad)**, your username, a reason, and the rule that found it. It is **idempotent** — a
scan already carrying a proposed or accepted flag for this variable is skipped, so
re-running this notebook never stacks duplicates. Nothing you propose changes any data:
that takes a curator's `accepted`.
```{python}
n_sp = cc.cc_propose_flags(
con, spikes.scan_id, variable, qual_code=4,
reason=f"single-scan spike >0.5 {units}, neighbours agree within 0.5",
rule_key="ctd_spike_v1")
n_ob = cc.cc_propose_flags(
con, oob.scan_id, variable, qual_code=9,
reason="outside declared physical bounds [-3, 40] — sentinel or conversion error",
rule_key="ctd_value_out_of_range_v1")
print(f"proposed: {n_sp} spike + {n_ob} out-of-bounds flags "
f"(0s mean an earlier run already proposed them)")
```
The ledger as it now stands for this cruise — every row carries proposer, timestamp,
and, once reviewed, the reviewer and verdict (full history in `ctd.flag_audit`):
```{python}
ledger = cc.cc_flags(con, study=study)
print(ledger.status.value_counts().to_dict())
ledger.tail(6)
```
**Curators**: accepting is one UPDATE, from psql, pgAdmin, R or here —
```python
# curators only (calcofi_curator role); everyone else gets a clean permission error
con.execute("""
UPDATE ctd.flag SET status = 'accepted', review_note = 'agree — heave spikes'
WHERE flag_id = ANY(%s)""", ([1234, 1235],))
con.commit()
```
## Derive the clean 1 m bins
`ctd.v_scan_clean` is the originals with **accepted** fixes substituted and
accepted-bad values NULLed — the ledger's verdicts, nothing else. Binning it gives the
product; anyone can regenerate it at any time, which is the point of never editing
originals. `write_table=` also publishes it as a real table in the shared `work`
schema, so colleagues (R, Python, pgAdmin) can query it by name.
```{python}
tbl = f"clean_1m_{study.lower()}_{variable}"
bins = cc.cc_bin_1m(con, study, variable, cast_dir="D", write_table=tbl)
print(f"{len(bins):,} (cast × 1 m depth) bins -> work.{tbl}")
bins.head(4)
```
```{python}
#| label: fig-clean
#| fig-cap: "Raw downcast scans (grey) vs the clean 1 m binned product (blue) for the worst cast of @tbl-triage. Until curators accept flags the two coincide; each acceptance moves only the blue line — the grey is immutable."
import plotly.graph_objects as go
w = scans.query("cast_seq == @worst and cast_dir == 'D'").cast_id.iloc[0]
raw = scans.query("cast_id == @w").dropna(subset=[variable])
bw = bins.query("cast_id == @w")
fig = go.Figure()
fig.add_scatter(x=raw[variable], y=raw.depth, mode="markers", name="raw scans",
marker=dict(size=3, color="lightgrey"))
fig.add_scatter(x=bw.value, y=bw.depth_m, mode="lines", name="clean 1 m bins",
line=dict(color="#1f77b4", width=2))
fig.update_layout(height=600, title=f"{study} cast {worst} — raw vs clean 1 m {variable}",
xaxis_title=f"{variable} ({units})", yaxis_title="depth (m)")
fig.update_yaxes(autorange="reversed")
fig
```
```{python}
con.close()
cc.cc_pg_tunnel_close()
```
## Reproducibility {#sec-repro}
If this rendered page is kept as the archive of a cleaning pass, the block below is its
provenance: the software (above all `calcofi4py`), the git commit of the QC rule registry
this repo carries (`metadata/qc_rules/` — with a loud note if the working tree had
uncommitted rule changes), and the database server it ran against.
::: {.callout-note collapse="true" title="Session info — software and rule versions behind this archive"}
```{python}
print(cc.cc_session_info(
repos={
"workflows": (".", None), # this repo (the notebook + registries)
"qc_rules": (".", "metadata/qc_rules"), # the QC rule definitions specifically
},
extra={
"postgresql": pg_version,
"database": "calcofi @ ssh.calcofi.io (schema ctd)",
"release": cc.cc_resolve_version("latest"),
}))
```
:::
## Where to go from here
- **Change the parameters**: any cruise in `ctd.file` (142 studies, 1993→present), any
measurement column in `ctd.scan_column` — salinity (`salt1`; `saltave_corr` is empty
in sensor-only preliminary tiers like this one), oxygen (`ox1`), fluorescence
(`fluorv`), …
- **Tune the thresholds** — the rule registry's values are starting points; Q25 in
[`questions.csv`](https://github.com/CalCOFI/workflows/blob/main/metadata/calcofi/ctd-cast/questions.csv)
asks the team to confirm the flag vocabulary and the curator list.
- **Review**: proposed flags wait in `ctd.flag` (`SELECT * FROM ctd.flag WHERE status =
'proposed'`, or the ledger table above) — accept, reject, or propose better values
(`qual_code = 5` + `proposed_value`).
- **Everything downstream is automatic**: accepted flags flow into `ctd.v_scan_qc` /
`ctd.v_scan_clean` immediately, and each night the ledger is published to
[`gs://calcofi-db/qc/ctd/`](https://storage.calcofi.io/calcofi-db/qc/ctd/) so the
public release ingest can apply it — no live database dependency.
- API docs for every helper used here: [calcofi.io/calcofi4py](https://calcofi.io/calcofi4py/reference/).