Telemetry¶

The parquet index of one local artifact store: an incremental update over its committed directories, plus query.

Methods:

• update –

  Index every artifact directory whose mtime moved since the last pass — all of them when rebuild — and return the rows appended.

• query –

  A fluent query over the index, reading no sidecar until records asks for one.

128
129
130
131
132

def __init__(self, telemetry: Telemetry, store: LocalArtifactStore) -> None:
    """Track the committed artifacts of ``store`` in the artifact table of ``telemetry``."""
    self.telemetry: Telemetry = telemetry
    self.store: LocalArtifactStore = store
    self.hive = telemetry.artifacts

update(*, rebuild: bool = False) -> int

167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188

def update(self, *, rebuild: bool = False) -> int:
    """Index every artifact directory whose mtime moved since the last pass — all of them when ``rebuild`` — and return the rows appended."""
    candidates: list[tuple[str, Path, datetime]] = self._candidates()
    if not candidates:
        return 0

    pending = candidates if rebuild else self._changed(candidates)

    rows: list[dict[str, object]] = []
    for _, directory, mtime in pending:
        artifact: StoredArtifact | None = self.store.read(directory)
        if artifact is None:
            continue

        rows.append(artifact_row(artifact, store_root=self.store.root, dir_mtime=mtime))

    if not rows:
        return 0

    _ = self.hive.append(pl.DataFrame(rows, schema=ARTIFACT_INDEX_SCHEMA))

    return len(rows)

query() -> ArtifactQuery

190
191
192

def query(self) -> ArtifactQuery:
    """A fluent query over the index, reading no sidecar until ``records`` asks for one."""
    return ArtifactQuery(self.hive.scan(), self.store)

A fluent filter over the artifact index; each method returns a new query and nothing runs until frame.

Methods:

• where –

  A new query narrowed by predicate; the other filters are named shorthands for this.

• kind –

  Artifacts of any of kinds.

• spec_type –

  Artifacts whose spec model's module:QualName ends with name, so the bare class name is enough.

• since –

  Artifacts published at or after when.

• key_prefix –

  Artifacts whose key starts with prefix.

• newest –

  Keep only the count most recently published artifacts, applied after the duplicate rows collapse.

• frame –

  The matching rows, newest first, one per stored artifact (the latest row wins).

• records –

  The matching artifacts read back from the store, skipping any whose directory is gone.

Attributes:

• lazy (LazyFrame) –

  The artifact index rows the query filters.

• store (LocalArtifactStore) –

  The store records reads the matching artifacts back from.

• limit (int | None) –

  How many of the newest artifacts to keep; None keeps them all.

lazy: LazyFrame

store: LocalArtifactStore

limit: int | None = None

where(predicate: Expr) -> ArtifactQuery

76
77
78

def where(self, predicate: pl.Expr) -> ArtifactQuery:
    """A new query narrowed by ``predicate``; the other filters are named shorthands for this."""
    return replace(self, lazy=self.lazy.filter(predicate))

kind(*kinds: str) -> ArtifactQuery

80
81
82

def kind(self, *kinds: str) -> ArtifactQuery:
    """Artifacts of any of ``kinds``."""
    return self.where(pl.col(ArtifactColumn.KIND).is_in(list(kinds)))

spec_type(name: str) -> ArtifactQuery

84
85
86

def spec_type(self, name: str) -> ArtifactQuery:
    """Artifacts whose spec model's ``module:QualName`` ends with ``name``, so the bare class name is enough."""
    return self.where(pl.col(ArtifactColumn.SPEC_TYPE).str.ends_with(name))

since(when: datetime) -> ArtifactQuery

88
89
90

def since(self, when: datetime) -> ArtifactQuery:
    """Artifacts published at or after ``when``."""
    return self.where(pl.col(ArtifactColumn.CREATED_AT) >= when)

key_prefix(prefix: str) -> ArtifactQuery

92
93
94

def key_prefix(self, prefix: str) -> ArtifactQuery:
    """Artifacts whose key starts with ``prefix``."""
    return self.where(pl.col(ArtifactColumn.KEY).str.starts_with(prefix))

newest(count: int) -> ArtifactQuery

96
97
98

def newest(self, count: int) -> ArtifactQuery:
    """Keep only the ``count`` most recently published artifacts, applied after the duplicate rows collapse."""
    return replace(self, limit=count)

frame() -> DataFrame

100
101
102
103
104
105

