eculocate

commit 2448b243ac187b180588af824679bfdaa21e25e9
parent ad57be45af926241bc49a8c729e4cd088a6074fa
Author: Daniel Barlow <dan@telent.net>
Date:   Fri, 20 Feb 2026 00:41:28 +0000

ble client: extract classes

Diffstat:
M
client/ble.py
 | 
85
++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------

1 file changed, 52 insertions(+), 33 deletions(-)
diff --git a/client/ble.py b/client/ble.py
@@ -1,6 +1,3 @@
-import code
-import readline
-import rlcompleter
 import asyncio
 from bleak import BleakClient,BleakScanner
 import sys
@@ -12,10 +9,19 @@ from nacl.public import PrivateKey, Box, PublicKey
 from nacl.encoding import RawEncoder
 import sys
 
-readline.parse_and_bind("tab: complete")
 
 address = "FF:E4:05:1A:8F:FF"
 
+
+class Network():
+    def __init__(self, bytestr):
+        self.rssi = bytestr[0]
+        self.ssid = bytestr[1:7]
+        self.name = bytestr[8:]
+
+    def __str__(self):
+        return f"<Network {self.name} ssid:{self.ssid} rssi:{self.rssi}>"
+
 UUIDS = {
     'connecting':       "76be25c2-0801-44fc-a97c-bcb56c69c926",
     'max_index':        "76be25c2-0802-44fc-a97c-bcb56c69c926",
@@ -24,21 +30,16 @@ UUIDS = {
     'secret':           "76be25c2-0805-44fc-a97c-bcb56c69c926",
 }
 
-networks = []
-max_index = 0
 
-async def show_networks(client):
-    global max_index
-    global networks
-    max_index = ord(await client.read_gatt_char(UUIDS['max_index']))
-    print(f"{max_index} networks available")
-    networks = [None]
+class Encrypter():
+    def __init__(self, private_key, public_key):
+        self.sbox = Box(private_key, public_key)
 
-    for i in range(1, max_index + 1):
-        await client.write_gatt_char(UUIDS['current_index'], bytes([i]))
-        current_network = await client.read_gatt_char(UUIDS['current_network'])
-        networks.append(current_network)
-        print(f"{i}: {current_network}")
+    def encrypt(self, plaintext):
+        mybytes = self.sbox.encrypt(plaintext.encode('UTF-8'))
+        r = bytearray(mybytes)
+        r.insert(0, len(mybytes))
+        return r
 
 with open("keys/client.key", "rb") as f:
     secret = PrivateKey(f.read(), encoder=RawEncoder)
@@ -46,21 +47,35 @@ with open("keys/client.key", "rb") as f:
 with open("keys/device.pub", "rb") as f:
     rcpt = PublicKey(f.read(), encoder=RawEncoder)
 
-sbox = Box(secret, rcpt)
+ENCRYPTER = Encrypter(secret, rcpt)
+
+class Device():
+    def __init__(self, ble_client):
+        self.client = ble_client
+        self._networks = None
+        self.max_index = 0
+
+    async def networks(self):
+        if self._networks != None:
+            return self._networks
+
+        self.max_index = ord(await self.client.read_gatt_char(UUIDS['max_index']))
+        self._networks = []
 
-def encrypt_password(plaintext):
-    mybytes = sbox.encrypt(plaintext.encode('UTF-8'))
-    r = bytearray(mybytes)
-    r.insert(0, len(mybytes))
-    return r
+        for i in range(1, self.max_index + 1):
+            await self.client.write_gatt_char(UUIDS['current_index'], bytes([i]))
+            current_network = await self.client.read_gatt_char(UUIDS['current_network'])
+            self._networks.append(Network(current_network))
+        return self._networks
 
-async def choose_network(client, name):
-    buf = bytearray(name)
-    buf.insert(0, len(name))
-    secret = encrypt_password(os.getenv('PASSWORD'))
-    print(f"writing {len(buf)}, {len(secret)}")
-    await client.write_gatt_char(UUIDS['secret'], buf + secret)
-    await client.write_gatt_char(UUIDS['connecting'], bytes([1]))
+    async def choose_network(self, name, password):
+        client = self.client
+        secret = ENCRYPTER.encrypt(password)
+        buf = bytearray(name)
+        buf.insert(0, len(name))
+        print(f"writing {len(buf)}, {len(secret)}")
+        await client.write_gatt_char(UUIDS['secret'], buf + secret)
+        await client.write_gatt_char(UUIDS['connecting'], bytes([1]))
 
 
 async def main(address):
@@ -70,10 +85,14 @@ async def main(address):
     for d in devices:
         print(d)
 
-    async with BleakClient(address, timeout=30) as client:
-        await show_networks(client)
+    async with BleakClient(address, timeout=60) as client:
+        device = Device(client)
+        networks = await device.networks()
+        for n in networks:
+            print(f"network {n}")
         if preferred != None:
-            await choose_network(client, preferred.encode(encoding="utf-8"))
+            name_bytes = preferred.encode(encoding="utf-8")
+            await device.choose_network(name_bytes, os.getenv('PASSWORD'))
 
 
 asyncio.run(main(address))