18 Commits
dev ... master

Author SHA1 Message Date
Shin'ichiro Kawasaki
ed4df282b1 Merge pull request #39 from rcy/master-1
Update README.md to correct micro:bit hex file download link.
2023-08-15 11:28:26 +09:00
Ryan Yeske
00154f2b51 Update README.md
Remove version number

Upstream url changed on https://scratch.mit.edu/microbit
2023-08-05 17:08:21 -07:00
Shin'ichiro Kawasaki
dc2ee5a22f Tag version 0.2.8
Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-07-02 21:52:08 +09:00
Shin'ichiro Kawasaki
caa344ecbb scratch_link.py: Support COMPLETE_LOCAL_NAME for 'namePrefix' filter
Current implementation of match() does filter matching for 'namePrefix'
keyword only for SHORT_LOCAL_NAME of the device. When the device does
not provide SHORT_LOCAL_NAME but provides COMPLETE_LOCAL_NAME, match()
for 'namePrefix' fails. For example, MicroBit More v2 [1] does not
provide the SHORT_LOCAL_NAME.

To support 'namePrefix' filter match with COMPLETE_LOCAL_NAME, add
support for it. While at it, replace the SHORT_LOCAL_NAME identifier
constant with the definition in btle.ScanEntry.

[1] https://lab.yengawa.com/project/microbit-more-v2/

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-07-02 21:52:08 +09:00
Shin'ichiro Kawasaki
0f9ccd3b63 Tag version 0.2.7
Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-03-21 17:17:31 +09:00
Shin'ichiro Kawasaki
d1f7f58ca2 README.md: Add description about -r option for scan failure
Add an item to "In Case You Fail to Connect" section to describe that
the "-r" option may help to address device scan failure.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-03-21 16:54:59 +09:00
Shin'ichiro Kawasaki
7050016ee6 scratch_link.py: Add -r option to retry BLE scan
Recently, scratch link connection failure is reported. Similar failure
is observed with my environment. I found BTLEDisconnectError happens at
scanner.scan(). Though the true cause of the failure is not yet known, I
found simple retry of scanner.scan() avoids the failure. As a temporary
work around, implement -r or --scan_retry option which specify how many
times to retry the scan. I confirmed that "-r 2" option avoided the
failure.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-03-21 15:53:34 +09:00
Shin'ichiro Kawasaki
d845d69bb8 gencert.py: Support Snap Firefox and Chromium
Recently Ubuntu provides Firefox and Chromium in the form of Snap
package, and they place NSS DB at different path from non-Snap package.
However, current gencert.py implementation supports only the NSS DB
paths for non-Snap packages. This results in HTTPS communication failure
between the browsers and scratch_link.

Support the NSS DB paths for the Snap packages. Add a new function
prep_cert_for_app() which takes application name and its NSS DB search
path. Call this function for list of browsers, covering both non-Snap
and Snap packages.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2023-03-19 20:59:58 +09:00
Shin'ichiro Kawasaki
751190935a release.sh: add example command lines
I tend to forget the command lines to release new packages. Record it in
the release script.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 11:12:46 +09:00
Shin'ichiro Kawasaki
3956c81869 Tag version 0.2.6
Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 10:58:32 +09:00
Shin'ichiro Kawasaki
caaec40935 release.sh: update package build commands
Recent changes in python eco-system require different command to build
python packages. Update the commands to fit to the latest environment.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 10:58:32 +09:00
Shin'ichiro Kawasaki
845b37707e README.md: remove descriptions about EV3
pyscrlink no longer support LEGO Mindstorm EV3. Remove the description
about EV3 and explain why it is not supported. I wish it will be
supported again some day.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 10:53:57 +09:00
Shin'ichiro Kawasaki
a5d19fa2ba setup.py: remove dependency to pybluez
pybluez can not be installed with pip. Remove dependency to it.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 10:53:57 +09:00
Shin'ichiro Kawasaki
865a4d6f09 scratch_link.py: remove pybluez dependency
Recently pybluez is not maintained actively and causing troubles. One of
the troubles is pip installation failure [1] and this made pyscrlink
uninstallable via pip. To avoid this, remove pybluez dependency from
pyscrlink. This removes Bluetooth Classic protocol support, then LEGO
Mindstorm EV3 is no longer supported.

