06 · Data has a schema¶

A file is not “just numbers”. It is a contract between the program that wrote it and every program that will interpret it.

Decision: does the input mean what the analysis assumes?

Output: a validated named schema and a reduced data product.

Frame¶

The teaching log promises:

Column	Meaning	Validity
`t`	dimensionless time	finite, strictly increasing
`dt`	time step	finite, positive
`kinetic_energy`	integrated kinetic energy	finite, non-negative
`h_min`	minimum length scale	finite, positive

Validation turns that prose into executable boundaries.

Predict¶

For each corrupted input, decide whether to stop, skip a row, or record missing data:

h_min column absent;
one kinetic_energy value is "failed";
time is repeated after a restart;
h_min is zero;
the file has a header but no rows.

There is no universal answer. The point is to choose deliberately and record the policy.

Use names, not positions¶

with path.open(newline="", encoding="utf-8") as stream:
    reader = csv.DictReader(stream)
    rows = list(reader)

Now row["kinetic_energy"] carries meaning. row[7] carries an undocumented dependency on column order.

Check the header before conversion:

required = {"t", "dt", "kinetic_energy", "h_min"}
missing = required.difference(reader.fieldnames or [])
if missing:
    raise ValueError(f"missing required columns: {sorted(missing)}")

Validate at the boundary¶

The reference reader converts and checks once:

log = load_basilisk_log("data/basilisk_log.csv")

Downstream functions receive a SimulationLog, not an arbitrary dictionary. That does not make invalid data impossible, but it creates one visible place where the contract is enforced.

"""Validated readers for the small teaching datasets."""

from __future__ import annotations

import csv
from dataclasses import dataclass
from pathlib import Path

import numpy as np
from numpy.typing import NDArray

FloatArray = NDArray[np.float64]


@dataclass(frozen=True)
class SimulationLog:
    """A minimal, explicit schema for a Basilisk-style time-series log."""

    time: FloatArray
    time_step: FloatArray
    kinetic_energy: FloatArray
    minimum_length: FloatArray


@dataclass(frozen=True)
class RegimeCase:
    """One synthetic case in the teaching regime map."""

    case_id: str
    ohnesorge: float
    bond: float
    outcome: str


def _rows(path: Path) -> tuple[list[str], list[dict[str, str]]]:
    if not path.is_file():
        raise FileNotFoundError(path)
    with path.open(newline="", encoding="utf-8") as stream:
        reader = csv.DictReader(stream)
        if reader.fieldnames is None:
            raise ValueError(f"{path} has no header")
        return reader.fieldnames, list(reader)


def _require_columns(path: Path, fields: list[str], required: set[str]) -> None:
    missing = sorted(required.difference(fields))
    if missing:
        raise ValueError(f"{path} is missing required columns: {', '.join(missing)}")


def load_basilisk_log(path: str | Path) -> SimulationLog:
    """Load and validate a four-column Basilisk-style teaching log."""

    source = Path(path)
    fields, rows = _rows(source)
    required = {"t", "dt", "kinetic_energy", "h_min"}
    _require_columns(source, fields, required)
    if not rows:
        raise ValueError(f"{source} contains no data rows")

    columns: dict[str, list[float]] = {name: [] for name in required}
    for line_number, row in enumerate(rows, start=2):
        for name in required:
            try:
                columns[name].append(float(row[name]))
            except (TypeError, ValueError) as error:
                raise ValueError(
                    f"{source}:{line_number}: {name} is not a number",
                ) from error

    time = np.asarray(columns["t"])
    time_step = np.asarray(columns["dt"])
    kinetic_energy = np.asarray(columns["kinetic_energy"])
    minimum_length = np.asarray(columns["h_min"])

    if not all(
        np.all(np.isfinite(values))
        for values in (time, time_step, kinetic_energy, minimum_length)
    ):
        raise ValueError(f"{source} contains non-finite values")
    if np.any(np.diff(time) <= 0):
        raise ValueError(f"{source}: t must be strictly increasing")
    if np.any(time_step <= 0):
        raise ValueError(f"{source}: dt must be strictly positive")
    if np.any(kinetic_energy < 0):
        raise ValueError(f"{source}: kinetic_energy cannot be negative")
    if np.any(minimum_length <= 0):
        raise ValueError(f"{source}: h_min must be strictly positive")

    return SimulationLog(time, time_step, kinetic_energy, minimum_length)

Notice the line number in conversion errors. “Could not convert string to float” is much less useful than “run.csv:417: h_min is not a number”.

Preserve raw and reduced data¶

A good workflow does not overwrite the source.

raw log.csv
   ↓ validated reader
reduced summary.csv
   ↓ figure script
figure.pdf

Reduced data should include enough identifiers and metadata to trace each row back to a case. A figure should not be the only surviving representation of the result.

The public Jumping-Drops/postProcess pipeline documents a 12-column schema, per-snapshot processing, energy integration, and reduced CSV/PNG outputs. The reusable idea is not its exact columns. It is that the schema and execution policy are written down.

Paths are data too¶

from pathlib import Path

root = Path(__file__).resolve().parents[1]
input_path = root / "data" / "log.csv"

This anchors a repository-relative path to the script, rather than to whichever directory the caller happened to be in. Avoid hard-coded /Users/name/... paths in shared code.

Command-line paths are even better when the input is a genuine choice:

python analyse.py run/log.csv --output figures/run-12.png

Record provenance¶

At minimum, a reduced dataset should state:

source file or case identifier;
source checksum or immutable commit where practical;
schema and units;
program version or commit;
parameters controlling the reduction;
creation date.

For small public capsules, a README plus manifest.sha256 is enough.

Large data need the same contract¶

Do not load a 100 GB file merely because pandas.read_csv can be typed in one line. Decide:

which columns are required;
whether rows can be streamed or chunked;
whether one case can be processed independently;
where reduced outputs belong;
how a partial run resumes.

On HPC, process independent snapshots in a single-node batch job or suitable data-transfer node. Never turn the login node into an analysis workstation.

Verify¶

Complete Exercise 05. After the tests pass, add one test for a repeated time value. Decide whether the starter's summary function should own that check or whether a separate reader should.

Reflect¶

Write the schema for one dataset you currently use. If the column meaning lives only in your memory or a plotting script, the dataset is not yet self-describing enough to collaborate on.