Streaming laboratory data with physical units

// Synthetic prefix/full-source comparison with physical units and missing values.
// Requires the updated native Aner development runtime and the supplied sensor fixture.
// Run from the directory containing examples/fixtures/data_execution/sensor_trials.csv.
// Counts/statistics are deterministic; elapsed time and read-ahead are not benchmarks.
import aner.data
import aner.units
import aner.math

fn close(actual: Float64, expected: Float64, tolerance: Float64 = 0.000000001) -> Bool {
    return math.abs(actual - expected) <= tolerance
}
fn verify(label: String, condition: Bool) -> Unit {
    print(label)
    print(condition)
}

var schema = data.schema()
schema = data.column(schema, "row_id", "String", false)
schema = data.column(schema, "site", "String", false)
schema = schema.unit_column("temperature_f", "degF", true)
schema = schema.unit_column("distance_cm", "cm", false)
schema = schema.unit_column("elapsed_s", "s", false)
schema = data.column(schema, "signal", "Float64", true)
schema = data.column(schema, "cohort", "String", false)
let source = data.scan_csv("examples/fixtures/data_execution/sensor_trials.csv", schema, "NA")
    .convert_unit("temperature_f", "degC")
    .convert_unit("distance_cm", "m")
let execution: DataExecution = data.execution(timeout_ms: 30000)
let controlled = data.with_execution(source, execution)

// The rare category first appears at row ten. A six-row cap bounds this search.
let rare = data.where_eq(controlled, "cohort", "rare")
let rare_prefix = data.bounded(rare, max_scanned_rows: 6, max_read_bytes: 0, timeout_ms: 30000)
let peek = data.preview(rare_prefix, max_rows: 2, max_bytes: 1048576, batch_rows: 2)
verify("Rare preview finds no rows in the bounded prefix", data.rows(peek) == 0 && data.selected_rows(peek) == 0 && data.returned_rows(peek) == 0)
verify("Rare preview reports six validated rows and an incomplete stop", data.scanned_rows(peek) == 6 && !data.complete(peek) && data.stop_reason(peek) == "scan_rows")
verify("Preview receipt reports the operation and accounting fields", data.receipt(peek).contains("preview") && data.receipt(peek).contains("header_validated") && data.receipt(peek).contains("peak_bytes"))

// Bounded summaries also describe only their explicitly validated prefix.
let prefix = data.bounded(controlled, max_scanned_rows: 6, max_read_bytes: 0, timeout_ms: 30000)
let partial = data.summarize(prefix, max_bytes: 1048576, batch_rows: 2)
verify("Prefix summary is incomplete with six selected aggregate rows", !data.complete(partial) && data.stop_reason(partial) == "scan_rows" && data.scanned_rows(partial) == 6 && data.selected_rows(partial) == 6 && data.returned_rows(partial) == 0)
verify("Prefix statistics retain present and missing counts", data.valid(partial, "temperature_f") == 5 && data.missing(partial, "temperature_f") == 1 && data.valid(partial, "signal") == 5 && data.missing(partial, "signal") == 1)
verify("Prefix temperature mean and population spread use only observed rows", close(data.unit_mean(partial, "temperature_f", "degC").value(), 24.0) && close(data.unit_std(partial, "temperature_f", "delta_degC").value(), math.sqrt(8.0)))

// An output-row cap stops immediately after two matching, fully validated records.
let routine = data.where_eq(prefix, "cohort", "routine")
let first_two = data.preview(routine, max_rows: 2, max_bytes: 1048576, batch_rows: 2)
verify("Output-capped preview preserves the first two text identifiers", data.rows(first_two) == 2 && data.text(first_two, 0, "row_id") == "0001" && data.text(first_two, 1, "row_id") == "0002")
verify("Output cap is not EOF or whole-source validation", !data.complete(first_two) && data.stop_reason(first_two) == "preview_rows" && data.scanned_rows(first_two) == 2 && data.selected_rows(first_two) == 2 && data.returned_rows(first_two) == 2)