def frame(self) -> pl.DataFrame:
    """The matching rows, newest first, one per stored artifact (the latest row wins)."""
    deduped = self.lazy.sort(ArtifactColumn.INDEXED_AT, maintain_order=True).unique(subset=[ArtifactColumn.KIND, ArtifactColumn.KEY], keep="last", maintain_order=True)
    ordered = deduped.sort(ArtifactColumn.CREATED_AT, descending=True, nulls_last=True, maintain_order=True)

    return (ordered.head(self.limit) if self.limit is not None else ordered).collect()

records() -> list[StoredArtifact]

107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

def records(self) -> list[StoredArtifact]:
    """The matching artifacts read back from the store, skipping any whose directory is gone."""
    rows: pl.DataFrame = self.frame()
    roots: list[str] = cast("list[str]", rows[ArtifactColumn.STORE_ROOT].to_list())
    kinds: list[str] = cast("list[str]", rows[ArtifactColumn.KIND].to_list())
    names: list[str] = cast("list[str]", rows[ArtifactColumn.NAME].to_list())

    found: list[StoredArtifact] = []
    for root, kind, name in zip(roots, kinds, names, strict=True):
        artifact: StoredArtifact | None = self.store.read(Path(root) / kind / name)
        if artifact is None:
            continue

        found.append(artifact)

    return found

Bases: NamedTuple

Which sweep, and which point of it, the current context belongs to.

Attributes:

• sweep_id (str) –

  The sweep's id.

• point_key (str) –

  The point's key within the sweep.

sweep_id: str

point_key: str

Bases: BaseModel

The iteration caps and tolerances one run's sif authored, written beside its convergence log so the log reads in context.

Config:

• use_attribute_docstrings: True

Fields:

• loops (list[LoopTarget])

loops: list[LoopTarget]

of(targets: ConvergenceTargets) -> Self

58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

@classmethod
def of(cls, targets: ConvergenceTargets) -> Self:
    """The persistable form of the mapping ``convergence_targets`` builds from a solver stack."""
    loops: list[LoopTarget] = []
    for (solver, loop), target in targets.items():
        loops.append(
            LoopTarget(
                solver=solver,
                loop=ConvergenceLoop(loop),
                max_iterations=target.max_iterations,
                tolerance=target.tolerance,
            )
        )

    return cls(loops=loops)

load(path: str | Path) -> Self

74
75
76
77
78
79
80
81
82
83
84

@classmethod
def load(cls, path: str | Path) -> Self:
    """Load a bundle's ``convergence_targets.json`` (pass the bundle directory or the file itself); a bundle without one reads as no targets."""
    file: Path = Path(path)
    if file.is_dir():
        file = file / BundleFile.CONVERGENCE_TARGETS

    if not file.is_file():
        return cls()

    return cls.model_validate_json(file.read_text(encoding="utf-8"))

write(sim_dir: str | Path) -> Path

86
87
88
89
90
91

def write(self, sim_dir: str | Path) -> Path:
    """Write the targets into ``sim_dir`` as ``convergence_targets.json`` and return the path."""
    path: Path = Path(sim_dir) / BundleFile.CONVERGENCE_TARGETS
    write_atomic(path, self.model_dump_json(indent=2))

    return path

targets() -> ConvergenceTargets

 93
 94
 95
 96
 97
 98
 99
100

def targets(self) -> ConvergenceTargets:
    """The renderer's mapping again, keyed the way Elmer tags ComputeChange lines."""
    mapping: dict[tuple[str, Literal["NS", "SS"]], LoopTargets] = {}
    for entry in self.loops:
        loop: Literal["NS", "SS"] = "NS" if entry.loop is ConvergenceLoop.NS else "SS"
        mapping[(normalize_solver_key(entry.solver), loop)] = LoopTargets(max_iterations=entry.max_iterations, tolerance=entry.tolerance)

    return mapping

Bases: BaseModel

One solver loop's authored iteration cap and tolerance, in the form the bundle persists them.

Config:

• use_attribute_docstrings: True

Fields:

• solver (str)
• loop (ConvergenceLoop)
• max_iterations (int | None)
• tolerance (float | None)

solver: str

loop: ConvergenceLoop

max_iterations: int | None = None

tolerance: float | None = None

One hive-partitioned parquet table: a typed scan, an atomic append of one shard per partition, and a settling compact.

Parameters:

• root (Path) –

  The table's directory, holding one directory level per partition key.

• schema (Schema) –

  Every column of the table, partition columns included.

