abr_sdk.core

Wrapper classes around the object hierarchy exposed by the ABR C ABI.

Author: Andreas Stöckel (Applied Brain Research)

Module Contents

Classes

BufferDirection	Enum describing how a buffer is used.
EventFlags	Flags that influence how Event instances behave.
Handle	Base class for all objects exposed by the ABR SDK C ABI.
Library	Loaded shared library with access to metadata.
Buffer	FIFO buffer for I/O.
EventSet	A set of events that can be polled together.
Event	An event that can be polled for a condition.
Application	A time-series application instance.

Functions

_emit_log	Log a raw C-side msg at level, clamped to at least DEBUG.
_variant_to_python
get_fingerprint	Return this device’s fingerprint, as reported by the library.
install_license	Verify and install a signed license envelope for this device.

Data

default_logger
DEFAULT_LICENSE_DIR
VariantDict

API

abr_sdk.core.default_logger

‘getLogger(…)’

abr_sdk.core._emit_log(logger: logging.Logger, level: int, msg: bytes | None) → None

Log a raw C-side msg at level, clamped to at least DEBUG.

abr_sdk.core.DEFAULT_LICENSE_DIR

None

abr_sdk.core._variant_to_python(v: abr_sdk.cabi.Variant) → bool | int | str | None

class abr_sdk.core.BufferDirection(*args, **kwds)

Bases: enum.Enum

Enum describing how a buffer is used.

Initialization

Input

‘auto(…)’

Buffer receives data from the Python application.

Output

‘auto(…)’

Buffer provides data to the Python application.

class abr_sdk.core.EventFlags

Flags that influence how Event instances behave.

auto_disable: bool

False

If true, then automatically disables events that are triggered during a call to “poll” or while waiting for the platform event handle.

This is useful if the calling code needs to wait for multiple events to be triggered before some action can be performed. Without auto-disabling, events must be manually disabled to avoid “poll” constantly returning, which complicates the code.

disabled: bool

False

If true, then the event is created in the “disabled” state.

That means that the event is not included in the set of events that can cause poll to return.

property bitset: int

Return a bitset that encodes the flags stored in this class.

The result value may be directly fed into C ABI functions that take an “event flags” parameter.

abr_sdk.core.VariantDict

None

class abr_sdk.core.Handle(cabi: abr_sdk.core.Handle.__init__.cabi, handle: ctypes.c_void_p | None, *, handle_may_be_none: bool = False, class_: str | None = None)

Base class for all objects exposed by the ABR SDK C ABI.

Provides access to the properties attached to objects, the parent object, and helper functions for lifetime management.

Initialization

Create a new object instance wrapping the given handle and ABI instance.

Parameters

cabi CABI instance providing access to the low-level C functions. A C ABI object may be obtained by loading an ABR SDK shared library. handle Pointer at the ABR SDK object that should be wrapped by the new Handle instance. Must be non-None unless handle_may_be_none is True. handle_may_be_none If True, handle may be None. Used for access to the library metadata. class_ String containing the expected object class. If not set to None, then the given string is compared to the “class” property of the handle.

_ensure_alive() → None

property properties: abr_sdk.core.VariantDict

All properties as a {name: value} dict.

get_parent() → Optional[abr_sdk.core.Handle]

Return the parent object, or None if this is a root object.

get_property(name: str | bytes) → Any

Return a property value by name. Raises KeyError if not found.

class abr_sdk.core.Library(lib_path: str | pathlib.Path, *, lib_search_paths: Iterable[str | pathlib.Path] | None = None, use_default_lib_search_paths: bool = True, app_type: str | None = None)

Bases: abr_sdk.core.Handle

Loaded shared library with access to metadata.

Initialization

Attempt to load the library at lib_path.

Parameters

lib_path Absolute or relative path pointing at the library to load. If a relative path is given, then this function searches in the provided search paths. lib_search_paths Search paths to use for the library. Only used if lib_path is not absolute. use_default_lib_search_paths If set to True, use a set of default search paths from the environment. Set this to False if the environment is not trusted or pass an absolute path to lib_path. app_type Expected application type, or None to accept any application type.

