Added playback mode ready for offline processing
CoreConfig how has a playback mode option which can be set to reprocess or read. Setting it to reprocess means that all derived signals are reprocessed during playback, in this case the PlaybackEngine only plays back the signals whose source is set to "origin" Setting it to read means that all derived signals are playedback from the recorded buffer, this whill be the preferred mode once offline processing is set up which will allow for more complex data processing offline once a recording is completed, which would require the read playback mode as it wouldn't support realtime processing as reprocess does
This commit is contained in:
11 files changed
+52
-689
No files matched your search
@@ -61,7 +61,9 @@ class Core:
|
||||
)
|
||||
self._input_worker_thread.start()
|
||||
|
||||
self._playback_engine = PlaybackEngine(self._data_input_queue)
|
||||
self._playback_engine = PlaybackEngine(
|
||||
self._core_config, self._data_input_queue
|
||||
)
|
||||
|
||||
self._derive_registry = DeriveRegistry(
|
||||
self._data_input_queue, self._get_live_value_descriptor
|
||||
@@ -332,21 +334,30 @@ class Core:
|
||||
signal: SignalDescriptor | None = None
|
||||
if self._mode.value == "realtime":
|
||||
signal = self._connector_registry.get_signal_descriptor(signal_id)
|
||||
if signal is None:
|
||||
signal = self._derive_registry.get_signal_descriptor(signal_id)
|
||||
|
||||
elif self._mode.value == "playback":
|
||||
signal = self._playback_engine.get_signal_descriptor(signal_id)
|
||||
|
||||
if signal is None:
|
||||
signal = self._derive_registry.get_signal_descriptor(signal_id)
|
||||
if signal is None and self._core_config.playback_mode == "reprocess":
|
||||
signal = self._derive_registry.get_signal_descriptor(signal_id)
|
||||
|
||||
return signal
|
||||
|
||||
def get_all_signal_descriptors(self) -> list[SignalDescriptor]:
|
||||
root_signals: list[SignalDescriptor] = []
|
||||
derive_signals: list[SignalDescriptor] = []
|
||||
if self._mode.value == "realtime":
|
||||
root_signals = self._connector_registry.get_all_signal_descriptors()
|
||||
elif self._mode.value == "playback":
|
||||
root_signals = self._playback_engine.get_all_signal_descriptors()
|
||||
derive_signals = self._derive_registry.get_all_signal_descriptors()
|
||||
|
||||
if (
|
||||
self._core_config.playback_mode == "reprocess"
|
||||
or self._mode.value == "realtime"
|
||||
):
|
||||
derive_signals = self._derive_registry.get_all_signal_descriptors()
|
||||
return root_signals + derive_signals
|
||||
|
||||
def get_live_value(self, signal_id: UUID) -> float | None:
|
||||
@@ -467,7 +478,11 @@ class Core:
|
||||
message.timestamp, message.signal_id, message.value
|
||||
)
|
||||
|
||||
self._derive_registry.put_data(message)
|
||||
if (
|
||||
self._mode.value == "realtime"
|
||||
or self._core_config.playback_mode == "reprocess"
|
||||
):
|
||||
self._derive_registry.put_data(message)
|
||||
except Exception as error:
|
||||
log.exception(
|
||||
"Core input worker failed",
|
||||
|
||||
@@ -2,13 +2,17 @@
|
||||
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
from typing import Literal
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
PlaybackMode = Literal["reprocess", "read"]
|
||||
|
||||
|
||||
class CoreConfig(BaseModel):
|
||||
host: str = "127.0.0.1"
|
||||
port: int = Field(default=58763, ge=0, le=65535)
|
||||
thread_debug: bool = False
|
||||
playback_mode: PlaybackMode = "reprocess"
|
||||
|
||||
def bind_str(self) -> str:
|
||||
return f"{self.host}:{self.port}"
|
||||
@@ -9,6 +9,7 @@ import threading
|
||||
import time
|
||||
from uuid import UUID
|
||||
|
||||
from dynalab_core.config import CoreConfig
|
||||
from dynalab_core.dlpak import DLPak
|
||||
from dynalab_core.errors import PlaybackError
|
||||
from dynalab_core.protocols.packets import ProtocolMessage
|
||||
@@ -19,7 +20,10 @@ log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PlaybackEngine:
|
||||
def __init__(self, data_queue: Queue[ProtocolMessage]) -> None:
|
||||
def __init__(
|
||||
self, core_config: CoreConfig, data_queue: Queue[ProtocolMessage]
|
||||
) -> None:
|
||||
self._core_config = core_config
|
||||
self._data: DLPak | None = None
|
||||
self._data_queue: Queue[ProtocolMessage] = data_queue
|
||||
self._playing: threading.Event = threading.Event()
|
||||
@@ -65,16 +69,24 @@ class PlaybackEngine:
|
||||
if self._data is not None and self._data._manifest is not None:
|
||||
for signal in self._data._manifest.signals:
|
||||
if signal.id == uuid:
|
||||
if (
|
||||
signal.origin == "derived"
|
||||
and self._core_config.playback_mode == "reprocess"
|
||||
):
|
||||
return None
|
||||
return signal
|
||||
return None
|
||||
|
||||
def get_all_signal_descriptors(self) -> list[SignalDescriptor]:
|
||||
if self._data is not None and self._data._manifest is not None:
|
||||
return [
|
||||
signal
|
||||
for signal in self._data._manifest.signals
|
||||
if signal.origin != "derived"
|
||||
]
|
||||
if self._core_config.playback_mode == "reprocess":
|
||||
return [
|
||||
signal
|
||||
for signal in self._data._manifest.signals
|
||||
if signal.origin != "derived"
|
||||
]
|
||||
elif self._core_config.playback_mode == "read":
|
||||
return self._data._manifest.signals
|
||||
|
||||
return list[SignalDescriptor]()
|
||||
|
||||
@@ -90,11 +102,12 @@ class PlaybackEngine:
|
||||
code="incomplete_data",
|
||||
)
|
||||
self._data = data
|
||||
self._ignored_signals = {
|
||||
signal.id
|
||||
for signal in self._data._manifest.signals
|
||||
if signal.origin == "derived"
|
||||
}
|
||||
if self._core_config.playback_mode == "reprocess":
|
||||
self._ignored_signals = {
|
||||
signal.id
|
||||
for signal in self._data._manifest.signals
|
||||
if signal.origin == "derived"
|
||||
}
|
||||
|
||||
def _update_internal_timestamp(self) -> None:
|
||||
self._internal_timestamp = time.monotonic_ns() - self._start_timestamp
|
||||
|
||||
Reference in new issue
Block a user