ldtc.runtime¶

Fixed-Δt real-time loop primitives. Two pieces:

scheduler: a daemon-thread FixedScheduler that runs a tick callback every Δt seconds and tracks per- tick jitter for the Δt guard.
windows: a SlidingWindow ring buffer that collects state vectors for the next estimator pass.

Together these define the harness's heartbeat. The CLI constructs both and hands them to the run loop in run_baseline.

Runtime utilities for fixed-interval scheduling and sliding windows.

runtime provides the small mechanical pieces that LDTC's verification loops are built on:

scheduler exposes FixedScheduler, a Δt-enforcing scheduler with jitter metrics and audit hooks.
windows exposes SlidingWindow for per-channel telemetry buffering and block_bootstrap_indices for CI estimation.

These primitives are intentionally tiny and decoupled. They have no knowledge of NC1 or SC1; they exist so that the CLI run loop can stream telemetry into the lmeas estimators on a stable cadence.

scheduler¶

scheduler ¶

Fixed-interval scheduler.

A lightweight Δt-enforcing scheduler with jitter metrics and optional audit hooks. Used by CLI runs and the verification harness loops to guarantee that measurements happen on a stable cadence regardless of estimator runtime.

The scheduler runs the user-provided tick callback in a daemon thread, so the caller stays responsive (for example, to handle keyboard interrupts) while ticks fire in the background. Jitter is recorded per tick and exposed via stats.jitter_* properties so that the dt_guard can decide whether the current schedule is still healthy.

TickStats `dataclass` ¶

TickStats(dt_target: float, ticks: int = 0, jitter_abs_sum: float = 0.0, jitter_max: float = 0.0, start_time: float = perf_counter(), jitters: list[float] = list())

Per-run jitter and tick counters for a FixedScheduler.

Attributes:

Name	Type	Description
`dt_target`	`float`	Target tick period in seconds.
`ticks`	`int`	Number of ticks recorded so far.
`jitter_abs_sum`	`float`	Sum of `\|actual_dt - dt_target\|` over all ticks.
`jitter_max`	`float`	Largest absolute jitter observed.
`start_time`	`float`	`perf_counter()` value captured at construction.
`jitters`	`list[float]`	Per-tick absolute jitter samples, kept for percentile queries.

Methods:

Name	Description
`record`	Record a single tick's actual interval.
`jitter_percentile_abs`	Nearest-rank percentile of the absolute jitter distribution.

elapsed `property` ¶

elapsed: float

Wall-clock seconds since the scheduler started.

jitter_mean_abs `property` ¶

jitter_mean_abs: float

Mean absolute jitter across all recorded ticks.

jitter_p95_abs `property` ¶

jitter_p95_abs: float

Convenience alias for jitter_percentile_abs(0.95).

record ¶

record(actual_dt: float) -> None

Record a single tick's actual interval.

Parameters:

Name	Type	Description	Default
`actual_dt`	`float`	Measured interval in seconds since the previous tick.	required

jitter_percentile_abs ¶

jitter_percentile_abs(q: float = 0.95) -> float

Nearest-rank percentile of the absolute jitter distribution.

Parameters:

Name	Type	Description	Default
`q`	`float`	Percentile in `[0, 1]`. Values outside the range are clamped.	`0.95`

Returns:

Type	Description
`float`	The `q`-th percentile of `\|actual_dt - dt_target\|`, in
`float`	seconds. Returns `0.0` if no ticks have been recorded yet.

FixedScheduler ¶

FixedScheduler(dt: float, tick_fn: Callable[[float], None], on_start: Optional[Callable[[], None]] = None, on_stop: Optional[Callable[[TickStats], None]] = None, audit_hook: Optional[Callable[[str, Dict], None]] = None)

Fixed-interval scheduler.

Enforces a constant sampling interval Δt and invokes a tick callback every period until stopped. Tracks jitter statistics and emits optional audit events through a user-provided hook.

Parameters:

Name	Type	Description	Default
`dt`	`float`	Target period in seconds (`Δt > 0`).	required
`tick_fn`	`Callable[[float], None]`	Callback invoked each tick with the current `perf_counter` timestamp.	required
`on_start`	`Optional[Callable[[], None]]`	Optional hook executed before the worker thread begins.	`None`
`on_stop`	`Optional[Callable[[TickStats], None]]`	Optional hook executed after stop; receives the final `TickStats`.	`None`
`audit_hook`	`Optional[Callable[[str, Dict], None]]`	Optional callable taking `(event: str, details: Dict)` for emitting audit records (typically wired up to `AuditLog.append`).	`None`

Notes