_app_type_mismatch(app_type: str | None) → str | None

Describe how this library’s application type differs from app_type.

Returns None if app_type is None, matches, or the library exports no app_type property.

static get_default_lib_search_paths() → list[pathlib.Path]

Return default library search paths from the environment.

static find(lib_path: str | pathlib.Path, *, lib_search_paths: Iterable[str | pathlib.Path] | None = None, use_default_lib_search_paths: bool = True) → Iterable[pathlib.Path]

Resolve lib_path to a list of candidates.

See init for a detailed description of the parameters.

property path: pathlib.Path

Location of the loaded library.

property abi_version: tuple[int, int]

ABI version as a (major, minor) tuple.

class abr_sdk.core.Buffer(cabi: abr_sdk.core.Buffer.__init__.cabi, handle: ctypes.c_void_p)

Bases: abr_sdk.core.Handle

FIFO buffer for I/O.

Write data into input buffers via push, read data from output buffers via pull.

Initialization

Create a new object instance wrapping the given handle and ABI instance.

Parameters

cabi CABI instance providing access to the low-level C functions. A C ABI object may be obtained by loading an ABR SDK shared library. handle Pointer at the ABR SDK object that should be wrapped by the new Handle instance. Must be non-None unless handle_may_be_none is True. handle_may_be_none If True, handle may be None. Used for access to the library metadata. class_ String containing the expected object class. If not set to None, then the given string is compared to the “class” property of the handle.

property dir: abr_sdk.core.BufferDirection

Buffer input/output direction.

property size: int

Total buffer capacity in bytes.

property level: int

Bytes that can be pushed, or pulled up to the next flush marker.

push(data: bytes | bytearray | memoryview, flush: bool = False) → int

Push data into the buffer.

Parameters

data Data to push onto the buffer. flush Instructs the application to flush the processing pipeline once all bytes in data have been processed. Note that the “flush” flag only takes effect if there is enough space for all data in “data” in the buffer.

Returns

int The number of bytes actually written. This may be less than len(data) if the buffer is full.

pull(max_n_bytes: int) → tuple[bytes, bool]

Pull data from the buffer.

Parameters

max_n_bytes Maximum number of bytes to read from the buffer at once. Fewer bytes than this may be returned if the buffer is empty or a “flush” boundary is reached.

Returns

data Data pulled from the buffer.

flush Return true if the retrieved data is at a flush boundary.

class abr_sdk.core.EventSet(cabi: abr_sdk.core.EventSet.__init__.cabi, handle: ctypes.c_void_p, parent: abr_sdk.core.Application)

Bases: abr_sdk.core.Handle

A set of events that can be polled together.

Create events with :meth:create_buffer_level_event, then call

Meth:

poll to wait.

Initialization

Create a new object instance wrapping the given handle and ABI instance.

Parameters

cabi CABI instance providing access to the low-level C functions. A C ABI object may be obtained by loading an ABR SDK shared library. handle Pointer at the ABR SDK object that should be wrapped by the new Handle instance. Must be non-None unless handle_may_be_none is True. handle_may_be_none If True, handle may be None. Used for access to the library metadata. class_ String containing the expected object class. If not set to None, then the given string is compared to the “class” property of the handle.

property platform_handle: int

Platform-specific handle (e.g. a file descriptor on POSIX).

poll(timeout_ms: int) → bool

Block until an event fires or timeout_ms elapses.

A negative timeout waits forever; zero returns immediately. Returns True if at least one event triggered. Raises InterruptedError on signal interruption.

interrupt() → None

Interrupts a blocking call to poll.

create_buffer_level_event(buf: abr_sdk.core.Buffer, threshold: int, flags: abr_sdk.core.EventFlags | None = None) → abr_sdk.core.Event

Register a buffer level event for the given buffer.

The I/O event is triggered if there are threshold bytes available for reading from or writing to the buffer, or if an output buffer holds a flush marker.

close() → None

Release this event set and all its events.

__enter__() → abr_sdk.core.EventSet