// Raise the source cap; the same controller can be reused sequentially.
let full_plan = data.bounded(controlled, max_scanned_rows: 100, max_read_bytes: 0, timeout_ms: 30000)
let full = data.summarize(full_plan, max_bytes: 1048576, batch_rows: 2)
verify("Full summary proves EOF and validates all twelve source records", data.complete(full) && data.stop_reason(full) == "eof" && data.scanned_rows(full) == 12 && data.rows(full) == 12)
verify("Full streaming summary returns counters rather than table records", data.selected_rows(full) == 12 && data.returned_rows(full) == 0)
verify("Full missing counts remain explicit", data.valid(full, "temperature_f") == 10 && data.missing(full, "temperature_f") == 2 && data.valid(full, "signal") == 10 && data.missing(full, "signal") == 2)
verify("Point minima and maxima retain Celsius type and meaning", type(data.unit_min(full, "temperature_f", "degC")) == "Point<degC>" && close(data.unit_min(full, "temperature_f", "degC").value(), 20.0) && close(data.unit_max(full, "temperature_f", "degC").value(), 38.0))
verify("Full temperature mean is 29 degC and spread is sqrt(33) delta_degC", close(data.unit_mean(full, "temperature_f", "degC").value(), 29.0) && data.unit_std(full, "temperature_f", "delta_degC").unit() == "delta_degC" && close(data.unit_std(full, "temperature_f", "delta_degC").value(), math.sqrt(33.0)))
verify("Primitive signal extrema and mean use ten present observations", data.min(full, "signal") == 1.0 && data.max(full, "signal") == 12.0 && close(data.mean(full, "signal"), 6.7))
verify("Primitive signal population spread is sqrt(12.81)", close(data.std(full, "signal"), math.sqrt(12.81)))
verify("Distance summaries retain meter Quantities", type(data.unit_max(full, "distance_cm", "m")) == "Quantity<m>" && close(data.unit_min(full, "distance_cm", "m").value(), 1.0) && close(data.unit_max(full, "distance_cm", "m").value(), 3.2) && close(data.unit_mean(full, "distance_cm", "m").value(), 2.1))
verify("Distance population spread is sqrt(143/300) meters", close(data.unit_std(full, "distance_cm", "m").value(), math.sqrt(143.0 / 300.0)))
verify("Receipt counters report managed memory, read bytes and elapsed milliseconds", data.peak_bytes(full) > 0 && data.peak_bytes(full) <= 1048576 && data.read_bytes(full) > 0 && data.elapsed_ms(full) >= 0)
verify("Controller latest-run receipt does not alter earlier result receipts", data.complete(execution) && data.scanned_rows(execution) == 12 && data.scanned_rows(peek) == 6 && data.stop_reason(peek) == "scan_rows" && !data.complete(partial))

// Rare rows exist in the tail, so the empty prefix did not establish their absence.
let rare_full = data.summarize(data.where_eq(full_plan, "cohort", "rare"), max_bytes: 1048576, batch_rows: 2)
verify("Full rare summary finds the two late observations", data.complete(rare_full) && data.scanned_rows(rare_full) == 12 && data.selected_rows(rare_full) == 2 && data.returned_rows(rare_full) == 0)
verify("Rare temperature mean and spread are 35 degC and 1 delta_degC", close(data.unit_mean(rare_full, "temperature_f", "degC").value(), 35.0) && close(data.unit_std(rare_full, "temperature_f", "delta_degC").value(), 1.0))
verify("Rare receipt contains completion, status and stop reason", data.receipt(rare_full).contains("complete") && data.receipt(rare_full).contains("status") && data.receipt(rare_full).contains("stop_reason"))

// Source code stays synchronous: cancellation inspection does not start a job.
let cancelled = data.execution(timeout_ms: 0)
let alias = cancelled
data.cancel(cancelled)
verify("Controller aliases share sticky cancellation", data.is_cancelled(cancelled) && data.is_cancelled(alias) && data.scanned_rows(alias) == 0)