CTD QA/QC, end to end — a read-only walkthrough¶

This article strings the calcofi4py helpers together the way the CTD team's parameterized notebook (clean_ctd_cruise-var.qmd) does — one cruise, one variable, from raw casts to a clean 1 m product — without writing anything. The two steps that change the database (proposing flags, publishing a work.* table) are shown as code but not run; everything else executes for real.

Out[1]:

The loop this article walks: originals are loaded once and never edited; rules propose, curators decide, products are derived.

How this page is built. It is the Python equivalent of a pre-computed R vignette: someone with a server account runs scripts/render_articles.sh, which executes this notebook through their SSH tunnel and ~/.pgpass, and commits the executed notebook (docs/articles/*.ipynb). GitHub Pages then publishes the stored outputs as-is — the site build holds no database credentials and never touches the server. No secret ever appears in the notebook: cc_pg_connect() reads the password from libpq's ~/.pgpass, and the session-info block at the end records exactly what software and server the outputs came from.

In [2]:

import pandas as pd
import calcofi4py as cc
from IPython.display import HTML, display

print("calcofi4py", cc.__version__)     # the version every output below was produced with

pd.options.display.max_rows = 10
pd.options.display.width = 120

def show(fig):
    """Plotly figure as self-contained HTML, so it renders in the static site."""
    display(HTML(fig.to_html(include_plotlyjs="cdn", full_html=False)))

study, variable, units, sensor_pair = "2607SH", "tempave", "degC", ("temp1", "temp2")

import pandas as pd import calcofi4py as cc from IPython.display import HTML, display print("calcofi4py", cc.__version__) # the version every output below was produced with pd.options.display.max_rows = 10 pd.options.display.width = 120 def show(fig): """Plotly figure as self-contained HTML, so it renders in the static site.""" display(HTML(fig.to_html(include_plotlyjs="cdn", full_html=False))) study, variable, units, sensor_pair = "2607SH", "tempave", "degC", ("temp1", "temp2")

calcofi4py 0.3.6

1. The public release — no account needed¶

The other data store first, because anyone can run this part. Releases are immutable, versioned Parquet on a public bucket; DuckDB reads only what a query touches.

In [3]:

versions = cc.cc_list_versions()
print(f"{len(versions)} releases; latest = {cc.cc_resolve_version('latest')}")

rel = cc.cc_get_db(tables=["sample", "cruise"])          # release tables as DuckDB views
rel.execute("SET enable_progress_bar = false")            # no Jupyter progress-bar widgets in a static page
rel.sql("""
  SELECT dataset_key, count(*) AS samples, min(datetime) AS first, max(datetime) AS last
  FROM sample WHERE dataset_key LIKE 'calcofi_%'
  GROUP BY 1 ORDER BY 2 DESC
""").df()

versions = cc.cc_list_versions() print(f"{len(versions)} releases; latest = {cc.cc_resolve_version('latest')}") rel = cc.cc_get_db(tables=["sample", "cruise"]) # release tables as DuckDB views rel.execute("SET enable_progress_bar = false") # no Jupyter progress-bar widgets in a static page rel.sql(""" SELECT dataset_key, count(*) AS samples, min(datetime) AS first, max(datetime) AS last FROM sample WHERE dataset_key LIKE 'calcofi_%' GROUP BY 1 ORDER BY 2 DESC """).df()

27 releases; latest = v2026.08.14

Out[3]:

	dataset_key	samples	first	last
0	calcofi_bottle	931015	1949-02-28 22:42:00	2021-05-13 20:37:17
1	calcofi_mets	77795	2004-01-05 19:07:43	2022-11-21 00:47:00
2	calcofi_ctd-cast	18250	1992-04-18 22:09:40	2026-07-13 15:05:36
3	calcofi_dic	3263	1983-06-15 00:00:00	2021-07-20 00:00:00
4	calcofi_phyllosoma	1859	1951-07-03 00:00:00	2009-07-30 00:00:00
5	calcofi_phytoplankton	409	NaT	NaT

2. Connect to the team's PostgreSQL database¶

cc_pg_connect(tunnel=True) opens ssh -N calcofi for you (your ~/.ssh/config alias) and authenticates from ~/.pgpass. Nothing here is a secret: the alias name, the database name, and the server version.

In [4]:

con = cc.cc_pg_connect(tunnel=True)
pg_version = con.execute("SHOW server_version").fetchone()[0]
n_files, n_scans = con.execute(
    "SELECT (SELECT count(*) FROM ctd.file), (SELECT count(*) FROM ctd.scan)").fetchone()
print(f"PostgreSQL {pg_version}: {n_files:,} cast files, {n_scans:,} scans, all verbatim and immutable")

con = cc.cc_pg_connect(tunnel=True) pg_version = con.execute("SHOW server_version").fetchone()[0] n_files, n_scans = con.execute( "SELECT (SELECT count(*) FROM ctd.file), (SELECT count(*) FROM ctd.scan)").fetchone() print(f"PostgreSQL {pg_version}: {n_files:,} cast files, {n_scans:,} scans, all verbatim and immutable")

something already listens on localhost:5432 — using it as-is. If that is not the CalCOFI server, use local_port=15432.

PostgreSQL 18.6 (Debian 18.6-1.pgdg13+2): 409 cast files, 10,812,360 scans, all verbatim and immutable

3. The cruise at a glance¶

Every station occupation is a numbered cast_seq (from cast_id: "2607_001d" → 1) with a down- and an upcast at the same position. That number cross-references the map, the section, the triage table and the profiles.

In [5]:

casts = cc.cc_ctd_casts(con, study)
print(f"{study} (cruise_key {casts.cruise_key.iloc[0]}): {casts.cast_seq.nunique()} occupations, "
      f"{len(casts)} casts, {casts.datetime_utc.min():%Y-%m-%d} → {casts.datetime_utc.max():%Y-%m-%d}, "
      f"max depth {casts.depth_max.max():.0f} m, {casts.n_scans.sum():,} scans")
casts.head(4)

casts = cc.cc_ctd_casts(con, study) print(f"{study} (cruise_key {casts.cruise_key.iloc[0]}): {casts.cast_seq.nunique()} occupations, " f"{len(casts)} casts, {casts.datetime_utc.min():%Y-%m-%d} → {casts.datetime_utc.max():%Y-%m-%d}, " f"max depth {casts.depth_max.max():.0f} m, {casts.n_scans.sum():,} scans") casts.head(4)

2607SH (cruise_key 2026-07-3322): 72 occupations, 144 casts, 2026-06-30 → 2026-07-13, max depth 747 m, 56,993 scans

Out[5]:

	cast_seq	file_id	cast_id	cast_dir	ord_occ	sta_id	line	sta	datetime_utc	lat	lon	n_scans	depth_min	depth_max	cruise_key	data_stage	is_best_stage
0	1	559	2607_001d	D	001	093.3 028.0	93.3	28.0	2026-06-30 01:23:31	32.909441	-117.397487	517	2.0	518.0	2026-07-3322	preliminary_without_bottle	True
1	1	560	2607_001u	U	001	093.3 028.0	93.3	28.0	2026-06-30 01:35:26	32.907397	-117.400851	517	2.0	518.0	2026-07-3322	preliminary_without_bottle	True
2	2	559	2607_002d	D	002	093.4 026.4	93.4	26.4	2026-06-30 04:03:02	32.946628	-117.283882	20	2.0	21.0	2026-07-3322	preliminary_without_bottle	True
3	2	560	2607_002u	U	002	093.4 026.4	93.4	26.4	2026-06-30 04:04:46	32.946617	-117.283746	21	1.0	21.0	2026-07-3322	preliminary_without_bottle	True

In [6]:

show(cc.cc_station_map(casts, title=f"{study} — station occupations by cast_seq"))

show(cc.cc_station_map(casts, title=f"{study} — station occupations by cast_seq"))

cc_ctd_scans() reads from ctd.v_scan_qc, so each measurement column arrives with its accepted QC flag (*_qc) and accepted correction (*_fix) — empty until curators accept something, by construction.

In [7]:

scans = cc.cc_ctd_scans(con, study, columns=[variable, *sensor_pair])
print(f"{len(scans):,} scans × {scans.shape[1]} columns; QC companions: "
      f"{[c for c in scans.columns if c.endswith(('_qc', '_fix'))]}")
show(cc.cc_section_plot(scans, casts, variable, units=units, title=f"{study} — {variable} section"))

scans = cc.cc_ctd_scans(con, study, columns=[variable, *sensor_pair]) print(f"{len(scans):,} scans × {scans.shape[1]} columns; QC companions: " f"{[c for c in scans.columns if c.endswith(('_qc', '_fix'))]}") show(cc.cc_section_plot(scans, casts, variable, units=units, title=f"{study} — {variable} section"))

56,993 scans × 18 columns; QC companions: ['tempave_qc', 'tempave_fix', 'temp1_qc', 'temp1_fix', 'temp2_qc', 'temp2_fix']

4. Run the portable QC rules¶

Each rule returns evidence — a DataFrame of suspect scans — and changes nothing. They are the same SQL the ctd-qaqc app runs (metadata/qc_rules/).

In [8]:

spikes = cc.cc_qc_spike(con, study, variable, spike_threshold=0.5, neighbour_tol=0.5)
pair   = cc.cc_qc_sensor_pair(con, study, *sensor_pair, threshold=0.05)
oob    = cc.cc_qc_range(con, study, variable, valid_min=-3, valid_max=40)
print(f"spikes: {len(spikes)} on {spikes.cast_id.nunique()} casts "
      f"({(spikes.cast_dir == 'U').mean():.0%} on upcasts)\n"
      f"sensor pair |{sensor_pair[0]} - {sensor_pair[1]}| > 0.05: {len(pair)} scans\n"
      f"outside [-3, 40] {units}: {len(oob)} values")
spikes.sort_values("excursion", ascending=False).head(5)

spikes = cc.cc_qc_spike(con, study, variable, spike_threshold=0.5, neighbour_tol=0.5) pair = cc.cc_qc_sensor_pair(con, study, *sensor_pair, threshold=0.05) oob = cc.cc_qc_range(con, study, variable, valid_min=-3, valid_max=40) print(f"spikes: {len(spikes)} on {spikes.cast_id.nunique()} casts " f"({(spikes.cast_dir == 'U').mean():.0%} on upcasts)\n" f"sensor pair |{sensor_pair[0]} - {sensor_pair[1]}| > 0.05: {len(pair)} scans\n" f"outside [-3, 40] {units}: {len(oob)} values") spikes.sort_values("excursion", ascending=False).head(5)

spikes: 414 on 9 casts (59% on upcasts)
sensor pair |temp1 - temp2| > 0.05: 3942 scans
outside [-3, 40] degC: 197 values

Out[8]:

	scan_id	cast_id	cast_dir	depth	value	value_above	value_below	excursion	neighbour_gap
25	7645702	2607_005d	D	219.0	7.8735	10.7156	10.6163	2.7925	0.0993
204	7673968	2607_005u	U	516.0	0.4526	-1.9045	-2.3695	2.5896	0.4650
125	7674320	2607_005u	U	164.0	2.2746	0.2813	0.2593	2.0043	0.0220
103	7674413	2607_005u	U	71.0	-0.2618	1.8092	1.6752	2.0040	0.1340
134	7674280	2607_005u	U	204.0	0.4755	2.3674	2.4872	1.9518	0.1198

5. Triage — which casts most need a human¶

In [9]:

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.head(8)

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.head(8)

Out[9]:

	cast_seq	n_flags	depth_min	depth_max	value_min	value_max	n_ctd_spike_v1	n_ctd_value_out_of_range_v1	pct_scans_flagged
0	6	325	7.0	514.0	-5.0872	11.8330	206	119	31.46
1	5	258	11.0	516.0	-10.9123	10.2011	180	78	25.05
2	1	23	29.0	226.0	9.0743	17.0211	23	0	2.22
3	4	2	10.0	11.0	18.3892	18.8822	2	0	5.13
4	20	1	62.0	62.0	16.0585	16.0585	1	0	0.10
5	59	1	17.0	17.0	13.5915	13.5915	1	0	2.50
6	70	1	9.0	9.0	14.3537	14.3537	1	0	0.10

In [10]:

show(cc.cc_profile_explorer(scans, variable, flags=candidates, units=units, default=worst,
                            title=f"{study} — {variable} profiles (dropdown: any cast_seq)"))

show(cc.cc_profile_explorer(scans, variable, flags=candidates, units=units, default=worst, title=f"{study} — {variable} profiles (dropdown: any cast_seq)"))

6. The ledger as it stands (read)¶

cc_flags() is the flag ledger joined to its scans — who proposed what, and its fate. Reading it is free; only the curators' accepted verdicts affect any product.

In [11]:

ledger = cc.cc_flags(con, study=study)
print(ledger.status.value_counts().to_dict())
ledger[["flag_id", "cast_id", "depth", "variable", "qual_label", "rule_key", "status", "created_by"]].tail(5)

ledger = cc.cc_flags(con, study=study) print(ledger.status.value_counts().to_dict()) ledger[["flag_id", "cast_id", "depth", "variable", "qual_label", "rule_key", "status", "created_by"]].tail(5)

{'proposed': 574}

Out[11]:

	flag_id	cast_id	depth	variable	qual_label	rule_key	status	created_by
569	988	2607_006d	183.0	tempave	missing	ctd_value_out_of_range_v1	proposed	bebest
570	989	2607_006d	189.0	tempave	missing	ctd_value_out_of_range_v1	proposed	bebest
571	990	2607_006d	200.0	tempave	missing	ctd_value_out_of_range_v1	proposed	bebest
572	991	2607_006d	201.0	tempave	missing	ctd_value_out_of_range_v1	proposed	bebest
573	992	2607_006d	239.0	tempave	missing	ctd_value_out_of_range_v1	proposed	bebest

7. What proposing looks like — shown, not run¶

This is the one step that writes. It is idempotent (a scan already carrying a proposed or accepted flag for the variable is skipped), it records your username, and it is reversible by you while still proposed:

n = 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")

mine = cc.cc_flags(con, study, status="proposed").query("created_by == @user")
cc.cc_withdraw_flags(con, mine.flag_id, note="second thoughts")   # undo: withdrawn, still audited

Curators accept or reject with one UPDATE (pgAdmin, psql, R or Python); everyone else gets a clean permission error — see the notebook.

8. Derive the clean 1 m bins — computed, not published¶

cc_bin_1m() bins ctd.v_scan_clean — originals with accepted fixes substituted and accepted-bad values NULLed. Without write_table= it is a pure computation that returns a DataFrame; with it, the same result lands as a shared table in the work schema.

In [12]:

import plotly.graph_objects as go

bins = cc.cc_bin_1m(con, study, variable, cast_dir="D")       # no write_table -> nothing written
print(f"{len(bins):,} (cast × 1 m) bins")

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=520, 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")
show(fig)

import plotly.graph_objects as go bins = cc.cc_bin_1m(con, study, variable, cast_dir="D") # no write_table -> nothing written print(f"{len(bins):,} (cast × 1 m) bins") 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=520, 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") show(fig)

28,479 (cast × 1 m) bins

Until curators accept flags the two coincide; each acceptance moves only the blue line.

9. Both stores in one query¶

The release is what the public sees; PostgreSQL is the work in progress. They share cruise_key, and DuckDB can read both at once — cc_pg_attach() attaches PostgreSQL (read-only by default) as catalog pg next to the release views.

In [13]:

cc.cc_pg_attach(rel)
rel.sql("""
  WITH qc AS (                                    -- PostgreSQL: flags per cruise
    SELECT fi.cruise_key, fi.study,
           count(*) FILTER (WHERE f.status = 'proposed') AS flags_proposed,
           count(*) FILTER (WHERE f.status = 'accepted') AS flags_accepted
    FROM pg.ctd.flag f JOIN pg.ctd.file fi USING (file_id)
    GROUP BY 1, 2)
  SELECT qc.study, qc.cruise_key, qc.flags_proposed, qc.flags_accepted,
         count(s.sample_key) AS casts_in_release  -- release: what is public today
  FROM qc LEFT JOIN sample s
    ON s.cruise_key = qc.cruise_key AND s.dataset_key = 'calcofi_ctd-cast'
  GROUP BY ALL ORDER BY qc.cruise_key DESC
""").df()

cc.cc_pg_attach(rel) rel.sql(""" WITH qc AS ( -- PostgreSQL: flags per cruise SELECT fi.cruise_key, fi.study, count(*) FILTER (WHERE f.status = 'proposed') AS flags_proposed, count(*) FILTER (WHERE f.status = 'accepted') AS flags_accepted FROM pg.ctd.flag f JOIN pg.ctd.file fi USING (file_id) GROUP BY 1, 2) SELECT qc.study, qc.cruise_key, qc.flags_proposed, qc.flags_accepted, count(s.sample_key) AS casts_in_release -- release: what is public today FROM qc LEFT JOIN sample s ON s.cruise_key = qc.cruise_key AND s.dataset_key = 'calcofi_ctd-cast' GROUP BY ALL ORDER BY qc.cruise_key DESC """).df()

Out[13]:

	study	cruise_key	flags_proposed	flags_accepted	casts_in_release
0	2607SH	2026-07-3322	574	0	122
1	2304SH	2023-04-3322	1	0	226

In [14]:

con.close()
cc.cc_pg_tunnel_close()

con.close() cc.cc_pg_tunnel_close()

Session info¶

The provenance of every output above — package versions and the server they came from.

In [15]:

print(cc.cc_session_info(extra={"postgresql": pg_version,
                                "database": "calcofi @ ssh.calcofi.io (schema ctd)",
                                "release": cc.cc_resolve_version("latest")}))

print(cc.cc_session_info(extra={"postgresql": pg_version, "database": "calcofi @ ssh.calcofi.io (schema ctd)", "release": cc.cc_resolve_version("latest")}))

rendered_utc    2026-08-21 20:17:13 UTC
python          3.14.5 (CPython)
platform        macOS-15.7.1-arm64-arm-64bit-Mach-O
executable      /Users/bbest/Github/CalCOFI/calcofi4py/.venv/bin/python3.14

packages
  calcofi4py    0.3.6
  duckdb        1.5.5
  psycopg       3.3.4
  pandas        3.0.5
  plotly        6.9.0
  numpy         2.5.2
  jupyter_core  5.9.1
  ipykernel     7.3.0

environment
  postgresql    18.6 (Debian 18.6-1.pgdg13+2)
  database      calcofi @ ssh.calcofi.io (schema ctd)
  release       v2026.08.14