The run registry¶

cn.runs records the computations you may want to find, compare, cite or re-run later, with their numerics, code and machine.

It needs no service: an SQLite index plus one directory per run (a JSON record, an HDF5 file for series and fields, solver checkpoints) under a root you declare. Tutorial: 11. Record, query, reproduce: the run registry.

Where runs go¶

Set $CN_RUNS to the directory you choose (or pass root=); without it, a scratch directory of the project is used. A default that resolves under your home directory is refused: runs belong on scratch storage. cn.runs.explain() prints the root and every declared constant.

Recording¶

import numpy as np
import conatus as cn


def heat(kappa):
    """u_t = kappa u_xx, u(x, 0) = sin(3x): the amplitude decays as exp(-9 kappa t)."""
    m = cn.Model(cn.Box(x=cn.periodic(2 * np.pi)), name="heat", kappa=kappa)
    m.scalar("u")
    m.equations("dt(u) = kappa*lap(u)")
    return m.simulate(N=32, scheme="RK443", dt=1e-3).initial(u=lambda x: np.sin(3 * x))


def decay_rate(sim):
    g = sim.grid("u")
    x = np.linspace(0, 2 * np.pi, g.size, endpoint=False)
    return -np.log(2 * np.mean(g * np.sin(3 * x))) / sim.t

sim = heat(0.2)
with cn.runs.track("heat_k0.2", tags=["decay"], params={"note": "first try"}) as run:
    sim.solver.run(200, record=run)       # problem hash, numerics, explain(), start/end state
    rate = decay_rate(sim)
    run.log(t=sim.t, rate=rate)           # a row of series (in memory, written at the end)
    run.result(rate=rate, rel_error=abs(rate - 1.8) / 1.8)   # scalar results, queryable
    run.save_field("u", sim.grid("u"), compression="gzip")   # compression has no default
run.certify("decay rate = 9 kappa (closed form)", evidence={"rel_error": run.results["rel_error"]})

Other ways in:

Call	Records
`solver.run(n, record=True)` (or a name, or a dict of options)	one run per call: numerics, problem hash, parameters, code, hardware, Slurm job, start/end state; returns the run
`sim.run(until=..., record=True)` (native layer; also `cache=True`)	the same, through the native simulation; returns the run
`cn.runs.record(obj, name=..., results=..., fields=..., compression=...)`	a finished object in one call; an `EVPResult` gives eigenvalues, residuals, `growth_max`, `residual_max`
`@cn.runs.track("name")` on a function	one run per call: arguments as parameters, the return value as results / fields

What is recorded, always: the structural hash of the problem, the values of its parameters, the full numerical configuration and explain() text, the executed code (commit, dirty paths, sha256 of the imported tree and of the compiled core), versions, host, GPU, batch job, the command line. Nothing is guessed: an unreadable field is recorded as such.

Recording never blocks or breaks the computation. Scalars and series are kept in memory; fields, files and states are copied and written by a background thread through a bounded queue (a full queue drops and counts, never waits); a disk error is recorded and reported by a RegistryWriteWarning when the run closes, and the numbers are bit-for-bit those of an unrecorded run (tested with an unwritable root).

Statuses: provisional (default), certified, retracted. Changing one needs a reason (and optional evidence such as residuals); the history is kept, a retracted run is never deleted and never reused. A run failing inside a with block is recorded as failed and the exception propagates.

Querying¶

import numpy as np
import conatus as cn


def heat(kappa):
    """u_t = kappa u_xx, u(x, 0) = sin(3x): the amplitude decays as exp(-9 kappa t)."""
    m = cn.Model(cn.Box(x=cn.periodic(2 * np.pi)), name="heat", kappa=kappa)
    m.scalar("u")
    m.equations("dt(u) = kappa*lap(u)")
    return m.simulate(N=32, scheme="RK443", dt=1e-3).initial(u=lambda x: np.sin(3 * x))


def decay_rate(sim):
    g = sim.grid("u")
    x = np.linspace(0, 2 * np.pi, g.size, endpoint=False)
    return -np.log(2 * np.mean(g * np.sin(3 * x))) / sim.t

import socket

for kappa in (0.2, 0.4):                             # two recorded runs to query
    sim = heat(kappa)
    with cn.runs.track(f"heat_k{kappa}", tags=["decay"]) as run:
        sim.solver.run(100, record=run)
        run.log(t=sim.t, rate=decay_rate(sim))
        run.result(rate=decay_rate(sim))
    run.certify("decay rate = 9 kappa", evidence={"rate": run.results["rate"]})

print(cn.runs.find(problem=sim.solver.plan.problem, kappa__gt=0.15, status="certified",
                   tags=["decay"]))
