Feat/data api

[wyzula-jan]

Description

This PR introduces the Data API (bec_lib.data_api) — a unified, client-side data access layer for scan data. It
gives every consumer (plotting widgets, scripts, services) one subscription contract for three kinds of data that
previously required three different code paths: the currently running scan, finished scans from the scan history, and
scan-independent device streams.

Architecture

Facade. client.data_api.subscribe(sources, scan, callback, ...) is the single entry point.
sources is a list of (device, entry) pairs; scan selects the scope: "live" follows the active scan and every
scan after it, a concrete scan id serves exactly that scan, and None
subscribes to scan-less device streams (readbacks, monitor_1d monitors, preview signals).

Immutable columnar updates. Every delivery is a full-state SubscriptionUpdate snapshot, not a delta: consumers
render the newest update and never accumulate data themselves, which makes emission coalescing lossless (at most one
update per min_emit_interval, runtime-adjustable).
update.aligned() returns equal-length value columns; update.axis(...) resolves an x-axis (index / timestamp /
device); update.reason distinguishes "live", "backfill", "rebind"
and "history" deliveries.

Ordinal alignment. Points are keyed by an ordinal — the scan point id for monitored signals, the per-scan async
message index for async signals, an arrival counter for legacy unindexed devices — instead of arrival order. Late points
fill their hole, repeated points overwrite in place, and one lagging source can never silently shift a curve against
another. Sources are partitioned into correlation groups (scan, async:<tag>, standalone), and only sources within
a group are aligned with each other.

Plugin routing. Three data source plugins claim scans by priority: live (10), history (50, file-backed on a worker
thread), device streams (90). A scan="live" subscription moves through a defined lifecycle: bind + backfill of
already-recorded points → coalesced live updates →
"rebind" on scan change → terminal flush → automatic re-route to the authoritative history snapshot once the file is
published. Subscribing mid-scan therefore never misses data, and the final state is identical to a later history read.

Bounded resources. max_points caps retention for endless streams; size_limit_bytes arms a size gate — the
history plugin estimates a scan's payload from stored metadata before reading anything, and an over-limit load is
withheld until confirm_size().

Additional improvements on top of the core

• Production hardening: per-client instance registry with automatic facade re-creation after
  close(), unavailable sources retried on the next scan status instead of failing the subscription, async signal-info
  cache invalidation on device config updates.
• Performance: incremental sorted columns with cached snapshots (constant-time repeated
  aligned() calls); numpy-native bulk ingest for history reads — file columns stay ndarray
  end to end, removing multi-second lock holds and O (n) tuple round-trips on multi-million-point scans.
• Chunked history reads with progress reporting: large datasets are read in slabs and a
  progress_callback receives the load fraction, so GUIs can show a progress bar instead of a frozen window.
• Deadlock fix: subscribe() vs. the scan-segment dispatcher callback could deadlock on lock order (Data API lock
  vs. scan-manager lock). Entry points now prefetch scan items before taking the API lock; a deterministic regression
  test reproduces the deadlock on the old code.
• fix(bec_lib): ScanHistory now stores a new entry before firing the update event — previously a callback
  querying the history from the event could miss the scan that triggered it (pre-existing issue, surfaced by the history
  re-route).

Documentation (Diátaxis: learn / how-to / reference) is prepared in a bec_docs PR.

Type of Change

• New feature (bec_lib.data_api package)
• Bug fix (scan_history update-event ordering)

How to test

pytest --random-order bec_lib/tests/test_data_api.py bec_lib/tests/test_data_api_alignment.py \
       bec_lib/tests/test_data_api_history.py bec_lib/tests/test_scan_history.py

• End-to-end coverage was extended in test_scans_lib_e2e.py.
• Try examples mention in the documentation branch in IPython client
• test with Feat/data api bec_widgets#1269, try all plotting widgets there
• The performance of most widgets (especially ones using Curve) is heavily improved

Potential side effects

• No message formats were changed (messages.py untouched) — servers and GUIs remain independently deployable; existing
  consumers of scan_item / scan_history are unaffected -> for full compatibility of BEC Widgets with DataAPI if we
  decide to deploy BEC first and widgets later this PR has to be merged as well add PR