[1] https://github.com/pybluez/pybluez/issues/431

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-18 10:24:05 +09:00
Shin'ichiro Kawasaki
bf154e0a14 .gitignore: add more directories to ignore
Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2022-07-17 10:51:32 +09:00
Shin'ichiro Kawasaki
7cc9ccac2e Tag version 0.2.5
Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2021-11-21 14:25:56 +09:00
Shin'ichiro Kawasaki
4558ea43df BLESession: handle device UUIDs as list
In the debug for the issue #30, it turned out that the bluepy API
ScanEntry.getValueText() for adtypes may return multiple UUIDs
concatenated with comma. Current code assumes that the API returns
single UUID, then multiple UUIDs cause the "Error: Non-hexadecimal digit
found" for the comma.

To fix it, replace ScanEntry.getValueText() call with getValue(), and
handle the result as a list of UUIDs. Also rename the helper function
_get_dev_uuid() to _get_dev_uuids() accordingly.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2021-11-21 14:21:50 +09:00
Shin'ichiro Kawasaki
641b84a86e BLESession: Enrich logs for device UUID check
To debug the GitHub issue #30, add more debug logs to functions to check
UUID.

Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
2021-11-21 14:21:42 +09:00
6 changed files with 119 additions and 343 deletions

6
.gitignore vendored
View File

@@ -1 +1,5 @@
/__pycache__
__pycache__
build
dist
pyscrlink/__pycache__
pyscrlink.egg-info

109
README.md
View File

