Skip to content

API reference

Everything is importable from the top level: import calcofi4py as cc.

Public database releases

cc_get_db 

cc_get_db(version='latest', tables=None, supplemental=False, con=None)

DuckDB connection with every release table registered as a view.

Parameters mirror calcofi4r::cc_get_db():

  • version: "latest" (default) or a pinned "vYYYY.MM.DD" — pin for reproducibility, releases are immutable.
  • tables: restrict to these table names (also the way to opt in to a single supplemental table by name).
  • supplemental: include the supplemental tables (obs_ctd_full ~216M rows, obs_mets_full ~20M) that are hosted + cataloged but excluded by default.
  • con: register the views on an existing DuckDB connection (e.g. one that already has the PostgreSQL database attached) instead of a new in-memory one.

con = cc_get_db() con.sql("SELECT count(*) FROM sample").fetchone()

Source code in src/calcofi4py/release.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def cc_get_db(
    version: str = "latest",
    tables: list[str] | None = None,
    supplemental: bool = False,
    con: duckdb.DuckDBPyConnection | None = None,
) -> duckdb.DuckDBPyConnection:
    """DuckDB connection with every release table registered as a view.

    Parameters mirror ``calcofi4r::cc_get_db()``:

    - ``version``: ``"latest"`` (default) or a pinned ``"vYYYY.MM.DD"`` —
      pin for reproducibility, releases are immutable.
    - ``tables``: restrict to these table names (also the way to opt in to a
      single supplemental table by name).
    - ``supplemental``: include the supplemental tables (``obs_ctd_full``
      ~216M rows, ``obs_mets_full`` ~20M) that are hosted + cataloged but
      excluded by default.
    - ``con``: register the views on an existing DuckDB connection (e.g. one
      that already has the PostgreSQL database attached) instead of a new
      in-memory one.

    >>> con = cc_get_db()
    >>> con.sql("SELECT count(*) FROM sample").fetchone()
    """
    version = cc_resolve_version(version)
    catalog = cc_catalog(version)
    if con is None:
        con = duckdb.connect()

    tbls = catalog["tables"]
    if tables is not None:
        tbls = [t for t in tbls if t["name"] in set(tables)]
    elif not supplemental:
        tbls = [t for t in tbls if not t.get("supplemental")]

    if any(t.get("partitioned") for t in tbls):
        _setup_gcs_httpfs(con)
    else:
        con.execute("INSTALL httpfs; LOAD httpfs;")

    for t in tbls:
        name = t["name"]
        if t.get("partitioned"):
            src = (
                f"read_parquet('{BASE_S3}/{version}/parquet/{name}/**/*.parquet',"
                " hive_partitioning = true)"
            )
        else:
            src = f"read_parquet('{BASE_HTTPS}/{version}/parquet/{name}.parquet')"
        con.execute(f'CREATE OR REPLACE VIEW "{name}" AS SELECT * FROM {src}')

    return con

cc_query 

cc_query(sql, version='latest')

One-shot SQL against a release; returns a duckdb relation.

cc_query(...).df() for a pandas DataFrame, .fetchall() for tuples.

Source code in src/calcofi4py/release.py 
109
110
111
112
113
114
def cc_query(sql: str, version: str = "latest"):
    """One-shot SQL against a release; returns a ``duckdb`` relation.

    ``cc_query(...).df()`` for a pandas DataFrame, ``.fetchall()`` for tuples.
    """
    return cc_get_db(version).sql(sql)

cc_list_versions 

cc_list_versions()

All published release versions (newest first), from versions.json.

Source code in src/calcofi4py/release.py 
25
26
27
def cc_list_versions() -> list[dict]:
    """All published release versions (newest first), from ``versions.json``."""
    return json.loads(_fetch_text(f"{BASE_HTTPS}/versions.json"))["versions"]

cc_catalog 

cc_catalog(version='latest')

The release catalog.json: table names, row counts, partitioned/supplemental flags.

Source code in src/calcofi4py/release.py 
39
40
41
42
def cc_catalog(version: str = "latest") -> dict:
    """The release ``catalog.json``: table names, row counts, partitioned/supplemental flags."""
    version = cc_resolve_version(version)
    return json.loads(_fetch_text(f"{BASE_HTTPS}/{version}/catalog.json"))

cc_resolve_version 

cc_resolve_version(version='latest')

Resolve "latest" to the promoted version string (e.g. v2026.08.14).

Source code in src/calcofi4py/release.py 
30
31
32
33
34
35
36
def cc_resolve_version(version: str = "latest") -> str:
    """Resolve ``"latest"`` to the promoted version string (e.g. ``v2026.08.14``)."""
    if version == "latest":
        return _fetch_text(f"{BASE_HTTPS}/latest.txt").strip().splitlines()[0]
    if not version.startswith("v"):
        raise ValueError(f"version must be 'latest' or like 'v2026.08.14', got {version!r}")
    return version

PostgreSQL (CTD QA/QC working database)

cc_pg_connect 

cc_pg_connect(dbname='calcofi', host=None, port=None, user=None, tunnel=False, **kwargs)

psycopg connection to the CalCOFI PostgreSQL database, defaults resolved.

  • host: postgis on the CalCOFI server (RStudio/Jupyter there), otherwise localhost — the local end of your SSH tunnel. PGHOST overrides.
  • user: PGUSER if set, else the role in your ~/.pgpass for this host/port/db, else your OS user name.
  • password: never passed — libpq reads ~/.pgpass.
  • tunnel=True starts ssh -N calcofi for you first (off-server only).

con = cc_pg_connect(tunnel=True) con.execute("SELECT count(*) FROM ctd.cast").fetchone()

