src.bluetooth_sig.stream.pairing

Stream helpers for pairing dependent characteristic notifications.

This module provides a generic, backend-agnostic buffer that correlates dependent characteristic notifications based on caller-defined grouping keys. Useful for Bluetooth SIG profiles where characteristics must be paired by sequence numbers, timestamps, or other identifiers.

Classes

Name

Description

DependencyPairingBuffer

Buffer and pair dependent characteristic notifications.

Module Contents

class src.bluetooth_sig.stream.pairing.DependencyPairingBuffer(*, translator: src.bluetooth_sig.core.translator.BluetoothSIGTranslator, required_uuids: set[str], group_key: Callable[[str, Any], collections.abc.Hashable], on_pair: Callable[[dict[str, Any]], None])

Buffer and pair dependent characteristic notifications.

Buffers incoming notifications until all required UUIDs for a grouping key are present, then batch-parses and invokes the callback. Order-independent.

Parameters:
  • translator – BluetoothSIGTranslator instance for parsing characteristics.

  • required_uuids – Set of UUID strings that must be present to form a complete pair.

  • group_key – Function that extracts a grouping key from each parsed notification. Called as group_key(uuid, parsed_result) and must return a hashable value.

  • on_pair – Callback invoked with complete parsed pairs as on_pair(results: dict[str, Any]).

Note

Does not manage BLE subscriptions. Callers handle connection and notification setup.

ingest(uuid: str, data: bytes) None

Ingest a single characteristic notification.

Parameters:
  • uuid – Characteristic UUID string (16-bit or 128-bit).

  • data – Raw bytes from the characteristic notification.