print(cn.runs.find(rate__ge=1.0, host__contains=socket.gethostname()))   # results, index columns
a = cn.runs.find(name="heat_k0.2")[0]                 # or cn.runs.get(id or unique id prefix)
b = cn.runs.find(name="heat_k0.4")[0]
print(cn.runs.compare(a, b).explain())               # parameters, numerics, results, code
cn.runs.compare(a, b).plot("rate", x="t")            # both curves, each on its own abscissae
print(cn.runs.lineage(a).explain())                  # parents, forks, reproductions, cache reuse

cn.runs.to_dataframe(tags="decay") gives the same rows as a pandas table (pandas is optional).

Filters are key=value or key__op=value with op in eq, ne, gt, ge, lt, le, in, contains. A key is an index column (host, gpu, git_head, slurm_job, telemetry_id, …), else a parameter, else a result. A key that is both a parameter and a result is refused (write param_Re or result_Re), and an unknown key raises with a suggestion: a typo never returns an empty list.

Reproducing¶

import numpy as np
import conatus as cn


def heat(kappa):
    """u_t = kappa u_xx, u(x, 0) = sin(3x): the amplitude decays as exp(-9 kappa t)."""
    m = cn.Model(cn.Box(x=cn.periodic(2 * np.pi)), name="heat", kappa=kappa)
    m.scalar("u")
    m.equations("dt(u) = kappa*lap(u)")
    return m.simulate(N=32, scheme="RK443", dt=1e-3).initial(u=lambda x: np.sin(3 * x))


def decay_rate(sim):
    g = sim.grid("u")
    x = np.linspace(0, 2 * np.pi, g.size, endpoint=False)
    return -np.log(2 * np.mean(g * np.sin(3 * x))) / sim.t

run = heat(0.2).run(steps=200, record="heat_k0.2")   # a recorded run
solver = heat(0.2).solver             # solver built by your script for the same problem
report = run.reproduce(solver)
print(report.explain())               # bit-for-bit identical?  code identical?

The solver must have the same problem hash, parameter values and numerics; any difference is refused with the list, never adapted. The recorded start state is loaded, the same number of steps is taken in a new run (relation="reproduce"), and the whole end state (the state, every multistep history array, the time and the iteration) is compared bit for bit: same dtype and same bytes, so -0.0 differs from 0.0. A run recorded by 0.1.0 does not hold the given-field data, the user functions and six numerical choices that 0.1.1 records: an IdentityWarning names them (they are not compared) and the end state is still compared bit for bit. If the executed code changed since the run, a CodeChangedWarning says how (require_same_code=True makes it an error). A decorated function is re-imported and called with the recorded arguments.

The cache: only when you ask¶

import numpy as np
import conatus as cn


def heat(kappa):
    """u_t = kappa u_xx, u(x, 0) = sin(3x): the amplitude decays as exp(-9 kappa t)."""
    m = cn.Model(cn.Box(x=cn.periodic(2 * np.pi)), name="heat", kappa=kappa)
    m.scalar("u")
    m.equations("dt(u) = kappa*lap(u)")
    return m.simulate(N=32, scheme="RK443", dt=1e-3).initial(u=lambda x: np.sin(3 * x))


def decay_rate(sim):
    g = sim.grid("u")
    x = np.linspace(0, 2 * np.pi, g.size, endpoint=False)
    return -np.log(2 * np.mean(g * np.sin(3 * x))) / sim.t

heat(0.4).solver.run(200, record=True)                # computed and recorded
again = heat(0.4).solver.run(200, record=True, cache=True)   # read back: CacheReuseWarning
print(again.cache_report["hit"])

With cache=True, an identical computation already in the registry is read instead of computed, and a CacheReuseWarning says which run was reused and what was read (run.cache_report has the key and its parts). Identical means the same identity of the computation (the one the checkpoint guards and the HDF5 append also compare: problem hash, parameter values, the data of every given field by content, user functions by code, every numerical choice including time_parameter, bc_time and f_backend), exact start state (and multistep history), number of steps, the series functions and series_every (a hit returns the source run’s series), and executed code (tree and compiled core hashes, Python/NumPy/SciPy versions, architecture). For a decorated function, identical also covers what it reads besides its arguments: its closure values and the module globals its code names (by content; library modules, functions and classes by name and version; the attributes it names of a user module or a user class, such as cfg.A or Cfg.A, by content), followed into the user functions it calls, so two closures make(1) and make(2) of one factory, or one function under @scale(2) and @scale(3), are two computations (your own decorators under @track are part of what is called and of the key). The cache is off by default; it needs record=; it is refused with a callback (a hit would never call it); retracted runs are never reused; an argument, or a value read by the function or by a user pointwise function, without a content hash makes the call uncachable (said with a CacheReuseWarning naming it, then computed). A hit returns values of the recorded type (a NumPy scalar keeps its dtype, the string "nan" stays a string), and its series and fields are read from the source run through the new run (run.series(...)).

