use http api
This commit is contained in:
@@ -1,64 +1,74 @@
|
|||||||
import time
|
import requests
|
||||||
|
import random
|
||||||
|
import string
|
||||||
import json
|
import json
|
||||||
import socket
|
|
||||||
|
|
||||||
GOVEE_API_BASE_URL = "https://openapi.api.govee.com/router/api/v1/"
|
GOVEE_API_BASE_URL = "https://openapi.api.govee.com/router/api/v1/"
|
||||||
|
|
||||||
class GoveeApi:
|
class GoveeApi:
|
||||||
device_ip = ""
|
device_ip = ""
|
||||||
|
|
||||||
def __init__(self, device_ip):
|
def __init__(self, key):
|
||||||
self.device_ip = device_ip
|
self.key = key
|
||||||
|
|
||||||
def send_scene(self, scene_code):
|
def get_random_string(self):
|
||||||
power_payload = {
|
# Choose from all lowercase, uppercase letters, and digits
|
||||||
"msg": {
|
characters = string.ascii_letters + string.digits
|
||||||
"cmd": "turn",
|
result_str = ''.join(random.choices(characters, k=4))
|
||||||
"data": {
|
return result_str
|
||||||
"value": 1 # 1 = on, 0 = off
|
|
||||||
}
|
def get_diy_scenes(self, sku, device):
|
||||||
}
|
headers = {
|
||||||
|
'Govee-API-Key': self.key
|
||||||
}
|
}
|
||||||
self.send_over_socket(power_payload)
|
|
||||||
time.sleep(0.5)
|
|
||||||
|
|
||||||
segments = scene_code.split(",")
|
|
||||||
for segment in segments:
|
|
||||||
payload = {
|
|
||||||
"msg": {
|
|
||||||
"cmd": "ptReal",
|
|
||||||
"data": {
|
|
||||||
"command": [segment]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
self.send_over_socket(payload)
|
|
||||||
|
|
||||||
def set_to_original_color(self):
|
|
||||||
payload = {
|
payload = {
|
||||||
"msg": {
|
"requestId": self.get_random_string(),
|
||||||
"cmd": "colorwc",
|
"payload": {
|
||||||
"data": {
|
"sku": sku,
|
||||||
"color": {"r": 255, "g": 0, "b": 0},
|
"device": device
|
||||||
"colorTemInKelvin": 0
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
res = requests.post(GOVEE_API_BASE_URL + 'device/diy-scenes', json=payload, headers=headers)
|
||||||
|
res.raise_for_status()
|
||||||
|
print(json.dumps(res.json(), indent=4))
|
||||||
|
|
||||||
|
def set_diy_scene(self, sku, device):
|
||||||
|
headers = {
|
||||||
|
'Govee-API-Key': self.key
|
||||||
|
}
|
||||||
|
payload = {
|
||||||
|
'requestId': self.get_random_string(),
|
||||||
|
'payload': {
|
||||||
|
'sku': sku,
|
||||||
|
'device': device,
|
||||||
|
'capability': {
|
||||||
|
'type': 'device.capabilities.dynamic_scene',
|
||||||
|
'instance': 'diyScene',
|
||||||
|
'value': 22757907
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
self.send_over_socket(payload)
|
|
||||||
|
req = requests.post(GOVEE_API_BASE_URL + 'device/control', headers=headers, json=payload)
|
||||||
def send_over_socket(self, payload, retries=2, delay=0.1):
|
req.raise_for_status()
|
||||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
||||||
sock.settimeout(1)
|
def set_to_original_color(self, sku, device):
|
||||||
data = json.dumps(payload).encode()
|
headers = {
|
||||||
|
'Govee-API-Key': self.key
|
||||||
for attempt in range(retries):
|
}
|
||||||
sock.sendto(data, (self.device_ip, 4003))
|
payload = {
|
||||||
try:
|
'requestId': self.get_random_string(),
|
||||||
response, _ = sock.recvfrom(1024)
|
'payload': {
|
||||||
print(f"Got response on attempt {attempt + 1}")
|
'sku': sku,
|
||||||
break # stop retrying once we get an ack
|
'device': device,
|
||||||
except socket.timeout:
|
'capability': {
|
||||||
if attempt < retries - 1:
|
'type': 'devices.capabilities.color_setting',
|
||||||
time.sleep(delay)
|
'instance': 'colorRgb',
|
||||||
|
'value': 16711680
|
||||||
sock.close()
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
req = requests.post(GOVEE_API_BASE_URL + 'device/control', json=payload, headers=headers)
|
||||||
|
req.raise_for_status()
|
||||||
+1
-4
@@ -44,14 +44,12 @@ LOGO_DIR = "/usr/local/share/logos"
|
|||||||
SABRES_ABBR = "BUF"
|
SABRES_ABBR = "BUF"
|
||||||
GOVEE_DEVICE = "3D:22:D7:94:40:46:2F:72"
|
GOVEE_DEVICE = "3D:22:D7:94:40:46:2F:72"
|
||||||
GOVEE_SKU = "H6168"
|
GOVEE_SKU = "H6168"
|
||||||
GOVEE_AWS = "owABAgT/AGQMACr//+YAADb//8k=,o//QAAIDAwAAAAAAAAAAAAAAAI4=,MwUKdwAAAAAAAAAAAAAAAAAAAEs="
|
|
||||||
GOVEE_IP = "172.16.0.15"
|
|
||||||
|
|
||||||
# --- Logo cache ---
|
# --- Logo cache ---
|
||||||
logo_cache = {}
|
logo_cache = {}
|
||||||
|
|
||||||
# --- Govee API ---
|
# --- Govee API ---
|
||||||
govee_api = govee.GoveeApi(device_ip=GOVEE_IP)
|
govee_api = govee.GoveeApi(key="")
|
||||||
|
|
||||||
def render_goal_frame(text, text_scale, bg_color, text_color):
|
def render_goal_frame(text, text_scale, bg_color, text_color):
|
||||||
big_h = max(8, int(32 * text_scale))
|
big_h = max(8, int(32 * text_scale))
|
||||||
@@ -293,7 +291,6 @@ def run():
|
|||||||
while True:
|
while True:
|
||||||
play_goal_celebration()
|
play_goal_celebration()
|
||||||
govee_api.send_scene(GOVEE_AWS)
|
govee_api.send_scene(GOVEE_AWS)
|
||||||
govee_api.set_to_original_color()
|
|
||||||
|
|
||||||
# while True:
|
# while True:
|
||||||
# now = time.time()
|
# now = time.time()
|
||||||
|
|||||||
Reference in New Issue
Block a user