padne.context

Process-wide context for cross-cutting concerns. Currently hosts the stage-timing API used to measure how long named stages take across the codebase.

Usage:

with context.timing_session() as session:

with context.stage_timer(“connectivity”):

… # -> “padne.solver.connectivity”

@context.stage_timer def assemble_system(…): … # -> “padne.solver.assemble_system”

@context.stage_timer(“custom_name”) def f(…): … # -> “<f’s module>.custom_name”

with session.durations as d:

… # dict[str, Session.Duration]

The module-level stage_timer is a no-op when called outside an active timing_session.

Session exposes the same polymorphic stage_timer API directly, so once you have a session reference you can also do:

@session.stage_timer # key = “<func module>.<qualname>” def f(): …

with session.stage_timer(“padne.solver.something”):

…

When invoked on a Session directly the string is the full key with no caller-module prefix. The module-level wrapper is what adds the prefix.

Attributes

T

Classes

LockedObject	Wraps a value so it cannot be accessed without holding the lock.
Session	A single bag of stage timings shared across all threads that run

Functions

timing_session()	Open a timing session. While active, module-level stage_timer
merge_durations(→ None)	Merge a durations snapshot into the active session. No-op when no
stage_timer(arg)	Module-level wrapper that adds the caller-module prefix and defers

Module Contents

padne.context.T

class padne.context.LockedObject(obj: T)

Bases: Generic[T]

Wraps a value so it cannot be accessed without holding the lock.

owned = LockedObject([]) with owned as items:

items.append(…)

class padne.context.Session

A single bag of stage timings shared across all threads that run inside one timing_session() block.

class Duration

perf_time: float = 0.0

process_time: float = 0.0

call_count: int = 0

durations

stage_timer(arg)

Polymorphic: with session.stage_timer(“full.key”): … # CM @session.stage_timer # bare decorator @session.stage_timer(“full.key”) # parameterized decorator

snapshot() → dict[str, Session]

Return an independent copy of the recorded durations. Used to ship timings across process boundaries (see padne.parallel).

merge(snapshot: dict[str, Session]) → None

Accumulate durations from another session’s snapshot into this one. Times sum and call counts add, same as repeated stage entries.

format_summary() → str

Return a tree-shaped table of recorded stages.

Keys are split on ‘.’ to form a hierarchy; siblings are sorted by descending perf time (a branch’s sort key is the sum of its descendants’ perf time, with leaf keys contributing their own recorded perf time).

padne.context.timing_session()

Open a timing session. While active, module-level stage_timer entries record into it.

TODO: support multiple concurrent independent sessions. Right now we only allow one active session per process and raise if a second is opened. The check-and-set below races for that case; it’s only safe today because the single-session restriction means nobody opens two in parallel.

padne.context.merge_durations(snapshot: dict[str, Session]) → None

Merge a durations snapshot into the active session. No-op when no session is active, mirroring the stage_timer behavior.

padne.context.stage_timer(arg)

Module-level wrapper that adds the caller-module prefix and defers session lookup to call/enter time.

with stage_timer(“name”): # key = “<caller module>.name”

…

@stage_timer # key = “<func module>.<func qualname>” def f(): …

@stage_timer(“name”) # key = “<caller module>.name” def g(): …