From db5d509b49c80c362683affab95823c6d80059ab Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Mon, 19 Oct 2026 19:35:10 +0200 Subject: [PATCH] Version 0.2.1 Version 0.2.1 now has derive unit context for derivative calculus --- docs/API.md | 27 +++-- docs/Processing.md | 152 +++++++++++++++++++++++++++ docs/Protocol.md | 2 +- docs/README.md | 2 + src/dynalab_core/__init__.py | 16 ++- src/dynalab_core/constants.py | 2 +- src/dynalab_core/derive.py | 182 ++++++++++++++++++++++++++++----- src/dynalab_core/playback.py | 5 + src/dynalab_core/processing.py | 117 +++++++++++++++++++++ uv.lock | 4 +- 10 files changed, 473 insertions(+), 36 deletions(-) create mode 100644 docs/Processing.md create mode 100644 src/dynalab_core/processing.py diff --git a/docs/API.md b/docs/API.md index 4ab7010..53d0038 100644 --- a/docs/API.md +++ b/docs/API.md @@ -127,10 +127,13 @@ 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 +in sample order. Legacy alignment uses per-signal binary searches; contexts +advance bounded histories. +For functions without a processing context, other inputs select the nearest +recorded value, preferring the earlier sample when distances tie; this can +select future samples. Functions accepting `DeriveContext` use causal history +and preserve each input's timestamp instead. See [processing and derivatives](Processing.md). +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 @@ -141,8 +144,10 @@ 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 +`DeriveError(code="dependency_cycle")`. A unit that produces no values raises +`DeriveError(code="no_values")` unless its return annotation allows `None`. +An optional-output unit may produce an empty series; units requiring that output +are skipped with a diagnostic. 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. @@ -366,6 +371,14 @@ Derive source operates on internal `ValueDescriptor` and `SignalDescriptor` dataclasses, not their `Json*` equivalents. Source is executed with `exec()`; only load units from trusted sources. +Functions may optionally accept a `DeriveContext` between their input values and +the final output descriptor. `ctx.derivative(value, order=1, max_gap_s=None)` +estimates time derivatives of orders 1–3 using bounded history in realtime, +offline processing, and recalculated playback. Annotate the return type as +`ValueDescriptor | None` to skip outputs during warm-up. Existing functions keep +their original signatures and behavior. See [DLUnit processing and derivatives](Processing.md) +for examples, numerical behavior, input alignment, and reset rules. + ### `validate_derive_units() -> None` Checks the configured graph against signals available in the current mode. @@ -375,6 +388,8 @@ raise `DeriveError(code="missing_inputs")`; dependency cycles 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"`. +Functions annotated `-> ValueDescriptor | None` may also return `None` to emit +no sample for that invocation. Structural validation runs before a unit is registered, so cycles and duplicate outputs cannot begin realtime execution. Units can be bound before their sources diff --git a/docs/Processing.md b/docs/Processing.md new file mode 100644 index 0000000..4314ade --- /dev/null +++ b/docs/Processing.md @@ -0,0 +1,152 @@ +# DLUnit processing and derivatives + +DLUnits can optionally request a `DeriveContext` to calculate time derivatives +of their declared inputs. The same interface works during realtime acquisition, +offline `Core.reprocess()`, and playback with `playback_mode="reprocess"`. +Playback with `playback_mode="read"` reads previously saved outputs instead. + +## Example: speed to acceleration + +Use this as the DLUnit's `unit.py` source: + +```python +from dynalab_core.processing import DeriveContext +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +def process( + speed: ValueDescriptor, + ctx: DeriveContext, + output: SignalDescriptor, +) -> ValueDescriptor | None: + acceleration = ctx.derivative(speed, order=1, max_gap_s=0.5) + if acceleration is None: + return None + return ValueDescriptor(output.id, acceleration, speed.timestamp) +``` + +Bind it using the existing API: + +```python +from uuid import uuid4 +from dynalab_core.dlunit import DLUnit +from dynalab_core.protocols.packets.handshake import SignalDescriptor + +acceleration_signal = SignalDescriptor( + id=uuid4(), name="Acceleration (m/s²)", type="number", origin="derived" +) +unit = DLUnit() +unit.set_manifest(uuid4(), [speed_signal], acceleration_signal) +unit.set_data(source_text) # The unit.py source above. +core.bind_derive_unit(dlunit=unit) +# Optionally persist it with unit.write(directory, "acceleration.dlunit"). +``` + +`speed_signal` must describe the existing speed input. If its values are in m/s, +the output values are in m/s². Core does not infer physical units or rename +signals automatically. Derivative outputs are ordinary derived signals: they +can feed another DLUnit, be recorded, and be saved in a DLPak. + +## Function contract + +A function takes its `ValueDescriptor` inputs in manifest order, optionally one +`DeriveContext`, then one final `SignalDescriptor` describing the output. Use +the actual imported classes in annotations, without postponed/string +annotations. The source must expose exactly one Python function; importing the +context class is supported. + +Existing functions such as `process(a, b, output) -> ValueDescriptor` continue +to work. To skip output, explicitly annotate `-> ValueDescriptor | None` and +return `None`. This also works without a context. Returning `None` with the +original `-> ValueDescriptor` annotation remains an invalid result. + +No placeholder is emitted when a function returns `None`: live values are not +updated, recording gets no sample, and downstream units receive no event for +that output. A previously published live output remains subject to its normal +timeout; skipping an output does not explicitly clear it. + +An offline unit annotated to allow `None` may produce an empty series. Core logs +`derive.offline_no_values` and skips units requiring that empty output, logging +`derive.offline_upstream_empty`. Reprocessing still succeeds, and descriptors +are retained for the empty outputs. Units that do not allow `None` retain the +`DeriveError(code="no_values")` behavior if they execute but emit nothing. + +## `ctx.derivative(value, order=1, *, max_gap_s=None) -> float | None` + +- `value` is one of the function's supplied input descriptors. +- `order` is an integer: 1, 2, or 3. Other values raise `ValueError`. +- `max_gap_s` is an optional finite positive maximum interval between + consecutive samples in the derivative window. Invalid limits raise + `ValueError`. The default places no gap limit. +- The result uses seconds: input units/s, input units/s², or input units/s³. +- Warm-up requires 2, 3, or 4 distinct timestamps, respectively. Until then, + the result is `None`. A window containing a gap above `max_gap_s` also returns + `None`; estimates resume after enough samples arrive following that gap. +- Repeated calls are read-only and return the same estimate. You can request + different orders for the same input in one invocation. + +The estimator differentiates the degree-`order` polynomial interpolating the +most recent `order + 1` samples. For order 1 this is the usual backward secant: +`(new_value - previous_value) / elapsed_seconds`. Orders 2 and 3 use the actual +unequal sample intervals rather than repeatedly dividing by a nominal period. +Timestamp differences are taken as integers before converting to seconds. + +These are numerical estimates without smoothing. They use only present and +past samples, and higher orders can magnify measurement noise. Very uneven +spacing and floating-point precision also limit accuracy. A non-finite +computed estimate returns `None`. + +## History and alignment + +Core retains at most four samples per declared input, per context-enabled unit. +The router advances history only when that input itself updates and queues a +snapshot for each invocation. Later arrivals cannot alter an already queued +context, and calls to `derivative()` do not advance history. + +Context-enabled units wait until all declared inputs have a sample at or before +the triggering event's timestamp. Each input retains its own original timestamp. +Only the most recent four accepted samples are available, so severely late +events may no longer have a companion sample in history and are skipped. +Equal or older timestamps for a signal are ignored by that unit's history and +do not trigger its processing function. A newer non-finite value clears that +signal's history and skips the invocation; finite samples restart warm-up. +These rules do not remove or alter raw recording samples. + +When another input changes, a derivative of an unchanged input reuses that +input's existing estimate. Its history does not gain another sample. If the +output combines multiple inputs, choose its timestamp explicitly, for example +`max(a.timestamp, b.timestamp)`. Companion values are retained in the bounded +history; they do not expire through the realtime live-value timeout. Apply any +additional age policy in the function using the supplied timestamps. + +Offline processing uses the same bounded history and causal alignment, with a +fresh history on every pass. Functions without a context retain the original +offline nearest-neighbor alignment, including possible future samples and +timestamps assigned from the triggering event. + +Online and offline estimates match for the same chronological accepted input +sequence. Offline processing sorts stored samples; online out-of-order arrivals +can therefore differ. Queue overload can also change the observed sequence: +router drops omit samples from history, while a unit queue drop omits that +invocation but its sample remains in later history snapshots. Asynchronous +multi-input derived chains can observe a different event ordering online. +Offline processing does not drop samples due to queue capacity. + +Starting or stopping recording does not reset ongoing online history. If a +recording begins midstream, its first online derivative outputs may depend on +samples preceding the recording. Offline recalculation warms up from the +recorded samples and can omit those early estimates. + +## Lifecycle + +Online histories are invalidated when Core changes mode, a source connector +registers/disconnects, processing or playback data is replaced, or a playback +control changes its generation (play/resume, pause, seek, or restart). Connector +changes conservatively invalidate all units' histories. Queued requests and +results from an older revision are discarded, including results completing +after the reset. New data warms up the estimator again. + +Offline histories are local to the processing pass and never share online +state. DLUnit archives save source and manifest, not history. DLPak and connector +packet formats are unchanged. No numerical-library dependency is required. diff --git a/docs/Protocol.md b/docs/Protocol.md index 547bcc8..651b365 100644 --- a/docs/Protocol.md +++ b/docs/Protocol.md @@ -83,7 +83,7 @@ packet classes. { "type": "dynalab_hello", "instance_id": "2f3c7f62-0a45-4d66-9f32-7fbeb92f9d36", - "core_version": {"type": "alpha", "major": 0, "minor": 1, "patch": 0}, + "core_version": {"type": "alpha", "major": 0, "minor": 2, "patch": 1}, "protocol_version": {"type": "alpha", "major": 0, "minor": 0, "patch": 1}, "heartbeat_interval_ms": 1000, "heartbeat_timeout_ms": 5000 diff --git a/docs/README.md b/docs/README.md index 539c3bb..5788a53 100644 --- a/docs/README.md +++ b/docs/README.md @@ -12,4 +12,6 @@ This documentation is ordered in the following sections [**Protocol**](Protocol.md) - Documents the data ingress protocol, this is useful for developing connectors to pipe data into the DynaLab Core +[**Processing and derivatives**](Processing.md) - DLUnit context API, first through third time derivatives, warm-up, and online/offline behavior + [**Internal**](Internal/README.md) - Documents the inner working of DynaLab Core including data flow, internal API's, custom file types and much more diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 23a9259..a14bccd 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -22,7 +22,11 @@ import psutil from dynalab_core.buffer import ValueBuffer from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION -from dynalab_core.derive import DeriveRegistry +from dynalab_core.derive import ( + DeriveRegistry, + DerivedValueDescriptor, + DerivedPlaybackValueDescriptor, +) from dynalab_core.dlpak import DLPak from dynalab_core.dlunit import DLUnit from dynalab_core.errors import CoreError @@ -67,6 +71,7 @@ class Core: self._input_error: str | None = None self._recording_signals: dict[UUID, SignalDescriptor] = {} self._mode: CoreModeState = CoreModeState() + self._derive_epoch = 0 self._recording = threading.Event() self._recording_buffer = ValueBuffer() self._recording_timestamp: datetime = datetime.now(timezone.utc) @@ -282,14 +287,17 @@ class Core: if self._mode.value == "playback" and mode != "playback": self._playback_engine.pause() self._mode.value = mode + self._derive_epoch += 1 self._live_values.clear() def _get_source_revision(self) -> object: return ( + self._derive_epoch, self._mode.value, id(self._processing_buffer), id(self._playback_engine._data), self._connector_registry.get_revision(), + self._playback_engine.get_generation(), ) def get_connectors(self) -> list[ConnectorStatus]: @@ -771,6 +779,12 @@ class Core: continue if isinstance(message, ValueDescriptor): with self._live_values_lock: + if isinstance( + message, + (DerivedValueDescriptor, DerivedPlaybackValueDescriptor), + ): + if not self._derive_registry.is_current_result(message): + continue if isinstance(message, PlaybackValueDescriptor): if ( self._mode.value != "playback" diff --git a/src/dynalab_core/constants.py b/src/dynalab_core/constants.py index 6981736..606f787 100644 --- a/src/dynalab_core/constants.py +++ b/src/dynalab_core/constants.py @@ -12,7 +12,7 @@ from dynalab_core.protocols.packets.handshake import ( ) -CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) +CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=2, patch=1) HELLO_PACKET = JsonDynaLabHello( instance_id=uuid4(), diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 648da30..167d4b9 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -4,6 +4,7 @@ from collections.abc import Callable from datetime import datetime, timezone +from dataclasses import dataclass from enum import IntEnum import inspect import logging @@ -23,10 +24,28 @@ from dynalab_core.protocols.packets.handshake import SignalDescriptor from dynalab_core.types import CoreModeState, DeriveUnitStatus, WorkerStatus from dynalab_core.values import SignalValueDescriptor from dynalab_core.operations import ProgressCallback +from dynalab_core.processing import DeriveContext, _SampleHistory log = logging.getLogger(__name__) +@dataclass(slots=True) +class DerivedValueDescriptor(SignalValueDescriptor): + revision: object + + +@dataclass(slots=True) +class DerivedPlaybackValueDescriptor(PlaybackValueDescriptor): + revision: object + + +@dataclass +class _DeriveRequest: + inputs: list[ValueDescriptor] + context: DeriveContext | None + revision: object + + class ParserState(IntEnum): INIT = 0 VALUES_DONE = 1 @@ -37,13 +56,14 @@ class DeriveUnit: def __init__( self, unit_uuid: UUID, - processing_function: Callable[..., ValueDescriptor], + processing_function: Callable[..., ValueDescriptor | None], input_signals: list[SignalDescriptor], return_signal: SignalDescriptor, return_queue: Queue[ValueDescriptor], is_current_playback_value: Callable[[PlaybackValueDescriptor], bool] | None = None, queue_capacity: int = 16384, + get_revision: Callable[[], object] | None = None, ) -> None: self._id: UUID = unit_uuid self._process_function: Callable = processing_function @@ -51,7 +71,10 @@ class DeriveUnit: self._input_signals: list[SignalDescriptor] = input_signals self._stop_event: threading.Event = threading.Event() self._stopped_event: threading.Event = threading.Event() - self._input_queue: Queue[list[ValueDescriptor]] = Queue(queue_capacity) + self._input_queue: Queue[_DeriveRequest] = Queue(queue_capacity) + self._get_revision = get_revision or (lambda: None) + self._history = _SampleHistory() + self._history_revision: object = object() self._error: str | None = None self._dropped_samples = 0 self._status_lock = threading.Lock() @@ -60,6 +83,7 @@ class DeriveUnit: self._num_input_args: int = 0 self._parser_state: int = ParserState.INIT + self._uses_context = False sig = inspect.signature(self._process_function) for name, param in sig.parameters.items(): @@ -70,6 +94,9 @@ class DeriveUnit: self._parser_state = ParserState.VALUES_DONE if self._parser_state == ParserState.VALUES_DONE: + if param.annotation is DeriveContext and not self._uses_context: + self._uses_context = True + continue if param.annotation is SignalDescriptor: self._parser_state = ParserState.SIGNAL_DONE continue @@ -85,9 +112,10 @@ class DeriveUnit: code="invalid_signature", ) - if sig.return_annotation is not ValueDescriptor: + self._allows_none = sig.return_annotation == (ValueDescriptor | None) + if sig.return_annotation is not ValueDescriptor and not self._allows_none: raise DeriveError( - "The derive function must return ValueDescriptor", + "The derive function must return ValueDescriptor or ValueDescriptor | None", code="invalid_signature", ) @@ -134,9 +162,12 @@ class DeriveUnit: try: while not self._stop_event.is_set(): try: - input_args = self._input_queue.get(timeout=0.1) + request = self._input_queue.get(timeout=0.1) except Empty: continue + input_args = request.inputs + if request.revision != self._get_revision(): + continue if len(input_args) != self._num_input_args: log.warning( @@ -156,25 +187,31 @@ class DeriveUnit: if self._has_stale_playback_input(input_args): continue try: - value = self._process_function(*input_args, self._return_signal) - self._validate_result(value) + value = self._process(input_args, request.context) except Exception: - if self._has_stale_playback_input(input_args): + if ( + request.revision != self._get_revision() + or self._has_stale_playback_input(input_args) + ): continue raise + if value is None or request.revision != self._get_revision(): + continue if input_args and isinstance(input_args[0], PlaybackValueDescriptor): - value = PlaybackValueDescriptor( + value = DerivedPlaybackValueDescriptor( signal_id=value.signal_id, value=value.value, timestamp=value.timestamp, generation=input_args[0].generation, + revision=request.revision, ) else: - value = SignalValueDescriptor( + value = DerivedValueDescriptor( value.signal_id, value.value, value.timestamp, self._return_signal, + request.revision, ) while not self._stop_event.is_set(): try: @@ -207,11 +244,17 @@ class DeriveUnit: }, ) - def put_data(self, input: list[ValueDescriptor]) -> bool: + def put_data( + self, + input: list[ValueDescriptor], + *, + context: DeriveContext | None = None, + revision: object = None, + ) -> bool: if self._stop_event.is_set() or self._stopped_event.is_set(): return False try: - self._input_queue.put_nowait(input) + self._input_queue.put_nowait(_DeriveRequest(input, context, revision)) return True except Full: with self._status_lock: @@ -243,7 +286,19 @@ class DeriveUnit: worker, ) - def _validate_result(self, value: ValueDescriptor) -> None: + def _process( + self, + inputs: list[ValueDescriptor], + context: DeriveContext | None, + ) -> ValueDescriptor | None: + args = [*inputs, context] if self._uses_context else inputs + value = self._process_function(*args, self._return_signal) + self._validate_result(value) + return value + + def _validate_result(self, value: ValueDescriptor | None) -> None: + if value is None and self._allows_none: + return if ( not isinstance(value, ValueDescriptor) or value.signal_id != self._return_signal.id @@ -294,10 +349,26 @@ class DeriveUnit: buffer.sort() input_ids = {signal.id for signal in self._input_signals} output_buffer = ValueBuffer() + history = _SampleHistory() for timestamp, signal_id, value in buffer.get_samples(): if signal_id not in input_ids: continue + if self._uses_context: + if not history.update(ValueDescriptor(signal_id, value, timestamp)): + continue + args = history.inputs( + [signal.id for signal in self._input_signals], timestamp + ) + if args is None: + continue + result = self._process(args, history.snapshot(timestamp)) + if result is not None: + output_buffer.append( + result.timestamp, self._return_signal.id, result.value + ) + continue + input_args: list[ValueDescriptor] = [] for signal in self._input_signals: if signal.id == signal_id: @@ -308,9 +379,11 @@ class DeriveUnit: ) input_args.append(ValueDescriptor(signal.id, input_value, timestamp)) - result = self._process_function(*input_args, self._return_signal) - self._validate_result(result) - output_buffer.append(result.timestamp, self._return_signal.id, result.value) + result = self._process(input_args, None) + if result is not None: + output_buffer.append( + result.timestamp, self._return_signal.id, result.value + ) return output_buffer @@ -332,7 +405,9 @@ class DeriveRegistry: self._get_live_value = get_live_value self._get_all_core_signal_descriptors = get_all_core_signal_descriptors self._is_current_playback_value = is_current_playback_value - self._routing_queue: Queue[ValueDescriptor] = Queue(queue_capacity) + self._routing_queue: Queue[tuple[object, ValueDescriptor]] = Queue( + queue_capacity + ) self._queue_capacity = queue_capacity self._get_source_revision = get_source_revision self._units_revision = 0 @@ -420,6 +495,7 @@ class DeriveRegistry: self._core_input_queue, self._is_current_playback_value, self._queue_capacity, + self._current_revision, ) self._units[unit_id] = unit self._units_revision += 1 @@ -520,7 +596,7 @@ class DeriveRegistry: if self._routing_stop_event.is_set() or self._routing_stopped_event.is_set(): return False try: - self._routing_queue.put_nowait(message) + self._routing_queue.put_nowait((self._current_revision(), message)) return True except Full: with self._lock: @@ -529,6 +605,15 @@ class DeriveRegistry: log.warning("Derive routing queue is full; dropping new requests") return False + def _current_revision(self) -> object: + return self._get_source_revision() if self._get_source_revision else None + + def is_current_result( + self, + message: DerivedValueDescriptor | DerivedPlaybackValueDescriptor, + ) -> bool: + return message.revision == self._current_revision() + def stop(self) -> bool: log.debug( "Stopping derive registry", @@ -661,11 +746,12 @@ class DeriveRegistry: try: while not self._routing_stop_event.is_set(): try: - message = self._routing_queue.get(timeout=0.1) + revision, message = self._routing_queue.get(timeout=0.1) except Empty: continue - self._route_data(message) + if revision == self._current_revision(): + self._route_data(message, revision) except Exception as error: self._error = str(error) log.exception( @@ -686,7 +772,7 @@ class DeriveRegistry: }, ) - def _route_data(self, message: ValueDescriptor) -> None: + def _route_data(self, message: ValueDescriptor, revision: object = None) -> None: if ( isinstance(message, PlaybackValueDescriptor) and self._is_current_playback_value is not None @@ -705,6 +791,32 @@ class DeriveRegistry: continue args: list[ValueDescriptor] = [] + context = None + if unit._uses_context: + if unit._history_revision != revision: + unit._history = _SampleHistory() + unit._history_revision = revision + if not unit._history.update(message): + continue + inputs = unit._history.inputs( + [signal.id for signal in input_signals], message.timestamp + ) + if inputs is None: + continue + context = unit._history.snapshot(message.timestamp) + if isinstance(message, PlaybackValueDescriptor): + inputs = [ + PlaybackValueDescriptor( + value.signal_id, + value.value, + value.timestamp, + message.generation, + ) + for value in inputs + ] + unit.put_data(inputs, context=context, revision=revision) + continue + for signal in input_signals: if signal.id == message.signal_id: args.append(message) @@ -726,7 +838,7 @@ class DeriveRegistry: ) for value in args ] - unit.put_data(args) + unit.put_data(args, revision=revision) def process_offline( self, buffer: DLPak, progress: ProgressCallback | None = None @@ -740,6 +852,7 @@ class DeriveRegistry: for timestamp, signal_id, value in buffer._data._samples: working_buffer.append(timestamp, signal_id, value) processed_buffer = ValueBuffer() + empty_outputs: set[UUID] = set() for layer_index, layer in enumerate(layers): layer_units = [self._units[uuid] for uuid in layer] @@ -750,6 +863,23 @@ class DeriveRegistry: total_units, f"Processing derive unit {unit.uuid()}", ) + if any( + signal.id in empty_outputs for signal in unit.get_input_signals() + ): + log.info( + "Skipping derive unit %s: upstream unit produced no values", + unit.uuid(), + extra={"event": "derive.offline_upstream_empty"}, + ) + empty_outputs.add(unit.return_signal().id) + completed_units += 1 + if progress: + progress( + completed_units, + total_units, + "Upstream signal has no values", + ) + continue returned_buffer = unit.process_offline(working_buffer) completed_units += 1 if progress: @@ -764,9 +894,11 @@ class DeriveRegistry: "layer_index": layer_index, }, ) - raise DeriveError( - "Offline processing returned no values", code="no_values" - ) + if not unit._allows_none: + raise DeriveError( + "Offline processing returned no values", code="no_values" + ) + empty_outputs.add(unit.return_signal().id) log.debug( "Offline derive unit %s produced %d sample(s)", unit.uuid(), diff --git a/src/dynalab_core/playback.py b/src/dynalab_core/playback.py index 0a3b7d0..bf186af 100644 --- a/src/dynalab_core/playback.py +++ b/src/dynalab_core/playback.py @@ -113,6 +113,11 @@ class PlaybackEngine: not self._stop_event.is_set() and message.generation == self._generation ) + def get_generation(self) -> int: + """Revision used to invalidate processing history across playback controls.""" + with self._condition: + return self._generation + def get_status(self) -> WorkerStatus: return WorkerStatus( self._worker_thread.name, diff --git a/src/dynalab_core/processing.py b/src/dynalab_core/processing.py new file mode 100644 index 0000000..856d47a --- /dev/null +++ b/src/dynalab_core/processing.py @@ -0,0 +1,117 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + +"""Causal sample history and numerical derivatives for DLUnit functions.""" + +from collections import deque +from math import factorial, fsum, isfinite, prod +from types import MappingProxyType +from uuid import UUID + +from dynalab_core.protocols.packets.data import ValueDescriptor + + +class DeriveContext: + """Read-only history snapshot supplied by Core to one processing call.""" + + def __init__(self, samples: dict[UUID, tuple[tuple[int, float], ...]]) -> None: + self._samples = MappingProxyType(dict(samples)) + + def derivative( + self, + value: ValueDescriptor, + order: int = 1, + *, + max_gap_s: float | None = None, + ) -> float | None: + """Estimate d**order(value)/dt**order in seconds, or return None. + + Uses the last order + 1 distinct samples, including ``value``. Repeated + queries do not advance history. No smoothing or future samples are used. + ``max_gap_s`` optionally restarts warm-up across excessive sample gaps. + """ + if ( + not isinstance(order, int) + or isinstance(order, bool) + or order not in (1, 2, 3) + ): + raise ValueError("Derivative order must be 1, 2, or 3") + if max_gap_s is not None and (not isfinite(max_gap_s) or max_gap_s <= 0): + raise ValueError("max_gap_s must be finite and positive") + samples = self._samples.get(value.signal_id, ()) + if len(samples) < order + 1 or samples[-1] != (value.timestamp, value.value): + return None + samples = samples[-(order + 1) :] + if max_gap_s is not None and any( + (right[0] - left[0]) / 1_000_000_000 > max_gap_s + for left, right in zip(samples, samples[1:]) + ): + return None + + # The nth derivative of the degree-n interpolant is constant. Normalize + # time to the window span and subtract the latest value to limit loss of + # precision from large absolute timestamps and constant value offsets. + span_ns = samples[-1][0] - samples[0][0] + span_s = span_ns / 1_000_000_000 + try: + result = ( + factorial(order) + * fsum( + (sample - samples[-1][1]) + / prod( + (timestamp - other_timestamp) / span_ns + for j, (other_timestamp, _) in enumerate(samples) + if i != j + ) + for i, (timestamp, sample) in enumerate(samples) + ) + / span_s**order + ) + except (ArithmeticError, ValueError): + # Extreme value ranges can overflow even when each input is finite. + return None + return result if isfinite(result) else None + + +class _SampleHistory: + """Mutable history owned by a unit's router, or by one offline pass.""" + + def __init__(self) -> None: + self._samples: dict[UUID, deque[tuple[int, float]]] = {} + + def update(self, value: ValueDescriptor) -> bool: + samples = self._samples.setdefault(value.signal_id, deque(maxlen=4)) + if samples and value.timestamp <= samples[-1][0]: + return False + if not isfinite(value.value): + samples.clear() + return False + samples.append((value.timestamp, value.value)) + return True + + def snapshot(self, timestamp: int) -> DeriveContext: + return DeriveContext( + { + key: tuple(sample for sample in samples if sample[0] <= timestamp) + for key, samples in self._samples.items() + } + ) + + def inputs( + self, signal_ids: list[UUID], timestamp: int + ) -> list[ValueDescriptor] | None: + values = [] + for key in signal_ids: + sample = next( + ( + sample + for sample in reversed(self._samples.get(key, ())) + if sample[0] <= timestamp + ), + None, + ) + if sample is None: + return None + values.append(ValueDescriptor(key, sample[1], sample[0])) + return values diff --git a/uv.lock b/uv.lock index 5df7bb1..feac460 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.13" [[package]] @@ -22,7 +22,7 @@ wheels = [ [[package]] name = "dynalab-core" -version = "0.1.0" +version = "0.2.1" source = { editable = "." } dependencies = [ { name = "psutil" },