Source code in src/calcofi4py/postgres.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def cc_pg_connect(
    dbname: str = "calcofi",
    host: str | None = None,
    port: int | None = None,
    user: str | None = None,
    tunnel: bool = False,
    **kwargs,
) -> psycopg.Connection:
    """psycopg connection to the CalCOFI PostgreSQL database, defaults resolved.

    - **host**: ``postgis`` on the CalCOFI server (RStudio/Jupyter there),
      otherwise ``localhost`` — the local end of your SSH tunnel. ``PGHOST`` overrides.
    - **user**: ``PGUSER`` if set, else the role in your ``~/.pgpass`` for this
      host/port/db, else your OS user name.
    - **password**: never passed — libpq reads ``~/.pgpass``.
    - ``tunnel=True`` starts ``ssh -N calcofi`` for you first (off-server only).

    >>> con = cc_pg_connect(tunnel=True)
    >>> con.execute("SELECT count(*) FROM ctd.cast").fetchone()
    """
    host = host or _nz(os.environ.get("PGHOST")) or ("postgis" if cc_on_server() else "localhost")
    port = int(port or _nz(os.environ.get("PGPORT")) or 5432)
    if tunnel and host in ("localhost", "127.0.0.1"):
        cc_pg_tunnel(local_port=port)
    user = (
        user
        or _nz(os.environ.get("PGUSER"))
        or cc_pgpass_user(host, port, dbname)
        or os.environ.get("USER", os.environ.get("USERNAME", ""))
    )
    return psycopg.connect(dbname=dbname, host=host, port=port, user=user, **kwargs)

cc_pg_tunnel 

cc_pg_tunnel(ssh_host='calcofi', local_port=5432, remote_port=5432, wait=10.0)

Open ssh -N -L {local_port}:localhost:{remote_port} {ssh_host} in the background.

Uses your ~/.ssh/config alias (host, user, key) so no credentials are handled here; Windows 10+ has ssh.exe built in. Reused while alive; close with :func:cc_pg_tunnel_close. If something already listens on local_port it is left alone (use local_port=15432 in both places if that is not the CalCOFI tunnel).

Source code in src/calcofi4py/postgres.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def cc_pg_tunnel(
    ssh_host: str = "calcofi",
    local_port: int = 5432,
    remote_port: int = 5432,
    wait: float = 10.0,
) -> subprocess.Popen | None:
    """Open ``ssh -N -L {local_port}:localhost:{remote_port} {ssh_host}`` in the background.

    Uses your ``~/.ssh/config`` alias (host, user, key) so no credentials are
    handled here; Windows 10+ has ``ssh.exe`` built in. Reused while alive;
    close with :func:`cc_pg_tunnel_close`. If something already listens on
    ``local_port`` it is left alone (use ``local_port=15432`` in both places if
    that is not the CalCOFI tunnel).
    """
    key = f"{ssh_host}:{local_port}"
    p = _TUNNELS.get(key)
    if p is not None and p.poll() is None:
        return p
    if _port_open("127.0.0.1", local_port):
        print(
            f"something already listens on localhost:{local_port} — using it as-is. "
            "If that is not the CalCOFI server, use local_port=15432."
        )
        return None
    ssh = shutil.which("ssh")
    if not ssh:
        raise RuntimeError(
            "no `ssh` on PATH (Windows: Settings > Optional features > OpenSSH Client)"
        )
    p = subprocess.Popen(  # noqa: S603
        [ssh, "-N", "-o", "ExitOnForwardFailure=yes", "-o", "BatchMode=yes",
         "-L", f"{local_port}:localhost:{remote_port}", ssh_host],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    t0 = time.monotonic()
    while not _port_open("127.0.0.1", local_port):
        if p.poll() is not None:
            err = (p.stderr.read() if p.stderr else b"").decode()
            raise RuntimeError(
                f"ssh exited before the tunnel came up:\n{err}\n"
                f"Check `ssh {ssh_host}` works in a terminal first."
            )
        if time.monotonic() - t0 > wait:
            p.kill()
            raise TimeoutError(f"tunnel did not open within {wait} s")
        time.sleep(0.25)
    _TUNNELS[key] = p
    print(f"SSH tunnel up: localhost:{local_port} -> {ssh_host}:{remote_port}")
    return p

cc_pg_tunnel_close 

cc_pg_tunnel_close(ssh_host='calcofi', local_port=5432)

Stop a tunnel started by :func:cc_pg_tunnel.

Source code in src/calcofi4py/postgres.py 
168
169
170
171
172
173
def cc_pg_tunnel_close(ssh_host: str = "calcofi", local_port: int = 5432) -> None:
    """Stop a tunnel started by :func:`cc_pg_tunnel`."""
    p = _TUNNELS.pop(f"{ssh_host}:{local_port}", None)
    if p is not None and p.poll() is None:
        p.kill()
        print(f"tunnel closed ({ssh_host}:{local_port})")

cc_pg_attach 

cc_pg_attach(con, alias='pg', dbname='calcofi', host=None, port=None, user=None, read_only=True)

ATTACH the PostgreSQL database inside a DuckDB connection.

One DuckDB query can then join the public release tables (from :func:calcofi4py.cc_get_db) with the team's PostgreSQL tables (pg.ctd.flag, pg.work.*). The password comes from ~/.pgpass (DuckDB's postgres extension uses libpq). read_only=False also allows bulk writes from Parquet into PostgreSQL.

con = cc_get_db() cc_pg_attach(con) con.sql("SELECT count(*) FROM pg.ctd.flag").fetchone()

Source code in src/calcofi4py/postgres.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def cc_pg_attach(
    con,
    alias: str = "pg",
    dbname: str = "calcofi",
    host: str | None = None,
    port: int | None = None,
    user: str | None = None,
    read_only: bool = True,
):
    """ATTACH the PostgreSQL database inside a DuckDB connection.

    One DuckDB query can then join the public release tables (from
    :func:`calcofi4py.cc_get_db`) with the team's PostgreSQL tables
    (``pg.ctd.flag``, ``pg.work.*``). The password comes from ``~/.pgpass``
    (DuckDB's postgres extension uses libpq). ``read_only=False`` also allows
    bulk writes from Parquet into PostgreSQL.

    >>> con = cc_get_db()
    >>> cc_pg_attach(con)
    >>> con.sql("SELECT count(*) FROM pg.ctd.flag").fetchone()
    """
    host = host or _nz(os.environ.get("PGHOST")) or ("postgis" if cc_on_server() else "localhost")
    port = int(port or _nz(os.environ.get("PGPORT")) or 5432)
    user = (
        user
        or _nz(os.environ.get("PGUSER"))
        or cc_pgpass_user(host, port, dbname)
        or os.environ.get("USER", os.environ.get("USERNAME", ""))
    )
    con.execute("INSTALL postgres; LOAD postgres;")
    opts = "TYPE postgres" + (", READ_ONLY" if read_only else "")
    con.execute(
        f"ATTACH IF NOT EXISTS 'dbname={dbname} host={host} port={port} user={user}' "
        f"AS {alias} ({opts})"
    )
    return con