@@ -9,11 +9,16 @@ micro:bit.
Pyscrlink allows you to connect Scratch and bluetooth devices with the Linux
OSes. It uses the Linux Bluetooth protocol stack [Bluez](http://www.bluez.org/)
and its python interfaces [pybluez](https://github.com/pybluez/pybluez) to
handle Bluetooth, and [bluepy](https://github.com/IanHarvey/bluepy) to handle
Bluetooth Low Energy (BLE) connections. It has been reported that pyscrlink
connects Scratch 3.0 with micro:bit, LEGO Mindstorms EV3, LEGO WeDo, LEGO
Boost, Intelino Smart Train and toio.
and [bluepy](https://github.com/IanHarvey/bluepy) to handle Bluetooth Low Energy
(BLE) connections. It has been reported that pyscrlink connects Scratch 3.0 with
micro:bit, LEGO WeDo, LEGO Boost and toio.
Until version v0.2.5, pyscrlink supported Bluetooth Classic protocol using
[pybluez](https://github.com/pybluez/pybluez). Unfortunately, pybluez is not
well maintained and caused technical troubles. Then Bluetooth Classic protocol
support is dropped from pyscrlink. This means that LEGO Mindstorm EV3 can not
be connected with pyscrlink. Bluetooth Classic support is the improvement
opportunity of pyscrlink.
To use websockets, pyscrlink requires python version 3.6 or later. If your
system has python older than version 3.6, install newer version. If your
@@ -39,7 +44,6 @@ Devices:
Linux distros:
* Arch Linux
* elementary OS 5.1 Hera
Browsers:
* Firefox
@@ -49,10 +53,8 @@ It was reported that pyscrlink (former bluepy-scratch-link) working with
following devices and Linux distros.
Devices:
* LEGO Mindstorm EV3 by @chrisglencross
* LEGO WeDo by @zhaowe, @KingBBQ
* LEGO Boost and compatible devices by @laurentchar, @miguev, @jacquesdt, @n3storm
* Intelino Smart Train by @ErrorJan
* toio by @shimodash
Linux distros:
@@ -89,7 +91,6 @@ Installation
```
4. Set bluepy-helper capability.
This step is required for most of devices except LEGO Mindstorm EV3.
```
$ bluepy_helper_cap
@@ -101,74 +102,25 @@ Installation
5. For micro:bit, install Scratch-link hex on your device.
* Download and unzip the [micro:bit Scratch Hex file](https://downloads.scratch.mit.edu/microbit/scratch-microbit-1.1.0.hex.zip).
* Download and unzip the [micro:bit Scratch Hex file](https://downloads.scratch.mit.edu/microbit/scratch-microbit.hex.zip).
* Flash the micro:bit over USB with the Scratch Hex File, you will see the
five character name of the micro:bit scroll across the screen such as
'zo9ev'.
Usage
-----
1. For LEGO Mindstorms EV3, pair your Linux PC to the EV3 brick.
First, turn on the EV3 and ensure Bluetooth is enabled.
Then, pair using your Linux desktop's the Bluetooth settings.
If using Gnome:
* Settings -> Bluetooth
* Click on the EV3 device name
* Accept the connection on EV3 brick
* Enter a matching PIN on EV3 brick and Linux PC. '1234' is the value Scratch suggests.
* Confirm EV3 status is "Disconnected" in Bluetooth settings
With a Raspberry Pi default Raspbian desktop, click the Bluetooth logo in the top right of the screen and
Add Device. Then follow the Gnome instructions. You will be warned that the Raspberry Pi
does not know how to talk to this device; that is not a problem.
Alternatively you can perform pairing from the command-line:
```shell script
$ bluetoothctl
[bluetooth]# power on
Changing power on succeeded
[bluetooth]# pairable on
Changing pairable on succeeded
[bluetooth]# agent KeyboardOnly
Agent registered
[bluetooth]# devices
...
Device 00:16:53:53:D3:19 EV3
...
[bluetooth]# pair 00:16:53:53:D3:19
Attempting to pair with 00:16:53:53:D3:19
# Confirm pairing on the EV3 display, set PIN to 1234
Request PIN code
[agent] Enter PIN code: 1234
[CHG] Device 00:16:53:53:D3:19 Connected: yes
[CHG] Device 00:16:53:53:D3:19 Paired: yes
Pairing successful
[bluetooth]# quit
```
2. Start scratch-link python script.
1. Start scratch-link python script.
```sh
$ scratch_link
```
If your device is toio, add "-s 1" option to the scratch_link command. It
allows the toio Do Visual Programming to connect to toio automatically.
3. Connect scratch to the target device such as micro:bit or LEGO Mindstorms:
2. Connect scratch to the target device such as micro:bit:
* Open FireFox or Chrome. (Make sure to run as the same user for scratch-link python script.)
* Access [Scratch 3.0](https://scratch.mit.edu/) and create your project.
* Select the "Add Extension" button.
* Select the extension for your device (e.g., micro:bit or Lego Mindstorms EV3 extension) and follow the prompts to connect.
* Select the extension for your device (e.g., micro:bit) and follow the prompts to connect.
* Build your project with the extension blocks.
In Case You Fail to Connect
@@ -189,10 +141,22 @@ In Case You Fail to Connect
* If the service is not working, refer guide of your distro to set it up.
* If the service is working, also check that /etc/bluetooth/main.conf sets AutoEnable=true.
3. If scratch_link.py says "failed to connect to BT device: [Errno 13] Permission denied",
3. If device scan still fails, use -r option to retry device scan.
The command line below does device scan twice. Each scan takes 10 seconds.
```
$ scratch_link -r 2
```
It would be good to use -s option together to reduce each scan duration.
The command line below does 3 seconds device scan twice.
```
$ scratch_link -r 2 -s 3
```
4. If scratch_link.py says "failed to connect to BT device: [Errno 13] Permission denied",
make sure to pair the bluetooth device to your PC before connecting to Scratch.
4. To connect to multiple devices at the same time, make all the target devices
5. To connect to multiple devices at the same time, make all the target devices
ready for scan at the first device scan. This is important for toio. The toio
allows a single project to connect to two toio devices.
* When the second device was prepared after the first device was connected, device scan can not find the second device.
@@ -206,6 +170,23 @@ Please file issues to [GitHub issue tracker](https://github.com/kawasaki/pyscrli
Releases
--------
Release 0.2.8
* Supported Microbit More v2
Release 0.2.7
* Supported Snap Firefox and Chromium
* Added -r option to retry BLE scan
Release 0.2.6
* Removed Bluetooth Classic and LEGO Mindstorm EV3 support
Release 0.2.5
* Fixed handling of multiple UUIDs for LEGO Boost
Release 0.2.4
* Added -s option to specify BLE scan duration

View File

@@ -152,36 +152,37 @@ def prep_nss_cert(dir, cert, nickname):
remove_cert(dir, nickname)
add_cert(dir, cert, nickname)
def prep_cert():
# Generate certification and key
gen_cert(cert_file_path, key_file_path)
# Add certificate to FireFox
def prep_cert_for_app(cert, app, search_path):
"""
Find a NSS DB in the search_path for the app and prepare the cert in the DB.
"""
nssdb = None
firefox_nss_path = os.path.join(homedir, ".mozilla/firefox/")
for root, dirs, files in os.walk(firefox_nss_path):
for root, dirs, files in os.walk(os.path.join(homedir, search_path)):
for name in files:
if not re.match("key.*\.db", name):
continue
nssdb = root
if prep_nss_cert(nssdb, cert_file_path, SCRATCH_CERT_NICKNAME):
logger.error(f"Failed to add certificate to FireFox NSS DB: {nssdb}")
if prep_nss_cert(nssdb, cert, SCRATCH_CERT_NICKNAME):
logger.error(f"Failed to add certificate to {app}: {nssdb}")
sys.exit(3)
else:
logger.info(f"Certificate is ready in FireFox NSS DB: {nssdb}")
logger.info(f"Certificate is ready in {app} NSS DB: {nssdb}")
if not nssdb:
logger.info("FireFox NSS DB not found. Do not add certificate.")
logger.debug(f"NSS DB for {app} not found. Do not add certificate.")
# Add certificate to Chrome
nssdb = os.path.join(homedir, ".pki/nssdb")
if os.path.isdir(nssdb):
if prep_nss_cert(nssdb, cert_file_path, SCRATCH_CERT_NICKNAME):
logger.error(f"Failed to add certificate to Chrome")
sys.exit(4)
else:
logger.info("Certificate is ready for Chrome")
else:
logger.info("Chrome NSS DB not found. Do not add certificate.")
def prep_cert():
# Generate certification and key
gen_cert(cert_file_path, key_file_path)
nss_dbs = {
"FireFox": ".mozilla/firefox/",
"FireFox(Snap)": "snap/firefox/common/.mozilla/firefox/",
"Chrome": ".pki",
"Chromium(Snap)": "snap/chromium",
}
[ prep_cert_for_app(cert_file_path, k, nss_dbs[k]) for k in nss_dbs ]
if __name__ == "__main__":
prep_cert()

View File

@@ -17,11 +17,8 @@ import signal
import traceback
import argparse
# for Bluetooth (e.g. Lego EV3)
import bluetooth
# for BLESession (e.g. BBC micro:bit)
from bluepy.btle import Scanner, UUID, Peripheral, DefaultDelegate
from bluepy.btle import Scanner, UUID, Peripheral, DefaultDelegate, ScanEntry
from bluepy.btle import BTLEDisconnectError, BTLEManagementError
from pyscrlink import bluepy_helper_cap
@@ -47,8 +44,6 @@ logger.propagate = False
HOSTNAME="device-manager.scratch.mit.edu"
scan_seconds=10.0
CCCD_UUID = 0x2902
class Session():
"""Base class for BTSession and BLESession"""
def __init__(self, websocket, loop):
@@ -140,204 +135,6 @@ class Session():
class BTSession(Session):
"""Manage a session for Bluetooth device"""
INITIAL = 1
DISCOVERY = 2
DISCOVERY_COMPLETE = 3
CONNECTED = 4
DONE = 5
# Split this into discovery thread and communication thread
# discovery thread should auto-terminate
class BTThread(threading.Thread):
"""
Separated thread to control notifications to Scratch.
It handles device discovery notification in DISCOVERY status
and notifications from bluetooth devices in CONNECTED status.
"""
class BTDiscoverer(bluetooth.DeviceDiscoverer):
def __init__(self, major_class, minor_class):
super().__init__()
self.major_class = major_class
self.minor_class = minor_class
self.found_devices = {}
self.done = False
def pre_inquiry(self):
self.done = False
def device_discovered(self, address, device_class, rssi, name):
logger.debug(f"Found device {name} addr={address} class={device_class} rssi={rssi}")
major_class = (device_class & 0x1F00) >> 8
minor_class = (device_class & 0xFF) >> 2
if "LEGO Hub" in name:
minor_class = 1
logger.info(f"Pretend to be LEGO EV3 with LEGO Hub: class={major_class/minor_class}")
if major_class == self.major_class and minor_class == self.minor_class:
self.found_devices[address] = (name, device_class, rssi)
def inquiry_complete(self):
self.done = True
def __init__(self, session, major_device_class, minor_device_class):
threading.Thread.__init__(self)
self.session = session
self.major_device_class = major_device_class
self.minor_device_class = minor_device_class
self.cancel_discovery = False
self.ping_time = None
def discover(self):
major = self.major_device_class
minor = self.minor_device_class
if major == 8 and minor == 1:
logger.info(f"Search LEGO Hub instead of LEGO EV3")
minor = 4
logger.debug(f"BT discover: class={major}/{minor}")
discoverer = self.BTDiscoverer(major, minor)
discoverer.find_devices(lookup_names=True)
while self.session.status == self.session.DISCOVERY and not discoverer.done and not self.cancel_discovery:
readable = select.select([discoverer], [], [], 0.5)[0]
if discoverer in readable:
discoverer.process_event()
for addr, (device_name, device_class, rssi) in discoverer.found_devices.items():
logger.debug(f"notifying discovered {addr}: {device_name}")
params = {"rssi": rssi, 'peripheralId': addr, 'name': device_name.decode("utf-8")}
self.session.notify('didDiscoverPeripheral', params)
discoverer.found_devices.clear()
if not discoverer.done:
discoverer.cancel_inquiry()
def run(self):
while self.session.status != self.session.DONE:
logger.debug(f"loop in BT thread: session status={self.session.status}")
current_time = int(round(time.time()))
if self.session.status == self.session.DISCOVERY and not self.cancel_discovery:
logger.debug("in discovery status:")
try:
self.discover()
self.ping_time = current_time + 5
finally:
self.session.status = self.session.DISCOVERY_COMPLETE
elif self.session.status == self.session.CONNECTED:
logger.debug("in connected status:")
sock = self.session.sock
try:
ready = select.select([sock], [], [], 1)
if ready[0]:
header = sock.recv(2)
[msg_len] = struct.unpack("<H", header)
msg_data = sock.recv(msg_len)
data = header + msg_data
params = {'message': base64.standard_b64encode(data).decode('utf-8'), "encoding": "base64"}
self.session.notify('didReceiveMessage', params)
self.ping_time = current_time + 5
except Exception as e:
logger.error(e)
self.session.close()
break
# To avoid repeated lock by this single thread,
# yield CPU to other lock waiting threads.
time.sleep(0)
else:
# Nothing to do:
time.sleep(1)
# Terminate if we have lost websocket connection to Scratch (e.g. browser closed)
if self.ping_time is None or self.ping_time <= current_time:
try:
self.session.notify('ping', {})
self.ping_time = current_time + 5
except Exception as e:
logger.error(e)
self.session.close()
break
def __init__(self, websocket, loop):
super().__init__(websocket, loop)
self.status = self.INITIAL
self.sock = None
self.bt_thread = None
def close(self):
self.status = self.DONE
if self.sock:
logger.info(f"disconnect to BT socket: {self.sock}")
self.sock.close()
self.sock = None
def __del__(self):
self.close()
def handle_request(self, method, params):
"""Handle requests from Scratch"""
logger.debug("handle request to BT device")
logger.debug(method)
if len(params) > 0:
logger.debug(params)
res = { "jsonrpc": "2.0" }
if self.status == self.INITIAL and method == 'discover':
logger.debug("Starting async discovery")
self.status = self.DISCOVERY
self.bt_thread = self.BTThread(self, params["majorDeviceClass"], params["minorDeviceClass"])
self.bt_thread.start()
res["result"] = None
elif self.status in [self.DISCOVERY, self.DISCOVERY_COMPLETE] and method == 'connect':
# Cancel discovery
while self.status == self.DISCOVERY:
logger.debug("Cancelling discovery")
self.bt_thread.cancel_discovery = True
time.sleep(1)
addr = params['peripheralId']
logger.debug(f"connecting to the BT device {addr}")
try:
self.sock = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
self.sock.connect((addr, 1))
logger.info(f"connected to BT device: {addr}")
except bluetooth.BluetoothError as e:
logger.error(f"failed to connect to BT device: {e}", exc_info=e)
self.status = self.DONE
self.sock = None
if self.sock:
res["result"] = None
self.status = self.CONNECTED
else:
err_msg = f"BT connect failed: {addr}"
res["error"] = { "message": err_msg }
self.status = self.DONE
elif self.status == self.CONNECTED and method == 'send':
logger.debug("handle send request")
if params['encoding'] != 'base64':
logger.error("encoding other than base 64 is not "
"yet supported: ", params['encoding'])
msg_bstr = params['message'].encode('ascii')
data = base64.standard_b64decode(msg_bstr)
self.sock.send(data)
res['result'] = len(data)
logger.debug(res)
return res
def end_request(self):
logger.debug(f"end_request of BTSession {self}")
return self.status == self.DONE
class BLESession(Session):
"""
Manage a session for Bluetooth Low Energy device such as micro:bit
@@ -364,14 +161,6 @@ class BLESession(Session):
scan_lock = threading.RLock()
scan_started = False
@staticmethod
def _getDevName(dev):
"""
Get AdType 0x09 (Completed local name). If it is not available,
get AdType 0x08 (Shortened local name).
"""
return dev.getValueText(0x9) or dev.getValueText(0x8)
class BLEThread(threading.Thread):
"""
Separated thread to control notifications to Scratch.
@@ -391,7 +180,7 @@ class BLESession(Session):
for d in devices:
params = { 'rssi': d.rssi }
params['peripheralId'] = devices.index(d)
params['name'] = BLESession._getDevName(d)
params['name'] = d.getValueText(0x9) or d.getValueText(0x8)
self.session.notify('didDiscoverPeripheral', params)
time.sleep(1)
elif self.session.status == self.session.CONNECTED:
@@ -408,8 +197,7 @@ class BLESession(Session):
logger.debug("after waitForNotification")
logger.debug("released lock for waitForNotification")
except Exception as e:
logger.error(f"Exception in waitForNotifications: "
f"{type(e).__name__}: {e}")
logger.error(e)
self.session.close()
break
else:
@@ -433,8 +221,7 @@ class BLESession(Session):
self.restart_notification_event.set()
def add_handle(self, serviceId, charId, handle):
logger.debug(f"add handle for notification: "
f"{serviceId} {charId} {handle}")
logger.debug(f"add handle for notification: {handle}")
params = { 'serviceId': UUID(serviceId).getCommonName(),
'characteristicId': charId,
'encoding': 'base64' }
@@ -442,14 +229,6 @@ class BLESession(Session):
def handleNotification(self, handle, data):
logger.debug(f"BLE notification: {handle} {data}")
if handle not in self.handles:
logger.error(f"Notification with unknown handle: {handle}")
keys = list(self.handles.keys())
if keys and len(keys) == 1:
logger.debug(f"Debug: override {handle} with {keys[0]}")
handle = keys[0]
else:
return
params = self.handles[handle].copy()
params['message'] = base64.standard_b64encode(data).decode('ascii')
self.session.notify('characteristicDidChange', params)
@@ -513,15 +292,19 @@ class BLESession(Session):
logger.debug("match...")
return True
if 'namePrefix' in f:
# 0x08: Shortened Local Name
deviceName = self._getDevName(dev)
if not deviceName:
continue
logger.debug(f"Name of \"{deviceName}\" begins with: \"{f['namePrefix']}\"?")
if(deviceName.startswith(f['namePrefix'])):
logger.debug("Yes")
return True
logger.debug("No")
logger.debug(f"given namePrefix: {f['namePrefix']}")
deviceName = dev.getValueText(ScanEntry.SHORT_LOCAL_NAME)
if deviceName:
logger.debug(f"SHORT_LOCAL_NAME: {deviceName}")
if deviceName.startswith(f['namePrefix']):
logger.debug(f"match...")
return True
deviceName = dev.getValueText(ScanEntry.COMPLETE_LOCAL_NAME)
if deviceName:
logger.debug(f"COMPLETE_LOCAL_NAME: {deviceName}")
if deviceName.startswith(f['namePrefix']):
logger.debug(f"match...")
return True
if 'name' in f or 'manufactureData' in f:
logger.error("name/manufactureData filters not implemented")
# TODO: implement other filters defined:
@@ -539,16 +322,23 @@ class BLESession(Session):
BLESession.found_devices.clear()
for i in range(self.MAX_SCANNER_IF):
scanner = Scanner(iface=i)
try:
logger.debug(f"start BLE scan: {scan_seconds} seconds")
devices = scanner.scan(scan_seconds)
for dev in devices:
if self.matches(dev, params['filters']):
BLESession.found_devices.append(dev)
found = True
logger.debug(f"BLE device found with iface #{i}");
except BTLEManagementError as e:
logger.debug(f"BLE iface #{i}: {e}");
for j in range(scan_retry):
try:
logger.debug(f"start BLE scan: {scan_seconds} seconds")
devices = scanner.scan(scan_seconds)
for dev in devices:
if self.matches(dev, params['filters']):
BLESession.found_devices.append(dev)
found = True
logger.debug(f"BLE device found with iface #{i}");
if found:
break
except BTLEDisconnectError as de:
logger.debug(f"BLE iface #{i}: {de}");
except BTLEManagementError as me:
logger.debug(f"BLE iface #{i}: {me}");
if found:
break
else:
found = len(BLESession.found_devices) > 0
return found
@@ -630,7 +420,7 @@ class BLESession(Session):
elif self.status == self.DISCOVERY and method == 'connect':
logger.debug("connecting to the BLE device")
self.device = BLESession.found_devices[params['peripheralId']]
self.deviceName = self._getDevName(self.device)
self.deviceName = self.device.getValueText(0x9) or self.device.getValueText(0x8)
try:
self.perip = Peripheral(self.device)
logger.info(f"connected to the BLE peripheral: {self.deviceName}")
@@ -707,19 +497,11 @@ class BLESession(Session):
service = self._get_service(service_id)
c = self._get_characteristic(chara_id)
handle = c.getHandle()
# get CCCD or Client Characterstic Configuration Descriptor
cccd = None
for d in c.getDescriptors():
if d.uuid == UUID(CCCD_UUID):
cccd = d
if not cccd:
logger.error("Characteristic {char_id} does not have CCCD")
return
# prepare notification handler
self.delegate.add_handle(service_id, chara_id, handle)
# request notification to the BLE device
with self.lock:
self.perip.writeCharacteristic(cccd.handle, value, True)
self.perip.writeCharacteristic(handle + 1, value, True)
def startNotifications(self, service_id, chara_id):
logger.debug(f"start notification for {chara_id}")
@@ -763,11 +545,14 @@ def stack_trace():
def main():
global scan_seconds
global scan_retry
parser = argparse.ArgumentParser(description='start Scratch-link')
parser.add_argument('-d', '--debug', action='store_true',
help='print debug messages')
parser.add_argument('-s', '--scan_seconds', type=float, default=10.0,
help='specifiy duration to scan BLE devices in seconds')
parser.add_argument('-r', '--scan_retry', type=int, default=1,
help='specifiy retry times to scan BLE devices')
args = parser.parse_args()
if args.debug:
print("Print debug messages")
@@ -775,7 +560,9 @@ def main():
handler.setLevel(logLevel)
logger.setLevel(logLevel)
scan_seconds = args.scan_seconds
scan_retry = args.scan_retry
logger.debug(f"set scan_seconds: {scan_seconds}")
logger.debug(f"set scan_retry: {scan_retry}")
# Prepare certificate of the WSS server
gencert.prep_cert()

View File

@@ -4,13 +4,17 @@ usage() {
echo "Usage: ${0} COMMAND"
echo -e "COMMAND:"
echo -e "\tbuild"
echo -e "\tupload FILES"
echo -e "\tupload-testpypi FILES"
echo -e "examples:"
echo -e "\t${0} build"
echo -e "\t${0} upload dist/*0.2.6*"
exit 1
}
case ${1} in
build)
python setup.py sdist bdist_wheel
python -m build
;;
upload)
shift

View File

@@ -5,7 +5,7 @@ with open("README.md", "r") as fh:
setuptools.setup(
name="pyscrlink",
version="0.2.4",
version="0.2.8",
author="Shin'ichiro Kawasaki",
author_email='kawasaki@juno.dti.ne.jp',
description='Scratch-link for Linux with Python',
@@ -22,7 +22,6 @@ setuptools.setup(
install_requires=[
'websockets',
'bluepy',
'pybluez',
'pyOpenSSL',
],
entry_points={