• New API surface only; nothing existing was removed. The companion BEC Widgets PR (Feat/data api bec_widgets#1269) builds
  on this branch and requires it.

Additional Comments

The subscription callback may run on background threads (dispatcher / history worker); Qt consumers should go through
the QtDataSubscription bridge added in the widgets PR bec-project/bec_widgets#1269.

[Copilot]

Pull request overview

This PR introduces a new DataAPI v2 facade in bec_lib that unifies access to live scan data, terminal/history scan data (file-backed), and scan-less device streams behind a single subscription interface with ordinal-based alignment. It also adjusts scan-history update ordering to support consumers that read from the history registry during callbacks, and adds broad test coverage (unit, integration over fakeredis, and an end-to-end scan test).

Changes:

• Add bec_lib.data_api (facade + plugins + alignment models) to serve live/history/device-stream data via a unified subscription API.
• Fix ScanHistory._on_scan_history_update ordering so registry storage occurs before firing SCAN_HISTORY_UPDATE.
• Add BECClient.data_api lazy property and extensive new tests (unit + integration + e2e).

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
bec_lib/bec_lib/scan_history.py	Store scan history message before firing update callback to support registry reads in handlers.
bec_lib/bec_lib/data_api/alignment.py	New ordinal-keyed alignment engine and bundle snapshot builder.
bec_lib/bec_lib/data_api/models.py	New immutable emission contract models (SubscriptionUpdate, SourceData).
bec_lib/bec_lib/data_api/plugin_base.py	Plugin contract and request/spec dataclasses for DataAPI routing.
bec_lib/bec_lib/data_api/live_plugin.py	Live scan plugin (scan_segment + async streams ingestion).
bec_lib/bec_lib/data_api/history_plugin.py	History plugin (metadata-only routing + worker-thread file read + live fallback).
bec_lib/bec_lib/data_api/device_plugin.py	Scan-less device stream plugin (readback/monitor/preview).
bec_lib/bec_lib/data_api/data_api.py	DataAPI facade + subscription lifecycle, routing, rate limiting, and size gating.
bec_lib/bec_lib/data_api/init.py	Public exports for the DataAPI package.
bec_lib/bec_lib/client.py	Add BECClient.data_api lazy property caching a per-client DataAPI instance.
bec_lib/tests/test_scan_history.py	Add regression test asserting update event fires after registry store.
bec_lib/tests/test_data_api.py	New unit/integration tests for live/device-stream subscriptions and lifecycle behaviors.
bec_lib/tests/test_data_api_history.py	New tests for history plugin behavior and live→history handover.
bec_lib/tests/test_data_api_alignment.py	New tests for alignment engine semantics (gaps, grouping, completeness).
bec_lib/tests/test_data_api_benchmarks.py	New performance regression-guard tests for alignment/emission scaling.
bec_ipython_client/tests/end-2-end/test_scans_lib_e2e.py	New end-to-end test exercising monitored + monitored-async bundling in a real scan.

Suppressed comments (1)

bec_lib/bec_lib/data_api/live_plugin.py:364

• For async "add" updates, the ordinal is looked up in async_indices using spec.entry (obj_name). async_indices keys come from the message’s signals keys (typically storage_name), so ordinals may be missed and the series will fall back to arrival-counter ordinals, breaking alignment.

        ordinal = async_indices.get(spec.entry)
        series.insert(ordinal if isinstance(ordinal, int) else None, value, timestamp)

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov]

Codecov Report

❌ Patch coverage is 85.15929% with 191 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
bec_lib/bec_lib/data_api/data_api.py	84.66%	27 Missing and 23 partials ⚠️
bec_lib/bec_lib/data_api/live_plugin.py	81.34%	24 Missing and 23 partials ⚠️
bec_lib/bec_lib/data_api/history_plugin.py	79.27%	28 Missing and 18 partials ⚠️
bec_lib/bec_lib/data_api/device_plugin.py	78.72%	13 Missing and 17 partials ⚠️
bec_lib/bec_lib/data_api/alignment.py	96.29%	6 Missing and 1 partial ⚠️
bec_lib/bec_lib/scan_data_container.py	86.04%	4 Missing and 2 partials ⚠️
bec_lib/bec_lib/data_api/models.py	92.06%	4 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!