-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
112 lines (86 loc) · 3.02 KB
/
main.py
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
112
import argparse
import asyncio
import logging
from datetime import datetime, timedelta
from uuid import UUID
from bleak import BleakClient
from amazfit_pyclient.chunked_encoder import ChunkedDecoder, ChunkedEncoder
from amazfit_pyclient.chunked_encoder.handlers import (
AuthHandler,
BatteryClient,
ConnectionClient,
HeartRateClient,
HttpClient,
LogsClient,
StepsClient,
)
from amazfit_pyclient.fetch import FetchActivity
CHARACTERISTIC_HR = UUID("00002a37-0000-1000-8000-00805f9b34fb")
def print_chars(client: BleakClient):
for service in client.services:
print(f"{service.uuid}: {service.description}")
for char in service.characteristics:
print(f"\t{char.uuid}: {char.description}")
async def main(address: str, key: str):
disconnect_event = asyncio.Event()
async with BleakClient(
address,
disconnected_callback=lambda x: disconnect_event.set(),
timeout=30,
) as client:
decoder = ChunkedDecoder(client)
encoder = ChunkedEncoder(client)
eh = AuthHandler(key, encoder, decoder)
http = HttpClient(encoder, decoder)
hr = HeartRateClient(encoder, decoder)
steps = StepsClient(encoder, decoder)
conn = ConnectionClient(encoder, decoder)
logs = LogsClient(encoder, decoder)
battery = BatteryClient(encoder, decoder)
await decoder.start_notify()
await eh.autenticate()
# await steps.get_steps()
await conn.get_mtu()
# await battery.request_status()
# print_chars(client)
await notify_hr(client)
# df = FetchActivity(client)
# await df.start(
# since=datetime.now().astimezone() - timedelta(days=1),
# )
# async with df.in_progress:
# pass
# df = FetchTemperature(client)
# await df.start(since=datetime.now().astimezone() - timedelta(days=1))
# await client.disconnect()
await disconnect_event.wait()
print("Disconnected")
async def notify_hr(client):
def heart_rate(char, data: bytearray):
print(f"Heart rate: {int.from_bytes(data)}")
heart_rate_measurement_char = client.services.get_characteristic(CHARACTERISTIC_HR)
await client.start_notify(heart_rate_measurement_char, heart_rate)
if __name__ == "__main__":
logging.getLogger("bleak").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%H:%M:%S",
)
parser = argparse.ArgumentParser(description="Python amazfit client")
parser.add_argument(
"-a",
"--address",
required=True,
help="Bluetooth address",
)
parser.add_argument(
"-k",
"--key",
required=True,
help="Bluetooth key",
)
args = parser.parse_args()
asyncio.run(main(address=args.address, key=args.key))