v/Lego-Boost-Python-Library
1
0
Fork 0
mirror of https://github.com/undera/pylgbst.git synced 2020-11-18 19:37:26 -08:00
Code Issues Releases Wiki Activity

A bit of refactoring

Browse Source
This commit is contained in:
Andrey Pokhilko 2019-12-20 14:39:34 +03:00
parent 8b970c0792
commit c11e8fbd18
5 changed files with 24 additions and 25 deletions

8
pylgbst/comms/__init__.py

@ -46,6 +46,14 @@ class Connection(object):
def enable_notifications(self):
self.write(ENABLE_NOTIFICATIONS_HANDLE, ENABLE_NOTIFICATIONS_VALUE)
def _is_device_matched(self, address, name, hub_mac):
log.debug("Checking device name: %s, MAC: %s", name, address)
if address != "00:00:00:00:00:00":
if (not hub_mac and name == LEGO_MOVE_HUB) or hub_mac.lower() == address.lower():
log.info("Found %s at %s", name, address)
return True
return False
class DebugServer(object):
"""

11
pylgbst/comms/cbluepy.py

@ -4,7 +4,7 @@ from threading import Thread, Event
from bluepy import btle
from pylgbst.comms import Connection, LEGO_MOVE_HUB
from pylgbst.comms import Connection
from pylgbst.utilities import str2hex, queue
log = logging.getLogger('comms-bluepy')
@ -100,13 +100,10 @@ class BluepyConnection(Connection):
address = dev.addr
addressType = dev.addrType
name = dev.getValueText(COMPLETE_LOCAL_NAME_ADTYPE)
log.debug("Found dev, name: {}, address: {}".format(name, address))
if address != "00:00:00:00:00:00":
if (not hub_mac and name == LEGO_MOVE_HUB) or hub_mac == address:
logging.info("Found %s at %s", name, address)
self._peripheral = BluepyThreadedPeripheral(address, addressType, self._controller)
break
if self._is_device_matched(address, name, hub_mac):
self._peripheral = BluepyThreadedPeripheral(address, addressType, self._controller)
break
return self

11
pylgbst/comms/cgatt.py

@ -5,7 +5,7 @@ from time import sleep
import gatt
from pylgbst.comms import Connection, LEGO_MOVE_HUB, MOVE_HUB_HW_UUID_SERV, MOVE_HUB_HW_UUID_CHAR, \
from pylgbst.comms import Connection, MOVE_HUB_HW_UUID_SERV, MOVE_HUB_HW_UUID_CHAR, \
MOVE_HUB_HARDWARE_HANDLE
from pylgbst.utilities import str2hex
@ -100,12 +100,9 @@ class GattConnection(Connection):
for dev in devices:
address = dev.mac_address
name = dev.alias()
logging.debug("Device %s at %s", name, address)
if address != "00:00:00:00:00:00":
if (not hub_mac and name == LEGO_MOVE_HUB) or hub_mac == address:
logging.info("Found %s at %s", name, address)
self._device = CustomDevice(address, self._manager)
break
if self._is_device_matched(address, name, hub_mac):
self._device = CustomDevice(address, self._manager)
break
if not self._device:
sleep(1)

10
pylgbst/comms/cgattlib.py

@ -5,7 +5,7 @@ from threading import Thread
from gattlib import DiscoveryService, GATTRequester
from pylgbst.comms import Connection, LEGO_MOVE_HUB
from pylgbst.comms import Connection
from pylgbst.utilities import queue, str2hex
log = logging.getLogger('comms-gattlib')
@ -70,11 +70,9 @@ class GattLibConnection(Connection):
log.debug("Devices: %s", devices)
for address, name in devices.items():
if address != "00:00:00:00:00:00":
if (not hub_mac and name == LEGO_MOVE_HUB) or hub_mac == address:
logging.info("Found %s at %s", name, address)
self.requester = Requester(address, True, self._iface)
break
if self._is_device_matched(address, name, hub_mac):
self.requester = Requester(address, True, self._iface)
break
if self.requester:
break

9
pylgbst/comms/cpygatt.py

@ -33,11 +33,9 @@ class GattoolConnection(Connection):
for dev in devices:
address = dev['address']
name = dev['name']
if address != "00:00:00:00:00:00":
if (not hub_mac and name == LEGO_MOVE_HUB) or hub_mac == address:
logging.info("Found %s at %s", name, address)
self._conn_hnd = adapter.connect(address)
break
if self._is_device_matched(address, name, hub_mac):
self._conn_hnd = adapter.connect(address)
break
if self._conn_hnd:
break
@ -58,6 +56,7 @@ class GattoolConnection(Connection):
return True
class BlueGigaConnection(GattoolConnection):
def __init__(self):
super(BlueGigaConnection, self).__init__()