• partition_keys (Sequence[str]) –

  The columns whose values name the partition directories, outermost first.

Methods:

• shards –

  Every shard file under the root, in sorted order.

• partitions –

  The directories holding shards, one per distinct partition-key combination.

• scan –

  A lazy frame over every shard, each typed onto the schema, or a typed empty frame when nothing has been written.

• append –

  Write frame as one new shard per partition and return the shard paths; existing shards are untouched.

• compact –

  Fold each partition's settled shards into one merged shard and delete the originals; returns the merged shards.

Attributes:

• hive_schema (Schema) –

  The partition columns, whose values live in the directory names rather than the file bodies.

• body_schema (Schema) –

  The columns actually stored inside each shard: everything but the partition keys.

103
104
105
106
107
108
109
110
111
112
113

def __init__(self, root: Path, schema: pl.Schema, partition_keys: Sequence[str]) -> None:
    """Address the table stored under ``root``; nothing touches disk until a read or write.

    Args:
        root: The table's directory, holding one directory level per partition key.
        schema: Every column of the table, partition columns included.
        partition_keys: The columns whose values name the partition directories, outermost first.
    """
    self.root: Path = root
    self.schema: pl.Schema = schema
    self.partition_keys: tuple[str, ...] = tuple(str(key) for key in partition_keys)

hive_schema: Schema

body_schema: Schema

shards() -> list[Path]

125
126
127
128
129
130

def shards(self) -> list[Path]:
    """Every shard file under the root, in sorted order."""
    if not self.root.is_dir():
        return []

    return sorted(self.root.rglob("*" + SHARD_SUFFIX))

partitions() -> list[Path]

132
133
134
135
136

def partitions(self) -> list[Path]:
    """The directories holding shards, one per distinct partition-key combination."""
    directories = {shard.parent for shard in self.shards()}

    return sorted(directories)

scan() -> LazyFrame

138
139
140
141
142
143
144
145
146
147

def scan(self) -> pl.LazyFrame:
    """A lazy frame over every shard, each typed onto the schema, or a typed empty frame when nothing has been written."""
    shards = self.shards()
    if not shards:
        return pl.LazyFrame(schema=self.schema)

    # polars refuses to unify Enum columns whose categories differ, so only shards of one schema share a scan, and each run is typed before they meet.
    typed = [coerce(pl.scan_parquet(paths, hive_partitioning=True, hive_schema=self.hive_schema), self.schema) for paths in schema_runs(shards)]

    return pl.concat(typed, how="vertical")

append(frame: DataFrame) -> list[Path]

149
150
151
152
153
154
155
156
157
158
159
160
161
162

def append(self, frame: pl.DataFrame) -> list[Path]:
    """Write ``frame`` as one new shard per partition and return the shard paths; existing shards are untouched."""
    if frame.is_empty():
        return []

    typed = frame.cast(self.schema).select(self.schema.names())
    groups = cast("dict[tuple[object, ...], pl.DataFrame]", typed.partition_by(self.partition_keys, as_dict=True, include_key=False))

    written: list[Path] = []
    for values, body in groups.items():
        segments = (f"{key}={value}" for key, value in zip(self.partition_keys, values, strict=True))
        written.append(write_shard(self.root.joinpath(*segments), body))

    return written

compact(settle_s: float = SETTLE_S) -> list[Path]

164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

def compact(self, settle_s: float = SETTLE_S) -> list[Path]:
    """Fold each partition's settled shards into one merged shard and delete the originals; returns the merged shards."""
    cutoff = time.time() - settle_s

    merged: list[Path] = []
    for directory in self.partitions():
        shards = sorted(directory.glob("*" + SHARD_SUFFIX))
        settled = [shard for shard in shards if shard.stat().st_mtime <= cutoff]
        if len(settled) < 2:
            continue

        typed = [coerce(pl.scan_parquet(paths), self.body_schema) for paths in schema_runs(settled)]
        body = pl.concat(typed, how="vertical").collect()
        merged.append(write_shard(directory, body))

        for shard in settled:
            shard.unlink()

    return merged

A process's own append-only JSONL journal: every record is written and flushed immediately, then folded into parquet by checkpoint.

Methods:

• write –

  Append one record and flush it, so a reader tailing the journal sees it before the next parquet flush.

• checkpoint –

  Append the journal's records to store and truncate it; returns the shards written.

• close –

  Close the journal handle and delete the (now empty) file; idempotent.

