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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
[](https://codecov.io/gh/home-assistant-libs/aioshelly)
[](https://github.com/home-assistant-libs/aioshelly/actions/workflows/test.yml?query=branch%3Amain)
# Aioshelly
Asynchronous library to control Shelly devices
**This library is under development**
## Requirements
- Python >= 3.11
- bluetooth-data-tools
- aiohttp
- orjson
## Install
```bash
pip install aioshelly
```
## Install from Source
Run the following command inside this folder
```bash
pip install --upgrade .
```
## Install development requirements
Run the following command inside this folder
```bash
pip install .[dev] .[lint]
```
## Examples
### Gen1 Device (Block/CoAP) example:
```python
import asyncio
from pprint import pprint
import aiohttp
from aioshelly.block_device import COAP, BlockDevice
from aioshelly.common import ConnectionOptions
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError
async def test_block_device():
"""Test Gen1 Block (CoAP) based device."""
options = ConnectionOptions("192.168.1.165", "username", "password")
async with aiohttp.ClientSession() as aiohttp_session, COAP() as coap_context:
try:
device = await BlockDevice.create(aiohttp_session, coap_context, options)
except InvalidAuthError as err:
print(f"Invalid or missing authorization, error: {repr(err)}")
return
except DeviceConnectionError as err:
print(f"Error connecting to {options.ip_address}, error: {repr(err)}")
return
for block in device.blocks:
print(block)
pprint(block.current_values())
print()
if __name__ == "__main__":
asyncio.run(test_block_device())
```
### Gen2 and Gen3 (RPC/WebSocket) device example:
```python
import asyncio
from pprint import pprint
import aiohttp
from aioshelly.common import ConnectionOptions
from aioshelly.exceptions import DeviceConnectionError, InvalidAuthError
from aioshelly.rpc_device import RpcDevice, WsServer
async def test_rpc_device():
"""Test Gen2/Gen3 RPC (WebSocket) based device."""
options = ConnectionOptions("192.168.1.188", "username", "password")
ws_context = WsServer()
await ws_context.initialize(8123)
async with aiohttp.ClientSession() as aiohttp_session:
try:
device = await RpcDevice.create(aiohttp_session, ws_context, options)
except InvalidAuthError as err:
print(f"Invalid or missing authorization, error: {repr(err)}")
return
except DeviceConnectionError as err:
print(f"Error connecting to {options.ip_address}, error: {repr(err)}")
return
pprint(device.status)
if __name__ == "__main__":
asyncio.run(test_rpc_device())
```
## Example script
The repository includes example script to quickly try it out.
### Connect to a device and print its status whenever we receive a state change:
```
python3 tools/example.py -ip <ip> [-u <username>] [-p <password] -i
```
### Connect to all the devices in `devices.json` at once and print their status:
```
python3 tools/example.py -d -i
```
### Show usage help:
```
python3 tools/example.py -h
```
## Contribution guidelines
Object hierarchy and property/method names should match the [Shelly API](https://shelly-api-docs.shelly.cloud/).
|