__exit__(*exc: Any) → None

class abr_sdk.core.Event(cabi: abr_sdk.core.Event.__init__.cabi, handle: Any, parent: abr_sdk.core.EventSet)

Bases: abr_sdk.core.Handle

An event that can be polled for a condition.

Created by :class:EventSet factory methods.

Initialization

Create a new object instance wrapping the given handle and ABI instance.

Parameters

cabi CABI instance providing access to the low-level C functions. A C ABI object may be obtained by loading an ABR SDK shared library. handle Pointer at the ABR SDK object that should be wrapped by the new Handle instance. Must be non-None unless handle_may_be_none is True. handle_may_be_none If True, handle may be None. Used for access to the library metadata. class_ String containing the expected object class. If not set to None, then the given string is compared to the “class” property of the handle.

property is_triggered: bool

True if this event’s condition currently holds.

enable() → None

Enable this event so it can cause :meth:EventSet.poll to return.

disable() → None

Ignore this event in :meth:EventSet.poll.

close() → None

Release this event.

__enter__() → abr_sdk.core.Event

__exit__(*exc: Any) → None

class abr_sdk.core.Application(lib_or_path: str | pathlib.Path | abr_sdk.core.Library, *, lib_search_paths: list[str | pathlib.Path] | None = None, use_default_lib_search_paths: bool = True, app_type: str | None = None, resources_dir: str | pathlib.Path | None = None, litert_delegate: str | None = None, cache_dir: str | pathlib.Path | None = None, config: abr_sdk.core.VariantDict | None = None, logger: logging.Logger | None = None)

Bases: abr_sdk.core.Handle

A time-series application instance.

Can be constructed directly with keyword arguments (auto-discovers and loads a suitable library).

Initialization

Create a new object instance wrapping the given handle and ABI instance.

Parameters

cabi CABI instance providing access to the low-level C functions. A C ABI object may be obtained by loading an ABR SDK shared library. handle Pointer at the ABR SDK object that should be wrapped by the new Handle instance. Must be non-None unless handle_may_be_none is True. handle_may_be_none If True, handle may be None. Used for access to the library metadata. class_ String containing the expected object class. If not set to None, then the given string is compared to the “class” property of the handle.

static make_log_callback(logger: logging.Logger | None) → Any

Return a log callback that may be called from C.

static make_config(config: abr_sdk.core.VariantDict) → tuple[Any, int]

Convert a Python dictionary into a C-compatible config object.

create_event_set() → abr_sdk.core.EventSet

Create a new :class:EventSet for polling buffer/network events.

close() → None

Release all resources held by this instance.

__enter__() → abr_sdk.core.Application

__exit__(*exc: Any) → None

abr_sdk.core.get_fingerprint(library: abr_sdk.core.Library, *, license_dir: str | pathlib.Path, logger: logging.Logger | None = None) → str

Return this device’s fingerprint, as reported by the library.

Parameters

library A loaded SDK :class:Library. license_dir Directory the SDK may write a random identity into. logger Optional logger to receive the SDK’s native log messages.

Returns

str The fingerprint, as 64 lowercase hexadecimal characters.

Notes

Needs no application instance. Any failure raises :class:AbrSdkError.

abr_sdk.core.install_license(library: abr_sdk.core.Library, envelope: str, *, license_dir: str | pathlib.Path, logger: logging.Logger | None = None) → None

Verify and install a signed license envelope for this device.

Parameters

library A loaded SDK :class:Library. Activation goes through its abr_activate_license, which verifies the envelope’s Ed25519 signature against the library’s built-in key and writes the device-bound license + integrity files under license_dir. envelope The signed machine-file envelope (the {enc, sig, alg} JSON string). license_dir Directory the SDK writes the license and integrity files into. logger Optional logger to receive the SDK’s native log messages.

Notes

Activation needs no application instance, and hence no model or resources: license_dir is the only configuration the SDK reads. Any activation failure raises :class:AbrSdkError.

The backend’s log output is buffered and replayed (to logger, or the default logger) only if activation fails. On success it is discarded.