Studio and export¶

The The Studio lists, filters and compares registry runs (read-only), and a live run links to its registry record when it was recorded with cn.runs.track(..., telemetry=tel).
cn.runs.export(dir) writes runs as small JSON files (filters as in find); cn.runs.export_mlflow(runs) copies runs to MLflow if you have it (never installed by Conatus).

Reference¶

conatus.runs.track(name=None, *, tags=(), params=None, obj=None, parent=None, relation=None, status='provisional', root=None, telemetry=None, capacity=256, notes=None, cache=False, compression=None)[source]¶

Record a block (with track("rb", tags=[""]) as run: ...) or every call of a function (@track("eig", cache=True)). compression applies to arrays returned by a decorated function (None, ‘gzip’, ‘lzf’; all lossless).

Parameters:

name (str | None)
tags (Iterable[str])
params (Mapping[str, Any] | None)
obj (Any)
parent (Any)
relation (str | None)
status (str)
root (Any)
telemetry (Any)
capacity (int)
notes (str | None)
cache (bool)
compression (str | None)

Return type:

Tracker

conatus.runs.record(obj=None, *, name=None, params=None, results=None, series=None, fields=None, files=(), compression=REQUIRED, tags=(), status='provisional', parent=None, relation=None, root=None, telemetry=None, notes=None)[source]¶

Record one finished computation in one call and return the closed run.

obj: a Problem, Plan or solver (numerics, problem hash, explain()), an EVPResult (eigenvalues and residuals stored as fields, n_modes, growth_max, residual_max as results), or any object with explain(). series: {name: 1-D sequence}, all the same length. fields: {name: array}, which requires compression (None, ‘gzip’, ‘lzf’).

Parameters:

obj (Any)
name (str | None)
params (Mapping[str, Any] | None)
results (Mapping[str, Any] | None)
series (Mapping[str, Any] | None)
fields (Mapping[str, Any] | None)
files (Iterable[Any])
compression (Any)
tags (Iterable[str])
status (str)
parent (Any)
relation (str | None)
root (Any)
telemetry (Any)
notes (str | None)

Return type:

Run

conatus.runs.find(problem=None, *, name=None, status=None, state=None, tags=None, parent=None, since=None, until=None, limit=None, order='-created', root=None, **filters)[source]¶

Runs matching every criterion, newest first.

problem: a Problem / Plan / solver (its structural hash), a hash prefix (>= 8 hex digits) or a problem name. filters: key=value or key__op=value with op in eq, ne, gt, ge, lt, le, in, contains. A key is an index column (host, gpu, git_head, slurm_job, …), else a parameter, else a result; a key that is both a parameter and a result is refused (write param_<key> or result_<key>), and an unknown key is an error (typos never return an empty list silently), except in an empty registry.

Parameters:

problem (Any)
name (str | None)
status (str | Iterable[str] | None)
state (str | Iterable[str] | None)
tags (str | Iterable[str] | None)
parent (Any)
since (Any)
until (Any)
limit (int | None)
order (str)
root (Any)
filters (Any)

Return type:

RunList

conatus.runs.get(run_id, *, root=None)[source]¶

One run by id or unique id prefix.

Parameters:

run_id (Any)
root (Any)

Return type:

Run

conatus.runs.compare(a, b, *, root=None)[source]¶

Differences between two runs (ids or Run objects): parameters, numerics, results, code.

Parameters:

a (Any)
b (Any)
root (Any)

Return type:

Comparison

conatus.runs.lineage(run, *, root=None)[source]¶

Parent chain and descendants (fork, restart, reproduce, cache reuse) of a run.

Parameters:

run (Any)
root (Any)

Return type:

Lineage

conatus.runs.reproduce(run, solver=None, *, fn=None, record=True, require_same_code=False)[source]¶

Run the recorded computation again and compare with the recorded output.

IVP runs: pass a solver built for the same problem and numerics (checked: problem hash, parameters, bases, scheme, dt… ; any difference is refused, never adapted); the recorded start state is loaded, the same number of steps is taken, the end state is compared bit for bit. Decorated calls: the function is re-imported (or fn=) and called with the recorded arguments. If the executed code differs from the recorded one, a CodeChangedWarning says how (or an error with require_same_code=True).

Parameters:

run (Run)
solver (Any)
fn (Callable | None)
record (bool)
require_same_code (bool)

Return type:

ReproduceReport

conatus.runs.export_mlflow(runs, *, experiment=None, tracking_uri=None)[source]¶

Copy runs to MLflow (optional; mlflow imported here only, never installed by Conatus): parameters, numeric results as metrics, series as metric histories, tags, status, and the full record as an artifact. Returns {conatus id: mlflow run id}.

