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

Alternate implementation

Browse Source
This commit is contained in:
Andrey Pokhilko 2020-06-30 12:16:44 +03:00
parent afdbe4b2e0
commit 812ca1ad15
6 changed files with 108 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
examples/demo.py
View File

@ -207,7 +207,7 @@ def get_options():
'-c', '--connection',
default='auto://',
help='''Specify connection URL to use, `protocol://mac?param=X` with protocol in:
"gatt","pygatt","gattlib","gattool", "bluepy","bluegiga"'''
"gatt", "pygatt", "gattlib", "gattool", "bluepy", "bluegiga", "bleak"'''
)
arg_parser.add_argument(
'-d', '--demo',
@ -246,7 +246,7 @@ def connection_from_url(url):
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO, format='%(relativeCreated)d\t%(levelname)s\t%(name)s\t%(message)s')
logging.basicConfig(level=logging.DEBUG, format='%(relativeCreated)d\t%(levelname)s\t%(name)s\t%(message)s')
parser = get_options()
options = parser.parse_args()
parameters = {}
@ -256,7 +256,7 @@ if __name__ == '__main__':
except ValueError as err:
parser.error(err.args[0])
hub = MoveHub(**parameters)
hub = MoveHub(get_connection_bleak(hub_name='LEGO Move Hub'))
try:
demo = DEMO_CHOICES[options.demo]
demo(hub)

3
pylgbst/__init__.py
View File

@ -2,6 +2,7 @@ import logging
import traceback
from pylgbst.comms import DebugServer
from pylgbst.comms.cbleak import BleakConnection, BleakConnection2
log = logging.getLogger('pylgbst')
@ -48,7 +49,7 @@ def get_connection_bleak(controller='hci0', hub_mac=None, hub_name=None):
del controller # to prevent code analysis warning
from pylgbst.comms.cbleak import BleakDriver
return BleakDriver(hub_mac, hub_name)
return BleakConnection2().connect(hub_mac, hub_name)
def get_connection_auto(controller='hci0', hub_mac=None, hub_name=None):

2
pylgbst/comms/__init__.py
View File

@ -24,7 +24,7 @@ MOVE_HUB_HARDWARE_HANDLE = 0x0E
class Connection(object):
def connect(self, hub_mac=None):
def connect(self, hub_mac=None, hub_name=None):
pass
@abstractmethod

93
pylgbst/comms/cbleak.py
View File

@ -15,6 +15,91 @@ resp_queue = queue.Queue()
req_queue = queue.Queue()
class BleakConnection2(Connection):
"""
:type _client: BleakClient
"""
def __init__(self) -> None:
super().__init__()
self._abort = False
self._loop = asyncio.new_event_loop()
asyncio.set_event_loop(self._loop)
self._client = None # noqa
self._notify_queue = queue.Queue()
def connect(self, hub_mac=None, hub_name=None):
#logging.getLogger('bleak.backends.dotnet.client').setLevel(logging.WARNING)
#logging.getLogger('bleak.backends.bluezdbus.client').setLevel(logging.WARNING)
#logging.getLogger('bleak.backends.dotnet.discovery').setLevel(logging.WARNING)
#logging.getLogger('bleak.backends.bluezdbus.discovery').setLevel(logging.WARNING)
log.info("Discovering devices... Press green button on Hub")
for i in range(0, 30):
devices = self._loop.run_until_complete(discover(timeout=1))
log.debug("Devices: %s", devices)
for dev in devices:
log.debug(dev)
address = dev.address
name = dev.name
if self._is_device_matched(address, name, hub_mac, hub_name):
log.info('Device matched: %r', dev)
device = dev
break
else:
continue
break
else:
raise ConnectionError('Device not found.')
self._client = BleakClient(device.address, self._loop)
status = self._loop.run_until_complete(self._client.connect())
log.debug('Connection status: %s', status)
def enqueue(handle, data):
log.debug("Put into queue: %s", data)
self._notify_queue.put((handle, data))
self._loop.run_until_complete(self._client.start_notify(MOVE_HUB_HW_UUID_CHAR, enqueue))
return self
def disconnect(self):
self._abort = True
if self.is_alive():
log.debug("Disconnecting bleak connection")
self._loop.run_until_complete(self._client.disconnect())
def is_alive(self):
log.debug("Checking if bleak conn is alive")
return self._loop.run_until_complete(self._client.is_connected())
def write(self, handle, data):
desc = self._client.services.get_descriptor(handle)
if not isinstance(data, bytearray):
data = bytearray(data)
if desc is None:
# dedicated handle not found, try to send by using LEGO Move Hub default characteristic
self._loop.run_until_complete(self._client.write_gatt_char(MOVE_HUB_HW_UUID_CHAR, data))
else:
self._loop.run_until_complete(self._client.write_gatt_char(desc.characteristic_uuid, data))
def set_notify_handler(self, handler):
def _processing():
while not self._abort:
handle, data = self._notify_queue.get(block=True)
handler(handle, data)
log.info("Processing thread has exited")
threading.Thread(target=_processing, daemon=True).start()
class BleakDriver(object):
"""Driver that provides interface between API and Bleak."""
@ -68,7 +153,7 @@ class BleakDriver(object):
data = req_queue.get()
await bleak.write(data[0], data[1])
logging.info("Communications thread has exited")
log.info("Communications thread has exited")
@staticmethod
def _safe_handler(handler, data):
@ -81,7 +166,7 @@ class BleakDriver(object):
self._handler(msg[0], msg[1])
time.sleep(0.01)
logging.info("Processing thread has exited")
log.info("Processing thread has exited")
def write(self, handle, data):
"""
@ -122,7 +207,7 @@ class BleakConnection(Connection):
def __init__(self):
"""Initialize new instance of BleakConnection class."""
Connection.__init__(self)
super().__init__(self)
self.loop = asyncio.get_event_loop()
self._device = None
@ -209,4 +294,4 @@ class BleakConnection(Connection):
This method does nothing.
:return: None.
"""
pass
return self._client.is_connected()

23
pylgbst/hub.py
View File

@ -143,10 +143,8 @@ class Hub(object):
log.info("Attached peripheral: %s", self.peripherals[msg.port])
if msg.event == msg.EVENT_ATTACHED:
hw_revision = reversed([usbyte(msg.payload, x) for x in range(2, 6)])
sw_revision = reversed([usbyte(msg.payload, x) for x in range(6, 10)])
# what to do with this info? it's useless, I guess
del hw_revision, sw_revision
self.peripherals[port].hw_revision = reversed([usbyte(msg.payload, x) for x in range(2, 6)])
self.peripherals[port].sw_revision = reversed([usbyte(msg.payload, x) for x in range(6, 10)])
elif msg.event == msg.EVENT_ATTACHED_VIRTUAL:
self.peripherals[port].virtual_ports = (usbyte(msg.payload, 2), usbyte(msg.payload, 3))
@ -197,14 +195,7 @@ class MoveHub(Hub):
# noinspection PyTypeChecker
def __init__(self, connection=None):
if connection is None:
connection = get_connection_auto(hub_name="LEGO Move Hub")
super(MoveHub, self).__init__(connection)
self.info = {}
# shorthand fields
self.button = Button(self)
self.led = None
self.current = None
self.voltage = None
@ -217,6 +208,14 @@ class MoveHub(Hub):
self.port_C = None
self.port_D = None
self.info = {}
if connection is None:
connection = get_connection_auto(hub_name="LEGO Move Hub")
super(MoveHub, self).__init__(connection)
self.button = Button(self)
self._wait_for_devices()
self._report_status()
@ -230,7 +229,7 @@ class MoveHub(Hub):
log.debug("All devices are present: %s", devices)
return
log.debug("Waiting for builtin devices to appear: %s", devices)
time.sleep(0.1)
time.sleep(0.2)
log.warning("Got only these devices: %s", get_dev_set())
def _report_status(self):

2
pylgbst/peripherals.py
View File

@ -55,6 +55,8 @@ class Peripheral(object):
:type port: int
"""
super(Peripheral, self).__init__()
self.sw_revision = None
self.hw_revision = None
self.virtual_ports = ()
self.hub = parent
self.port = port