Workflow (parameter sets)¶

This notebook demonstrates how to use the kosh_workflow CLI to document and monitor a parameterized workflow:

Note: --ensemble is optional; if omitted, kosh_workflow searches/updates across all datasets in the store. For simple parameter names you can now pass filters directly as --name=value; keep using --param key=value for names with exotic characters such as spaces or semicolons. If a near-match gets flagged as a typo, add --no-typo-check to force it through.

1. Create an ensemble of datasets keyed by a parameter set.
2. Advance a dataset through workflow steps and report how long it has been at the current step.
3. Query datasets by one parameter within a range.
4. Query datasets by multiple parameters within ranges.
5. Attempt to add a duplicate (within rtol/atol) and upsert the existing dataset instead.

In [1]:

import os
import sys
import tempfile

os.environ.setdefault("PATH", "/usr/bin:/bin")

import kosh
from sina.utils import DataRange

import os import sys import tempfile os.environ.setdefault("PATH", "/usr/bin:/bin") import kosh from sina.utils import DataRange

In [2]:

tmpdir = tempfile.TemporaryDirectory()
store_path = os.path.join(tmpdir.name, "store.sql")
ensemble_name = "designs"

store_path

tmpdir = tempfile.TemporaryDirectory() store_path = os.path.join(tmpdir.name, "store.sql") ensemble_name = "designs" store_path

Out[2]:

'/tmp/tmp4zbu9knm/store.sql'

1) Create datasets and add them to an ensemble¶

Each call with --step init creates a dataset (unless --upsert-init matches an existing one). In the examples below, mach and aoa use the direct --name=value form because their names are simple CLI-safe identifiers. Use --no-typo-check if you need to force through a name that is too close to a real option.

In [3]:

for mach, aoa in [(0.7, 2.0), (0.8, 2.0), (0.9, 2.0), (0.8, 3.0)]:
    !{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --mach={mach} --aoa={aoa}

store = kosh.connect(store_path)
ensemble = list(store.find_ensembles(name=ensemble_name))[0]
datasets = list(ensemble.find_datasets())
len(datasets)

for mach, aoa in [(0.7, 2.0), (0.8, 2.0), (0.9, 2.0), (0.8, 3.0)]: !{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --mach={mach} --aoa={aoa} store = kosh.connect(store_path) ensemble = list(store.find_ensembles(name=ensemble_name))[0] datasets = list(ensemble.find_datasets()) len(datasets)

Matched dataset ids: f7819760b642401999a8a3635fc39ac6
Matched dataset ids: 9b173ef7f15f45f0aa6d21e3fa346a9b
Matched dataset ids: ac313c7d3bff49cdada4d4073f1a1b29
Matched dataset ids: c3b88b9c7c4148eeb231895194ef12af

Out[3]:

4

2) Query: one parameter within a range¶

Find datasets with mach in [0.79, 0.81].

In [4]:

matches = list(ensemble.find_datasets(mach=DataRange(0.79, 0.81, max_inclusive=True)))
[(d.id, getattr(d, "mach"), getattr(d, "aoa")) for d in matches]

matches = list(ensemble.find_datasets(mach=DataRange(0.79, 0.81, max_inclusive=True))) [(d.id, getattr(d, "mach"), getattr(d, "aoa")) for d in matches]

Out[4]:

[('9b173ef7f15f45f0aa6d21e3fa346a9b', 0.8, 2.0),
 ('c3b88b9c7c4148eeb231895194ef12af', 0.8, 3.0)]

3) Query: multiple parameters within ranges¶

Find datasets with mach near 0.8 and aoa near 2.0.

In [5]:

matches = list(
    ensemble.find_datasets(
        mach=DataRange(0.79, 0.81, max_inclusive=True),
        aoa=DataRange(1.99, 2.01, max_inclusive=True),
    )
)
[(d.id, getattr(d, "mach"), getattr(d, "aoa")) for d in matches]

matches = list( ensemble.find_datasets( mach=DataRange(0.79, 0.81, max_inclusive=True), aoa=DataRange(1.99, 2.01, max_inclusive=True), ) ) [(d.id, getattr(d, "mach"), getattr(d, "aoa")) for d in matches]

Out[5]:

