ctddump 0.25.0

Convert oceanographic CTD (Conductivity, Temperature, Depth) data from NetCDF to Parquet or YAML
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! YAML report: for each source file in a merged header YAML, record which core
//! physical columns exist and which biogeochemical/biological measurement
//! variables are present.

use std::error::Error;
use std::path::Path;

use polars::prelude::*;

use crate::cli::ReportFormat;
use crate::report::format;

/// Core physical measurement variables, excluded from the BGC list.
const CORE: [&str; 4] = ["TEMP", "PSAL", "PRES", "DEPH"];

/// Summarise `src` and write the report in `format` to `dest` (or stdout).
pub fn run(format_: ReportFormat, src: &Path, dest: Option<&Path>) -> Result<(), Box<dyn Error>> {
    let content = std::fs::read_to_string(src)
        .map_err(|e| format!("Cannot read {}: {}", src.display(), e))?;
    let doc: serde_yaml::Value = serde_yaml::from_str(&content)
        .map_err(|e| format!("Cannot parse {}: {}", src.display(), e))?;
    let map = doc
        .as_mapping()
        .ok_or_else(|| format!("{}: expected a top-level YAML mapping", src.display()))?;

    let mut filename = Vec::new();
    let mut has_temp = Vec::new();
    let mut has_psal = Vec::new();
    let mut has_pres = Vec::new();
    let mut has_deph = Vec::new();
    let mut has_time_qc = Vec::new();
    let mut has_position_qc = Vec::new();
    let mut extra_params = Vec::new();

    for (key, value) in map {
        let name = key.as_str().unwrap_or("").to_string();
        let vars = value.get("variables");
        let present = |v: &str| vars.and_then(|m| m.get(v)).is_some();

        filename.push(name);
        has_temp.push(present("TEMP"));
        has_psal.push(present("PSAL"));
        has_pres.push(present("PRES"));
        has_deph.push(present("DEPH"));
        // The two profile-level QC flags, as a pair. These are what `dropqc`
        // filters on, and match the `time_qc` / `position_qc` output columns.
        has_time_qc.push(present("TIME_QC"));
        has_position_qc.push(present("POSITION_QC"));
        extra_params.push(detect_extra_params(vars).join(";"));
    }

    let df = DataFrame::new(vec![
        Series::new("filename".into(), filename),
        Series::new("has_temp".into(), has_temp),
        Series::new("has_psal".into(), has_psal),
        Series::new("has_pres".into(), has_pres),
        Series::new("has_deph".into(), has_deph),
        Series::new("has_time_qc".into(), has_time_qc),
        Series::new("has_position_qc".into(), has_position_qc),
        Series::new("extra_params".into(), extra_params),
    ])?;

    format::write_report(&df, format_, dest)?;
    Ok(())
}

/// Detect the extra measurement parameters by complement: a `Float` variable
/// dimensioned `(TIME, DEPTH)`, not a `_QC` flag and not a core physical
/// (`TEMP/PSAL/PRES/DEPH`). This captures biogeochemical/biological parameters
/// (DOXY, FLU2, TUR3, …) as well as other non-core measurements (CNDC, SVEL, …).
/// Returns sorted variable names.
fn detect_extra_params(vars: Option<&serde_yaml::Value>) -> Vec<String> {
    let Some(mapping) = vars.and_then(|v| v.as_mapping()) else {
        return Vec::new();
    };
    let mut out: Vec<String> = Vec::new();
    for (vk, vv) in mapping {
        let Some(vname) = vk.as_str() else { continue };
        if vname.ends_with("_QC") || CORE.contains(&vname) {
            continue;
        }
        let is_float = vv
            .get("data_type")
            .and_then(|d| d.as_str())
            .is_some_and(|d| d.starts_with("Float"));
        let dims_ok = vv
            .get("dimensions")
            .and_then(|d| d.as_sequence())
            .is_some_and(|seq| {
                let dims: Vec<&str> = seq.iter().filter_map(|x| x.as_str()).collect();
                dims == ["TIME", "DEPTH"]
            });
        if is_float && dims_ok {
            out.push(vname.to_string());
        }
    }
    out.sort();
    out
}