Custom Transport (Without SDK Scanner)

Use your own BLE discovery and connection instead of the SDK scanner: scan in your app, create a Device with your transport, then connect it.

Transport requirements

The SDK splits commands into frames and chooses the write mode. Your transport must:

• Write frames in the order it receives them, with the requested write mode (withResponse, response in Python).
• Not retry, split or merge writes. If a write fails, stop writing and disconnect.
• Report the write size: in Python, mtu_size returns the negotiated ATT MTU (for example 247); on Android, Apple platforms and Flutter, mtuSize returns the maximum characteristic value length (for example 244).
• Report every lost link through the disconnection callback, or the disconnection stream in Python, including disconnects your app did not request.

In Python, notification and disconnection streams may be async generators or coroutines that return an async iterator.

Examples

The Flutter example's platform-transport option imports the underlying BLE client directly. Add it as an explicit dependency with flutter pub add flutter_reactive_ble.

Python

import asyncio
from typing import AsyncIterator

from aidlab import AidlabTransport, DataType, Device, DeviceDelegate


class MyTransport(AidlabTransport):
    @property
    def mtu_size(self) -> int:
        return 247

    async def connect(self) -> None:
        ...

    async def disconnect(self) -> None:
        ...

    async def read_char(self, uuid: str) -> bytes:
        ...

    async def write_char(self, uuid: str, data: bytes, response: bool = True) -> None:
        ...

    async def notifications(self, uuid: str) -> AsyncIterator[bytes]:
        ...

    async def disconnections(self) -> AsyncIterator[None]:
        ...


class Delegate(DeviceDelegate):
    async def did_connect(self, device):
        await device.collect([DataType.RESPIRATION], [])


async def run():
    # discovered by your own scanner
    address = "AA:BB:CC:DD:EE:FF"
    name = "Aidlab 2"

    device = Device(address=address, name=name, transport=MyTransport())
    await device.connect(Delegate())

    while True:
        await asyncio.sleep(1)


asyncio.run(run())

Android

import com.aidlab.sdk.*
import java.util.UUID

class MyTransport(
    private val addressValue: String,
    private val nameValue: String?,
) : AidlabTransport {
    override val address: String = addressValue
    override val name: String? = nameValue
    // Negotiated ATT MTU 247 minus the 3-byte ATT header.
    override val mtuSize: Int = 244

    override var onDisconnect: ((DisconnectReason, Throwable?) -> Unit)? = null

    override fun connect(
        onSuccess: () -> Unit,
        onError: (Throwable) -> Unit,
    ) {
        // your BLE connect
        // call onSuccess() when connected or onError(e) on failure
    }

    override fun disconnect() {
        // Disconnect and report the terminal event through onDisconnect.
    }

    override fun readCharacteristic(
        uuid: UUID,
        onSuccess: (ByteArray) -> Unit,
        onError: (Throwable) -> Unit,
    ) {
        // your BLE read
    }

    override fun writeCharacteristic(
        uuid: UUID,
        data: ByteArray,
        withResponse: Boolean,
        onSuccess: () -> Unit,
        onError: (Throwable) -> Unit,
    ) {
        // your BLE write
    }

    override fun startNotifications(
        uuid: UUID,
        onData: (ByteArray) -> Unit,
        onError: (Throwable) -> Unit,
    ) {
        // your BLE notifications
    }

    override fun stopNotifications(uuid: UUID) {
        // stop notifications
    }
}

// discovered by your own scanner
val transport = MyTransport("AA:BB:CC:DD:EE:FF", "Aidlab 2")
val device = Device(transport)
lifecycleScope.launch { device.connect(deviceDelegate) } // throws when the connection fails

Apple Platforms

import Aidlab
import CoreBluetooth
import Foundation

final class MyTransport: AidlabTransport {
    let address: UUID
    let name: String?
    var rssi: NSNumber
    // Negotiated ATT MTU 247 minus the 3-byte ATT header.
    let mtuSize: Int = 244

    var onDisconnect: ((DisconnectReason, Error?) -> Void)?

    init(address: UUID, name: String?, rssi: NSNumber) {
        self.address = address
        self.name = name
        self.rssi = rssi
    }

    func connect(completion: @escaping (Result<Void, Error>) -> Void) {
        // Connect using your BLE stack, then call completion(.success(()))
        // or completion(.failure(error)).
    }

    func disconnect() {
        // Disconnect and report the terminal event through onDisconnect.
    }

    func readCharacteristic(
        _ uuid: CBUUID,
        completion: @escaping (Result<Data, Error>) -> Void
    ) {
        // Read with your BLE stack and complete with Data or Error.
    }

    func writeCharacteristic(
        _ uuid: CBUUID,
        data: Data,
        withResponse: Bool,
        completion: @escaping (Result<Void, Error>) -> Void
    ) {
        // Preserve frame order and the requested write mode, then complete.
    }

    func startNotifications(
        _ uuid: CBUUID,
        onData: @escaping (Data) -> Void,
        onError: @escaping (Error) -> Void
    ) {
        // Forward characteristic values to onData and failures to onError.
    }

    func stopNotifications(_ uuid: CBUUID) {
        // Stop notifications for this characteristic.
    }
}

func connectDiscoveredDevice(
    address: UUID,
    name: String?,
    rssi: NSNumber,
    delegate: DeviceDelegate
) {
    let transport = MyTransport(address: address, name: name, rssi: rssi)
    let device = Device(transport: transport)
    Task { try await device.connect(delegate: delegate) } // throws when the connection fails
}

Flutter

import 'package:aidlab_sdk/aidlab_sdk.dart';
import 'package:flutter_reactive_ble/flutter_reactive_ble.dart' as frb;
import 'dart:typed_data';

final class MyTransport implements AidlabTransport {
  MyTransport({
    required this.address,
    required this.name,
  });

  @override
  final String address;

  @override
  final String? name;

  @override
  AidlabTransportDisconnectCallback? onDisconnect;

  @override
  Future<void> connect() async {
    // your BLE connect
  }

  @override
  Future<void> disconnect() async {
    // your BLE disconnect
  }

  @override
  // Negotiated ATT MTU 247 minus the 3-byte ATT header.
  Future<int> mtuSize() async => 244;

  @override
  Future<Uint8List> readCharacteristic(String uuid) async {
    // your BLE read
    return Uint8List(0);
  }

  @override
  Future<void> writeCharacteristic(
    String uuid,
    Uint8List data, {
    bool withResponse = true,
  }) async {
    // your BLE write
  }

  @override
  void startNotifications(
    String uuid, {
    required void Function(Uint8List data) onData,
    required void Function(Object error, StackTrace stackTrace) onError,
  }) {
    // your BLE notify setup
  }

  @override
  Future<void> stopNotifications(String uuid) async {
    // stop notifications for this characteristic
  }
}

Future<Device> buildDevice(String address, String? name) async {
  // Option A: fully custom transport
  final custom = Device.withTransport(
    transport: MyTransport(address: address, name: name),
  );

  // Option B: use the SDK's default BLE transport (flutter_reactive_ble),
  // but still keep your own scanner/discovery UI.
  final transport = ReactiveBleAidlabTransport(
    frb.FlutterReactiveBle(),
    address: address,
    name: name,
  );
  final platform = Device.withTransport(
    transport: transport,
  );

  return platform; // or return custom;
}

Platform Notes

• Unity: custom transport injection is not exposed. The SDK owns the platform transport and supports selecting one device by name.