[('9b173ef7f15f45f0aa6d21e3fa346a9b', 0.8, 2.0)]

4) Duplicate add attempt: upsert within tolerance¶

We try to add a "new" dataset for (mach, aoa)=(0.8000000005, 2.0).

With --upsert-init and a nonzero --atol/--rtol, the CLI will not create a new dataset; it will find the existing one and update it.

In [6]:

before = len(list(ensemble.find_datasets()))

!{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --upsert-init --rtol 1e-5 --atol 1e-8 --mach=0.8000000005 --aoa=2.0 --meta duplicate_attempt true

after = len(list(ensemble.find_datasets()))
before, after

before = len(list(ensemble.find_datasets())) !{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --upsert-init --rtol 1e-5 --atol 1e-8 --mach=0.8000000005 --aoa=2.0 --meta duplicate_attempt true after = len(list(ensemble.find_datasets())) before, after

Found existing dataset 9b173ef7f15f45f0aa6d21e3fa346a9b matching init params; upserting it.
Matched dataset ids: 9b173ef7f15f45f0aa6d21e3fa346a9b

Out[6]:

(4, 4)

In [7]:

matched = list(
    ensemble.find_datasets(
        mach=DataRange(0.79, 0.81, max_inclusive=True),
        aoa=DataRange(1.99, 2.01, max_inclusive=True),
    )
)

# The existing dataset should now carry the metadata update
[(d.id, getattr(d, "duplicate_attempt", None)) for d in matched]

matched = list( ensemble.find_datasets( mach=DataRange(0.79, 0.81, max_inclusive=True), aoa=DataRange(1.99, 2.01, max_inclusive=True), ) ) # The existing dataset should now carry the metadata update [(d.id, getattr(d, "duplicate_attempt", None)) for d in matched]

Out[7]:

[('9b173ef7f15f45f0aa6d21e3fa346a9b', 'true')]

5) Advance workflow steps and detect stalls¶

A common workflow pattern is to update a dataset's workflow_step as work progresses (e.g. queued → running → done).

With kosh_workflow --check, you can also retrieve the timestamp of the last update to that step, which is useful to detect and kill/restart stalled jobs.

In [8]:

import time

# Ensure the dataset exists (create it if missing, but avoid duplicates)
!{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --upsert-init --mach=0.8 --aoa=2.0

# Advance the workflow step for a known dataset (mach=0.8, aoa=2.0)
!{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step running --mach=0.8 --aoa=2.0

# Give the timestamp a chance to change visibly
time.sleep(1.0)

# --check prints: <id> <step> <formatted_time> <epoch> (tab-separated)
lines = !{sys.executable} -m kosh.workflow --check --check-time-format "%Y-%m-%d %H:%M:%S" --store {store_path} --ensemble {ensemble_name} --mach=0.8 --aoa=2.0

# Some environments emit extra startup warnings; select the first data line.
data_lines = [ln for ln in lines if ln.count("\t") >= 3]
if not data_lines:
    raise RuntimeError("Expected tab-separated output from --check, got:\n" + "\n".join(lines))
ds_id, step, ts_str, ts_epoch = data_lines[0].strip().split("\t", 3)

age_s = time.time() - float(ts_epoch)
print(f"{ds_id} step={step} since={ts_str} age_s={age_s:.1f}")

stalled_after_s = 3600
if age_s > stalled_after_s:
    print("stalled: consider killing/restarting the workflow")

import time # Ensure the dataset exists (create it if missing, but avoid duplicates) !{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step init --upsert-init --mach=0.8 --aoa=2.0 # Advance the workflow step for a known dataset (mach=0.8, aoa=2.0) !{sys.executable} -m kosh.workflow --store {store_path} --ensemble {ensemble_name} --step running --mach=0.8 --aoa=2.0 # Give the timestamp a chance to change visibly time.sleep(1.0) # --check prints: <id> <step> <formatted_time> <epoch> (tab-separated) lines = !{sys.executable} -m kosh.workflow --check --check-time-format "%Y-%m-%d %H:%M:%S" --store {store_path} --ensemble {ensemble_name} --mach=0.8 --aoa=2.0 # Some environments emit extra startup warnings; select the first data line. data_lines = [ln for ln in lines if ln.count("\t") >= 3] if not data_lines: raise RuntimeError("Expected tab-separated output from --check, got:\n" + "\n".join(lines)) ds_id, step, ts_str, ts_epoch = data_lines[0].strip().split("\t", 3) age_s = time.time() - float(ts_epoch) print(f"{ds_id} step={step} since={ts_str} age_s={age_s:.1f}") stalled_after_s = 3600 if age_s > stalled_after_s: print("stalled: consider killing/restarting the workflow")

Found existing dataset 9b173ef7f15f45f0aa6d21e3fa346a9b matching init params; upserting it.
Matched dataset ids: 9b173ef7f15f45f0aa6d21e3fa346a9b
Matched dataset ids: 9b173ef7f15f45f0aa6d21e3fa346a9b
9b173ef7f15f45f0aa6d21e3fa346a9b step=running since=2026-05-11 12:30:18 age_s=2.6

6) CLI --check and duplicate suppression (many samples)¶

This section demonstrates a common workflow pattern:

• For each parameter set, first query the store using kosh_workflow --check with --rtol/--atol (prints dataset id, current step, and step timestamp; customize with --check-time-format).
• If a match exists, skip work.
• Otherwise, record the new parameter set with kosh_workflow --step init.

We use 12 "attempts": 10 unique values and 2 near-duplicates (within tolerance). Only 10 datasets should be created in the store.

In [9]:

store_path2 = os.path.join(tmpdir.name, "store_check.sql")
sweep_ensemble = "p0_sweep"

p0_values = "1.0 2.0 3.0 4.0 5.0 6.0 7.0 8.0 9.0 10.0 1.0000000005 2.0000000005"
p1_value = "10.0"
rtol = "1e-5"
atol = "1e-8"

import subprocess

env = dict(os.environ)
env.setdefault("PATH", "/usr/bin:/bin")
env.update(
    PYTHON=sys.executable,
    STORE=store_path2,
    ENSEMBLE=sweep_ensemble,
    RTOL=rtol,
    ATOL=atol,
    P1=p1_value,
    P0_VALUES=p0_values,
)

script = r"""
set -e
created=0
skipped=0
for P0 in $P0_VALUES; do
  if $PYTHON -m kosh.workflow --check --store "$STORE" --ensemble "$ENSEMBLE" --rtol "$RTOL" --atol "$ATOL" --p0="$P0" --p1="$P1" >/dev/null; then
    echo "nothing to be done we already have this (p0=$P0)"
    skipped=$((skipped+1))
  else
    echo "New set of parameters we will do work (p0=$P0)"
    $PYTHON -m kosh.workflow --store "$STORE" --ensemble "$ENSEMBLE" --step init --upsert-init --rtol "$RTOL" --atol "$ATOL" --p0="$P0" --p1="$P1"
    created=$((created+1))
  fi
done
echo "created=$created skipped=$skipped"
"""

subprocess.run(["bash", "-c", script], check=True, env=env)

store_path2 = os.path.join(tmpdir.name, "store_check.sql") sweep_ensemble = "p0_sweep" p0_values = "1.0 2.0 3.0 4.0 5.0 6.0 7.0 8.0 9.0 10.0 1.0000000005 2.0000000005" p1_value = "10.0" rtol = "1e-5" atol = "1e-8" import subprocess env = dict(os.environ) env.setdefault("PATH", "/usr/bin:/bin") env.update( PYTHON=sys.executable, STORE=store_path2, ENSEMBLE=sweep_ensemble, RTOL=rtol, ATOL=atol, P1=p1_value, P0_VALUES=p0_values, ) script = r""" set -e created=0 skipped=0 for P0 in $P0_VALUES; do if $PYTHON -m kosh.workflow --check --store "$STORE" --ensemble "$ENSEMBLE" --rtol "$RTOL" --atol "$ATOL" --p0="$P0" --p1="$P1" >/dev/null; then echo "nothing to be done we already have this (p0=$P0)" skipped=$((skipped+1)) else echo "New set of parameters we will do work (p0=$P0)" $PYTHON -m kosh.workflow --store "$STORE" --ensemble "$ENSEMBLE" --step init --upsert-init --rtol "$RTOL" --atol "$ATOL" --p0="$P0" --p1="$P1" created=$((created+1)) fi done echo "created=$created skipped=$skipped" """ subprocess.run(["bash", "-c", script], check=True, env=env)

New set of parameters we will do work (p0=1.0)
Matched dataset ids: 763090c55e564dbc87013b8b8ab2f0e3
New set of parameters we will do work (p0=2.0)
Matched dataset ids: b410120cde3640ad9c6e777e1e23ea9f
New set of parameters we will do work (p0=3.0)
Matched dataset ids: e1ed16677f1e41b0ab7c61028c6b55f7
New set of parameters we will do work (p0=4.0)
Matched dataset ids: 5635acacd2714af7b4c827dfaa30b3c8
New set of parameters we will do work (p0=5.0)
Matched dataset ids: d23ac27658554a85a9ace2a295cbd4ce
New set of parameters we will do work (p0=6.0)
Matched dataset ids: 06e5c51ab9a84881b0b8ca2244b99364
New set of parameters we will do work (p0=7.0)
Matched dataset ids: b5e493dbd1114a6ab19bb18bd70bea19
New set of parameters we will do work (p0=8.0)
Matched dataset ids: 5cb7dc3e747843a596e2ecf2823280b9
New set of parameters we will do work (p0=9.0)
Matched dataset ids: f4525f97a5994202afc65d5fe1c10621
New set of parameters we will do work (p0=10.0)
Matched dataset ids: b75990e441a24d1ea9f2cf2432333e9a
nothing to be done we already have this (p0=1.0000000005)
nothing to be done we already have this (p0=2.0000000005)
created=10 skipped=2

Out[9]:

CompletedProcess(args=['bash', '-c', '\nset -e\ncreated=0\nskipped=0\nfor P0 in $P0_VALUES; do\n  if $PYTHON -m kosh.store_manager --check --store "$STORE" --ensemble "$ENSEMBLE" --rtol "$RTOL" --atol "$ATOL" --param "p0=$P0" --param "p1=$P1" >/dev/null; then\n    echo "nothing to be done we already have this (p0=$P0)"\n    skipped=$((skipped+1))\n  else\n    echo "New set of parameters we will do work (p0=$P0)"\n    $PYTHON -m kosh.store_manager --store "$STORE" --ensemble "$ENSEMBLE" --step init --upsert-init --rtol "$RTOL" --atol "$ATOL" --param "p0=$P0" --param "p1=$P1"\n    created=$((created+1))\n  fi\ndone\necho "created=$created skipped=$skipped"\n'], returncode=0)

In [10]:

ids = !{sys.executable} -m kosh.cli find --store {store_path2}
ids = [ln for ln in ids if ln.strip()]
print("Dataset ids:\n" + "\n".join(ids))
print("Dataset count in store:", len(ids))
assert len(ids) == 10

ensemble_count = !{sys.executable} -m kosh.workflow --size --store {store_path2} --ensemble {sweep_ensemble}
print("Dataset count in ensemble:", ensemble_count[0].strip())

ids = !{sys.executable} -m kosh.cli find --store {store_path2} ids = [ln for ln in ids if ln.strip()] print("Dataset ids:\n" + "\n".join(ids)) print("Dataset count in store:", len(ids)) assert len(ids) == 10 ensemble_count = !{sys.executable} -m kosh.workflow --size --store {store_path2} --ensemble {sweep_ensemble} print("Dataset count in ensemble:", ensemble_count[0].strip())

Dataset ids:
06e5c51ab9a84881b0b8ca2244b99364
b410120cde3640ad9c6e777e1e23ea9f
b5e493dbd1114a6ab19bb18bd70bea19
5635acacd2714af7b4c827dfaa30b3c8
763090c55e564dbc87013b8b8ab2f0e3
5cb7dc3e747843a596e2ecf2823280b9
f4525f97a5994202afc65d5fe1c10621
e1ed16677f1e41b0ab7c61028c6b55f7
b75990e441a24d1ea9f2cf2432333e9a
d23ac27658554a85a9ace2a295cbd4ce
Dataset count in store: 10
Dataset count in ensemble: 10

In [11]:

store.close()
tmpdir.cleanup()

store.close() tmpdir.cleanup()

In [ ]: