parent
dbed99b959
commit
a582539b41
@ -0,0 +1,14 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
|
||||||
|
class HueBridge():
|
||||||
|
def __init__(self, ipaddress: str, username: str):
|
||||||
|
self.mipaddress: str = ipaddress
|
||||||
|
self.musername: str = username
|
||||||
|
|
||||||
|
def get_ipaddress(self):
|
||||||
|
return self.mipaddress
|
||||||
|
|
||||||
|
def get_user(self):
|
||||||
|
return self.musername
|
@ -0,0 +1,26 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
|
||||||
|
class UserOrError:
|
||||||
|
def __init__(self):
|
||||||
|
self.muser: str = str()
|
||||||
|
self.merror: bool = True
|
||||||
|
self.mcode: int = 0
|
||||||
|
|
||||||
|
def get_error_code(self) -> int:
|
||||||
|
return self.mcode
|
||||||
|
|
||||||
|
def get_user(self) -> str:
|
||||||
|
return self.muser
|
||||||
|
|
||||||
|
def is_error(self) -> bool:
|
||||||
|
return self.merror
|
||||||
|
|
||||||
|
def set_error(self, code: int):
|
||||||
|
self.merror = True
|
||||||
|
self.mcode = code
|
||||||
|
|
||||||
|
def set_user(self, username: str):
|
||||||
|
self.merror = False
|
||||||
|
self.muser = username
|
@ -0,0 +1,89 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import time
|
||||||
|
import uuid
|
||||||
|
|
||||||
|
import requests
|
||||||
|
import toml
|
||||||
|
from upnpy import UPnP
|
||||||
|
|
||||||
|
import tinge
|
||||||
|
|
||||||
|
|
||||||
|
def connect(ipaddress: str, appid: uuid.UUID = uuid.uuid4()) -> UserOrError:
|
||||||
|
user_or_error = UserOrError()
|
||||||
|
body: dict = json.loads('{"devicetype":"{}#{}"}'.format("tinge", appid))
|
||||||
|
path: str = "api"
|
||||||
|
method: str = "POST"
|
||||||
|
response: requests.Response = make_request(ipaddress, path, method, body)
|
||||||
|
|
||||||
|
data: dict = response.json()[0]
|
||||||
|
if 'error' in data.keys():
|
||||||
|
user_or_error.set_error(data['error']['type'])
|
||||||
|
elif 'success' in data.keys():
|
||||||
|
user_or_error.set_user(data['success']['username'])
|
||||||
|
else:
|
||||||
|
user_or_error.set_error(9999)
|
||||||
|
return user_or_error
|
||||||
|
|
||||||
|
|
||||||
|
def is_valid_config(filename: str) -> bool:
|
||||||
|
return os.path.exists(filename) and os.path.getsize(filename) > 0
|
||||||
|
|
||||||
|
|
||||||
|
def make_request(ipaddress: str, path: str, method: str = "GET",
|
||||||
|
body: dict = json.loads('{}')) -> requests.Response:
|
||||||
|
rfunct = requests.get
|
||||||
|
url = "http://{}/{}".format(ipaddress, path)
|
||||||
|
if method == "PUT":
|
||||||
|
rfunct = requests.put
|
||||||
|
elif method == "POST":
|
||||||
|
rfunct = requests.post
|
||||||
|
elif method == "DELETE":
|
||||||
|
rfunct = requests.delete
|
||||||
|
response: requests.Response = requests.Response()
|
||||||
|
if body:
|
||||||
|
response = rfunct(url, data=body)
|
||||||
|
else:
|
||||||
|
response = rfunct(url)
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
class Tinge:
|
||||||
|
def __init__(self):
|
||||||
|
self.mbridges: list[HueBridge] = list()
|
||||||
|
self.mdiscovered: list[str] = list()
|
||||||
|
|
||||||
|
def discover_new_bridges(self):
|
||||||
|
upnp: UPnP = UPnP()
|
||||||
|
for device in upnp.discover():
|
||||||
|
if device.get_friendly_name().startswith("Philips hue") and device.host not in self.mdiscovered:
|
||||||
|
user_or_error: UserOrError = connect(device.host)
|
||||||
|
while user_or_error.is_error():
|
||||||
|
if user_or_error.get_error_code() == 101:
|
||||||
|
print("Please press the button on your Hue Bridge")
|
||||||
|
time.sleep(5)
|
||||||
|
user_or_error = connect(device.host)
|
||||||
|
bridge: HueBridge = HueBridge(device.host, user_or_error.get_user)
|
||||||
|
|
||||||
|
self.mbridges.append(bridge)
|
||||||
|
self.mdiscovered.append(device.host)
|
||||||
|
|
||||||
|
def get_bridges_from_file(self):
|
||||||
|
if is_valid_config(self.config):
|
||||||
|
with open(self.config, 'r') as configfile:
|
||||||
|
mbridges = toml.loads(configfile.read())
|
||||||
|
for bridge, value in mbridges.items():
|
||||||
|
print(bridge, value['user'])
|
||||||
|
if bridge not in self.mdiscovered:
|
||||||
|
bridge: HueBridge = HueBridge(bridge, value['user'])
|
||||||
|
bridge.connect()
|
||||||
|
self.mbridges.append(bridge)
|
||||||
|
self.mdiscovered.append(bridge)
|
||||||
|
|
||||||
|
def write_all_bridges_to_conf(self):
|
||||||
|
with open(self.config, 'w') as configfile:
|
||||||
|
for bridge in self.mbridges:
|
||||||
|
configfile.write('["{}"]\nuser = "{}"\n'.format(bridge.ip, bridge.username))
|
@ -0,0 +1,48 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
import os
|
||||||
|
|
||||||
|
import time
|
||||||
|
import toml
|
||||||
|
import wx
|
||||||
|
import HueBridge
|
||||||
|
from upnpy import UPnP
|
||||||
|
|
||||||
|
|
||||||
|
def connect_new_bridge(ipaddress) -> Bridge:
|
||||||
|
bridge: Bridge = Bridge(ipaddress)
|
||||||
|
connected: bool = False
|
||||||
|
while not connected:
|
||||||
|
try:
|
||||||
|
bridge.register_app()
|
||||||
|
except Exception as mexception: # PhueRegistrationException as mexception:
|
||||||
|
print(mexception)
|
||||||
|
time.sleep(5)
|
||||||
|
else:
|
||||||
|
connected = True
|
||||||
|
bridge.connect()
|
||||||
|
return bridge
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class Hui(wx.Frame):
|
||||||
|
def __init__(self, *args, **kw):
|
||||||
|
super().__init__(*args, **kw)
|
||||||
|
self.bridges: list[Bridge] = list()
|
||||||
|
self.discovered: list[str] = list()
|
||||||
|
self.config = os.path.join(os.getenv('HOME'), '.tinge')
|
||||||
|
if is_valid_config(self.config):
|
||||||
|
self.get_bridges_from_file()
|
||||||
|
self.discover_new_bridges()
|
||||||
|
self.write_all_bridges_to_conf()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
app = wx.App()
|
||||||
|
frm = Hui()
|
||||||
|
# frm.show()
|
||||||
|
app.MainLoop()
|
Loading…
Reference in new issue