Jitter metrics are accessible on the stats attribute after ticks have been recorded.
Thread-safe updates to Δt can be made via set_dt. The new period takes effect at the next tick boundary.

Initialize the scheduler. See class docstring for argument details.

Methods:

Name	Description
`start`	Start the worker thread.
`stop`	Stop the worker thread and return final stats.
`set_dt`	Change `Δt` at runtime in a thread-safe way.

start ¶

start() -> None

Start the worker thread.

No-op if a worker is already running. Calls on_start (if provided) before the thread begins, and emits a scheduler_started audit event.

stop ¶

stop() -> TickStats

Stop the worker thread and return final stats.

Joins the worker thread, calls on_stop (if provided), and emits a scheduler_stopped audit event with the summarized jitter metrics.

Returns:

Type	Description
`TickStats`	The final `TickStats` for
`TickStats`	the run.

set_dt ¶

set_dt(new_dt: float) -> float

Change Δt at runtime in a thread-safe way.

The new period is applied at the next tick boundary, so a tick currently in flight finishes against the previous interval.

Parameters:

Name	Type	Description	Default
`new_dt`	`float`	New period in seconds (`Δt > 0`).	required

Returns:

Type	Description
`float`	The previous `dt` value.

windows¶

windows ¶

Sliding windows and bootstrap indices.

A fixed-length per-channel sliding window plus a helper to generate circular block-bootstrap indices for confidence-interval estimation in the lmeas estimators.

SlidingWindow ¶

SlidingWindow(capacity: int, channel_order: List[str])

Fixed-length, per-channel sliding window.

Maintains a deque per channel and exposes a dense matrix view when the window is full. Useful for streaming estimators that require a fixed-size time-by-signal buffer.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	Number of samples to retain per channel.	required
`channel_order`	`List[str]`	Ordered list of channel names used for matrix columns. Order matters: the matrix columns appear in this order, and the column index is what the `PartitionManager` stores as a partition member.	required

Notes

append inserts a new sample dict; missing keys default to 0.0.
get_matrix returns a (T, N) numpy array in channel_order.

Initialize per-channel deques.

Parameters:

Name	Type	Description	Default
`capacity`	`int`	Maximum number of samples to retain per channel.	required
`channel_order`	`List[str]`	Ordered list of channel names; matrix columns follow this order.	required

Methods:

Name	Description
`append`	Append one sample across all channels.
`ready`	Return `True` once every channel deque is full.
`get_matrix`	Return the current window as a dense `(T, N)` matrix.
`clear`	Drop all buffered samples across every channel.

append ¶

append(sample: Dict[str, float]) -> None

Append one sample across all channels.

Parameters:

Name	Type	Description	Default
`sample`	`Dict[str, float]`	Mapping of channel name to scalar reading. Missing channels default to `0.0`.	required

ready ¶

ready() -> bool

Return True once every channel deque is full.

get_matrix ¶

get_matrix() -> ndarray

Return the current window as a dense (T, N) matrix.

Returns:

Type	Description
`ndarray`	Numpy array of shape `(capacity, len(channel_order))`.

Raises:

Type	Description
`RuntimeError`	If the window is not yet full.

clear ¶

clear() -> None

Drop all buffered samples across every channel.

block_bootstrap_indices ¶

block_bootstrap_indices(n: int, block: int, draws: int) -> List[ndarray]

Circular block-bootstrap indices for time series.

Generates draws index arrays, each of length n, by stitching together randomly-positioned blocks of length block with circular wrapping. This preserves short-range temporal dependence in the bootstrap samples used by estimate_L.

Parameters:

Name	Type	Description	Default
`n`	`int`	Length of the time series.	required
`block`	`int`	Block length for resampling.	required
`draws`	`int`	Number of bootstrap replicates to generate.	required

Returns:

Type	Description
`List[ndarray]`	List of index arrays (each of length `n`) representing bootstrap
`List[ndarray]`	samples with circular wrapping at boundaries.

Name	Description
`TickStats`	Per-run jitter and tick counters for a `FixedScheduler`.
`FixedScheduler`	Fixed-interval scheduler.

ldtc.runtime¶

scheduler¶

scheduler ¶

TickStats dataclass ¶

elapsed property ¶

jitter_mean_abs property ¶

jitter_p95_abs property ¶

record ¶

jitter_percentile_abs ¶

FixedScheduler ¶

start ¶

stop ¶

set_dt ¶

windows¶

windows ¶

SlidingWindow ¶

append ¶

ready ¶

get_matrix ¶

clear ¶

block_bootstrap_indices ¶

TickStats `dataclass` ¶

elapsed `property` ¶

jitter_mean_abs `property` ¶

jitter_p95_abs `property` ¶