138
139
140
141
142
143
144

def __init__(self, live_dir: Path, name: str) -> None:
    """Open this process's journal for the table ``name`` under ``live_dir``, creating the directory when needed."""
    live_dir.mkdir(parents=True, exist_ok=True)

    self.path: Path = live_dir / f"{name}-{HOST}-{os.getpid()}{JOURNAL_SUFFIX}"
    self._lock: threading.Lock = threading.Lock()
    self._handle = self.path.open("a", encoding="utf-8", newline="\n")

write(row: Mapping[str, object]) -> None

146
147
148
149
150
151
152
153
154
155

def write(self, row: Mapping[str, object]) -> None:
    """Append one record and flush it, so a reader tailing the journal sees it before the next parquet flush."""
    line = json.dumps(row, default=json_default)

    with self._lock:
        if self._handle.closed:
            return

        _ = self._handle.write(line + "\n")
        self._handle.flush()

checkpoint(store: HiveStore) -> list[Path]

157
158
159
160
161
162
163
164
165
166
167
168

def checkpoint(self, store: HiveStore) -> list[Path]:
    """Append the journal's records to ``store`` and truncate it; returns the shards written."""
    with self._lock:
        if self._handle.closed or self.path.stat().st_size == 0:
            return []

        rows = scan_journals(self.path.parent, self.path.name, store.schema).collect()
        written = store.append(rows)
        _ = self._handle.truncate(0)
        _ = self._handle.seek(0)

        return written

close() -> None

170
171
172
173
174
175
176
177
178

def close(self) -> None:
    """Close the journal handle and delete the (now empty) file; idempotent."""
    with self._lock:
        if self._handle.closed:
            return

        self._handle.close()

    self.path.unlink(missing_ok=True)

A telemetry root and the typed hive store for each table under it.

Methods:

• stores –

  Every table under this root, so a maintenance pass reaches them all without naming them one by one.

Attributes:

• live (Path) –

  The directory holding each process's write-ahead JSONL journal.

• logs (HiveStore) –

  Every log record, partitioned by date and process.

• runs (HiveStore) –

  The run index, partitioned by the month a run was created in.

• artifacts (HiveStore) –

  The artifact index, partitioned by artifact kind.

• sweeps (HiveStore) –

  A launch row per sweep plus a closing row once it finishes, partitioned by the month it started in.

• sweep_events (HiveStore) –

  Per-point state transitions, partitioned by sweep so one sweep's progress reads one directory.

79
80
81

def __init__(self, root: str | Path) -> None:
    """Address the tables under the telemetry root ``root``."""
    self.root: Path = Path(root)

live: Path

logs: HiveStore

runs: HiveStore

artifacts: HiveStore

sweeps: HiveStore

sweep_events: HiveStore

stores() -> list[HiveStore]

113
114
115

def stores(self) -> list[HiveStore]:
    """Every table under this root, so a maintenance pass reaches them all without naming them one by one."""
    return [self.logs, self.runs, self.artifacts, self.sweeps, self.sweep_events]

Bases: StrEnum

The subdirectories of a telemetry root: one per table, plus the journal directory the live tables are tailed from.

Attributes:

• LOGS –

  The log table.

• RUNS –

  The run index.

• ARTIFACTS –

  The artifact index.

• SWEEPS –

  The sweep table: a launch row per sweep, plus a closing row once it finishes.

• SWEEP_EVENTS –

  The sweep event table: one row per point transition.

• LIVE –

  The write-ahead journals the live tables are tailed from.

LOGS = 'logs'

RUNS = 'runs'

ARTIFACTS = 'artifacts'

SWEEPS = 'sweeps'

SWEEP_EVENTS = 'sweep_events'

LIVE = 'live'

Bases: ValueError

Neither an explicit path nor FEMFORGE_TELEMETRY names a telemetry root.

The parquet index of one runs directory: an incremental update over its bundles, the record a driver appends per run, and query.

Methods:

• update –

  Index every bundle whose manifest.json moved since the last pass — all of them when rebuild — and return the rows appended.

• record –

  Append one row for a run the driver just wrote, so the index is current without waiting for a scan.

• query –

  A fluent query over the whole index — every runs directory the root knows, not only this one — with the live claims one with_live away.

323
324
325
326
327

def __init__(self, telemetry: Telemetry, runs_dir: str | Path) -> None:
    """Track the bundles under ``runs_dir`` in the run table of ``telemetry``."""
    self.telemetry: Telemetry = telemetry
    self.runs_dir: Path = Path(runs_dir)
    self.store = telemetry.runs

