Completed initial API documentation and updated API to support derive unit

The initial API documentation is now complete and covers the whole Core
api ready for use

The underlying logic has been updated to correctly return derive unit
signals
This commit is contained in:
h3cx committed 2026-09-11 17:34:44 +02:00
1 parent c67316dd90
commit c0a3117be4
4 files changed
+275 -8

No files matched your search

+19 -5
View File
@@ -136,14 +136,24 @@ class Core:
self._processing_buffer.set_manifest(
self._recording_timestamp, signal_descriptors
)
# TODO: remove debug behavior default saving to output
self._processing_buffer.write("./", "output")
def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None:
return self._connector_registry.get_signal_descriptor(signal_id)
connector_signal = self._connector_registry.get_signal_descriptor(signal_id)
if connector_signal is None:
for unit in self._derive_units.values():
if unit._return_signal.id == signal_id:
return unit._return_signal
else:
return None
return connector_signal
def get_all_signal_descriptors(self) -> list[SignalDescriptor]:
return self._connector_registry.get_all_signal_descriptors()
connector_signals = self._connector_registry.get_all_signal_descriptors()
connector_signals.append(
unit._return_signal for unit in self._derive_units.values()
)
return connector_signals
def get_live_value(self, signal_id: UUID) -> float | None:
with self._live_values_lock:
@@ -186,7 +196,11 @@ class Core:
return unit_id
def unbind_derive_unit(self, unit_id: UUID) -> None:
self._derive_units[unit_id].stop()
try:
self._derive_units[unit_id].stop()
self._derive_units.pop(unit_id, None)
except KeyError:
return
def _input_worker(self) -> None:
while not self._stop_event.is_set():