ChipCoreBluetoothMgr.py

#
#    Copyright (c) 2020 Project CHIP Authors
#    Copyright (c) 2019-2020 Google LLC.
#    Copyright (c) 2015-2018 Nest Labs, Inc.
#    All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

#
#    @file
#      BLE Central support for Chip Device Controller via OSX CoreBluetooth APIs.
#

from __future__ import absolute_import, print_function

import logging
import queue
import time
from ctypes import CFUNCTYPE, PYFUNCTYPE, c_int, c_void_p, cast, pointer, pythonapi

import objc
from Foundation import CBUUID, CBCentralManager, NSDefaultRunLoopMode, NSRunLoop

from .ChipBleBase import ChipBleBase
from .ChipBleUtility import (BLE_ERROR_REMOTE_DEVICE_DISCONNECTED, BLE_SUBSCRIBE_OPERATION_SUBSCRIBE,
                             BLE_SUBSCRIBE_OPERATION_UNSUBSCRIBE, BleDisconnectEvent, BleDisconnectEventStruct, BleRxEvent,
                             BleRxEventStruct, BleSubscribeEvent, BleSubscribeEventStruct, BleTxEvent, BleTxEventStruct,
                             ParseServiceData)
from .ChipUtility import ChipUtility

try:
    objc.loadBundle(
        "CoreBluetooth",
        globals(),
        bundle_path=objc.pathForFramework(
            u"/System/Library/Frameworks/IOBluetooth.framework/Versions/A/Frameworks/CoreBluetooth.framework"
        ),
    )
except Exception:
    objc.loadBundle(
        "CoreBluetooth",
        globals(),
        bundle_path=objc.pathForFramework(
            u"/System/Library/Frameworks/CoreBluetooth.framework"
        ),
    )

BLE_PERIPHERAL_STATE_DISCONNECTED = 0
CBCharacteristicWriteWithResponse = 0
CBCharacteristicWriteWithoutResponse = 1

CHIP_SERVICE = CBUUID.UUIDWithString_(u"0000FFF6-0000-1000-8000-00805F9B34FB")
CHIP_SERVICE_SHORT = CBUUID.UUIDWithString_(u"FFF6")
CHIP_TX = CBUUID.UUIDWithString_(u"18EE2EF5-263D-4559-959F-4F9C429F9D11")
CHIP_RX = CBUUID.UUIDWithString_(u"18EE2EF5-263D-4559-959F-4F9C429F9D12")
CHROMECAST_SETUP_SERVICE = CBUUID.UUIDWithString_(
    u"0000FEA0-0000-1000-8000-00805F9B34FB"
)
CHROMECAST_SETUP_SERVICE_SHORT = CBUUID.UUIDWithString_(u"FEA0")


def _VoidPtrToCBUUID(ptr, len):
    try:
        ptr = ChipUtility.VoidPtrToByteArray(ptr, len)
        ptr = ChipUtility.Hexlify(ptr)
        ptr = (
            ptr[:8]
            + "-"
            + ptr[8:12]
            + "-"
            + ptr[12:16]
            + "-"
            + ptr[16:20]
            + "-"
            + ptr[20:]
        )
        ptr = CBUUID.UUIDWithString_(ptr)
    except Exception:
        print("ERROR: failed to convert void * to CBUUID")
        ptr = None

    return ptr


class LoopCondition:
    def __init__(self, op, timelimit, arg=None):
        self.op = op
        self.due = time.time() + timelimit
        self.arg = arg

    def TimeLimitExceeded(self):
        return time.time() > self.due


class BlePeripheral:
    def __init__(self, peripheral, advData):
        self.peripheral = peripheral
        self.advData = dict(advData)

    def __eq__(self, another):
        return self.peripheral == another.peripheral

    def getPeripheralDevIdInfo(self):
        # CHIP_SERVICE_SHORT
        if not self.advData:
            return None
        servDataDict = self.advData.get("kCBAdvDataServiceData", None)
        if not servDataDict:
            return None
        servDataDict = dict(servDataDict)
        for i in servDataDict.keys():
            if str(i).lower() == str(CHIP_SERVICE_SHORT).lower():
                return ParseServiceData(bytes(servDataDict[i]))
        return None


