Skip to main content

Documentation Index

Fetch the complete documentation index at: https://ahvn.top/llms.txt

Use this file to discover all available pages before exploring further.

Function utilities help with the runtime actions that are easy to get subtly wrong: reproducible random data, bounded parallel work, and command execution with explicit outputs.
Callable persistence and function-schema metadata are planned for the Capsule and prompt surfaces. They are not part of the general-purpose utility API yet.

​
1. Deterministic data with StableRNG

Use StableRNG for tests, demos, and benchmarks that generate mock data. A seed and a step path produce the same data every time. 
from heavenbase.utils import StableRNG

rng = StableRNG(seed=42).step("bench", "rows")
ids = rng.choice(["p1", "p2", "p3", "p4"], k=2, replace=False)
vec = rng.rnd_vec(dim=4)
Outside a context manager, each method call is stable for the current seed. Inside a context manager, the generator advances like a normal NumPy generator. Use a context when a sequence of draws should be coupled. 
from heavenbase.utils import StableRNG

assert StableRNG(seed=7).rnd_str(6) == StableRNG(seed=7).rnd_str(6)

with StableRNG(seed=7) as rng:
    first = rng.rnd_str(6)
    second = rng.rnd_str(6)
Use step(...) to derive independent child streams from one seed without mutating the parent generator: 
base = StableRNG(seed=42)
rows_rng = base.step("rows")
sample_rng = base.step("samples")
Most generators accept a batch shape through n: 
from heavenbase.utils import StableRNG

rng = StableRNG(seed=42)
one = rng.rnd_int(0, 10)        # int
many = rng.rnd_int(0, 10, n=3)  # list[int]
grid = rng.rnd_str(4, n=(2, 2)) # list[list[str]]
vecs = rng.rnd_vec(dim=8, n=5)  # five unit-length vectors
Available helpers include:
  • rnd, rnd_int, rnd_float, and rnd_normal for scalar or batched numeric values.
  • rnd_str for deterministic IDs or compact tokens.
  • choice, perm, and shuffled for sampled or reordered data.
  • hash_sample and hash_split for stable samples that do not depend on input order.
  • rnd_vec for unit-length synthetic vector embeddings.
  • abm and abm_path for simple Arithmetic Brownian Motion fixtures.
Benchmarks in HeavenBase use StableRNG(seed=42) by default so generated rows, samples, and vectors can be reproduced.

​
2. Bounded parallel work

Use pmap when work can run concurrently but the result order must match the input order. 
from heavenbase.utils import pmap

tasks = [{"row": row} for row in rows]
scores = pmap(lambda row: row["score"] * 2, tasks, max_workers=4)
Parallel helpers pass each item as keyword arguments, so each batch item must be a dict-like mapping. This keeps functions free to use clear parameters instead of unpacking one generic object. Use pforeach for side-effect work where no result list is needed. Set max_workers=1 for an explicit serial path. max_workers=None reads heavenbase.parallel.max_workers; when that config is unset, HeavenBase uses a CPU-based bounded default capped by heavenbase.parallel.fallback_max_workers. Use batch when you need per-item status, captured exceptions, elapsed time, or progress callbacks. It returns TaskResult records with index, item, ok, value, error, and elapsed_ms. 
from heavenbase.utils import batch

results = batch(lambda x: 10 / x, [{"x": 2}, {"x": 0}, {"x": 5}], max_workers=2, on_error="capture")
assert [result.ok for result in results] == [True, False, True]
assert results[1].error is not None
batch_stream yields TaskResult records as work finishes. Its default is ordered=False for completion order; pass ordered=True when consumers require input order. Ordered streaming still runs concurrently, but a later result can be held until earlier indexes are ready. 
from heavenbase.utils import batch_stream

for result in batch_stream(work, [{"payload": item} for item in items], max_workers=8, ordered=False):
    if result.ok:
        handle(result.value)
Error handling is explicit:
  • on_error="capture" keeps failed tasks as TaskResult(ok=False, error=...).
  • on_error="raise" raises the first observed task error and stops submitting more work; already-running thread tasks may finish during executor cleanup.
  • on_error="skip" omits failed task records from the output.
Async code uses the same contract through abatch and abatch_stream. Async helpers accept async callables directly; sync callables run through the event loop’s executor. 
from heavenbase.utils import abatch

async def enrich(row):
    return {**row, "score": row["score"] * 2}

results = await abatch(enrich, [{"row": row} for row in rows], max_workers=8, ordered=True)
payloads = [result.value for result in results if result.ok]

​
3. Command execution

Use cmd when a script needs to call a local command and inspect stdout, stderr, and the exit code. 
from heavenbase.utils import cmd

result = cmd(["python", "--version"])
if not result.ok:
    raise RuntimeError(result.err)
print(result.out)
For call sites that request specific fields, include=["stdout", "stderr", "returncode"] returns a small dictionary. A single include="out" returns just that value. 
from heavenbase.utils import cmd

version = cmd(["python", "--version"], include="out")
handle = cmd(["python", "-m", "http.server", "8000"], wait=False, include="handle")
Use check=True when a non-zero exit code should raise subprocess.CalledProcessError. Use browse(path) to open a local path with the platform default app, and clipboard(text) to copy text when the host provides a clipboard command.
Command helpers use Python’s subprocess under the hood and do not invoke a shell for sequence commands unless shell=True is explicitly passed.

​
Further Exploration

Related resources:
  • Utilities overview - choose the right helper group.
  • Common utilities - logging and debug helpers for command failures.