43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
import json
|
|
import asyncio
|
|
import socket
|
|
|
|
|
|
class WizDriver:
|
|
PORT = 38899
|
|
|
|
# Стандартные ID сцен WiZ
|
|
SCENES = {
|
|
"ocean": 1,
|
|
"romance": 2,
|
|
"party": 3,
|
|
"fireplace": 5,
|
|
"cozy": 6,
|
|
"forest": 10,
|
|
"bedtime": 13,
|
|
"warm_white": 33,
|
|
"daylight": 34,
|
|
}
|
|
|
|
async def send_udp(self, ip: str, payload: dict):
|
|
loop = asyncio.get_event_loop()
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(2.0)
|
|
data = json.dumps(payload).encode()
|
|
|
|
await loop.run_in_executor(None, sock.sendto, data, (ip, self.PORT))
|
|
|
|
try:
|
|
resp, _ = sock.recvfrom(1024)
|
|
return json.loads(resp.decode())
|
|
except socket.timeout:
|
|
return None
|
|
|
|
async def set_pilot(self, ip: str, params: dict):
|
|
payload = {"method": "setPilot", "params": params}
|
|
return await self.send_udp(ip, payload)
|
|
|
|
async def get_pilot(self, ip: str):
|
|
payload = {"method": "getPilot", "params": {}}
|
|
return await self.send_udp(ip, payload)
|