Parameters:

runs (Iterable[Run])
experiment (str | None)
tracking_uri (str | None)

Return type:

dict[str, str]

class conatus.runs.Run(registry, name, *, tags=(), params=None, obj=None, parent=None, relation=None, status='provisional', telemetry=None, capacity=256, notes=None, key_parts=None, _record=None)[source]¶

Bases: object

A registry run. Created live by conatus.runs.track() / conatus.runs.record() / solver.run(record=...), or read back by conatus.runs.get() / find.

Live methods: log() (series), result() (scalar results), save_field(), save_file(), attach() (a Problem / Plan / solver: numerics and explain()), certify() / retract(), close(). Read methods: params, results, series(), load_field(), file_path(), explain(), reproduce(), lineage().

Parameters:

registry (Registry)
name (str)
tags (Iterable[str])
params (Mapping[str, Any] | None)
obj (Any)
parent (Any)
relation (str | None)
status (str)
telemetry (Any)
capacity (int)
notes (str | None)
key_parts (dict | None)
_record (dict | None)

classmethod load(registry, run_id)[source]¶

Read-only view of a stored run.

Parameters:

registry (Registry)
run_id (str)

Return type:

Run

property name: str¶

property status: str¶

property state: str¶

property params: dict¶

property results: dict¶

property tags: list[str]¶

property parent: str | None¶

property path: Path¶

property cache_report: dict | None¶

attach(obj)[source]¶

Record the problem hash, the numerical configuration (conatus.provenance.numerics) and the explain() text of a Problem, Plan or solver. The problem’s parameter values (IR Parameter values, overridden by discretize(params=...)) join the run parameters; a parameter declared twice with different values is refused.

Parameters:: obj (Any)
Return type:: Run

log(step=None, **scalars)[source]¶

Append one row of series values (run.log(t=t, energy=E)); step is optional (the row number is always kept, so series logged together can be plotted against each other). In-memory only; written to data.h5 at close.

Parameters:

step (int | None)
scalars (Any)

Return type:

None

result(**values)[source]¶

Scalar results (numbers, strings, booleans, complex): stored in run.json and indexed, so find(Nu__gt=...) works.

Parameters:: values (Any)
Return type:: None

save_field(name, array, *, compression, level=None, t=None, step=None)[source]¶

Queue a copy of array for data.h5 (fields/<name>/<k>). compression has no default (None, ‘gzip’ or ‘lzf’, all lossless; level 0–9 for gzip) and is written in the dataset’s attributes. Returns False if the queue was full (dropped, counted).

Parameters:

name (str)
array (Any)
compression (str | None)
level (int | None)
t (float | None)
step (int | None)

Return type:

bool

save_file(path, name=None)[source]¶

Queue a copy of a file into files/ (sha256 recorded). The copy is taken by the writer thread: a file changed right after this call may be copied changed.

Parameters:

path (str | PathLike)
name (str | None)

Return type:

bool

set_status(status, reason, evidence=None)[source]¶

Change the status (history kept, never rewritten). reason is required.

Parameters:

status (str)
reason (str)
evidence (Any)

Return type:

Run

certify(reason, evidence=None)[source]¶

Mark the run certified (reason and optional evidence, e.g. residuals, recorded).

Parameters:

reason (str)
evidence (Any)

Return type:

Run

retract(reason)[source]¶

Mark the run retracted (kept, never deleted; never reused by the cache).

Parameters:: reason (str)
Return type:: Run

close(state='completed', error=None, timeout=600.0)[source]¶

Finish the run: series to data.h5, final run.json, index. Waits for the writer at most timeout s. Idempotent. Warns (RegistryWriteWarning) if anything was lost.

Parameters:

state (str)
error (str | None)
timeout (float)

Return type:

Run

series(name, x=None)[source]¶

{“row”, “step”, “value”} arrays of one series; with x (another series logged in the same rows, e.g. “t”), also “x” aligned on the rows both have (no interpolation).

Parameters:

name (str)
x (str | None)

Return type:

dict

load_field(name, index=-1)[source]¶

One saved snapshot of a field (default: the last one).

Parameters:

name (str)
index (int)

Return type:

ndarray

file_path(name)[source]¶

The attached file name of this run (a plain file name: no directory part, never “.” or “..”; security name=”../../x” read outside the run).

Parameters:: name (str)
Return type:: Path

refresh()[source]¶

Re-read run.json (e.g. after another process changed the status).

Return type:: Run

lineage()[source]¶

reproduce(solver=None, **kw)[source]¶

Run this computation again and compare (see conatus.runs.reproduce()).

Parameters:

solver (Any)
kw (Any)

explain()[source]¶

Return type:: str