From 0b1d49ccc8cc1e282fd997aa7ab9e494fd980f14 Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 19:00:33 +1100 Subject: [PATCH 1/8] Register BlueZ D-Bus pairing agent to fix AuthenticationCanceled on Linux --- README.md | 4 +- requirements.txt | 1 + retrieve_api_key.py | 151 ++++++++++++++++++++++++++++++-------------- 3 files changed, 109 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index 35dbd13..5f31ff1 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ This script uses your computer's Bluetooth to connect to your Powerpal. It authe - Your Powerpal's **MAC Address** (Format: `12:34:56:78:9A:BC`) - Your Powerpal's **6-digit Pairing Code** -> **Note:** The MAC address and pairing code can typically be found printed on the device itself, on the original packaging card, or mostly within the settings of **You can also find the MAC address within bluetooth settings on your phone or computer.** +> **Note:** The MAC address and pairing code can typically be found printed on the device itself, on the original packaging card, or within the settings of the Powerpal app. You can also find the MAC address within Bluetooth settings on your phone or computer. ## Installation @@ -53,6 +53,8 @@ Or you can provide your MAC address and Pairing code directly as command-line ar python retrieve_api_key.py 12:34:56:78:9A:BC 123456 ``` +> **Linux note:** The script automatically registers a BlueZ D-Bus pairing agent so that Bluetooth pairing completes without any OS prompts. On Windows an OS pairing dialog may appear on the first run — this is expected. + ### Script Execution 1. The script will pause and ask you to confirm that no other devices (like your phone) are currently connected to the Powerpal. Hit `Enter` to proceed. diff --git a/requirements.txt b/requirements.txt index b4069a4..fc5ce1c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ bleak requests +dbus-fast; sys_platform == "linux" diff --git a/retrieve_api_key.py b/retrieve_api_key.py index 5f5a87e..99b77a4 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -1,14 +1,67 @@ -from encodings import utf_8 import sys import asyncio import requests -from bleak import BleakClient, BleakScanner, BleakError +from bleak import BleakClient pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' powerpalSerialChar = '59DA0010-12F4-25A6-7D4F-55961DCE4205' +if sys.platform == 'linux': + from dbus_fast.aio import MessageBus + from dbus_fast.service import ServiceInterface, method + from dbus_fast import BusType + + AGENT_PATH = '/com/powerpalextractor/Agent' + + class PairingAgent(ServiceInterface): + """BlueZ pairing agent for Just Works bonding without user prompts. + + AuthorizeService and RequestAuthorization must return success or BlueZ + will abort the connection after bonding. + """ + + def __init__(self): + super().__init__('org.bluez.Agent1') + + @method() + def Release(self): + pass + + @method() + def RequestAuthorization(self, device: 'o'): # type: ignore[override] + pass + + @method() + def AuthorizeService(self, device: 'o', uuid: 's'): # type: ignore[override] + pass + + @method() + def Cancel(self): + pass + + + async def register_agent(bus: 'MessageBus') -> PairingAgent: + agent = PairingAgent() + bus.export(AGENT_PATH, agent) + introspection = await bus.introspect('org.bluez', '/org/bluez') + bluez_proxy = bus.get_proxy_object('org.bluez', '/org/bluez', introspection) + agent_mgr = bluez_proxy.get_interface('org.bluez.AgentManager1') + await agent_mgr.call_register_agent(AGENT_PATH, 'NoInputNoOutput') + await agent_mgr.call_request_default_agent(AGENT_PATH) + return agent + + + async def unregister_agent(bus: 'MessageBus'): + try: + introspection = await bus.introspect('org.bluez', '/org/bluez') + bluez_proxy = bus.get_proxy_object('org.bluez', '/org/bluez', introspection) + agent_mgr = bluez_proxy.get_interface('org.bluez.AgentManager1') + await agent_mgr.call_unregister_agent(AGENT_PATH) + except Exception: + pass + def convert_pairing_code(original_pairing_code): return int(original_pairing_code).to_bytes(4, byteorder='little') @@ -25,50 +78,56 @@ async def main(address, my_pairing_code): print("Pairing Code should be 6 digits...") input("Please confirm that you are NOT connected to the Powerpal via Bluetooth using any devices, and that bluetooth is enabled on your computer, and hit enter to continue...") - async with BleakClient(address) as client: - print(f"Connected: {client.is_connected}") - # Some Bleak backends (e.g. Windows WinRT) do not accept a pairing kind argument. - # Call pair() without arguments for maximum compatibility. - try: - paired = await client.pair() - print(f"Paired?: {paired}") - except TypeError: - # Older/newer Bleak versions may not support pair() on this backend at all. - # In that case, continue without explicit pairing and rely on connect(). - print("Pairing method not supported on this backend; continuing without explicit pair().") - - print(f"Authenticating with pairing_code: {my_pairing_code}, converted: {convert_pairing_code(my_pairing_code)}") - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - print('Auth Success\n') - - await asyncio.sleep(0.5) - - print(f"Attempting to retrieve Powerpal Serial Number...") - serial = await client.read_gatt_char(powerpalSerialChar) - formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() - print(f"Retrieved Device Serial: {formatted_serial}") - print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") - - print(f"Attempting to retrieve apikey (Powerpal UUID)...") - apikey = await client.read_gatt_char(powerpalUUIDChar) - joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) - formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() - print(f"Retrieved apikey: {formatted_apikey}\n") - - test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") - if test_endpoint.lower().startswith('y'): - url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" - headers = { 'Authorization': formatted_apikey } - - response = requests.request("GET", url, headers=headers) - if (response.status_code < 300): - print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") - print(response.text) - else: - print("Failure!") - print(f"{response.status_code} - {response.reason} - {response.text}") - else: - print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + # On Linux, register a BlueZ D-Bus pairing agent so that the passkey + # exchange during BLE bonding completes automatically without user prompts. + # On macOS/Windows the OS handles pairing through its own native dialogs. + bus = None + if sys.platform == 'linux': + bus = await MessageBus(bus_type=BusType.SYSTEM).connect() + await register_agent(bus) + + try: + # First connection: write pairing code, which triggers BLE pairing/bonding. + # The Powerpal requires a reconnect after bonding before reads are permitted. + async with BleakClient(address, timeout=30.0) as client: + print(f"Authenticating with pairing_code: {my_pairing_code}...") + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + print('Auth Success\n') + + # Second connection: bond is now established; write pairing code again to + # authenticate the session, then reads succeed. + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + + print(f"Attempting to retrieve Powerpal Serial Number...") + serial = await client.read_gatt_char(powerpalSerialChar) + formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() + print(f"Retrieved Device Serial: {formatted_serial}") + print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") + + print(f"Attempting to retrieve apikey (Powerpal UUID)...") + apikey = await client.read_gatt_char(powerpalUUIDChar) + joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) + formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() + print(f"Retrieved apikey: {formatted_apikey}\n") + + test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") + if test_endpoint.lower().startswith('y'): + url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" + headers = { 'Authorization': formatted_apikey } + response = requests.request("GET", url, headers=headers) + if (response.status_code < 300): + print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") + print(response.text) + else: + print("Failure!") + print(f"{response.status_code} - {response.reason} - {response.text}") + else: + print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + finally: + if bus is not None: + await unregister_agent(bus) + bus.disconnect() if __name__ == "__main__": asyncio.run(main((sys.argv[1] if len(sys.argv) >= 2 else None),(sys.argv[2] if len(sys.argv) == 3 else None))) \ No newline at end of file From 93f448a8805f49a2ee14794b4bebaa5ae34b70cc Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 21:50:45 +1100 Subject: [PATCH 2/8] Guard against transient BLE failures and bad reads BLE connections to the Powerpal can drop or return garbage data, especially while BlueZ is still settling after bonding. Without retries and validation the script silently produces wrong output or crashes on the first transient error. --- retrieve_api_key.py | 93 ++++++++++++++++++++++++++++++--------------- 1 file changed, 62 insertions(+), 31 deletions(-) diff --git a/retrieve_api_key.py b/retrieve_api_key.py index 99b77a4..c29b706 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -2,7 +2,7 @@ import asyncio import requests -from bleak import BleakClient +from bleak import BleakClient, BleakError pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' @@ -89,41 +89,72 @@ async def main(address, my_pairing_code): try: # First connection: write pairing code, which triggers BLE pairing/bonding. # The Powerpal requires a reconnect after bonding before reads are permitted. - async with BleakClient(address, timeout=30.0) as client: - print(f"Authenticating with pairing_code: {my_pairing_code}...") - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - print('Auth Success\n') + for attempt in range(1, 4): + try: + async with BleakClient(address, timeout=30.0) as client: + print(f"Authenticating with pairing_code: {my_pairing_code}...") + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + print('Auth Success\n') + break + except BleakError as e: + print(f"Pairing attempt {attempt} failed: {e}") + if attempt < 3: + await asyncio.sleep(3) + else: + raise + + # Give BlueZ a moment to settle after bonding before reconnecting. + await asyncio.sleep(2) # Second connection: bond is now established; write pairing code again to # authenticate the session, then reads succeed. - async with BleakClient(address, timeout=30.0) as client: - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - - print(f"Attempting to retrieve Powerpal Serial Number...") - serial = await client.read_gatt_char(powerpalSerialChar) - formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() - print(f"Retrieved Device Serial: {formatted_serial}") - print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") - - print(f"Attempting to retrieve apikey (Powerpal UUID)...") - apikey = await client.read_gatt_char(powerpalUUIDChar) - joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) - formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() - print(f"Retrieved apikey: {formatted_apikey}\n") - - test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") - if test_endpoint.lower().startswith('y'): - url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" - headers = { 'Authorization': formatted_apikey } - response = requests.request("GET", url, headers=headers) - if (response.status_code < 300): - print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") - print(response.text) + formatted_serial = None + formatted_apikey = None + for attempt in range(1, 4): + try: + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + + print(f"Attempting to retrieve Powerpal Serial Number...") + serial = await client.read_gatt_char(powerpalSerialChar) + if len(serial) != 4: + raise ValueError(f"Unexpected serial length: {len(serial)} bytes (expected 4)") + if not any(serial): + raise ValueError("Serial is all zeros") + formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() + print(f"Retrieved Device Serial: {formatted_serial}") + print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") + + print(f"Attempting to retrieve apikey (Powerpal UUID)...") + apikey = await client.read_gatt_char(powerpalUUIDChar) + if len(apikey) != 16: + raise ValueError(f"Unexpected apikey length: {len(apikey)} bytes (expected 16)") + if not any(apikey): + raise ValueError("API key is all zeros") + joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) + formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() + print(f"Retrieved apikey: {formatted_apikey}\n") + break + except (BleakError, ValueError) as e: + print(f"Read attempt {attempt} failed: {e}") + if attempt < 3: + await asyncio.sleep(3) else: - print("Failure!") - print(f"{response.status_code} - {response.reason} - {response.text}") + raise + + test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") + if test_endpoint.lower().startswith('y'): + url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" + headers = { 'Authorization': formatted_apikey } + response = requests.request("GET", url, headers=headers) + if (response.status_code < 300): + print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") + print(response.text) else: - print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + print("Failure!") + print(f"{response.status_code} - {response.reason} - {response.text}") + else: + print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') finally: if bus is not None: await unregister_agent(bus) From 3d8540b1b29ba60ae5727cefe308fce4ba08f25f Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 22:04:26 +1100 Subject: [PATCH 3/8] Auto-discover Powerpal via BLE scan when no address is given Manually finding the MAC address is a friction point for new users. Scanning by device name removes it entirely in the common case. --- retrieve_api_key.py | 27 +++++++++++++++++++++++---- 1 file changed, 23 insertions(+), 4 deletions(-) diff --git a/retrieve_api_key.py b/retrieve_api_key.py index c29b706..3408f7e 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -2,7 +2,7 @@ import asyncio import requests -from bleak import BleakClient, BleakError +from bleak import BleakClient, BleakError, BleakScanner pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' @@ -65,12 +65,31 @@ async def unregister_agent(bus: 'MessageBus'): def convert_pairing_code(original_pairing_code): return int(original_pairing_code).to_bytes(4, byteorder='little') +async def scan_for_powerpal() -> str: + """Scan repeatedly until a Powerpal is found, then return its address.""" + while True: + print("Scanning for Powerpal devices (5 s)...") + powerpals = [d for d in await BleakScanner.discover(timeout=5.0) + if d.name and d.name.startswith('Powerpal')] + if not powerpals: + print("No Powerpal devices found, retrying...") + elif len(powerpals) == 1: + print(f"Found: {powerpals[0].name} ({powerpals[0].address})") + return powerpals[0].address + else: + for i, d in enumerate(powerpals): + print(f" {i + 1}: {d.name} ({d.address})") + choice = input(f"Select device (1-{len(powerpals)}): ") + if choice.isdigit() and 1 <= int(choice) <= len(powerpals): + return powerpals[int(choice) - 1].address + async def main(address, my_pairing_code): - while address is None: - address = input("Your Powerpal MAC address: ") + if address is None: + address = await scan_for_powerpal() + else: if (address.count(':') != 5) or (len(address) != 17): - address = None print("Incorrect MAC address formatting, should look like -> 12:34:56:78:9A:BC") + address = await scan_for_powerpal() while my_pairing_code is None: my_pairing_code = int(input("Your Powerpal pairing code: ")) if not (0 <= my_pairing_code <= 999999): From bbe7c7c411ca4633a57a83dc7bc632b7d62cba87 Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 22:18:59 +1100 Subject: [PATCH 4/8] Use nice libraries to improve new user experience --- requirements.txt | 2 + retrieve_api_key.py | 141 ++++++++++++++++++++++++-------------------- 2 files changed, 78 insertions(+), 65 deletions(-) diff --git a/requirements.txt b/requirements.txt index fc5ce1c..e052451 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ bleak requests +rich +questionary dbus-fast; sys_platform == "linux" diff --git a/retrieve_api_key.py b/retrieve_api_key.py index 3408f7e..f688d8a 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -3,6 +3,12 @@ import requests from bleak import BleakClient, BleakError, BleakScanner +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +import questionary + +console = Console() pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' @@ -68,34 +74,36 @@ def convert_pairing_code(original_pairing_code): async def scan_for_powerpal() -> str: """Scan repeatedly until a Powerpal is found, then return its address.""" while True: - print("Scanning for Powerpal devices (5 s)...") - powerpals = [d for d in await BleakScanner.discover(timeout=5.0) - if d.name and d.name.startswith('Powerpal')] - if not powerpals: - print("No Powerpal devices found, retrying...") - elif len(powerpals) == 1: - print(f"Found: {powerpals[0].name} ({powerpals[0].address})") + with console.status("Scanning for Powerpal devices..."): + powerpals = [d for d in await BleakScanner.discover(timeout=5.0) + if d.name and d.name.startswith('Powerpal')] + if len(powerpals) == 1: + console.print(f"[green]Found:[/green] {powerpals[0].name} ({powerpals[0].address})") return powerpals[0].address - else: - for i, d in enumerate(powerpals): - print(f" {i + 1}: {d.name} ({d.address})") - choice = input(f"Select device (1-{len(powerpals)}): ") - if choice.isdigit() and 1 <= int(choice) <= len(powerpals): - return powerpals[int(choice) - 1].address + if len(powerpals) > 1: + choice = await questionary.select( + "Multiple Powerpal devices found. Select one:", + choices=[f"{d.name} ({d.address})" for d in powerpals], + ).ask_async() + if choice: + return next(d.address for d in powerpals if f"{d.name} ({d.address})" == choice) async def main(address, my_pairing_code): if address is None: address = await scan_for_powerpal() else: if (address.count(':') != 5) or (len(address) != 17): - print("Incorrect MAC address formatting, should look like -> 12:34:56:78:9A:BC") + console.print("[red]Incorrect MAC address format.[/red] Should look like 12:34:56:78:9A:BC") address = await scan_for_powerpal() - while my_pairing_code is None: - my_pairing_code = int(input("Your Powerpal pairing code: ")) - if not (0 <= my_pairing_code <= 999999): - my_pairing_code = None - print("Pairing Code should be 6 digits...") - input("Please confirm that you are NOT connected to the Powerpal via Bluetooth using any devices, and that bluetooth is enabled on your computer, and hit enter to continue...") + + if my_pairing_code is None: + raw = await questionary.text( + "Your Powerpal pairing code:", + validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", + ).ask_async() + if raw is None: + return + my_pairing_code = int(raw) # On Linux, register a BlueZ D-Bus pairing agent so that the passkey # exchange during BLE bonding completes automatically without user prompts. @@ -106,74 +114,77 @@ async def main(address, my_pairing_code): await register_agent(bus) try: - # First connection: write pairing code, which triggers BLE pairing/bonding. - # The Powerpal requires a reconnect after bonding before reads are permitted. for attempt in range(1, 4): try: - async with BleakClient(address, timeout=30.0) as client: - print(f"Authenticating with pairing_code: {my_pairing_code}...") - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - print('Auth Success\n') + with console.status(f"Pairing with {address}..."): + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + console.print(f"[green]✓[/green] Paired") break except BleakError as e: - print(f"Pairing attempt {attempt} failed: {e}") + console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") if attempt < 3: - await asyncio.sleep(3) + ok = await questionary.confirm( + "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", + default=True, + ).ask_async() + if not ok: + raise else: raise - # Give BlueZ a moment to settle after bonding before reconnecting. - await asyncio.sleep(2) + with console.status("Waiting for device to settle..."): + await asyncio.sleep(2) - # Second connection: bond is now established; write pairing code again to - # authenticate the session, then reads succeed. formatted_serial = None formatted_apikey = None for attempt in range(1, 4): try: - async with BleakClient(address, timeout=30.0) as client: - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - - print(f"Attempting to retrieve Powerpal Serial Number...") - serial = await client.read_gatt_char(powerpalSerialChar) - if len(serial) != 4: - raise ValueError(f"Unexpected serial length: {len(serial)} bytes (expected 4)") - if not any(serial): - raise ValueError("Serial is all zeros") - formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() - print(f"Retrieved Device Serial: {formatted_serial}") - print(f"Meaning your meter reading endpoint is: https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}\n") - - print(f"Attempting to retrieve apikey (Powerpal UUID)...") - apikey = await client.read_gatt_char(powerpalUUIDChar) - if len(apikey) != 16: - raise ValueError(f"Unexpected apikey length: {len(apikey)} bytes (expected 16)") - if not any(apikey): - raise ValueError("API key is all zeros") - joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) - formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() - print(f"Retrieved apikey: {formatted_apikey}\n") + with console.status("Authenticating and reading credentials..."): + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + serial = await client.read_gatt_char(powerpalSerialChar) + apikey = await client.read_gatt_char(powerpalUUIDChar) + if len(serial) != 4: + raise ValueError(f"Unexpected serial length: {len(serial)} bytes (expected 4)") + if not any(serial): + raise ValueError("Serial is all zeros") + formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() + if len(apikey) != 16: + raise ValueError(f"Unexpected apikey length: {len(apikey)} bytes (expected 16)") + if not any(apikey): + raise ValueError("API key is all zeros") + joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) + formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() break except (BleakError, ValueError) as e: - print(f"Read attempt {attempt} failed: {e}") + console.print(f"[yellow]Read attempt {attempt} failed:[/yellow] {e}") if attempt < 3: await asyncio.sleep(3) else: raise - test_endpoint = input(f"Would you like this script to validate your apikey by using it to make a request for your device information from https://readings.powerpal.net/api/v1/device/{formatted_serial}? (y/n): ") - if test_endpoint.lower().startswith('y'): - url = f"https://readings.powerpal.net/api/v1/device/{formatted_serial}" - headers = { 'Authorization': formatted_apikey } - response = requests.request("GET", url, headers=headers) + table = Table(show_header=False, box=None, padding=(0, 1)) + table.add_row("[bold]Serial[/bold] [dim](Device ID)[/dim]", formatted_serial) + table.add_row("[bold]API key[/bold] [dim](Authorisation Key)[/dim]", formatted_apikey) + table.add_row("[bold]Readings endpoint[/bold]", f"https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}") + console.print(Panel(table, title="[bold green]Powerpal Credentials[/bold green]", expand=False)) + + if await questionary.confirm( + f"Validate credentials by calling the Powerpal API for device {formatted_serial}?", + default=False, + ).ask_async(): + response = requests.get( + f"https://readings.powerpal.net/api/v1/device/{formatted_serial}", + headers={'Authorization': formatted_apikey}, + ) if (response.status_code < 300): - print("Success! Looks like your Device Serial and apikey were succesfully retrieved! Here is the device information returned from https://readings.powerpal.net/api/ :") - print(response.text) + console.print(f"[green]✓[/green] Credentials verified — your Device ID and Authorisation Key were successfully retrieved!") + console.print(Panel(response.text, title="[bold green]API Response[/bold green]")) else: - print("Failure!") - print(f"{response.status_code} - {response.reason} - {response.text}") + console.print(f"[red]Request failed:[/red] {response.status_code} {response.reason}\n{response.text}") else: - print(f'You may also confirm your apikey by using curl:\ncurl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + console.print(f'[dim]Tip:[/dim] curl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') finally: if bus is not None: await unregister_agent(bus) From 2be7c222c30e0453f9ef42bfb5ddbc6976b09453 Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 22:40:51 +1100 Subject: [PATCH 5/8] Validate pairing before stating paired --- retrieve_api_key.py | 48 ++++++++++++++++++++++++++++++--------------- 1 file changed, 32 insertions(+), 16 deletions(-) diff --git a/retrieve_api_key.py b/retrieve_api_key.py index f688d8a..5ad5252 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -114,24 +114,40 @@ async def main(address, my_pairing_code): await register_agent(bus) try: - for attempt in range(1, 4): - try: - with console.status(f"Pairing with {address}..."): - async with BleakClient(address, timeout=30.0) as client: - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + while True: + pairing_ok = False + for attempt in range(1, 4): + try: + with console.status(f"Pairing with {address}..."): + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) + await client.read_gatt_char(powerpalSerialChar) # validate code is correct console.print(f"[green]✓[/green] Paired") - break - except BleakError as e: - console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") - if attempt < 3: - ok = await questionary.confirm( - "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", - default=True, - ).ask_async() - if not ok: + pairing_ok = True + break + except BleakError as e: + if 'NotPermitted' in str(e): + console.print(f"[red]✗[/red] Incorrect pairing code.") + break + console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") + if attempt < 3: + ok = await questionary.confirm( + "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", + default=True, + ).ask_async() + if not ok: + return + else: raise - else: - raise + if pairing_ok: + break + raw = await questionary.text( + "Your Powerpal pairing code:", + validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", + ).ask_async() + if raw is None: + return + my_pairing_code = int(raw) with console.status("Waiting for device to settle..."): await asyncio.sleep(2) From 705cdb238b7fbdee361b08e947f829f35adcb4ad Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 23:15:41 +1100 Subject: [PATCH 6/8] Refactor to clean up the code --- retrieve_api_key.py | 260 ++++++++++++++++++++++++++------------------ 1 file changed, 156 insertions(+), 104 deletions(-) diff --git a/retrieve_api_key.py b/retrieve_api_key.py index 5ad5252..ad3ff0e 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -1,7 +1,10 @@ -import sys +from __future__ import annotations + import asyncio -import requests +import sys +from typing import Final +import requests from bleak import BleakClient, BleakError, BleakScanner from rich.console import Console from rich.panel import Panel @@ -10,9 +13,12 @@ console = Console() -pairingCodeChar = '59DA0011-12F4-25A6-7D4F-55961DCE4205' -powerpalUUIDChar ='59DA0009-12F4-25A6-7D4F-55961DCE4205' -powerpalSerialChar = '59DA0010-12F4-25A6-7D4F-55961DCE4205' +# Powerpal GATT characteristic UUIDs +PAIRING_CODE_CHAR: Final[str] = '59DA0011-12F4-25A6-7D4F-55961DCE4205' +POWERPAL_UUID_CHAR: Final[str] = '59DA0009-12F4-25A6-7D4F-55961DCE4205' +POWERPAL_SERIAL_CHAR: Final[str] = '59DA0010-12F4-25A6-7D4F-55961DCE4205' + +MAX_RETRIES: Final[int] = 3 if sys.platform == 'linux': from dbus_fast.aio import MessageBus @@ -28,7 +34,7 @@ class PairingAgent(ServiceInterface): will abort the connection after bonding. """ - def __init__(self): + def __init__(self) -> None: super().__init__('org.bluez.Agent1') @method() @@ -48,7 +54,7 @@ def Cancel(self): pass - async def register_agent(bus: 'MessageBus') -> PairingAgent: + async def register_agent(bus: 'MessageBus') -> 'PairingAgent': agent = PairingAgent() bus.export(AGENT_PATH, agent) introspection = await bus.introspect('org.bluez', '/org/bluez') @@ -59,7 +65,7 @@ async def register_agent(bus: 'MessageBus') -> PairingAgent: return agent - async def unregister_agent(bus: 'MessageBus'): + async def unregister_agent(bus: 'MessageBus') -> None: try: introspection = await bus.introspect('org.bluez', '/org/bluez') bluez_proxy = bus.get_proxy_object('org.bluez', '/org/bluez', introspection) @@ -68,8 +74,9 @@ async def unregister_agent(bus: 'MessageBus'): except Exception: pass -def convert_pairing_code(original_pairing_code): - return int(original_pairing_code).to_bytes(4, byteorder='little') +def convert_pairing_code(pairing_code: int) -> bytes: + return pairing_code.to_bytes(4, byteorder='little') + async def scan_for_powerpal() -> str: """Scan repeatedly until a Powerpal is found, then return its address.""" @@ -88,22 +95,132 @@ async def scan_for_powerpal() -> str: if choice: return next(d.address for d in powerpals if f"{d.name} ({d.address})" == choice) -async def main(address, my_pairing_code): + +async def get_pairing_code() -> int | None: + """Prompt the user for a valid 6-digit pairing code. Returns None if cancelled.""" + raw = await questionary.text( + "Your Powerpal pairing code:", + validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", + ).ask_async() + if raw is None: + return None + return int(raw) + + +async def pair_with_device(address: str, pairing_code: int) -> bool | None: + """Attempt to pair with the Powerpal. + + Returns: + True – paired successfully. + False – pairing code was wrong; caller should re-prompt. + None – user aborted the retry prompt; caller should exit. + Raises BleakError after exhausting MAX_RETRIES connection attempts. + """ + for attempt in range(1, MAX_RETRIES + 1): + try: + with console.status(f"Pairing with {address}..."): + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) + await client.read_gatt_char(POWERPAL_SERIAL_CHAR) # validate code is correct + console.print("[green]\u2713[/green] Paired") + return True + except BleakError as e: + if 'NotPermitted' in str(e): + console.print("[red]\u2717[/red] Incorrect pairing code.") + return False + console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") + if attempt < MAX_RETRIES: + ok = await questionary.confirm( + "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", + default=True, + ).ask_async() + if not ok: + return None # user chose to abort + else: + raise + return None # unreachable + + +def format_serial(data: bytes) -> str: + """Convert 4-byte little-endian serial to a lowercase hex string.""" + if len(data) != 4: + raise ValueError(f"Unexpected serial length: {len(data)} bytes (expected 4)") + if not any(data): + raise ValueError("Serial is all zeros") + return ''.join('{:02x}'.format(x) for x in reversed(data)) + + +def format_apikey(data: bytes) -> str: + """Convert 16-byte API key to lowercase UUID format string.""" + if len(data) != 16: + raise ValueError(f"Unexpected apikey length: {len(data)} bytes (expected 16)") + if not any(data): + raise ValueError("API key is all zeros") + h = ''.join('{:02x}'.format(x) for x in data) + return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:]}" + + +async def read_credentials(address: str, pairing_code: int) -> tuple[str, str]: + """Authenticate and read (serial, apikey) from the device, retrying up to MAX_RETRIES times.""" + for attempt in range(1, MAX_RETRIES + 1): + try: + with console.status("Authenticating and reading credentials..."): + async with BleakClient(address, timeout=30.0) as client: + await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) + serial_bytes = await client.read_gatt_char(POWERPAL_SERIAL_CHAR) + apikey_bytes = await client.read_gatt_char(POWERPAL_UUID_CHAR) + return format_serial(serial_bytes), format_apikey(apikey_bytes) + except (BleakError, ValueError) as e: + console.print(f"[yellow]Read attempt {attempt} failed:[/yellow] {e}") + if attempt < MAX_RETRIES: + await asyncio.sleep(3) + else: + raise + raise RuntimeError("unreachable") # satisfies type checker + + +def display_credentials(serial: str, apikey: str) -> None: + """Render the credentials panel to the console.""" + table = Table(show_header=False, box=None, padding=(0, 1)) + table.add_row("[bold]Serial[/bold] [dim](Device ID)[/dim]", serial) + table.add_row("[bold]API key[/bold] [dim](Authorisation Key)[/dim]", apikey) + table.add_row("[bold]Readings endpoint[/bold]", f"https://readings.powerpal.net/api/v1/meter_reading/{serial}") + console.print(Panel(table, title="[bold green]Powerpal Credentials[/bold green]", expand=False)) + + +async def validate_credentials(serial: str, apikey: str) -> None: + """Optionally validate credentials against the Powerpal API.""" + confirmed = await questionary.confirm( + f"Validate credentials by calling the Powerpal API for device {serial}?", + default=False, + ).ask_async() + if confirmed: + response = requests.get( + f"https://readings.powerpal.net/api/v1/device/{serial}", + headers={'Authorization': apikey}, + ) + if response.status_code < 300: + console.print("[green]\u2713[/green] Credentials verified \u2014 your Device ID and Authorisation Key were successfully retrieved!") + console.print(Panel(response.text, title="[bold green]API Response[/bold green]")) + else: + console.print(f"[red]Request failed:[/red] {response.status_code} {response.reason}\n{response.text}") + else: + console.print(f'[dim]Tip:[/dim] curl -H "Authorization: {apikey}" https://readings.powerpal.net/api/v1/device/{serial}') + + +async def main(address: str | None, pairing_code: int | None) -> None: + # Resolve device address if address is None: address = await scan_for_powerpal() - else: - if (address.count(':') != 5) or (len(address) != 17): - console.print("[red]Incorrect MAC address format.[/red] Should look like 12:34:56:78:9A:BC") - address = await scan_for_powerpal() - - if my_pairing_code is None: - raw = await questionary.text( - "Your Powerpal pairing code:", - validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", - ).ask_async() - if raw is None: + elif (address.count(':') != 5) or (len(address) != 17): + console.print("[red]Incorrect MAC address format.[/red] Should look like 12:34:56:78:9A:BC") + address = await scan_for_powerpal() + + # Resolve pairing code + if pairing_code is None: + pairing_code = await get_pairing_code() + if pairing_code is None: return - my_pairing_code = int(raw) # On Linux, register a BlueZ D-Bus pairing agent so that the passkey # exchange during BLE bonding completes automatically without user prompts. @@ -114,97 +231,32 @@ async def main(address, my_pairing_code): await register_agent(bus) try: + # Pairing loop — re-prompts for code on wrong-code errors while True: - pairing_ok = False - for attempt in range(1, 4): - try: - with console.status(f"Pairing with {address}..."): - async with BleakClient(address, timeout=30.0) as client: - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - await client.read_gatt_char(powerpalSerialChar) # validate code is correct - console.print(f"[green]✓[/green] Paired") - pairing_ok = True - break - except BleakError as e: - if 'NotPermitted' in str(e): - console.print(f"[red]✗[/red] Incorrect pairing code.") - break - console.print(f"[yellow]Pairing attempt {attempt} failed:[/yellow] {e}") - if attempt < 3: - ok = await questionary.confirm( - "Ensure Bluetooth is enabled and no other device is connected to the Powerpal. Ready to retry?", - default=True, - ).ask_async() - if not ok: - return - else: - raise - if pairing_ok: + result = await pair_with_device(address, pairing_code) + if result is True: break - raw = await questionary.text( - "Your Powerpal pairing code:", - validate=lambda v: True if (v.isdigit() and 0 <= int(v) <= 999999) else "Must be a 6-digit number", - ).ask_async() - if raw is None: + if result is None: + return # user aborted retry + # result is False: wrong code — ask again + pairing_code = await get_pairing_code() + if pairing_code is None: return - my_pairing_code = int(raw) with console.status("Waiting for device to settle..."): await asyncio.sleep(2) - formatted_serial = None - formatted_apikey = None - for attempt in range(1, 4): - try: - with console.status("Authenticating and reading credentials..."): - async with BleakClient(address, timeout=30.0) as client: - await client.write_gatt_char(pairingCodeChar, convert_pairing_code(my_pairing_code), response=False) - serial = await client.read_gatt_char(powerpalSerialChar) - apikey = await client.read_gatt_char(powerpalUUIDChar) - if len(serial) != 4: - raise ValueError(f"Unexpected serial length: {len(serial)} bytes (expected 4)") - if not any(serial): - raise ValueError("Serial is all zeros") - formatted_serial = ''.join('{:02x}'.format(x) for x in reversed(serial)).lower() - if len(apikey) != 16: - raise ValueError(f"Unexpected apikey length: {len(apikey)} bytes (expected 16)") - if not any(apikey): - raise ValueError("API key is all zeros") - joined_apikey = ''.join('{:02x}'.format(x) for x in apikey) - formatted_apikey = f"{joined_apikey[:8]}-{joined_apikey[8:12]}-{joined_apikey[12:16]}-{joined_apikey[16:20]}-{joined_apikey[20:]}".lower() - break - except (BleakError, ValueError) as e: - console.print(f"[yellow]Read attempt {attempt} failed:[/yellow] {e}") - if attempt < 3: - await asyncio.sleep(3) - else: - raise - - table = Table(show_header=False, box=None, padding=(0, 1)) - table.add_row("[bold]Serial[/bold] [dim](Device ID)[/dim]", formatted_serial) - table.add_row("[bold]API key[/bold] [dim](Authorisation Key)[/dim]", formatted_apikey) - table.add_row("[bold]Readings endpoint[/bold]", f"https://readings.powerpal.net/api/v1/meter_reading/{formatted_serial}") - console.print(Panel(table, title="[bold green]Powerpal Credentials[/bold green]", expand=False)) - - if await questionary.confirm( - f"Validate credentials by calling the Powerpal API for device {formatted_serial}?", - default=False, - ).ask_async(): - response = requests.get( - f"https://readings.powerpal.net/api/v1/device/{formatted_serial}", - headers={'Authorization': formatted_apikey}, - ) - if (response.status_code < 300): - console.print(f"[green]✓[/green] Credentials verified — your Device ID and Authorisation Key were successfully retrieved!") - console.print(Panel(response.text, title="[bold green]API Response[/bold green]")) - else: - console.print(f"[red]Request failed:[/red] {response.status_code} {response.reason}\n{response.text}") - else: - console.print(f'[dim]Tip:[/dim] curl -H "Authorization: {formatted_apikey}" https://readings.powerpal.net/api/v1/device/{formatted_serial}') + serial, apikey = await read_credentials(address, pairing_code) + display_credentials(serial, apikey) + await validate_credentials(serial, apikey) finally: if bus is not None: await unregister_agent(bus) bus.disconnect() + if __name__ == "__main__": - asyncio.run(main((sys.argv[1] if len(sys.argv) >= 2 else None),(sys.argv[2] if len(sys.argv) == 3 else None))) \ No newline at end of file + asyncio.run(main( + sys.argv[1] if len(sys.argv) >= 2 else None, + int(sys.argv[2]) if len(sys.argv) == 3 else None, + )) \ No newline at end of file From 591492f660e91514396ac489ec433f33f8cadb05 Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 23:21:06 +1100 Subject: [PATCH 7/8] Improve the CLI --- requirements.txt | 1 + retrieve_api_key.py | 56 ++++++++++++++++++++++++++++++--------------- 2 files changed, 38 insertions(+), 19 deletions(-) diff --git a/requirements.txt b/requirements.txt index e052451..9086ba0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ bleak requests rich questionary +typer dbus-fast; sys_platform == "linux" diff --git a/retrieve_api_key.py b/retrieve_api_key.py index ad3ff0e..39a13c2 100644 --- a/retrieve_api_key.py +++ b/retrieve_api_key.py @@ -5,6 +5,7 @@ from typing import Final import requests +import typer from bleak import BleakClient, BleakError, BleakScanner from rich.console import Console from rich.panel import Panel @@ -107,7 +108,7 @@ async def get_pairing_code() -> int | None: return int(raw) -async def pair_with_device(address: str, pairing_code: int) -> bool | None: +async def pair_with_device(device: str, pairing_code: int) -> bool | None: """Attempt to pair with the Powerpal. Returns: @@ -118,10 +119,10 @@ async def pair_with_device(address: str, pairing_code: int) -> bool | None: """ for attempt in range(1, MAX_RETRIES + 1): try: - with console.status(f"Pairing with {address}..."): - async with BleakClient(address, timeout=30.0) as client: + with console.status(f"Pairing with {device}..."): + async with BleakClient(device, timeout=30.0) as client: await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) - await client.read_gatt_char(POWERPAL_SERIAL_CHAR) # validate code is correct + await client.read_gatt_char(POWERPAL_SERIAL_CHAR) # validate pairing code is correct console.print("[green]\u2713[/green] Paired") return True except BleakError as e: @@ -160,12 +161,12 @@ def format_apikey(data: bytes) -> str: return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:]}" -async def read_credentials(address: str, pairing_code: int) -> tuple[str, str]: +async def read_credentials(device: str, pairing_code: int) -> tuple[str, str]: """Authenticate and read (serial, apikey) from the device, retrying up to MAX_RETRIES times.""" for attempt in range(1, MAX_RETRIES + 1): try: with console.status("Authenticating and reading credentials..."): - async with BleakClient(address, timeout=30.0) as client: + async with BleakClient(device, timeout=30.0) as client: await client.write_gatt_char(PAIRING_CODE_CHAR, convert_pairing_code(pairing_code), response=False) serial_bytes = await client.read_gatt_char(POWERPAL_SERIAL_CHAR) apikey_bytes = await client.read_gatt_char(POWERPAL_UUID_CHAR) @@ -208,13 +209,13 @@ async def validate_credentials(serial: str, apikey: str) -> None: console.print(f'[dim]Tip:[/dim] curl -H "Authorization: {apikey}" https://readings.powerpal.net/api/v1/device/{serial}') -async def main(address: str | None, pairing_code: int | None) -> None: +async def main(device: str | None, pairing_code: int | None) -> None: # Resolve device address - if address is None: - address = await scan_for_powerpal() - elif (address.count(':') != 5) or (len(address) != 17): + if device is None: + device = await scan_for_powerpal() + elif (device.count(':') != 5) or (len(device) != 17): console.print("[red]Incorrect MAC address format.[/red] Should look like 12:34:56:78:9A:BC") - address = await scan_for_powerpal() + device = await scan_for_powerpal() # Resolve pairing code if pairing_code is None: @@ -231,14 +232,14 @@ async def main(address: str | None, pairing_code: int | None) -> None: await register_agent(bus) try: - # Pairing loop — re-prompts for code on wrong-code errors + # Pairing loop — re-prompts for pairing code on wrong-code errors while True: - result = await pair_with_device(address, pairing_code) + result = await pair_with_device(device, pairing_code) if result is True: break if result is None: return # user aborted retry - # result is False: wrong code — ask again + # result is False: wrong pairing code — ask again pairing_code = await get_pairing_code() if pairing_code is None: return @@ -246,7 +247,7 @@ async def main(address: str | None, pairing_code: int | None) -> None: with console.status("Waiting for device to settle..."): await asyncio.sleep(2) - serial, apikey = await read_credentials(address, pairing_code) + serial, apikey = await read_credentials(device, pairing_code) display_credentials(serial, apikey) await validate_credentials(serial, apikey) finally: @@ -255,8 +256,25 @@ async def main(address: str | None, pairing_code: int | None) -> None: bus.disconnect() +app = typer.Typer(rich_markup_mode='rich') + + +@app.command() +def cli( + code: int | None = typer.Option( + default=None, + metavar="CODE", + help="6-digit pairing code from the card included with your Powerpal or the Powerpal app. If omitted, you will be prompted.", + ), + device: str | None = typer.Option( + default=None, + help="Bluetooth MAC address of the Powerpal (e.g. [bold]12:34:56:78:9A:BC[/bold]). " + "If omitted, the device will be located via scan.", + ), +) -> None: + """Extract the API key and serial number from a Powerpal energy monitor via Bluetooth.""" + asyncio.run(main(device, code)) + + if __name__ == "__main__": - asyncio.run(main( - sys.argv[1] if len(sys.argv) >= 2 else None, - int(sys.argv[2]) if len(sys.argv) == 3 else None, - )) \ No newline at end of file + app() \ No newline at end of file From 0faf98d3550ea21687afc697accd4e7fff7d6324 Mon Sep 17 00:00:00 2001 From: Adrian Cowan Date: Thu, 5 Mar 2026 23:55:28 +1100 Subject: [PATCH 8/8] Update readme to match new UX --- README.md | 84 +++++++++++++++++++++++++------------------------------ 1 file changed, 38 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index 5f31ff1..8183ba9 100644 --- a/README.md +++ b/README.md @@ -1,65 +1,57 @@ # Powerpal Auth Extractor -Because the powerpal mobile app is not open source, and does not provide a way to extract the API key and serial number from the device, this script was created to extract the API key and serial number from the device directly via Bluetooth Low Energy (BLE). +Extracts the **Device Serial** and **API Key** from a Powerpal energy monitor via Bluetooth, without needing the Powerpal app. Once you have these credentials you can query the Powerpal API directly, or use them with integrations like Home Assistant. -Once extracted, these credentials can be used to query your home's energy usage data straight from the official Powerpal API endpoints. Can be used with integrations like Home Assistant, Grafana, etc. +## Requirements -## What it does +- Python 3.10+ +- Bluetooth enabled on your computer +- Your Powerpal's **6-digit pairing code** (on the card that came with your Powerpal, or in the Powerpal app) -This script uses your computer's Bluetooth to connect to your Powerpal. It authenticates using your 6-digit pairing code, reads the specific GATT characteristics holding your Device Serial Number and API Key (UUID), and formats them for use. - -## Prerequisites - -- **Python 3.7+** -- A computer with **Bluetooth enabled** (Windows, macOS, or Linux) -- Your Powerpal's **MAC Address** (Format: `12:34:56:78:9A:BC`) -- Your Powerpal's **6-digit Pairing Code** - -> **Note:** The MAC address and pairing code can typically be found printed on the device itself, on the original packaging card, or within the settings of the Powerpal app. You can also find the MAC address within Bluetooth settings on your phone or computer. - -## Installation - -1. Clone or download this repository. -2. Install the required dependencies: +## Setup ```bash pip install -r requirements.txt ``` -## How to use it - -### IMPORTANT: When and How to Connect via Bluetooth - -**DO NOT** try to pair or connect to the Powerpal device through your computer's operating system Bluetooth menu. - -The Powerpal device can only maintain **one active Bluetooth connection at a time**. In order for this script to work, you must ensure the Powerpal is completely disconnected from any other devices. +## Usage -**Before running the script:** -1. Ensure your computer's Bluetooth hardware is **turned on**. -2. **Disconnect your phone** from the Powerpal device. The easiest way to do this is to fully close/kill the Powerpal mobile app on your smartphone, or temporarily turn off Bluetooth on your phone altogether. -3. The Python script will handle the Bluetooth connection directly. *make sure you are within range of the Powerpal device* - -### Running the script - -You can run the script and manually enter the information when prompted: +Close the Powerpal app on your phone (or disable phone Bluetooth) so the device is free to connect, then run: ```bash -python retrieve_api_key.py +$ python retrieve_api_key.py ``` -Or you can provide your MAC address and Pairing code directly as command-line arguments: - -```bash -python retrieve_api_key.py 12:34:56:78:9A:BC 123456 +``` +Found: Powerpal 000123ab (12:34:56:78:9A:BC) +? Your Powerpal pairing code: 123456 +✓ Paired +╭─────────────────────────────────── Powerpal Credentials ───────────────────────────────────╮ +│ Serial (Device ID) 000123ab │ +│ API key (Authorisation Key) 7e4a9f12-83cb-4d56-ae01-2f9b0c3d7e58 │ +│ Readings endpoint https://readings.powerpal.net/api/v1/meter_reading/000123ab │ +╰────────────────────────────────────────────────────────────────────────────────────────────╯ +? Validate credentials by calling the Powerpal API for device 000123ab? Yes +✓ Credentials verified — your Device ID and Authorisation Key were successfully retrieved! +╭──────────────────────────────────────── API Response ──────────────────────────────────────╮ +│ { │ +│ "serial_number": "000123ab", │ +│ "total_meter_reading_count": 312847, │ +│ "total_watt_hours": 2914830, │ +│ "total_cost": 1053.42, │ +│ ... │ +│ } │ +╰────────────────────────────────────────────────────────────────────────────────────────────╯ ``` -> **Linux note:** The script automatically registers a BlueZ D-Bus pairing agent so that Bluetooth pairing completes without any OS prompts. On Windows an OS pairing dialog may appear on the first run — this is expected. +You can pass arguments directly to skip the prompts: -### Script Execution +```bash +$ python retrieve_api_key.py --code 123456 +$ python retrieve_api_key.py --code 123456 --device 12:34:56:78:9A:BC +``` -1. The script will pause and ask you to confirm that no other devices (like your phone) are currently connected to the Powerpal. Hit `Enter` to proceed. -2. It attempts to connect to the Powerpal via BLE and authenticates with your pairing code. -3. It grabs your **Device Serial**. -4. It grabs your **API Key (UUID)**. -5. It will finally ask `(y/n)` if you'd like to test the credentials against `https://readings.powerpal.net/api/v1/device/{serial}`. -6. If the request succeeds, your Powerpal's configuration metadata will be printed to the console, and you now have the required keys to build your own integrations (like Home Assistant, Grafana, etc.). +| Option | Description | +| --- | --- | +| `--code` | 6-digit pairing code | +| `--device` | Bluetooth MAC address — skips scanning if you already know it |