class CoreBluetoothManager(ChipBleBase):
    def __init__(self, devCtrl, logger=None):
        if logger:
            self.logger = logger
        else:
            self.logger = logging.getLogger("ChipBLEMgr")
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s %(name)-12s %(levelname)-8s %(message)s",
            )
        self.manager = None
        self.peripheral = None
        self.service = None
        self.scan_quiet = False
        self.characteristics = {}
        self.peripheral_list = []
        self.peripheral_adv_list = []
        self.bg_peripheral_name = None
        self.chip_queue = queue.Queue()

        self.manager = CBCentralManager.alloc()
        self.manager.initWithDelegate_queue_options_(self, None, None)

        self.ready_condition = False
        self.loop_condition = (
            # indicates whether the cmd requirement has been met in the runloop.
            False
        )
        # reflects whether or not there is a connection.
        self.connect_state = False
        self.send_condition = False
        self.subscribe_condition = False

        self.runLoopUntil(LoopCondition("ready", 10.0))

        self.orig_input_hook = None
        self.hookFuncPtr = None

        self.setInputHook(self.readlineCB)
        self.devCtrl = devCtrl

        # test if any connections currently exist (left around from a previous run) and disconnect if need be.
        peripherals = self.manager.retrieveConnectedPeripheralsWithServices_(
            [CHIP_SERVICE_SHORT, CHIP_SERVICE]
        )

        if peripherals and len(peripherals):
            for periph in peripherals:
                self.logger.info("disconnecting old connection.")
                self.loop_condition = False
                self.manager.cancelPeripheralConnection_(periph)
                self.runLoopUntil(LoopCondition("disconnect", 5.0))

            self.connect_state = False
            self.loop_condition = False

    def __del__(self):
        self.disconnect()
        self.setInputHook(self.orig_input_hook)

    def devMgrCB(self):
        """A callback used by ChipDeviceCtrl.py to drive the OSX runloop while the
        main thread waits for the Chip thread to complete its operation."""
        runLoop = NSRunLoop.currentRunLoop()
        runLoop.limitDateForMode_(NSDefaultRunLoopMode)

    def readlineCB(self):
        """A callback used by readline to drive the OSX runloop while the main thread
        waits for commandline input from the user."""
        runLoop = NSRunLoop.currentRunLoop()
        runLoop.limitDateForMode_(NSDefaultRunLoopMode)

        if self.orig_input_hook:
            self.orig_input_hook()

    def setInputHook(self, hookFunc):
        """Set the PyOS_InputHook to call the specific function."""
        hookFunctionType = CFUNCTYPE(None)
        self.hookFuncPtr = hookFunctionType(hookFunc)
        pyos_inputhook_ptr = c_void_p.in_dll(pythonapi, "PyOS_InputHook")
        # save the original so that on del we can revert it back to the way it was.
        self.orig_input_hook = cast(
            pyos_inputhook_ptr.value, PYFUNCTYPE(c_int))
        # set the new hook. readLine will call this periodically as it polls for input.
        pyos_inputhook_ptr.value = cast(self.hookFuncPtr, c_void_p).value

    def shouldLoop(self, cond: LoopCondition):
        """ Used by runLoopUntil to determine whether it should exit the runloop. """

        if cond.TimeLimitExceeded():
            return False

        if cond.op == "ready":
            return not self.ready_condition
        elif cond.op == "scan":
            for peripheral in self.peripheral_adv_list:
                if cond.arg and str(peripheral.peripheral._.name) == cond.arg:
                    return False
                devIdInfo = peripheral.getPeripheralDevIdInfo()
                if devIdInfo and cond.arg and str(devIdInfo.discriminator) == cond.arg:
                    return False
        elif cond.op == "connect":
            return (not self.loop_condition)
        elif cond.op == "disconnect":
            return (not self.loop_condition)
        elif cond.op == "send":
            return (not self.send_condition)
        elif cond.op == "subscribe":
            return (not self.subscribe_condition)
        elif cond.op == "unsubscribe":
            return self.subscribe_condition

        return True

    def runLoopUntil(self, cond: LoopCondition):
        """Helper function to drive OSX runloop until an expected event is received or
        the timeout expires."""
        runLoop = NSRunLoop.currentRunLoop()
        nextfire = 1

        while nextfire and self.shouldLoop(cond):
            nextfire = runLoop.limitDateForMode_(NSDefaultRunLoopMode)

    def centralManagerDidUpdateState_(self, manager):
        """ IO Bluetooth initialization is successful."""

        state = manager.state()
        string = "BLE is ready!" if state > 4 else "BLE is not ready!"
        self.logger.info(string)
        self.manager = manager
        self.ready_condition = True if state > 4 else False

    def centralManager_didDiscoverPeripheral_advertisementData_RSSI_(
        self, manager, peripheral, data, rssi
    ):
        """ Called for each peripheral discovered during scan."""
        if self.bg_peripheral_name is None:
            if peripheral not in self.peripheral_list:
                if not self.scan_quiet:
                    self.logger.info("adding to scan list:")
                    self.logger.info("")
                    self.logger.info(
                        "{0:<16}= {1:<80}".format(
                            "Name", str(peripheral._.name))
                    )
                    self.logger.info(
                        "{0:<16}= {1:<80}".format(
                            "ID", str(peripheral._.identifier.UUIDString())
                        )
                    )
                    self.logger.info("{0:<16}= {1:<80}".format("RSSI", rssi))
                    devIdInfo = BlePeripheral(
                        peripheral, data).getPeripheralDevIdInfo()
                    if devIdInfo:
                        self.logger.info("{0:<16}= {1}".format(
                            "Pairing State", devIdInfo.pairingState))
                        self.logger.info("{0:<16}= {1}".format(
                            "Discriminator", devIdInfo.discriminator))
                        self.logger.info("{0:<16}= {1}".format(
                            "Vendor Id", devIdInfo.vendorId))
                        self.logger.info("{0:<16}= {1}".format(
                            "Product Id", devIdInfo.productId))
                    self.logger.info("ADV data: " + repr(data))
                    self.logger.info("")

                self.peripheral_list.append(peripheral)
                self.peripheral_adv_list.append(
                    BlePeripheral(peripheral, data))
        else:
            if (peripheral._.name == self.bg_peripheral_name) or (str(devIdInfo.discriminator) == self.bg_peripheral_name):
                if len(self.peripheral_list) == 0:
                    self.logger.info("found background peripheral")
                self.peripheral_list = [peripheral]
                self.peripheral_adv_list = [BlePeripheral(peripheral, data)]

    def centralManager_didConnectPeripheral_(self, manager, peripheral):
        """Called by CoreBluetooth via runloop when a connection succeeds."""
        self.logger.debug(repr(peripheral))
        # make this class the delegate for peripheral events.
        self.peripheral.setDelegate_(self)
        # invoke service discovery on the periph.
        self.logger.info("Discovering services")
        self.peripheral.discoverServices_([CHIP_SERVICE_SHORT, CHIP_SERVICE])

    def centralManager_didFailToConnectPeripheral_error_(
        self, manager, peripheral, error
    ):
        """Called by CoreBluetooth via runloop when a connection fails."""
        self.logger.info("Failed to connect error = " + repr(error))
        self.loop_condition = True
        self.connect_state = False

    def centralManager_didDisconnectPeripheral_error_(self, manager, peripheral, error):
        """Called by CoreBluetooth via runloop when a disconnect completes. error = None on success."""
        self.loop_condition = True
        self.connect_state = False
        if self.devCtrl:
            self.logger.info("BLE disconnected, error = " + repr(error))
            dcEvent = BleDisconnectEvent(BLE_ERROR_REMOTE_DEVICE_DISCONNECTED)
            self.chip_queue.put(dcEvent)
            self.devCtrl.DriveBleIO()

    def peripheral_didDiscoverServices_(self, peripheral, services):
        """Called by CoreBluetooth via runloop when peripheral services are discovered."""
        if len(self.peripheral.services()) == 0:
            self.logger.error("Chip service not found")
            self.connect_state = False
        else:
            # in debugging, we found connect being called twice. This
            # would trigger discovering the services twice, and
            # consequently, discovering characteristics twice.  We use the
            # self.service as a flag to indicate whether the
            # characteristics need to be invalidated immediately.
            if self.service == self.peripheral.services()[0]:
                self.logger.debug("didDiscoverServices already happened")
            else:
                self.service = self.peripheral.services()[0]
                self.characteristics[self.service.UUID()] = []
            # NOTE: currently limiting discovery to only the pair of Chip characteristics.
            self.peripheral.discoverCharacteristics_forService_(
                [CHIP_RX, CHIP_TX], self.service
            )

    def peripheral_didDiscoverCharacteristicsForService_error_(
        self, peripheral, service, error
    ):
        """Called by CoreBluetooth via runloop when a characteristic for a service is discovered."""
        self.logger.debug(
            "didDiscoverCharacteristicsForService:error "
            + str(repr(peripheral))
            + " "
            + str(repr(service))
        )
        self.logger.debug(repr(service))
        self.logger.debug(repr(error))

        if not error:
            self.characteristics[service.UUID()] = [
                char for char in self.service.characteristics()
            ]

            self.connect_state = True

        else:
            self.logger.error(
                "ERROR: failed to discover characteristics for service.")
            self.connect_state = False

        self.loop_condition = True

    def peripheral_didWriteValueForCharacteristic_error_(
        self, peripheral, characteristic, error
    ):
        """Called by CoreBluetooth via runloop when a write to characteristic
        operation completes. error = None on success."""
        self.logger.debug("didWriteValue error = " + repr(error))
        self.send_condition = True
        charId = bytearray(characteristic.UUID().data().bytes().tobytes())
        svcId = bytearray(CHIP_SERVICE.data().bytes().tobytes())

        if self.devCtrl:
            txEvent = BleTxEvent(
                charId=charId, svcId=svcId, status=True if not error else False
            )
            self.chip_queue.put(txEvent)
            self.devCtrl.DriveBleIO()

    def peripheral_didUpdateNotificationStateForCharacteristic_error_(
        self, peripheral, characteristic, error
    ):
        """Called by CoreBluetooth via runloop when a subscribe for notification operation completes.
        Error = None on success."""
        self.logger.debug("Receiving notifications")
        charId = bytearray(characteristic.UUID().data().bytes().tobytes())
        svcId = bytearray(CHIP_SERVICE.data().bytes().tobytes())
        # look at error and send True/False on Success/Failure
        success = True if not error else False
        if characteristic.isNotifying():
            operation = BLE_SUBSCRIBE_OPERATION_SUBSCRIBE
            self.subscribe_condition = True
        else:
            operation = BLE_SUBSCRIBE_OPERATION_UNSUBSCRIBE
            self.subscribe_condition = False

        self.logger.debug("Operation = " + repr(operation))
        self.logger.debug("success = " + repr(success))

        if self.devCtrl:
            subscribeEvent = BleSubscribeEvent(
                charId=charId, svcId=svcId, status=success, operation=operation
            )
            self.chip_queue.put(subscribeEvent)
            self.devCtrl.DriveBleIO()

    def peripheral_didUpdateValueForCharacteristic_error_(
        self, peripheral, characteristic, error
    ):
        """Called by CoreBluetooth via runloop when a new characteristic value is received for a
        characteristic to which this device has subscribed."""
        # len = characteristic.value().length()
        bytes = bytearray(characteristic.value().bytes().tobytes())
        charId = bytearray(characteristic.UUID().data().bytes().tobytes())
        svcId = bytearray(CHIP_SERVICE.data().bytes().tobytes())

        # Kick Chip thread to retrieve the saved packet.
        if self.devCtrl:
            # Save buffer, length, service UUID and characteristic UUID
            rxEvent = BleRxEvent(charId=charId, svcId=svcId, buffer=bytes)
            self.chip_queue.put(rxEvent)
            self.devCtrl.DriveBleIO()

        self.logger.debug("received")
        self.logger.debug(
            "received ("
            + str(len)
            + ") bytes: "
            + repr(characteristic.value().bytes().tobytes())
        )

    def GetBleEvent(self):
        """ Called by ChipDeviceMgr.py on behalf of Chip to retrieve a queued message."""
        if not self.chip_queue.empty():
            ev = self.chip_queue.get()

            if isinstance(ev, BleRxEvent):
                eventStruct = BleRxEventStruct.fromBleRxEvent(ev)
                return cast(pointer(eventStruct), c_void_p).value
            elif isinstance(ev, BleTxEvent):
                eventStruct = BleTxEventStruct.fromBleTxEvent(ev)
                return cast(pointer(eventStruct), c_void_p).value
            elif isinstance(ev, BleSubscribeEvent):
                eventStruct = BleSubscribeEventStruct.fromBleSubscribeEvent(ev)
                return cast(pointer(eventStruct), c_void_p).value
            elif isinstance(ev, BleDisconnectEvent):
                eventStruct = BleDisconnectEventStruct.fromBleDisconnectEvent(
                    ev)
                return cast(pointer(eventStruct), c_void_p).value

        return None

    def scan(self, line):
        """ API to initiatae BLE scanning for -t user_timeout seconds."""

        args = self.ParseInputLine(line, "scan")

        if not args:
            return
        self.scan_quiet = args[1]
        self.bg_peripheral_name = None
        del self.peripheral_list[:]
        del self.peripheral_adv_list[:]
        self.peripheral_list = []
        self.peripheral_adv_list = []
        # Filter on the service UUID Array or None to accept all scan results.
        self.manager.scanForPeripheralsWithServices_options_(
            [
                CHIP_SERVICE_SHORT,
                CHIP_SERVICE,
                CHROMECAST_SETUP_SERVICE_SHORT,
                CHROMECAST_SETUP_SERVICE,
            ],
            None,
        )
        # self.manager.scanForPeripheralsWithServices_options_(None, None)

        self.runLoopUntil(LoopCondition("scan", args[0], args[2]))

        self.manager.stopScan()
        self.logger.info("scanning stopped")

    def bgScanStart(self, name):
        """ API to initiate background BLE scanning."""
        self.logger.info("scanning started")
        self.bg_peripheral_name = name
        del self.peripheral_list[:]
        self.peripheral_list = []
        # Filter on the service UUID Array or None to accept all scan results.
        self.manager.scanForPeripheralsWithServices_options_(
            [
                CHIP_SERVICE_SHORT,
                CHIP_SERVICE,
                CHROMECAST_SETUP_SERVICE_SHORT,
                CHROMECAST_SETUP_SERVICE,
            ],
            None,
        )

    def bgScanStop(self):
        """ API to stop background BLE scanning."""
        self.manager.stopScan()
        self.bg_peripheral_name = None
        self.logger.info("scanning stopped")

    def ble_debug_log(self, line):
        args = self.ParseInputLine(line)
        if int(args[0]) == 1:
            self.logger.setLevel(logging.DEBUG)
            self.logger.debug("current logging level is debug")
        else:
            self.logger.setLevel(logging.INFO)
            self.logger.info("current logging level is info")
        return True

    def CloseBle(self, connObj):
        """ Called by Chip to close the BLE connection."""
        if self.peripheral:
            self.manager.cancelPeripheralConnection_(self.peripheral)
            self.characteristics = {}
            # del self.peripheral_list[:]
            # self.peripheral_list = []
            self.peripheral = None
            self.service = None
            self.connect_state = False

        return True

    def updateCharacteristic(self, bytes, svcId, charId):
        # TODO: implement this for Peripheral support.
        return False