blob: e480750b600af8105b1f586b705b6ab65364597f [file] [log] [blame]
#
# 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 Manager via BlueZ APIs.
#
from __future__ import absolute_import, print_function
import logging
import queue
import sys
import threading
import time
import traceback
import uuid
from ctypes import CFUNCTYPE, PYFUNCTYPE, c_int, c_void_p, cast, pythonapi
import dbus
import dbus.mainloop.glib
import dbus.service
from .ChipBleBase import ChipBleBase
from .ChipBleUtility import BLE_ERROR_REMOTE_DEVICE_DISCONNECTED, BleDisconnectEvent, ParseServiceData
try:
from gi.repository import GObject
except Exception:
logging.exception("Unable to find GObject from gi.repository")
from pgi.repository import GObject
chip_service = uuid.UUID("0000FFF6-0000-1000-8000-00805F9B34FB")
chip_tx = uuid.UUID("18EE2EF5-263D-4559-959F-4F9C429F9D11")
chip_rx = uuid.UUID("18EE2EF5-263D-4559-959F-4F9C429F9D12")
chip_service_short = uuid.UUID("0000FFF6-0000-0000-0000-000000000000")
chromecast_setup_service = uuid.UUID("0000FEA0-0000-1000-8000-00805F9B34FB")
chromecast_setup_service_short = uuid.UUID(
"0000FEA0-0000-0000-0000-000000000000")
BLUEZ_NAME = "org.bluez"
ADAPTER_INTERFACE = BLUEZ_NAME + ".Adapter1"
DEVICE_INTERFACE = BLUEZ_NAME + ".Device1"
SERVICE_INTERFACE = BLUEZ_NAME + ".GattService1"
CHARACTERISTIC_INTERFACE = BLUEZ_NAME + ".GattCharacteristic1"
DBUS_PROPERTIES = "org.freedesktop.DBus.Properties"
BLE_SCAN_CONNECT_GUARD_SEC = 2.0
BLE_STATUS_TRANSITION_TIMEOUT_SEC = 5.0
BLE_CONNECT_TIMEOUT_SEC = 15.0
BLE_SERVICE_DISCOVERY_TIMEOUT_SEC = 5.0
BLE_CHAR_DISCOVERY_TIMEOUT_SEC = 5.0
BLE_SUBSCRIBE_TIMEOUT_SEC = 5.0
BLE_WRITE_CHARACTERISTIC_TIMEOUT_SEC = 10.0
BLE_IDLE_DELTA = 0.1
def get_bluez_objects(bluez, bus, interface, prefix_path):
results = []
if bluez is None or bus is None or interface is None or prefix_path is None:
return results
for item in bluez.GetManagedObjects().items():
delegates = item[1].get(interface)
if not delegates:
continue
slice = {}
if item[0].startswith(prefix_path):
slice["object"] = bus.get_object(BLUEZ_NAME, item[0])
slice["path"] = item[0]
results.append(slice)
return results
class BluezDbusAdapter:
def __init__(self, bluez_obj, bluez, bus, logger=None):
self.logger = logger if logger else logging.getLogger("ChipBLEMgr")
self.object = bluez_obj
self.adapter = dbus.Interface(bluez_obj, ADAPTER_INTERFACE)
self.adapter_properties = dbus.Interface(bluez_obj, DBUS_PROPERTIES)
self.adapter_event = threading.Event()
self.bluez = bluez
self.bus = bus
self.path = self.adapter.object_path
self.signalReceiver = None
def __del__(self):
self.destroy()
def destroy(self):
self.logger.debug("destroy adapter")
self.adapter_unregister_signal()
self.adapter = None
self.adapter_properties = None
self.adapter_event.clear()
self.bluez = None
self.bus = None
self.object = None
self.path = None
self.signalReceiver = None
def adapter_register_signal(self):
if self.signalReceiver is None:
self.logger.debug("add adapter signal")
self.signalReceiver = self.bus.add_signal_receiver(
self.adapter_on_prop_changed_cb,
bus_name=BLUEZ_NAME,
dbus_interface=DBUS_PROPERTIES,
signal_name="PropertiesChanged",
path=self.path,
)
def adapter_unregister_signal(self):
if self.signalReceiver is not None:
self.logger.debug(" remove adapter signal")
self.bus.remove_signal_receiver(
self.signalReceiver,
signal_name="PropertiesChanged",
dbus_interface="org.freedesktop.DBus.Properties",
)
def adapter_on_prop_changed_cb(
self, interface, changed_properties, invalidated_properties
):
if len(changed_properties) == 0:
self.logger.debug("changed_properties is empty")
return
if len(invalidated_properties) > 0:
self.logger.debug(
"invalidated_properties is not empty %s" % str(
invalidated_properties)
)
return
if interface == ADAPTER_INTERFACE:
if "Discovering" in changed_properties:
self.adapter_event.set()
def adapter_bg_scan(self, enable):
self.adapter_event.clear()
action_flag = False
try:
if enable:
if not self.Discovering:
action_flag = True
self.logger.info("scanning started")
self.adapter.StartDiscovery()
else:
self.logger.info("it has started scanning")
else:
if self.Discovering:
action_flag = True
self.adapter.StopDiscovery()
self.logger.info("scanning stopped")
else:
print("it has stopped scanning")
if action_flag:
if not self.adapter_event.wait(BLE_STATUS_TRANSITION_TIMEOUT_SEC):
if enable:
self.logger.debug("scan start error")
else:
self.logger.debug("scan stop error")
self.adapter_event.clear()
except dbus.exceptions.DBusException as ex:
self.adapter_event.clear()
self.logger.debug(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
@property
def Address(self):
try:
result = self.adapter_properties.Get(ADAPTER_INTERFACE, "Address")
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def UUIDs(self):
try:
return self.adapter_properties.Get(ADAPTER_INTERFACE, "UUIDs")
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
def SetDiscoveryFilter(self, dict):
try:
self.adapter.SetDiscoveryFilter(dict)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
@property
def Discovering(self):
try:
result = self.adapter_properties.Get(
ADAPTER_INTERFACE, "Discovering")
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
def DiscoverableTimeout(self, timeoutSec):
try:
result = self.adapter_properties.Set(
ADAPTER_INTERFACE, "DiscoverableTimeout", timeoutSec
)
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
def Powered(self, enable):
try:
result = self.adapter_properties.Set(
ADAPTER_INTERFACE, "Powered", enable)
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
def find_devices(self, uuids):
devices = [
BluezDbusDevice(p["object"], self.bluez, self.bus, self.logger)
for p in get_bluez_objects(
self.bluez, self.bus, DEVICE_INTERFACE, self.path
)
]
found = []
for device in devices:
for i in device.uuids:
if i in uuids:
found.append(device)
break
# Some devices do not advertise their uuid lists, thus we should also check service data.
if device.ServiceData:
for i in device.ServiceData:
if uuid.UUID(str(i)) in uuids:
found.append(device)
break
return found
def clear_adapter(self):
devices = [
BluezDbusDevice(p["object"], self.bluez, self.bus, self.logger)
for p in get_bluez_objects(
self.bluez, self.bus, DEVICE_INTERFACE, self.path
)
]
for device in devices:
try:
if device.Connected:
device.device_bg_connect(False)
self.adapter.RemoveDevice(device.device.object_path)
except Exception:
pass
class BluezDbusDevice:
def __init__(self, bluez_obj, bluez, bus, logger=None):
self.logger = logger if logger else logging.getLogger("ChipBLEMgr")
self.object = bluez_obj
self.device = dbus.Interface(bluez_obj, DEVICE_INTERFACE)
self.device_properties = dbus.Interface(bluez_obj, DBUS_PROPERTIES)
self.path = self.device.object_path
self.device_event = threading.Event()
if self.Name:
try:
self.device_id = uuid.uuid3(uuid.NAMESPACE_DNS, self.Name)
except UnicodeDecodeError:
self.device_id = uuid.uuid3(
uuid.NAMESPACE_DNS, self.Name.encode("utf-8")
)
else:
self.device_id = uuid.uuid4()
self.bluez = bluez
self.bus = bus
self.signalReceiver = None
self.path = self.device.object_path
def __del__(self):
self.destroy()
def destroy(self):
self.logger.debug("destroy device")
self.device_unregister_signal()
self.device = None
self.device_properties = None
self.device_event = None
self.device_id = None
self.bluez = None
self.bus = None
self.object = None
self.signalReceiver = None
def device_register_signal(self):
if self.signalReceiver is None:
self.logger.debug("add device signal")
self.signalReceiver = self.bus.add_signal_receiver(
self.device_on_prop_changed_cb,
bus_name=BLUEZ_NAME,
dbus_interface=DBUS_PROPERTIES,
signal_name="PropertiesChanged",
path=self.path,
)
def device_unregister_signal(self):
if self.signalReceiver is not None:
self.logger.debug("remove device signal")
self.bus.remove_signal_receiver(
self.signalReceiver,
signal_name="PropertiesChanged",
dbus_interface=DBUS_PROPERTIES,
)
def device_on_prop_changed_cb(
self, interface, changed_properties, invalidated_properties
):
if len(changed_properties) == 0:
self.logger.debug("changed_properties is empty")
return
if len(invalidated_properties) > 0:
self.logger.debug(
"invalidated_properties is not empty %s" % str(
invalidated_properties)
)
return
if interface == DEVICE_INTERFACE:
if "Connected" in changed_properties:
self.device_event.set()
def device_bg_connect(self, enable):
time.sleep(BLE_SCAN_CONNECT_GUARD_SEC)
action_flag = False
self.device_event.clear()
try:
if enable:
if not self.Connected:
action_flag = True
self.device.Connect()
self.logger.info("BLE connecting")
else:
self.logger.info("BLE has connected")
else:
if self.Connected:
action_flag = True
self.device.Disconnect()
self.logger.info("BLE disconnected")
else:
self.logger.info("BLE has disconnected")
if action_flag:
if not self.device_event.wait(BLE_STATUS_TRANSITION_TIMEOUT_SEC):
if enable:
self.logger.info("BLE connect error")
else:
self.logger.info("BLE disconnect error")
self.device_event.clear()
except dbus.exceptions.DBusException as ex:
self.device_event.clear()
self.logger.info(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
def service_discover(self, gatt_dic):
self.logger.info("Discovering services")
try:
expired = time.time() + BLE_SERVICE_DISCOVERY_TIMEOUT_SEC
while time.time() < expired:
if self.ServicesResolved:
services = [
BluezDbusGattService(
p["object"], self.bluez, self.bus, self.logger
)
for p in get_bluez_objects(
self.bluez, self.bus, SERVICE_INTERFACE, self.path
)
]
for service in services:
if service.uuid in gatt_dic["services"]:
self.logger.info("Service discovering success")
return service
time.sleep(BLE_IDLE_DELTA)
self.logger.error("Service discovering fail")
return None
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def uuids(self):
try:
uuids = self.device_properties.Get(DEVICE_INTERFACE, "UUIDs")
uuid_result = []
for i in uuids:
if len(str(i)) == 4:
uuid_normal = "0000%s-0000-0000-0000-000000000000" % i
else:
uuid_normal = i
uuid_result.append(uuid.UUID(str(uuid_normal)))
return uuid_result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def Address(self):
try:
return self.device_properties.Get(DEVICE_INTERFACE, "Address")
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def Name(self):
try:
name = self.device_properties.Get(DEVICE_INTERFACE, "Name")
return name
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def Connected(self):
try:
result = self.device_properties.Get(DEVICE_INTERFACE, "Connected")
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
@property
def TxPower(self):
try:
return self.device_properties.Get(DEVICE_INTERFACE, "TxPower")
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def RSSI(self):
try:
result = self.device_properties.Get(DEVICE_INTERFACE, "RSSI")
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def Adapter(self):
try:
return self.device_properties.Get(DEVICE_INTERFACE, "Adapter")
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def ServiceData(self):
try:
return self.device_properties.Get(DEVICE_INTERFACE, "ServiceData")
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def ServicesResolved(self):
try:
result = self.device_properties.Get(
DEVICE_INTERFACE, "ServicesResolved")
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
class BluezDbusGattService:
def __init__(self, bluez_obj, bluez, bus, logger=None):
self.logger = logger if logger else logging.getLogger("ChipBLEMgr")
self.object = bluez_obj
self.service = dbus.Interface(bluez_obj, SERVICE_INTERFACE)
self.service_properties = dbus.Interface(bluez_obj, DBUS_PROPERTIES)
self.bluez = bluez
self.bus = bus
self.path = self.service.object_path
def __del__(self):
self.destroy()
def destroy(self):
self.logger.debug("destroy GattService")
self.service = None
self.service_properties = None
self.bluez = None
self.bus = None
self.object = None
self.path = None
@property
def uuid(self):
try:
result = uuid.UUID(
str(self.service_properties.Get(SERVICE_INTERFACE, "UUID"))
)
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
@property
def Primary(self):
try:
result = bool(self.service_properties.Get(
SERVICE_INTERFACE, "Primary"))
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
@property
def Device(self):
try:
result = self.service_properties.Get(SERVICE_INTERFACE, "Device")
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
def find_characteristic(self, uuid):
try:
expired = time.time() + BLE_CHAR_DISCOVERY_TIMEOUT_SEC
while time.time() < expired:
characteristics = [
BluezDbusGattCharacteristic(
p["object"], self.bluez, self.bus, self.logger
)
for p in get_bluez_objects(
self.bluez, self.bus, CHARACTERISTIC_INTERFACE, self.path
)
]
for characteristic in characteristics:
if characteristic.uuid == uuid:
return characteristic
time.sleep(BLE_IDLE_DELTA)
self.logger.error("Char discovering fail")
return None
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
class BluezDbusGattCharacteristic:
def __init__(self, bluez_obj, bluez, bus, logger=None):
self.logger = logger if logger else logging.getLogger("ChipBLEMgr")
self.object = bluez_obj
self.characteristic = dbus.Interface(
bluez_obj, CHARACTERISTIC_INTERFACE)
self.characteristic_properties = dbus.Interface(
bluez_obj, DBUS_PROPERTIES)
self.received = None
self.path = self.characteristic.object_path
self.bluez = bluez
self.bus = bus
self.signalReceiver = None
def __del__(self):
self.destroy()
def destroy(self):
self.logger.debug("destroy GattCharacteristic")
self.gattCharacteristic_unregister_signal()
self.characteristic = None
self.object = None
self.characteristic_properties = None
self.received = None
self.bluez = None
self.bus = None
self.path = None
self.signalReceiver = None
def gattCharacteristic_register_signal(self):
if not self.signalReceiver:
self.logger.debug("add GattCharacteristic signal")
self.signalReceiver = self.bus.add_signal_receiver(
self.gatt_on_characteristic_changed_cb,
bus_name=BLUEZ_NAME,
dbus_interface=DBUS_PROPERTIES,
signal_name="PropertiesChanged",
path=self.path,
)
def gattCharacteristic_unregister_signal(self):
if self.signalReceiver:
self.logger.debug("remove GattCharacteristic signal")
self.bus.remove_signal_receiver(
self.signalReceiver,
bus_name=BLUEZ_NAME,
signal_name="PropertiesChanged",
dbus_interface=DBUS_PROPERTIES,
path=self.path,
)
self.signalReceiver = None
def gatt_on_characteristic_changed_cb(
self, interface, changed_properties, invalidated_properties
):
self.logger.debug(
"property change in" +
str(self.characteristic) + str(changed_properties)
)
if len(changed_properties) == 0:
return
if len(invalidated_properties) > 0:
return
if interface == CHARACTERISTIC_INTERFACE:
if "Value" in changed_properties:
if self.received:
self.received(changed_properties["Value"])
def WriteValue(self, value, options, reply_handler, error_handler, timeout):
try:
self.characteristic.WriteValue(
value,
options,
reply_handler=reply_handler,
error_handler=error_handler,
timeout=timeout,
)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
@property
def uuid(self):
try:
result = uuid.UUID(
str(
self.characteristic_properties.Get(
CHARACTERISTIC_INTERFACE, "UUID")
)
)
return result
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return None
except Exception:
self.logger.debug(traceback.format_exc())
return None
def StartNotify(self, cbfunct, reply_handler, error_handler, timeout):
try:
if not cbfunct:
self.logger.info("please provide the notify callback function")
self.received = cbfunct
self.gattCharacteristic_register_signal()
self.characteristic.StartNotify(
reply_handler=reply_handler,
error_handler=error_handler,
timeout=timeout,
)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
def StopNotify(self, reply_handler, error_handler, timeout):
try:
self.logger.debug("stopping notifying")
self.characteristic.StopNotify(
reply_handler=reply_handler,
error_handler=error_handler,
timeout=timeout,
)
self.gattCharacteristic_unregister_signal()
self.received = None
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
except Exception:
self.logger.debug(traceback.format_exc())
@property
def Notifying(self):
try:
result = self.characteristic_properties.Get(
CHARACTERISTIC_INTERFACE, "Notifying"
)
return bool(result)
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
return False
except Exception:
self.logger.debug(traceback.format_exc())
return False
class BluezManager(ChipBleBase):
def __init__(self, devMgr, 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.scan_quiet = False
self.peripheral_list = []
self.device_uuid_list = []
self.chip_queue = queue.Queue()
self.Gmainloop = None
self.daemon_thread = None
self.adapter = None
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
GObject.threads_init()
dbus.mainloop.glib.threads_init()
self.bus = dbus.SystemBus()
self.bluez = dbus.Interface(
self.bus.get_object(
BLUEZ_NAME, "/"), "org.freedesktop.DBus.ObjectManager"
)
self.target = None
self.service = None
self.orig_input_hook = None
self.hookFuncPtr = None
self.connect_state = False
self.tx = None
self.rx = None
self.setInputHook(self.readlineCB)
self.devMgr = devMgr
self.devMgr.SetBlockingCB(self.devMgrCB)
def __del__(self):
self.disconnect()
self.setInputHook(self.orig_input_hook)
def ble_adapter_select(self, identifier=None):
if self.adapter:
self.adapter.destroy()
self.adapter = None
self.adapter = self.get_adapter_by_addr(identifier)
self.adapter.adapter_register_signal()
self.adapter.Powered(False)
self.adapter.Powered(True)
def get_adapters(self):
return [
BluezDbusAdapter(p["object"], self.bluez, self.bus, self.logger)
for p in get_bluez_objects(
self.bluez, self.bus, ADAPTER_INTERFACE, "/org/bluez"
)
]
def ble_adapter_print(self):
try:
adapters = [
BluezDbusAdapter(p["object"], self.bluez,
self.bus, self.logger)
for p in get_bluez_objects(
self.bluez, self.bus, ADAPTER_INTERFACE, "/org/bluez"
)
]
for adapter in adapters:
self.logger.info("AdapterName: %s AdapterAddress: %s" % (
adapter.path.replace("/org/bluez/", ""), adapter.Address))
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
def get_adapter_by_addr(self, identifier):
try:
adapters = [
BluezDbusAdapter(p["object"], self.bluez,
self.bus, self.logger)
for p in get_bluez_objects(
self.bluez, self.bus, ADAPTER_INTERFACE, "/org/bluez"
)
]
if identifier is None:
return adapters[0]
if len(adapters) > 0:
for adapter in adapters:
if (str(adapter.Address).upper() == str(identifier).upper() or
"/org/bluez/{}".format(identifier) == str(adapter.path)):
return adapter
self.logger.info(
"adapter %s cannot be found, expect the ble mac address" % (
identifier)
)
return None
except dbus.exceptions.DBusException as ex:
self.logger.debug(str(ex))
def runLoopUntil(self, target=None, **kwargs):
if target:
self.daemon_thread = threading.Thread(
target=self.running_thread, args=(target, kwargs)
)
self.daemon_thread.daemon = True
self.daemon_thread.start()
try:
self.Gmainloop = GObject.MainLoop()
self.Gmainloop.run()
except KeyboardInterrupt:
self.Gmainloop.quit()
sys.exit(1)
def running_thread(self, target, kwargs):
try:
while not self.Gmainloop or not self.Gmainloop.is_running():
time.sleep(0.00001)
target(**kwargs)
except Exception:
traceback.print_exc()
finally:
self.Gmainloop.quit()
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 runIdleLoop(self, **kwargs):
time.sleep(0)
def devMgrCB(self):
self.runLoopUntil(self.runIdleLoop)
def readlineCB(self):
self.runLoopUntil(self.runIdleLoop)
if self.orig_input_hook:
self.orig_input_hook()
def dump_scan_result(self, device):
self.logger.info("{0:<16}= {1}".format("Name", device.Name))
self.logger.info("{0:<16}= {1}".format("ID", device.device_id))
self.logger.info("{0:<16}= {1}".format("RSSI", device.RSSI))
self.logger.info("{0:<16}= {1}".format("Address", device.Address))
devIdInfo = self.get_peripheral_devIdInfo(device)
if devIdInfo is not None:
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))
if device.ServiceData:
for advuuid in device.ServiceData:
self.logger.info("{0:<16}= {1}".format(
"Adv UUID", str(advuuid)))
self.logger.info("{0:<16}= {1}".format(
"Adv Data", bytes(device.ServiceData[advuuid]).hex()))
else:
self.logger.info("")
self.logger.info("")
def scan_bg_implementation(self, **kwargs):
self.adapter.clear_adapter()
with self.chip_queue.mutex:
self.chip_queue.queue.clear()
self.adapter.adapter_bg_scan(True)
found = False
identifier = kwargs["identifier"]
timeout = kwargs["timeout"] + time.time()
self.device_uuid_list = []
self.peripheral_list = []
while time.time() < timeout:
scanned_peripheral_list = self.adapter.find_devices(
[
chip_service,
chip_service_short,
chromecast_setup_service,
chromecast_setup_service_short,
]
)
for device in scanned_peripheral_list:
try:
if not self.scan_quiet and device.Address not in self.device_uuid_list:
# display all scanned results
self.device_uuid_list.append(device.Address)
self.peripheral_list.append(device)
self.dump_scan_result(device)
devIdInfo = self.get_peripheral_devIdInfo(device)
if not devIdInfo:
# Not a chip device
continue
if identifier and (device.Name == identifier or str(device.Address).upper() == str(
identifier.upper()
) or str(devIdInfo.discriminator) == identifier):
if self.scan_quiet:
# only display the scanned target's info when quiet
self.dump_scan_result(device)
self.target = device
found = True
break
except Exception:
traceback.print_exc()
if found:
break
time.sleep(BLE_IDLE_DELTA)
self.adapter.adapter_bg_scan(False)
def scan(self, line):
args = self.ParseInputLine(line, "scan")
if not args:
return False
self.target = None
if not self.adapter:
self.logger.info("use default adapter")
self.ble_adapter_select()
del self.peripheral_list[:]
self.scan_quiet = args[1]
self.runLoopUntil(
self.scan_bg_implementation, timeout=args[0], identifier=args[2]
)
return True
def get_peripheral_devIdInfo(self, peripheral):
if not peripheral.ServiceData:
return None
for advuuid in peripheral.ServiceData:
if str(advuuid).lower() == str(chip_service).lower():
return ParseServiceData(bytes(peripheral.ServiceData[advuuid]))
return None
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."""
# Workaround: comment out disconnect because of hang when close, plz call disconnect explicitly after close
# Need to fix it
# self.disconnect()
if self.devMgr:
dcEvent = BleDisconnectEvent(BLE_ERROR_REMOTE_DEVICE_DISCONNECTED)
self.chip_queue.put(dcEvent)
self.devMgr.DriveBleIO()
return True