cc_pgpass_user 

cc_pgpass_user(host, port, dbname)

The role name recorded in ~/.pgpass for host:port:dbname (first match).

Lets a user who copied the file from the server connect with no PGUSER. Format per line: host:port:database:user:password (password may contain :; * wildcards; # comments).

Source code in src/calcofi4py/postgres.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def cc_pgpass_user(host: str, port: int | str, dbname: str) -> str | None:
    """The role name recorded in ``~/.pgpass`` for host:port:dbname (first match).

    Lets a user who copied the file from the server connect with no ``PGUSER``.
    Format per line: ``host:port:database:user:password`` (password may contain
    ``:``; ``*`` wildcards; ``#`` comments).
    """
    f = _pgpass_path()
    if not f.is_file():
        return None
    for ln in f.read_text().splitlines():
        ln = ln.strip()
        if not ln or ln.startswith("#"):
            continue
        parts = ln.split(":")
        if len(parts) < 5:
            continue
        h, p, d, u = parts[0], parts[1], parts[2], parts[3]
        if (
            h in ("*", host)
            and p in ("*", str(port))
            and d in ("*", dbname)
            and u
        ):
            return u
    return None

cc_on_server 

cc_on_server()

True inside the CalCOFI server containers, where the DB is host postgis.

Source code in src/calcofi4py/postgres.py 
32
33
34
35
36
def cc_on_server() -> bool:
    """True inside the CalCOFI server containers, where the DB is host ``postgis``."""
    return bool(os.environ.get("CALCOFI_ON_SERVER")) or (
        Path("/share/github/CalCOFI").is_dir() and platform.system() == "Linux"
    )

CTD QA/QC (ctd schema)

Read casts and scans, run the portable QC rules, propose flags, derive clean products, and plot — see the worked notebook clean_ctd_cruise-var.

cc_ctd_casts 

cc_ctd_casts(con, study, best_only=True)

One row per physical cast of a cruise, from ctd.cast.

Parameters:

Name Type Description Default
con

psycopg connection (:func:calcofi4py.cc_pg_connect)

 required
study str

source cruise id as in the files, e.g. "2607SH"

 required
best_only bool

only the best data stage per cruise × direction (default); False includes superseded preliminary/duplicate archives

 True

Returns:

Type Description
'pd.DataFrame'

DataFrame with cast_seq (the cast number shared by a down/up pair, from cast_id — "2607_001d" → 1), file_id, cast_id, cast_dir, sta_id, line, sta, datetime_utc, lat, lon, n_scans, depth_min, depth_max, cruise_key
Source code in src/calcofi4py/ctd.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def cc_ctd_casts(con, study: str, best_only: bool = True) -> "pd.DataFrame":
    """One row per physical cast of a cruise, from ``ctd.cast``.

    :param con: psycopg connection (:func:`calcofi4py.cc_pg_connect`)
    :param study: source cruise id as in the files, e.g. ``"2607SH"``
    :param best_only: only the best data stage per cruise × direction (default);
        ``False`` includes superseded preliminary/duplicate archives
    :return: DataFrame with ``cast_seq`` (the cast number shared by a down/up
        pair, from ``cast_id`` — ``"2607_001d"`` → 1), ``file_id, cast_id,
        cast_dir, sta_id, line, sta, datetime_utc, lat, lon, n_scans,
        depth_min, depth_max, cruise_key``
    """
    return _read_sql(con, """
        SELECT (regexp_match(cast_id, '_0*([0-9]+)'))[1]::int AS 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
        FROM ctd.cast WHERE study = %s AND (is_best_stage OR NOT %s)
        ORDER BY datetime_utc, cast_dir
        """, (study, best_only))

cc_ctd_scans 

cc_ctd_scans(con, study, columns=('tempave', 'salt1', 'ox1'), cast_id=None, qc=True)

Scan-level data for a cruise from ctd.v_scan_qc (best stage).

Parameters:

Name Type Description Default
columns Iterable[str]

measurement columns to include (see ctd.scan_column); with qc=True each also brings its <col>_qc (accepted IODE flag) and <col>_fix (accepted corrected value)

 ('tempave', 'salt1', 'ox1')
cast_id str | None

restrict to one cast (e.g. "2607_020d"); default all

 None
qc bool

read from ctd.v_scan_qc (default) vs raw ctd.v_scan_best

 True

Returns:

Type Description
'pd.DataFrame'

DataFrame keyed scan_id, with cast_id, cast_dir, depth, date_time_utc, lat_dec, lon_dec + the requested columns
Source code in src/calcofi4py/ctd.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def cc_ctd_scans(
    con,
    study: str,
    columns: Iterable[str] = ("tempave", "salt1", "ox1"),
    cast_id: str | None = None,
    qc: bool = True,
) -> "pd.DataFrame":
    """Scan-level data for a cruise from ``ctd.v_scan_qc`` (best stage).

    :param columns: measurement columns to include (see ``ctd.scan_column``);
        with ``qc=True`` each also brings its ``<col>_qc`` (accepted IODE flag)
        and ``<col>_fix`` (accepted corrected value)
    :param cast_id: restrict to one cast (e.g. ``"2607_020d"``); default all
    :param qc: read from ``ctd.v_scan_qc`` (default) vs raw ``ctd.v_scan_best``
    :return: DataFrame keyed ``scan_id``, with ``cast_id, cast_dir, depth,
        date_time_utc, lat_dec, lon_dec`` + the requested columns
    """
    cols = [_ident(c) for c in columns]
    view = "ctd.v_scan_qc" if qc else "ctd.v_scan_best"
    extra = ", ".join(
        c if not qc else f"{c}, {c}_qc, {c}_fix" for c in cols)
    where, params = "s.study = %s", [study]
    if cast_id is not None:
        where += " AND s.cast_id = %s"
        params.append(cast_id)
    return _read_sql(con, f"""
        SELECT s.scan_id, s.cast_id,
               (regexp_match(s.cast_id, '_0*([0-9]+)'))[1]::int AS cast_seq,
               f.cast_dir, s.row_num, s.depth,
               s.date_time_utc, s.lat_dec, s.lon_dec, {extra}
        FROM {view} s JOIN ctd.file f USING (file_id)
        WHERE {where}
        ORDER BY s.cast_id, s.depth
        """, params)

cc_qc_spike 

cc_qc_spike(con, study, column='tempave', spike_threshold=0.5, neighbour_tol=0.5)

Single-scan spikes against a locally smooth profile (ctd_spike.sql).

Neighbour agreement is the whole trick: a point qualifies only if it deviates from the midpoint of its neighbours by more than spike_threshold WHILE the neighbours agree with each other within neighbour_tol — otherwise every steep-but-smooth thermocline gradient fires. (Measured on one cruise: naive 92 hits, with neighbour agreement 19 — the other 73 were real gradients.)

Returns:

Type Description
'pd.DataFrame'

one row per suspect scan: scan_id, cast_id, cast_dir, depth, value, value_above, value_below, excursion, neighbour_gap
Source code in src/calcofi4py/ctd.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def cc_qc_spike(
    con,
    study: str,
    column: str = "tempave",
    spike_threshold: float = 0.5,
    neighbour_tol: float = 0.5,
) -> "pd.DataFrame":
    """Single-scan spikes against a locally smooth profile (``ctd_spike.sql``).

    Neighbour agreement is the whole trick: a point qualifies only if it deviates
    from the midpoint of its neighbours by more than ``spike_threshold`` WHILE
    the neighbours agree with each other within ``neighbour_tol`` — otherwise
    every steep-but-smooth thermocline gradient fires. (Measured on one cruise:
    naive 92 hits, with neighbour agreement 19 — the other 73 were real
    gradients.)

    :return: one row per suspect scan: ``scan_id, cast_id, cast_dir, depth,
        value, value_above, value_below, excursion, neighbour_gap``
    """
    c = _ident(column)
    return _read_sql(con, f"""
        WITH x AS (
          SELECT s.scan_id, s.cast_id, f.cast_dir, s.depth, s.{c} AS v,
                 LAG(s.{c})  OVER (PARTITION BY s.file_id, s.cast_id ORDER BY s.depth) AS v_above,
                 LEAD(s.{c}) OVER (PARTITION BY s.file_id, s.cast_id ORDER BY s.depth) AS v_below
          FROM ctd.v_scan_best s JOIN ctd.file f USING (file_id)
          WHERE s.study = %s AND s.{c} IS NOT NULL
        )
        SELECT scan_id, cast_id, cast_dir, depth,
               round(v::numeric, 4)                            AS value,
               round(v_above::numeric, 4)                      AS value_above,
               round(v_below::numeric, 4)                      AS value_below,
               round(abs(v - (v_above + v_below) / 2)::numeric, 4) AS excursion,
               round(abs(v_above - v_below)::numeric, 4)       AS neighbour_gap
        FROM x
        WHERE v_above IS NOT NULL AND v_below IS NOT NULL
          AND abs(v - (v_above + v_below) / 2) > %s
          AND abs(v_above - v_below)           < %s
        ORDER BY cast_id, depth
        """, (study, spike_threshold, neighbour_tol))

cc_qc_sensor_pair 

cc_qc_sensor_pair(con, study, column1='temp1', column2='temp2', threshold=0.05)

Primary vs secondary sensor disagreement (ctd_sensor1_vs_sensor2).

The source's own *Q codes 1/2 mean "use primary"/"use secondary" precisely because one sensor of a pair can misbehave; a persistent gap between the pair is how that is spotted.

Returns:

Type Description
'pd.DataFrame'

one row per scan where abs(col1 - col2) > threshold
Source code in src/calcofi4py/ctd.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def cc_qc_sensor_pair(
    con,
    study: str,
    column1: str = "temp1",
    column2: str = "temp2",
    threshold: float = 0.05,
) -> "pd.DataFrame":
    """Primary vs secondary sensor disagreement (``ctd_sensor1_vs_sensor2``).

    The source's own ``*Q`` codes 1/2 mean "use primary"/"use secondary"
    precisely because one sensor of a pair can misbehave; a persistent gap
    between the pair is how that is spotted.

    :return: one row per scan where ``abs(col1 - col2) > threshold``
    """
    c1, c2 = _ident(column1), _ident(column2)
    return _read_sql(con, f"""
        SELECT s.scan_id, s.cast_id, f.cast_dir, s.depth,
               s.{c1} AS v1, s.{c2} AS v2,
               round(abs(s.{c1} - s.{c2})::numeric, 4) AS gap
        FROM ctd.v_scan_best s JOIN ctd.file f USING (file_id)
        WHERE s.study = %s AND s.{c1} IS NOT NULL AND s.{c2} IS NOT NULL
          AND abs(s.{c1} - s.{c2}) > %s
        ORDER BY gap DESC
        """, (study, threshold))

cc_qc_range 

cc_qc_range(con, study, column, valid_min=None, valid_max=None)

Values outside declared physical bounds (ctd_value_out_of_range).

Bounds are deliberately generous — they catch the impossible (an unconverted -99 sentinel, a scaling error), they do not police oceanography.

Returns:

Type Description
'pd.DataFrame'

one row per out-of-bounds scan: scan_id, cast_id, depth, value
Source code in src/calcofi4py/ctd.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def cc_qc_range(
    con,
    study: str,
    column: str,
    valid_min: float | None = None,
    valid_max: float | None = None,
) -> "pd.DataFrame":
    """Values outside declared physical bounds (``ctd_value_out_of_range``).

    Bounds are deliberately generous — they catch the impossible (an unconverted
    ``-99`` sentinel, a scaling error), they do not police oceanography.

    :return: one row per out-of-bounds scan: ``scan_id, cast_id, depth, value``
    """
    c = _ident(column)
    viol = []
    if valid_min is not None:
        viol.append(f"s.{c} < %s")
    if valid_max is not None:
        viol.append(f"s.{c} > %s")
    if not viol:
        raise ValueError("give valid_min and/or valid_max")
    sql = f"""
        SELECT s.scan_id, s.cast_id, f.cast_dir, s.depth, s.{c} AS value
        FROM ctd.v_scan_best s JOIN ctd.file f USING (file_id)
        WHERE s.study = %s AND s.{c} IS NOT NULL AND ({' OR '.join(viol)})
        ORDER BY s.cast_id, s.depth
        """
    params = [study] + [b for b in (valid_min, valid_max) if b is not None]
    return _read_sql(con, sql, params)

cc_propose_flags 

cc_propose_flags(con, scan_ids, variable, qual_code, reason, rule_key=None, proposed_value=None, commit=True)

Propose QC flags: one ctd.flag row per scan × variable.

Idempotent: a scan that already carries a proposed or accepted flag for this variable (any proposer) is skipped, so re-running a notebook does not stack duplicates. Curators then accept/reject in SQL or pgAdmin; only accepted flags affect ctd.v_scan_qc / ctd.v_scan_clean.

Parameters:

Name Type Description Default
qual_code int

IODE code from ctd.qual_code — 3 probably_bad, 4 bad, 5 changed (requires proposed_value), 9 missing, …

 required
rule_key str | None

the QC rule that proposed it (e.g. "ctd_spike_v1"), for provenance

 None

Returns:

Type Description
int

number of flags actually inserted
Source code in src/calcofi4py/ctd.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def cc_propose_flags(
    con,
    scan_ids: Iterable[int],
    variable: str,
    qual_code: int,
    reason: str,
    rule_key: str | None = None,
    proposed_value: float | None = None,
    commit: bool = True,
) -> int:
    """Propose QC flags: one ``ctd.flag`` row per scan × variable.

    Idempotent: a scan that already carries a ``proposed`` or ``accepted`` flag
    for this variable (any proposer) is skipped, so re-running a notebook does
    not stack duplicates. Curators then accept/reject in SQL or pgAdmin; only
    **accepted** flags affect ``ctd.v_scan_qc`` / ``ctd.v_scan_clean``.

    :param qual_code: IODE code from ``ctd.qual_code`` — 3 probably_bad, 4 bad,
        5 changed (requires ``proposed_value``), 9 missing, …
    :param rule_key: the QC rule that proposed it (e.g. ``"ctd_spike_v1"``),
        for provenance
    :return: number of flags actually inserted
    """
    ids = [int(i) for i in scan_ids]
    if not ids:
        return 0
    cur = con.execute("""
        INSERT INTO ctd.flag (scan_id, variable, qual_code, proposed_value, rule_key, reason)
        SELECT s.scan_id, %(var)s, %(code)s, %(val)s, %(rule)s, %(reason)s
        FROM ctd.scan s
        WHERE s.scan_id = ANY(%(ids)s)
          AND NOT EXISTS (
            SELECT 1 FROM ctd.flag f
            WHERE f.scan_id = s.scan_id AND f.variable = %(var)s
              AND f.status IN ('proposed', 'accepted'))
        """, {"var": _ident(variable), "code": qual_code, "val": proposed_value,
              "rule": rule_key, "reason": reason, "ids": ids})
    n = cur.rowcount
    if commit:
        con.commit()
    return n

cc_flags 

cc_flags(con, study=None, status=None)

The QC ledger, joined to its scans: who proposed what, where, and its fate.

Parameters:

Name Type Description Default
study str | None

restrict to one cruise; default all

 None
status str | None

proposed / accepted / rejected / withdrawn; default all

 None
Source code in src/calcofi4py/ctd.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def cc_flags(con, study: str | None = None, status: str | None = None) -> "pd.DataFrame":
    """The QC ledger, joined to its scans: who proposed what, where, and its fate.

    :param study: restrict to one cruise; default all
    :param status: ``proposed`` / ``accepted`` / ``rejected`` / ``withdrawn``; default all
    """
    conds, params = ["true"], []
    if study is not None:
        conds.append("fi.study = %s")
        params.append(study)
    if status is not None:
        conds.append("f.status = %s")
        params.append(status)
    return _read_sql(con, f"""
        SELECT f.flag_id, f.scan_id, fi.study, s.cast_id, s.depth, f.variable, f.qual_code,
               q.label AS qual_label, f.proposed_value, f.rule_key, f.reason,
               f.status, f.created_by, f.created_at, f.reviewed_by, f.review_note
        FROM ctd.flag f
        JOIN ctd.qual_code q USING (qual_code)
        JOIN ctd.file fi ON fi.file_id = f.file_id
        LEFT JOIN ctd.scan s USING (scan_id)
        WHERE {' AND '.join(conds)}
        ORDER BY f.flag_id
        """, params)

cc_bin_1m 

cc_bin_1m(con, study, column='tempave', cast_dir='D', write_table=None, commit=True)

Clean 1 m binned averages per cast, from ctd.v_scan_clean.

"Clean" means: accepted fixes substituted, accepted-bad values NULLed — exactly the ledger's verdicts and nothing else. Regenerable at any time, which is the point of keeping originals immutable.

Parameters:

Name Type Description Default
cast_dir str

"D" downcast (default), "U" upcast, or "*" both

 'D'
write_table str | None

also write the result to work.<write_table> (replacing it), so colleagues can query it by name

 None

Returns:

Type Description
'pd.DataFrame'

DataFrame cast_id, cast_dir, depth_m, value, n, sd
Source code in src/calcofi4py/ctd.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def cc_bin_1m(
    con,
    study: str,
    column: str = "tempave",
    cast_dir: str = "D",
    write_table: str | None = None,
    commit: bool = True,
) -> "pd.DataFrame":
    """Clean 1 m binned averages per cast, from ``ctd.v_scan_clean``.

    "Clean" means: accepted fixes substituted, accepted-bad values NULLed —
    exactly the ledger's verdicts and nothing else. Regenerable at any time,
    which is the point of keeping originals immutable.

    :param cast_dir: ``"D"`` downcast (default), ``"U"`` upcast, or ``"*"`` both
    :param write_table: also write the result to ``work.<write_table>``
        (replacing it), so colleagues can query it by name
    :return: DataFrame ``cast_id, cast_dir, depth_m, value, n, sd``
    """
    c = _ident(column)
    conds, params = ["s.study = %s"], [study]
    if cast_dir != "*":
        conds.append("f.cast_dir = %s")
        params.append(cast_dir)
    d = _read_sql(con, f"""
        SELECT s.cast_id, f.cast_dir, floor(s.depth)::int AS depth_m,
               round(avg(s.{c})::numeric, 4)        AS value,
               count(s.{c})                          AS n,
               round(stddev_samp(s.{c})::numeric, 4) AS sd
        FROM ctd.v_scan_clean s JOIN ctd.file f USING (file_id)
        WHERE {' AND '.join(conds)} AND s.{c} IS NOT NULL
        GROUP BY 1, 2, 3
        ORDER BY 1, 3
        """, params)
    if write_table is not None:
        t = _ident(write_table)
        con.execute(f"DROP TABLE IF EXISTS work.{t}")
        con.execute(f"""
            CREATE TABLE work.{t} (
              cast_id text, cast_dir text, depth_m int,
              value double precision, n int, sd double precision)""")
        with con.cursor().copy(
                f"COPY work.{t} (cast_id, cast_dir, depth_m, value, n, sd) FROM STDIN") as cp:
            for row in d.itertuples(index=False):
                cp.write_row([None if pd.isna(v) else v for v in row])
        from psycopg import sql as _sql
        con.execute(_sql.SQL("COMMENT ON TABLE work.{} IS {}").format(
            _sql.Identifier(t),
            _sql.Literal(f"1 m binned {c} for {study} ({cast_dir}) from ctd.v_scan_clean — regenerable; calcofi4py.cc_bin_1m")))
        if commit:
            con.commit()
    return d

cc_station_map 

cc_station_map(casts, zoom=4.7, title=None)

Map of station occupations: one labeled marker per cast_seq.

Every occupation has a down- and an upcast at the same position, so plotting casts individually just stacks markers; here the pair collapses to one point (downcast position preferred) labeled with its cast_seq — the number to cross-reference in every other figure and table. Hover carries the station, time, both directions' scan counts and the max depth. Occupations whose source files carry -99 positions are absent from the map but present in every table.

One markers+text trace, and the labels MUST be strings: integer text serializes to plotly's binary typed-array encoding, which the symbol layer treats as icon names ("Image -15 could not be loaded") and drops. The opposite decomposition — a separate mode="text" trace — does not work either: plotly 3.7 never creates the symbol layer for a text-only scattermap trace, so the labels silently vanish. String labels on the combined trace is the one configuration that renders.

Parameters:

Name Type Description Default
casts 'pd.DataFrame'

from :func:cc_ctd_casts

 required
zoom float

initial map zoom (the CalCOFI grid fits at ~4.7)

 4.7
Source code in src/calcofi4py/ctd.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def cc_station_map(casts: "pd.DataFrame", zoom: float = 4.7, title: str | None = None):
    """Map of station occupations: **one labeled marker per** ``cast_seq``.

    Every occupation has a down- and an upcast at the same position, so plotting
    casts individually just stacks markers; here the pair collapses to one point
    (downcast position preferred) labeled with its ``cast_seq`` — the number to
    cross-reference in every other figure and table. Hover carries the station,
    time, both directions' scan counts and the max depth. Occupations whose
    source files carry ``-99`` positions are absent from the map but present in
    every table.

    One ``markers+text`` trace, and the labels MUST be strings: integer ``text``
    serializes to plotly's binary typed-array encoding, which the symbol layer
    treats as icon names ("Image -15 could not be loaded") and drops. The
    opposite decomposition — a separate ``mode="text"`` trace — does not work
    either: plotly 3.7 never creates the symbol layer for a text-only
    scattermap trace, so the labels silently vanish. String labels on the
    combined trace is the one configuration that renders.

    :param casts: from :func:`cc_ctd_casts`
    :param zoom: initial map zoom (the CalCOFI grid fits at ~4.7)
    """
    _px()  # ensures plotly is installed
    import plotly.graph_objects as go

    d = casts.dropna(subset=["lat", "lon"]).copy()
    agg = (d.sort_values("cast_dir")                      # D before U
           .groupby("cast_seq")
           .agg(lat=("lat", "first"), lon=("lon", "first"),
                station=("sta_id", "first"), time=("datetime_utc", "min"),
                directions=("cast_dir", lambda x: "+".join(x)),
                scans=("n_scans", "sum"), depth_max=("depth_max", "max"))
           .reset_index())
    labels = agg["cast_seq"].astype(int).astype(str).tolist()
    custom = agg[["station", "time", "directions", "scans", "depth_max"]].astype(str).values
    fig = go.Figure()
    fig.add_scattermap(
        lat=agg.lat, lon=agg.lon, mode="markers+text",
        marker=dict(size=10, color="#1f77b4"), name="",
        text=labels, textposition="top center",
        textfont=dict(size=9, color="#1f2d3d"),
        customdata=custom,
        hovertemplate=("<b>cast %{text}</b><br>station %{customdata[0]}<br>"
                       "%{customdata[1]}<br>casts: %{customdata[2]} · scans: %{customdata[3]}<br>"
                       "max depth %{customdata[4]} m<br>%{lat:.3f}, %{lon:.3f}<extra></extra>"))
    fig.update_layout(
        map=dict(style="carto-positron", zoom=zoom,
                 center=dict(lat=float(agg.lat.mean()), lon=float(agg.lon.mean()))),
        showlegend=False, height=560, title=title,
        margin=dict(l=0, r=0, t=40 if title else 0, b=0))
    return fig

cc_profile_plot 

cc_profile_plot(scans, column='tempave', cast_ids=None, flags=None, units='', title=None)

Depth profiles with down- and upcasts distinguished; optional flag overlay.

Parameters:

Name Type Description Default
scans 'pd.DataFrame'

from :func:cc_ctd_scans (needs depth, cast_id, cast_dir and column)

 required
cast_ids Iterable[str] | None

restrict to these casts (default: all in scans — one trace per cast × direction, thin, so a whole cruise reads as an envelope)

 None
flags 'pd.DataFrame | None'

rows with scan_id (e.g. from :func:cc_qc_spike or :func:cc_flags) drawn as violet × markers on top

 None
Source code in src/calcofi4py/ctd.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
def cc_profile_plot(
    scans: "pd.DataFrame",
    column: str = "tempave",
    cast_ids: Iterable[str] | None = None,
    flags: "pd.DataFrame | None" = None,
    units: str = "",
    title: str | None = None,
):
    """Depth profiles with down- and upcasts distinguished; optional flag overlay.

    :param scans: from :func:`cc_ctd_scans` (needs ``depth, cast_id, cast_dir``
        and ``column``)
    :param cast_ids: restrict to these casts (default: all in ``scans`` — one
        trace per cast × direction, thin, so a whole cruise reads as an envelope)
    :param flags: rows with ``scan_id`` (e.g. from :func:`cc_qc_spike` or
        :func:`cc_flags`) drawn as violet × markers on top
    """
    px = _px()
    d = scans.dropna(subset=[column]).copy()
    if cast_ids is not None:
        d = d[d.cast_id.isin(set(cast_ids))]
    d["direction"] = d["cast_dir"].map({"D": "down", "U": "up"})
    fig = px.line(
        d.sort_values(["cast_id", "depth"]),
        x=column, y="depth", color="direction", line_group="cast_id",
        color_discrete_map={"down": "#1f77b4", "up": "#ff7f0e"},
        hover_data={"cast_id": True, "depth": ":.1f", column: ":.3f"},
        height=600, title=title,
        labels={column: f"{column} ({units})" if units else column, "depth": "depth (m)"})
    fig.update_traces(line=dict(width=1), opacity=0.5)
    if flags is not None and len(flags):
        f = d[d.scan_id.isin(set(flags.scan_id))]
        fig.add_scatter(x=f[column], y=f.depth, mode="markers", name="flagged",
                        marker=dict(symbol="x", size=9, color="#9467bd"))
    fig.update_yaxes(autorange="reversed")
    return fig

cc_profile_explorer 

cc_profile_explorer(scans, column='tempave', flags=None, units='', title=None, default='all')

Depth profiles with a dropdown selector per cast_seq.

The all-casts envelope is unreadable at cruise scale, so this builds one down + up trace pair per occupation and a dropdown ("all casts" + every cast_seq) that isolates a single pair — with its flagged scans (red ×) when flags rows fall on it.

Parameters:

Name Type Description Default
flags 'pd.DataFrame | None'

rows with scan_id (from :func:cc_qc_spike, :func:cc_flags, …) overlaid per selected cast

 None
default int | str

initially selected option — "all" (default) or a cast_seq number (e.g. the worst from :func:cc_flag_summary)

 'all'
Source code in src/calcofi4py/ctd.py 
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
def cc_profile_explorer(
    scans: "pd.DataFrame",
    column: str = "tempave",
    flags: "pd.DataFrame | None" = None,
    units: str = "",
    title: str | None = None,
    default: int | str = "all",
):
    """Depth profiles with a **dropdown selector per** ``cast_seq``.

    The all-casts envelope is unreadable at cruise scale, so this builds one
    down + up trace pair per occupation and a dropdown ("all casts" + every
    ``cast_seq``) that isolates a single pair — with its flagged scans (red ×)
    when ``flags`` rows fall on it.

    :param flags: rows with ``scan_id`` (from :func:`cc_qc_spike`,
        :func:`cc_flags`, …) overlaid per selected cast
    :param default: initially selected option — ``"all"`` (default) or a
        ``cast_seq`` number (e.g. the worst from :func:`cc_flag_summary`)
    """
    px = _px()  # noqa: F841  (ensures plotly is installed)
    import plotly.graph_objects as go

    d = scans.dropna(subset=[column]).copy().sort_values(["cast_seq", "cast_dir", "depth"])
    flag_ids = set(flags.scan_id) if flags is not None and len(flags) else set()
    seqs = sorted(d.cast_seq.dropna().unique())
    fig = go.Figure()
    groups: list[int] = []          # trace -> cast_seq
    for seq in seqs:
        for cdir, color, name in (("D", "#1f77b4", "down"), ("U", "#ff7f0e", "up")):
            t = d[(d.cast_seq == seq) & (d.cast_dir == cdir)]
            if not len(t):
                continue
            fig.add_scatter(
                x=t[column], y=t.depth, mode="lines", legendgroup=name,
                name=f"{name} {int(seq)}", showlegend=False,
                line=dict(width=1, color=color), opacity=0.5,
                hovertemplate=f"cast {int(seq)}{cdir.lower()} · depth %{{y:.1f}} m · %{{x:.3f}}<extra></extra>")
            groups.append(int(seq))
        if flag_ids:
            f = d[(d.cast_seq == seq) & d.scan_id.isin(flag_ids)]
            if len(f):
                fig.add_scatter(
                    x=f[column], y=f.depth, mode="markers", name=f"flagged {int(seq)}",
                    showlegend=False,
                    marker=dict(symbol="x", size=9, color="#9467bd"),
                    hovertemplate=f"FLAG cast {int(seq)} · depth %{{y:.1f}} m · %{{x:.3f}}<extra></extra>")
                groups.append(int(seq))

    n = len(groups)
    buttons = [dict(label="all casts", method="update",
                    args=[{"visible": [True] * n}])]
    for seq in seqs:
        buttons.append(dict(
            label=f"cast {int(seq)}", method="update",
            args=[{"visible": [g == int(seq) for g in groups]}]))
    default_idx = 0 if default == "all" else 1 + seqs.index(default)
    if default != "all":
        vis = buttons[default_idx]["args"][0]["visible"]
        for tr, v in zip(fig.data, vis):
            tr.visible = v
    # the modebar appears on hover at the TOP-RIGHT — a menu there becomes unclickable,
    # so the selector lives top-left and the title moves to the center
    fig.update_layout(
        updatemenus=[dict(buttons=buttons, active=default_idx, x=0.0, xanchor="left",
                          y=1.15, yanchor="top")],
        height=600, title=dict(text=title, x=0.5, xanchor="center"),
        margin=dict(t=90),
        xaxis_title=f"{column} ({units})" if units else column,
        yaxis_title="depth (m)")
    fig.update_yaxes(autorange="reversed")
    return fig

cc_section_plot 

cc_section_plot(scans, casts, column='tempave', units='', title=None)

Section through the cruise: cast_seq (x) × depth (y), colored by value.

Downcasts only — a quick-look transect, not an interpolated product (each vertical stripe is one cast's scans). The x axis is the same cast_seq that labels :func:cc_station_map and indexes :func:cc_flag_summary.

Source code in src/calcofi4py/ctd.py 
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
def cc_section_plot(
    scans: "pd.DataFrame",
    casts: "pd.DataFrame",
    column: str = "tempave",
    units: str = "",
    title: str | None = None,
):
    """Section through the cruise: ``cast_seq`` (x) × depth (y), colored by value.

    Downcasts only — a quick-look transect, not an interpolated product (each
    vertical stripe is one cast's scans). The x axis is the same ``cast_seq``
    that labels :func:`cc_station_map` and indexes :func:`cc_flag_summary`.
    """
    px = _px()
    d = scans[(scans.cast_dir == "D") & scans[column].notna()].copy()
    fig = px.scatter(
        d, x="cast_seq", y="depth", color=column,
        color_continuous_scale="Viridis", height=520, title=title,
        hover_data={"cast_id": True, "cast_seq": True, "depth": ":.1f", column: ":.3f"},
        labels={"cast_seq": "cast_seq", "depth": "depth (m)",
                column: f"{column} ({units})" if units else column})
    fig.update_traces(marker=dict(size=3))
    fig.update_yaxes(autorange="reversed")
    return fig

cc_flag_summary 

cc_flag_summary(ledger, scans, column)

Flags rolled up per cast_seq — the triage table.

Joins the ledger (:func:cc_flags) to the scans it points at and answers "which casts most need a human": flags by rule, the depth span and value range flagged, and the share of the cast's scans affected. Sort descending and start at the top; casts absent from the table have no flags.

Parameters:

Name Type Description Default
ledger 'pd.DataFrame'

from :func:cc_flags (any statuses; filter first if wanted)

 required
scans 'pd.DataFrame'

from :func:cc_ctd_scans — supplies cast_seq and values

 required
column str

the measurement column the flags refer to

 required

Returns:

Type Description
'pd.DataFrame'

one row per flagged cast_seq: n_flags, per-rule_key counts, depth_min/depth_max, value_min/value_max, pct_scans_flagged
Source code in src/calcofi4py/ctd.py 
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
def cc_flag_summary(
    ledger: "pd.DataFrame",
    scans: "pd.DataFrame",
    column: str,
) -> "pd.DataFrame":
    """Flags rolled up per ``cast_seq`` — the triage table.

    Joins the ledger (:func:`cc_flags`) to the scans it points at and answers
    "which casts most need a human": flags by rule, the depth span and value
    range flagged, and the share of the cast's scans affected. Sort descending
    and start at the top; casts absent from the table have no flags.

    :param ledger: from :func:`cc_flags` (any statuses; filter first if wanted)
    :param scans: from :func:`cc_ctd_scans` — supplies ``cast_seq`` and values
    :param column: the measurement column the flags refer to
    :return: one row per flagged ``cast_seq``: ``n_flags``, per-``rule_key``
        counts, ``depth_min/depth_max``, ``value_min/value_max``,
        ``pct_scans_flagged``
    """
    j = ledger.merge(
        scans[["scan_id", "cast_seq", column]], on="scan_id", how="inner")
    if not len(j):
        return pd.DataFrame(columns=["cast_seq", "n_flags"])
    per_rule = (j.pivot_table(index="cast_seq", columns="rule_key",
                              values="flag_id", aggfunc="count", fill_value=0)
                .add_prefix("n_"))
    base = (j.groupby("cast_seq")
            .agg(n_flags=("flag_id", "count"),
                 depth_min=("depth", "min"), depth_max=("depth", "max"),
                 value_min=(column, "min"), value_max=(column, "max")))
    n_scans = scans.groupby("cast_seq").scan_id.count().rename("n_scans")
    out = (base.join(per_rule).join(n_scans)
           .assign(pct_scans_flagged=lambda x: (100 * x.n_flags / x.n_scans).round(2))
           .drop(columns="n_scans")
           .sort_values("n_flags", ascending=False)
           .reset_index())
    out["cast_seq"] = out["cast_seq"].astype(int)
    return out

cc_session_info 

cc_session_info(packages=_DEFAULT_PKGS, repos=None, extra=None)

The Python equivalent of R's devtools::session_info(), as printable text.

Made for the tail of a QA/QC notebook: when the rendered HTML is kept as the archive of a cleaning run, this block records exactly what produced it — Python, platform, package versions (calcofi4py above all), and the git commit of any data-rule directory the run depended on.

Parameters:

Name Type Description Default
packages tuple[str, ...]

distributions to report (missing ones are noted, not fatal)

 _DEFAULT_PKGS
repos dict[str, tuple[str, str | None]] | None

{label: (repo_path, subpath_or_None)} — each reports the last commit touching subpath (e.g. the QC rule registry), plus a dirty-tree warning so an uncommitted rule change cannot masquerade as a committed one

 None
extra dict[str, str] | None

extra {label: value} lines (e.g. the PostgreSQL server version captured while the connection was open)

print(cc_session_info(repos={"qc_rules": (".", "metadata/qc_rules")}))

 None
Source code in src/calcofi4py/session.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def cc_session_info(
    packages: tuple[str, ...] = _DEFAULT_PKGS,
    repos: dict[str, tuple[str, str | None]] | None = None,
    extra: dict[str, str] | None = None,
) -> str:
    """The Python equivalent of R's ``devtools::session_info()``, as printable text.

    Made for the tail of a QA/QC notebook: when the rendered HTML is kept as the
    **archive of a cleaning run**, this block records exactly what produced it —
    Python, platform, package versions (``calcofi4py`` above all), and the git
    commit of any data-rule directory the run depended on.

    :param packages: distributions to report (missing ones are noted, not fatal)
    :param repos: ``{label: (repo_path, subpath_or_None)}`` — each reports the
        last commit touching ``subpath`` (e.g. the QC rule registry), plus a
        dirty-tree warning so an uncommitted rule change cannot masquerade as a
        committed one
    :param extra: extra ``{label: value}`` lines (e.g. the PostgreSQL server
        version captured while the connection was open)

    >>> print(cc_session_info(repos={"qc_rules": (".", "metadata/qc_rules")}))
    """
    lines = [
        f"rendered_utc    {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC",
        f"python          {sys.version.split()[0]} ({platform.python_implementation()})",
        f"platform        {platform.platform()}",
        f"executable      {sys.executable}",
        "",
        "packages",
    ]
    for p in packages:
        try:
            lines.append(f"  {p:<14}{importlib.metadata.version(p)}")
        except importlib.metadata.PackageNotFoundError:
            lines.append(f"  {p:<14}(not installed)")
    if repos:
        lines += ["", "data-rule / repo versions (last commit touching the path)"]
        for label, (path, sub) in repos.items():
            lines.append(f"  {label:<14}{_repo_commit(path, sub)}")
    if extra:
        lines += ["", "environment"]
        for k, v in extra.items():
            lines.append(f"  {k:<14}{v}")
    return "\n".join(lines)