update(*, rebuild: bool = False) -> int

364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388

def update(self, *, rebuild: bool = False) -> int:
    """Index every bundle whose ``manifest.json`` moved since the last pass — all of them when ``rebuild`` — and return the rows appended."""
    candidates: list[tuple[Path, datetime]] = self._candidates()
    if not candidates:
        return 0

    pending = candidates if rebuild else self._changed(candidates)

    rows: list[dict[str, object]] = []
    for sim_dir, mtime in pending:
        rows.append(
            run_row(
                _shallow(sim_dir / BundleFile.MANIFEST),
                sim_dir,
                source=RunSource.SCAN,
                manifest_mtime=mtime,
            )
        )

    if not rows:
        return 0

    _ = self.store.append(self._inherit_sweep(pl.DataFrame(rows, schema=RUN_INDEX_SCHEMA)))

    return len(rows)

record(manifest: RunManifest, sim_dir: Path, source: RunSource = DRIVER) -> None

412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433

def record(self, manifest: RunManifest, sim_dir: Path, source: RunSource = RunSource.DRIVER) -> None:
    """Append one row for a run the driver just wrote, so the index is current without waiting for a scan.

    Args:
        manifest: The run's manifest.
        sim_dir: The bundle directory.
        source: How the row was produced.
    """
    try:
        mtime: float = (sim_dir / BundleFile.MANIFEST).stat().st_mtime
        written = datetime.fromtimestamp(mtime, tz=timezone.utc)
    except OSError:
        written = datetime.now(timezone.utc)

    row = run_row(
        cast("Mapping[str, object]", manifest.model_dump(mode="json")),
        sim_dir,
        source=source,
        manifest_mtime=written,
    )

    _ = self.store.append(pl.DataFrame([row], schema=RUN_INDEX_SCHEMA))

query() -> RunQuery

435
436
437

def query(self) -> RunQuery:
    """A fluent query over the whole index — every runs directory the root knows, not only this one — with the live claims one ``with_live`` away."""
    return RunQuery(self.store.scan())

A fluent filter over the run index; each method returns a new query and nothing runs until frame.

Methods:

• where –

  A new query narrowed by predicate, judged on each run's latest row; the other filters are named shorthands for this.

• status –

  Runs in any of statuses.

• tags –

  Runs whose manifest metadata carries every one of these tag pairs.

• text –

  Runs whose id, description or notes contain needle, case-insensitively.

• since –

  Runs created at or after when.

• sweep –

  Runs solved under the sweep sweep_id.

• newest –

  Keep only the count most recently created runs, applied after the duplicate rows collapse.

• with_live –

  Also carry the lock-owner columns, so a listing can mark which runs are in flight.

• frame –

  The matching rows, newest first, one per run (the latest row wins), joined with the live claims when asked.

• run_ids –

  The ids of the matching runs, newest first.

• runs –

  The matching bundles as Run facades, skipping the ones retention has since removed.

Attributes:

• lazy (LazyFrame) –

  The run index rows the query filters.

• predicates (tuple[Expr, ...]) –

  The filters each run's latest row must pass.

• limit (int | None) –

  How many of the newest runs to keep; None keeps them all.

• live (bool) –

  Whether frame joins the lock-owner columns.

lazy: LazyFrame

predicates: tuple[Expr, ...] = ()

limit: int | None = None

live: bool = False

where(predicate: Expr) -> RunQuery

233
234
235

def where(self, predicate: pl.Expr) -> RunQuery:
    """A new query narrowed by ``predicate``, judged on each run's latest row; the other filters are named shorthands for this."""
    return replace(self, predicates=(*self.predicates, predicate))

status(*statuses: str) -> RunQuery

237
238
239

def status(self, *statuses: str) -> RunQuery:
    """Runs in any of ``statuses``."""
    return self.where(pl.col(RunColumn.STATUS).is_in(list(statuses)))

tags(**pairs: str) -> RunQuery

241
242
243
244
245
246
247

def tags(self, **pairs: str) -> RunQuery:
    """Runs whose manifest metadata carries every one of these tag pairs."""
    query = self
    for key, value in pairs.items():
        query = query.where(pl.col(RunColumn.TAGS).str.json_path_match(f"$.{key}") == value)

    return query

text(needle: str) -> RunQuery

249
250
251
252
253
254
255
256
257
258
259
260