-
Notifications
You must be signed in to change notification settings - Fork 1
/
read_api_v2.py
137 lines (113 loc) · 4.35 KB
/
read_api_v2.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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""
Access API gen2 of Shelly device (Plus Plug S) using digest auth.
see https://shelly-api-docs.shelly.cloud/gen2/General/Authentication
"""
import hashlib
import json
import random
import sys
import requests
from credentials import password, username
from credentials import shelly2_ip as shelly_ip
# import time
# public endpoint with no auth required
# api_url = f"http://{shelly_ip}/shelly"
URL_SHELLY = f"http://{shelly_ip}/rpc"
method = "Switch.GetStatus"
payload_401 = {
"id": 1,
"method": method,
}
def extract_data_from_401(response_header: dict[str, str]) -> dict[str, str]:
"""
Extract data from Shelly 401 response and convert to dict.
"""
data_401: dict[str, str] = {}
s = response_header["WWW-Authenticate"]
s = s.replace("Digest qop", "qop")
# remove " from values
s = s.replace('"', "")
# extract key-value pairs of strings
for key_value in s.split(", "):
(key, value) = key_value.split("=")
data_401[key] = value
return data_401
# 1. request without auth, get a onetime-no and http 401
try:
response = requests.post(
URL_SHELLY,
timeout=3,
json=payload_401,
# data=json.dumps(payload_401),
)
if response.status_code == 401: # noqa: PLR2004
# print(response.headers)
data_401 = extract_data_from_401(dict(response.headers))
else:
print(
f"Failed to access the API. Status code: {response.status_code}, text: {response.text}", # noqa: E501
)
sys.exit()
# print(data_401)
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e!s}")
sys.exit()
except Exception as e: # noqa: BLE001
print(f"An error occurred: {e!s}")
sys.exit()
# 2. request via digest auth
try:
auth_parts = [username, data_401["realm"], password]
# Concatenate the auth_parts with ':' and compute the SHA-256 hash
ha1 = hashlib.sha256(":".join(auth_parts).encode()).hexdigest()
ha2 = hashlib.sha256(b"dummy_method:dummy_uri").hexdigest()
# print(ha1)
# print(ha2)
# cnonce = str(int(time.time()))
cnonce = str(random.randint(1000000, 9999999)) # noqa: S311
nc = "1" # number, nonce counter (returned only through websocket channel).
# It has value of 1 if it is not available in the response
s = ":".join((ha1, data_401["nonce"], nc, cnonce, "auth", ha2))
resp = hashlib.sha256(s.encode()).hexdigest()
d = {
"id": 1,
"method": method,
"params": {"id": 0}, # 0 = first switch/meter
"auth": {
"realm": data_401["realm"],
"username": username,
"nonce": data_401["nonce"],
"cnonce": cnonce,
"response": resp,
"algorithm": "SHA-256",
},
}
response = requests.post(f"http://{shelly_ip}/rpc", json=d, timeout=3)
res = json.loads(response.text)
if response.status_code == 200: # noqa: PLR2004
data = json.loads(response.text)
data = data["result"]
print(data)
# extract and convert relevant data
# api spec: Last measured instantaneous active power (in Watts) delivered to the attached load # noqa: E501
watt_now = float(data["apower"])
# api spec: Total energy consumed in Watt-hours
kWh_total = round(float(data["aenergy"]["total"] / 1000), 3) # noqa: N816
# api spec: Energy consumption by minute (in Milliwatt-hours) for the last three minutes # noqa: E501
past_minutes = [float(x) for x in data["aenergy"]["by_minute"]]
# convert to avg watt per min
watt_past_minutes = [round(x * 60 / 1000, 1) for x in past_minutes]
# api spec: Unix timestamp of the first second of the last minute
# TM: No, actually it is the current timestamp, not the timestamp related to past counters! # noqa: E501
timestamp = int(data["aenergy"]["minute_ts"])
# api spec: Temperature in Celsius (null if temperature is out of the measurement range) # noqa: E501
temp = float(data["temperature"]["tC"])
else:
print(
f"Error: Failed to access the API. Status code: {response.status_code}, text: {response.text}", # noqa: E501
)
sys.exit()
except requests.exceptions.RequestException as e:
print(f"Error in Request: {e!s}")
except Exception as e: # noqa: BLE001
print(f"Error: {e!s}")