pyboneco/example.py at main · DeKaN/pyboneco · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
import asyncio
import json
import logging

from aiohttp import ClientSession
from locale import getdefaultlocale
from pyboneco import BonecoAuth, BonecoClient, BonecoAuthState, check_firmware_update

logging.basicConfig(level=logging.INFO)


async def actions(auth: BonecoAuth):
    boneco_client = BonecoClient(auth)
    try:
        await boneco_client.connect()
        name = await boneco_client.get_device_name()
        print(f"User's device name is {name}")
        data = json.dumps(auth.save())
        print(f"Device json data: {data}")
        info = await boneco_client.get_device_info()
        print(f"Device info: {vars(info)}")
        state = await boneco_client.get_state()
        print(f"Device state: {vars(state)}")
        async with ClientSession() as websession:
            locale = getdefaultlocale()[0]
            country_code = locale.split("_")[1].lower() if locale is not None else "us"
            result, url = await check_firmware_update(
                websession,
                info.device.product_name,
                info.serial_number,
                info.software_version,
                country_code,
            )
            print(f"Check result={result}. url='{url}'")
    except Exception as e:
        print(e)
    finally:
        await boneco_client.disconnect()


def auth_state_callback(auth: BonecoAuth) -> None:
    print(
        f"Got new auth state: current={auth.current_state}, level={auth.current_auth_level}"
    )
    if auth.current_state == BonecoAuthState.CONFIRM_WAITING:
        print("Press button on device to confirm pairing")


async def find_device(address: str):
    scanned = await BonecoClient.find_boneco_devices()
    chosen = next((x for x in scanned.keys() if x.address == address), None)
    return chosen, scanned[chosen]


async def pair():
    scanned = await BonecoClient.find_boneco_devices()
    devices = list(scanned.keys())
    devices_text = "\n".join(
        [
            f"{n}) {value} (Pairing active = {scanned[value].pairing_active})"
            for n, value in enumerate(devices, start=1)
        ]
    )
    print(f"Scan results: \n{devices_text}\n")
    number = input(f"Choose device to pair [1-{len(scanned)}]: ")
    device = devices[int(number) - 1]
    advertisement = scanned[device]
    pairing_active = advertisement.pairing_active
    print(
        f'Chosen device "{device.name}" with address "{device.address}". Pairing active = {pairing_active}'
    )
    while not pairing_active:
        print("Put the device in pairing mode and press Enter")
        input()
        device, advertisement = await find_device(device.address)
        pairing_active = device and advertisement.pairing_active

    auth_data = BonecoAuth(device)
    auth_data.set_auth_state_callback(auth_state_callback)

    await actions(auth_data)


async def connect():
    print("Enter device json data")
    data = json.loads(input())
    device, advertisement = await find_device(data["address"])
    auth_data = BonecoAuth(device, data["key"])
    await actions(auth_data)


async def menu():
    while True:
        choosed = input(
            "\nChoose between (1) pairing new device and (2) connecting existing device: "
        )
        match choosed:
            case "1":
                await pair()
                break
            case "2":
                await connect()
                break
            case _:
                print("Not supported")
    print("Press Enter for exit")
    input()


if __name__ == "__main__":
    asyncio.run(menu())