dynalab-core/docs/API.md

# DynaLab Core API

`Core` runs a DynaLab Core instance: it accepts connector data, retains live
values, records data, prepares DLPak archives, plays recordings, and manages
derive units. Connector authors should also read the [protocol guide](Protocol.md).

> [!IMPORTANT]
> **Packet model change.** The default packet and signal classes are lightweight
> `@dataclass(slots=True)` objects. They intentionally do not validate or coerce
> input at construction time. Pydantic models are retained only at a JSON
> boundary and are named with the `Json` prefix, such as `JsonHeartbeat` and
> `JsonConnectorHello`. Use the `Json*` classes with JSON-over-TCP and convert
> with `to_pure()`; use the unprefixed classes inside Core, connectors, and
> derive code. See [Packet representations](Protocol.md#packet-representations).

> [!WARNING]
> DynaLab Core is in alpha. Its API and protocol can change between releases.

## Quick start

```python
from dynalab_core import Core
from dynalab_core.config import CoreConfig

core = Core(CoreConfig(host="127.0.0.1", port=58763))
try:
    core.start()
    core.wait()
finally:
    core.stop()
```

Constructing `Core` starts its internal workers. `start()` starts the TCP
server. Always call `stop()` during shutdown; a stopped core cannot be started
again.

## Configuration

`CoreConfig` is a Pydantic configuration model in `dynalab_core.config`.

| Field | Type | Default | Meaning |
| --- | --- | --- | --- |
| `host` | `str` | `"127.0.0.1"` | TCP bind address. |
| `port` | `int` | `58763` | TCP bind port; validated from 0 through 65535. |
| `thread_debug` | `bool` | `False` | Enables periodic per-thread CPU debug logging. |
| `derive_queue_capacity` | `int` | `16384` | Maximum queued requests in the derive router and each derive unit; minimum 1. |
| `playback_mode` | `"reprocess" \| "read"` | `"reprocess"` | Whether playback recalculates bound derived signals or reads their recorded values. Offline `reprocess()` requires `"read"`. |

`CoreConfig.bind_str() -> str` returns `"host:port"`.

## Lifecycle

### `Core(config: CoreConfig)`

Creates a core, its input, playback, and derive workers, plus the JSON/TCP
server object. It does not bind the network port until `start()`.

### `start() -> None`

Starts the configured JSON/TCP server. It may be called only once, while Core
is initialised; otherwise it raises `CoreError(code="invalid_state")`. A start
timeout or bind failure raises `JsonServerError` with code `"start_timeout"` or
`"startup_failed"`.

### `wait(timeout: float | None = None) -> None`

Blocks until `stop()` is called or the optional timeout elapses.

### `stop() -> bool`

Stops the server, connector registry, derive registry, input worker, and
playback worker. It also waits up to five seconds for an active data operation;
an operation still running makes shutdown incomplete. A processing operation
that finishes after stopping begins cannot commit a new processing buffer.
Returns `True` only when all required components stop within their time limits;
it returns `False` if shutdown is incomplete. Calling it on a constructed Core
before `start()`, or calling it again after shutdown, is supported.

## Modes and buffers

`CoreMode` is `"realtime" | "playback" | "offline"`; new instances start in
`"realtime"`. `set_mode()` clears cached live values and cannot run during an
active recording (`CoreError(code="recording_active")`).

| Mode | Source | Purpose |
| --- | --- | --- |
| `"realtime"` | Connected connectors | Live values and recording. |
| `"playback"` | Loaded DLPak | Time-based replay. |
| `"offline"` | Recorded or loaded DLPak | Prepare a processing buffer without replaying it. |

Endpoints remain connected in non-realtime modes, but only realtime connector
values are routed into Core.

### `set_mode(mode: CoreMode) -> None` / `get_mode() -> CoreMode`

Sets or returns the current mode. `CoreMode` is a type contract; callers must
pass one of the three values above. Leaving playback mode automatically pauses
it at its current position. Returning to playback mode does not automatically
resume it; call `play()` when ready.

### `load_playback_engine(input: Path | None = None, *, background: bool = False) -> UUID | None`

Available only in `"playback"` mode (`CoreError(code="invalid_mode")`). Pass a
`.dlpak` path, or omit `input` to load the current processing buffer. With no
processing buffer, omitting it raises `CoreError(code="no_playback_data")`.
Loading takes a deep copy of the archive, resets the playback position to zero,
and clears cached live values. It does not start playback; use `play()`.

### `load_processing_buffer(input: Path | None = None, *, background: bool = False) -> UUID | None`

Available only in `"offline"` mode (`CoreError(code="invalid_mode")`). Pass a
`.dlpak` path to load it, or omit `input` to turn the completed recording into
a processing buffer and create its manifest. Omitting `input` before a
recording is available raises `CoreError(code="no_playback_data")`.

Completed recordings, processing buffers, and loaded playback archives have
independent sample containers, indexes, locks, and manifest metadata. Loading
a processing buffer deep-copies the completed recording. Loading playback
deep-copies its input archive. Reprocessing, editing metadata, or beginning a
new recording therefore does not mutate earlier snapshots. `DLPak.set_data()`
also takes an independent data snapshot; `copy()` and `deepcopy()` of a DLPak
copy both its data and its nested manifest metadata.

### `reprocess(*, background: bool = False) -> UUID | None`

Recalculates derived signals in the current processing buffer using the bound
derive units. Call it in `"offline"` mode after `load_processing_buffer()`, with
`CoreConfig(playback_mode="read")`. Units with source inputs run first; units
that depend on their derived outputs run in dependency order. Each unit runs
in sample order using per-signal binary searches instead of full-buffer scans.
For other inputs it selects the nearest recorded value, preferring the earlier
sample when distances tie. This retains the previous offline alignment policy,
including selection of future samples. The operation
works on a deep copy, removes its previously recorded derived samples and
descriptors, and merges the new outputs. It replaces the processing buffer
only after all units finish successfully. Calling it again replaces those
outputs. It does not write an archive or start playback; use `write()` to save
the result, or switch to `"playback"` mode, call `load_playback_engine()`, and
then `play()`.

The mode, configuration, and buffer checks raise `CoreError` with codes
`"invalid_mode"`, `"invalid_config"`, and `"empty_buffer"`, respectively.
Missing inputs raise `DeriveError(code="missing_inputs")`, cycles raise
`DeriveError(code="dependency_cycle")`, and a unit that produces no values raises
`DeriveError(code="no_values")`. Errors from a derive function propagate to the
synchronous caller or are stored in the background operation status. Failure
leaves the original processing buffer, including its prior derived results
and metadata, unchanged.

## Signals and live values

`SignalDescriptor` is the lightweight internal signal type in
`dynalab_core.protocols.packets.handshake`. Its fields are `id`, `name`,
`type` (`"number"` or `"binary"`), `min_value`, `max_value`, `unit`,
`timeout_ms` (default `2000`), and `origin` (`"source"` or `"derived"`, default
`"source"`). Use `JsonSignalDescriptor` only when serializing a descriptor.

### `get_signal_descriptor(signal_id: UUID) -> SignalDescriptor | None`

Returns a copy of the matching descriptor or `None`. Realtime descriptors come
from connectors and bound derive units. Playback descriptors come from the
loaded archive; derived descriptors are recalculated only when
`playback_mode="reprocess"`.

### `get_all_signal_descriptors() -> list[SignalDescriptor]`

Returns descriptors available in the current mode. In `"read"` playback mode,
recorded derived descriptors are included; in `"reprocess"` mode, bound derive
units supply them instead.

### `get_live_value(signal_id: UUID) -> float | None`

Returns the most recent non-stale value for a signal, or `None` when unknown,
not yet received, or past the signal's `timeout_ms`.

### `get_all_live_values() -> dict[UUID, float]`

Returns a snapshot of current non-stale values. Signals without a live value
are omitted. Ingress timestamps use `time.monotonic_ns()`; see the protocol
guide for details.

## Recording and DLPak archives

### `start_recording() -> None` / `stop_recording() -> None`

Recording is available only in `"realtime"` mode. Start and stop insert markers
into Core's input queue and wait for the input worker to acknowledge them.
Samples queued before the start marker are excluded; samples between the start
and stop markers are included. This is a boundary at Core's input queue:
packets still buffered in an endpoint and derive results not yet queued when
the stop marker is inserted are outside the recording. There is no fixed sleep
or guess about when queued samples have finished.

Starting creates a new in-memory buffer. Stopping normalizes and sorts the
completed samples and creates an isolated offload archive. Signal descriptors
are retained throughout recording, including descriptors attached to incoming
values. A connector can join after recording starts and disconnect before its
queued samples are ingested without losing their metadata.

Errors use `"incorrect_mode"`, `"recording_active"`, or `"not_recording"` as
appropriate. A failed input worker raises `CoreError(code="worker_failed")`;
a full queue that cannot accept a control marker within five seconds raises
`CoreError(code="input_overload")`.

### `get_recording_state() -> bool`

Returns whether a recording is active.

### `write(dir: str | Path, filename: str, *, source: BufferSource = "processing", background: bool = False) -> UUID | None`

Writes the selected buffer as a `.dlpak` archive; the extension is appended
when absent. `BufferSource` is `"recording" | "processing" | "playback"`, defined
in `dynalab_core.types`. The default remains `"processing"` for compatibility.
Use `source="recording"` to save a completed recording directly, without
switching modes or loading the processing buffer. Stop recording first;
saving an active recording raises `CoreError(code="recording_active")`.
`source="playback"` saves the loaded archive, not the current playback position
or newly recalculated live values. Saving does not change mode or load data.

An unavailable buffer raises `CoreError(code="no_recording")`. Archive failures
raise `DLPakError(code="write_failed")`. Writes use a temporary file in the
destination directory and replace the destination only after the archive has
been completed successfully. The destination directory must already exist.

```python
from pathlib import Path

core.start_recording()
# Receive connector data, then finish and save it.
core.stop_recording()
core.write("recordings", "run-001", source="recording")

core.set_mode("offline")
core.load_processing_buffer(Path("recordings/run-001.dlpak"))
```

### DLPak format and standalone access

`DLPak` in `dynalab_core.dlpak` can also be used without Core:

```python
from pathlib import Path
from dynalab_core.dlpak import DLPak

archive = DLPak().read(Path("recordings/run-001.dlpak"))
series = archive.get_data_range(start_ns=0, end_ns=1_000_000_000)
archive.write("recordings", "run-001-copy")
```

Archives are ZIP files containing `manifest.json` and `data.csv`. The manifest
contains `format_version` (currently `1`), a UTC recording timestamp, signal
descriptors, and comments. Older manifests without `format_version` are read
as version 1; unsupported versions are rejected. Signal IDs in the manifest
must be unique, and every signal present in the CSV must have a descriptor.
Descriptors without samples are allowed. Invalid archives raise
`DLPakError(code="read_failed")`, with the underlying cause retained.

The CSV header is `timestamp_ns` followed by signal UUID columns. Empty cells
represent absent samples. Rows are ordered by timestamp. Repeated timestamps
preserve multiple values for the same signal; samples are not collapsed.

## Buffer queries for plotting

### `get_data_range(signal_ids: list[UUID] | None = None, start_ns: int | None = None, end_ns: int | None = None, *, source: BufferSource | None = None) -> dict[UUID, list[tuple[int, float]]]`

Returns an independent snapshot grouped by signal ID. Each list contains
`(timestamp_ns, value)` pairs in timestamp order, including duplicate
timestamps. Both bounds are inclusive. Omit a bound for an open-ended range,
or omit both for all samples. Omit `signal_ids` to select every signal; an
unknown requested ID maps to an empty list. An inverted range raises
`ValueError`. Data is returned at its original resolution without resampling
or downsampling.

By default, the selected buffer follows the current mode:

| Mode | Queried buffer |
| --- | --- |
| `"realtime"` | Active recording, or the latest completed recording when not recording. |
| `"offline"` | Loaded processing buffer. |
| `"playback"` | Loaded playback archive, including its recorded derived samples. |

Pass `source="recording"`, `"processing"`, or `"playback"` to query a specific
buffer without changing modes. An unavailable buffer raises
`CoreError(code="no_recording")`; an available empty buffer returns empty data.
No history is collected outside recordings. Playback queries access the
loaded archive, not newly recalculated live derived values.

Completed recordings use nanoseconds relative to their first sample. Active
recordings still contain ingress `time.monotonic_ns()` timestamps. Imported
archives retain their stored timestamps. Bounds and returned timestamps always
use the selected buffer's stored time base; stopping a recording normalizes it.

```python
bounds = core.get_time_range()
series = core.get_data_range(
    [rpm_signal_id, torque_signal_id],
    start_ns=1_000_000_000,
    end_ns=2_000_000_000,
)
# series[rpm_signal_id] is a list of (timestamp_ns, value) pairs.
```

### `get_time_range(*, source: BufferSource | None = None) -> tuple[int, int] | None`

Returns the minimum and maximum stored timestamps for the same buffer selection
as `get_data_range()`. An empty buffer returns `None`; a single sample returns
its timestamp for both bounds.

`DLPak` and `ValueBuffer` expose the same `get_data_range()` and
`get_time_range()` methods without the `source` argument. Buffer mutations
maintain per-signal indexes; out-of-order signal samples are sorted before
querying. Once sorted, a range query uses binary search plus copying the
selected samples, rather than scanning unrelated signals.

## Playback

### `play() -> None` / `pause() -> None`

Start or resume, and pause, a loaded playback engine. Both require
`"playback"` mode (`CoreError(code="invalid_mode")`). `play()` before loading
data raises `PlaybackError(code="no_data")`. Pause preserves the current
position and resume continues with the next pending sample. Controls interrupt
waits across long sample gaps; a full input queue does not block pause or stop.
Playback returns to timestamp zero after its last playable sample, and a later
`play()` restarts it. An empty archive completes without emitting samples.

Playback timing starts at the archive's first stored timestamp, even if it is
nonzero. The loaded archive itself keeps its original timestamps for range
queries and saving. In reprocess mode, recorded derived values are skipped and
bound derive units calculate their replacements.

### `seek(timestamp_ns: int) -> None`

Sets the playback position relative to the recording's first sample, in
nanoseconds. Works while playing or paused and preserves that state. The next
emitted sample is at or after the requested position, including all samples
exactly at that timestamp. Positions past the recording's end are clamped to
its duration. Seeking clears cached live values; fresh values arrive when
playback runs. Queued samples and derive results from before the seek are
discarded rather than entering the new timeline.

Requires `"playback"` mode; a negative timestamp raises
`PlaybackError(code="invalid_timestamp")`, and seeking before loading an archive
raises `PlaybackError(code="no_data")`.

### `get_playback_state() -> bool`

Returns `True` while playback is running, including waits between samples.

### `get_playback_position() -> int`

Returns the current playback position in nanoseconds relative to the archive's
first sample. The position advances while playing, remains fixed while paused,
and resets to zero when playback finishes or a new archive is loaded.

## Derived signals

### `bind_derive_unit(dlunit: DLUnit | None = None, file: Path | None = None) -> UUID`

Binds an in-memory `DLUnit` or loads one from a `.dlunit` path, then returns
its ID. Provide exactly one source. Missing or competing sources raise
`CoreError(code="bind_derive_failed")`; invalid source or function shape raises
`DeriveError` (for example `"invalid_source"` or `"invalid_signature"`).

Derive source operates on internal `ValueDescriptor` and `SignalDescriptor`
dataclasses, not their `Json*` equivalents. Source is executed with `exec()`;
only load units from trusted sources.

### `validate_derive_units() -> None`

Checks the configured graph against signals available in the current mode.
Missing inputs (including archive signals with no samples when reprocessing)
raise `DeriveError(code="missing_inputs")`; dependency cycles raise
`"dependency_cycle"`; duplicate outputs or outputs colliding with sources raise
`"duplicate_output"`. Outputs must have `origin="derived"`, otherwise validation
raises `"invalid_origin"`. A derive function must return a `ValueDescriptor`
whose `signal_id` is its configured output, or it raises `"invalid_result"`.

Structural validation runs before a unit is registered, so cycles and duplicate
outputs cannot begin realtime execution. Units can be bound before their sources
or upstream units connect; missing inputs are checked before execution. Realtime
routing revalidates when connectors, bound units, mode, or loaded data change.
An incomplete graph blocks derive routing until its dependencies become
available, and reports the reason in the routing worker's status. It does not
stop raw acquisition or recording. Offline reprocessing validates the entire
graph and sample availability before calling any derive function.

Connector signal IDs must also be unique across connectors and derive outputs;
registration rejects collisions with `ConnectorError(code="duplicate_signal")`.
Connector descriptors must have `origin="source"`.

### Saving DLUnit archives

`DLUnit.set_data(source: str)` accepts the Python source text.
`DLUnit.write(dir, filename)` writes `manifest.json` and the UTF-8 source as
`unit.py` inside a ZIP archive, appending `.dlunit` when needed.
`DLUnit().read(path)` restores both the manifest and source, ready to pass to
`bind_derive_unit()`. Saving without source or a manifest raises `DLUnitError`
with code `"no_data"` or `"no_manifest"`, respectively. Writes use a temporary
file followed by atomic replacement, so a failed write preserves an existing
unit archive.

### `unbind_derive_unit(unit_id: UUID) -> None`

Stops and removes a bound derive unit. An unknown ID has no effect.

## Background operations and progress

`load_processing_buffer()`, `load_playback_engine()`, `write()`, and `reprocess()`
remain synchronous by default: they return `None` on success and raise errors
on failure. Pass `background=True` to return an operation UUID immediately.
Only one data operation can run at a time. Data operations are rejected while
recording; mode changes, recording controls, derive binding/unbinding, and
playback controls are rejected while a data operation is active, with
`CoreError(code="operation_active")`. Queries and status polling remain available.

### `get_operation(operation_id: UUID | None = None) -> OperationStatus | None`

Returns an immutable snapshot from `dynalab_core.operations`. With no ID, it
returns the most recently started operation, or `None` before any operation.
Fields are `id`, `name`, `state` (`"running"`, `"completed"`, or `"failed"`),
`completed`, `total`, `message`, `error_code`, and `error_message`. The `progress`
property is the fraction `completed / total`. Load/save progress reports coarse
archive stages; reprocessing reports completed derive units, rather than a
predicted remaining time. Completion is determined by `state`, not the progress
fraction alone. The latest 100 operations are retained.

### `wait_operation(operation_id: UUID, timeout: float | None = None) -> OperationStatus`

Waits for completion or the timeout. On timeout, returns the current running
status. On failure, raises the original exception; the failed status remains
available through `get_operation()`. An unknown or expired ID raises
`CoreError(code="unknown_operation")`.

```python
operation_id = core.reprocess(background=True)
status = core.get_operation(operation_id)  # Poll from the GUI timer.
if status.state == "failed":
    show_error(status.error_message)
# Outside the GUI event loop, wait and propagate any error:
core.wait_operation(operation_id)
```

## Connector, derive, and worker status

### `get_state() -> str`

Returns Core's lifecycle state: `"initd"`, `"started"`, `"stopping"`, `"stopped"`,
or `"stop_incomplete"` after construction.

### `get_connectors() -> list[ConnectorStatus]`

Returns immutable snapshots of registered endpoints: `id`, `name`, `version`,
`signal_ids`, and `state` (`"connected"`, `"stopping"`, or `"failed"`). Endpoints
are removed after disconnect cleanup; a GUI can compare snapshots to detect
connection and disconnection.

### `get_derive_units() -> list[DeriveUnitStatus]`

Returns each bound unit's `id`, `input_signal_ids`, `output_signal_id`, and
`worker` status. Failed units retain their error until unbound; unbind and rebind
a repaired unit to start a new worker.

### `get_worker_statuses() -> list[WorkerStatus]`

Returns the input, JSON server, playback, derive router, and individual derive
workers. Each status contains `name`, `state` (`"running"`, `"stopped"`,
`"failed"`, or `"blocked"`), `error`, `queue_size`, `queue_capacity`, and
`dropped_samples`. A blocked routing worker reports missing dependencies.
A server not started yet is stopped. Worker status describes worker health;
use `get_playback_state()` to distinguish playing from paused.
The status dataclasses are defined in `dynalab_core.types`.

### Derive overload policy

`CoreConfig.derive_queue_capacity` bounds both the routing queue and every
unit's input queue. Full derive queues drop the newest request, increment their
`dropped_samples` counter, and log a warning on the first drop. They do not block
raw ingestion or discard raw samples already accepted into Core's recording
buffer. Unit output waits for space in Core's bounded input queue and remains
interruptible during shutdown. Core's input queue has its own existing limit;
connector input overload disconnects the affected endpoint and is logged.
Offline processing does not use these derive queues and does not drop samples.

## Exceptions

Expected library errors inherit from `DynaLabError` in `dynalab_core.errors`
and expose a stable `code` attribute. Catch a specific subsystem class for
recovery, or `DynaLabError` at an application boundary. Operating-system and
validation causes are retained through `error.__cause__` where applicable.

| Exception | Area |
| --- | --- |
| `CoreError` | Lifecycle, modes, recording, buffer preparation, derive binding. |
| `DLPakError` | DLPak archive read/write and completeness. |
| `DLUnitError` | DLUnit archive read/write and completeness. |
| `DeriveError` | Derive source, signature, arguments, registration. |
| `PlaybackError` | Playback data, state, and timestamps. |
| `ConnectorError` | Connector queues and registration. |
| `JsonServerError` | JSON/TCP server startup and timeouts. |