micke
/
tinge
1
0
Fork 0
Code Issues 9 Pull Requests Releases 2 Wiki Activity

Add test program, HueLight and HueUtils and fixup HueBridge

Browse Source
pull/17/head
Micke Nordin 4 years ago
parent a582539b41
commit 91675efb16
Signed by: micke
GPG Key ID: 014B273D614BE877
7 changed files with 269 additions and 116 deletions
Show Stats Download Patch File Download Diff File

11
main.py
Unescape View File

@ -0,0 +1,11 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from tinge import Tinge
if __name__ == "__main__":
m_tinge = Tinge()
bridge = m_tinge.get_bridges()[0]
for light in bridge.get_lights():
print("Light: {} is reachable: {} and on:{}".format(light, light.is_reachable(), light.is_on()))
light.toggle()
print("Light: {} is reachable: {} and on:{}".format(light, light.is_reachable(), light.is_on()))

29
tinge/HueBridge/__init__.py
Unescape View File

@ -1,14 +1,35 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
from ..HueLight import HueLight
from ..HueUtils import make_request
class HueBridge():
def __init__(self, ipaddress: str, username: str):
self.mipaddress: str = ipaddress
self.musername: str = username
self.m_ipaddress: str = ipaddress
self.m_username: str = username
self.m_lights: list[HueLight] = self.discover_lights()
def discover_lights(self) -> list[HueLight]:
path: str = "{}/lights".format(self.m_username)
response = make_request(self.m_ipaddress, path)
lights: list[HueLight] = list()
for key, value in json.loads(response.text).items():
lights.append(HueLight(key, value, self.get_ipaddress(), self.get_user()))
return lights
def get_lights(self):
return self.m_lights
def get_light_by_id(self, id: int) -> HueLight:
for light in self.m_lights:
if light.get_id() == id:
return light
def get_ipaddress(self):
return self.mipaddress
return self.m_ipaddress
def get_user(self):
return self.musername
return self.m_username

143
tinge/HueLight/__init__.py
Unescape View File

@ -0,0 +1,143 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
from datetime import datetime
import requests
from ..HueUtils import make_request
class HueLight:
class State:
def __init__(self, data_slice: dict):
keys = data_slice.keys()
self.m_on: bool = data_slice['on']
if 'bri' in keys:
self.m_bri: int = data_slice['bri']
else:
self.m_bri: int = 0
if 'ct' in keys:
self.m_ct: int = data_slice['ct']
else:
self.m_ct: int = 0
self.m_alert: str = data_slice['alert']
if 'colormode' in keys:
self.m_colormode: str = data_slice['colormode']
else:
self.m_colormode: str = ""
self.m_mode: str = data_slice['mode']
self.m_reachable: bool = data_slice['reachable']
def is_on(self) -> bool:
return self.m_on
def is_reachable(self) -> bool:
return self.m_reachable
class SwUpdate:
def __init__(self, data_slice: dict):
self.m_state: str = data_slice['state']
self.m_lastinstall: datetime = datetime.strptime(data_slice['lastinstall'], "%Y-%m-%dT%H:%M:%S")
class Capabilites:
class Control:
class ColorTemp:
def __init__(self, data_slice: dict):
keys = data_slice.keys()
if 'min' in keys:
self.m_min: int = data_slice['min']
else:
self.m_min: int = 0
if 'max' in keys:
self.m_max: int = data_slice['max']
else:
self.m_max: int = 0
def __init__(self, data_slice: dict):
keys = data_slice.keys()
if 'mindimlevel' in keys:
self.m_mindimlevel: int = data_slice['mindimlevel']
else:
self.m_mindimlevel: int = 0
if 'maxlumen' in keys:
self.m_maxlumen: int = data_slice['maxlumen']
else:
self.m_maxlumen: int = 0
if 'ct' in data_slice.keys():
self.m_ct = HueLight.Capabilites.Control.ColorTemp(data_slice['ct'])
else:
self.m_ct = HueLight.Capabilites.Control.ColorTemp({})
class Streaming:
def __init__(self, data_slice: dict):
self.m_renderer: bool = data_slice['renderer']
self.m_proxy: bool = data_slice['proxy']
def __init__(self, data_slice: dict):
self.m_certified: bool = data_slice['certified']
self.m_control = HueLight.Capabilites.Control(data_slice['control'])
self.m_streaming = HueLight.Capabilites.Streaming(data_slice['streaming'])
class Config:
class Startup:
def __init__(self, data_slice: dict):
self.m_mode: str = data_slice['mode']
self.m_configured: bool = data_slice['configured']
def __init__(self, data_slice: dict):
self.m_archetype: str = data_slice['archetype']
self.m_function: str = data_slice['function']
self.m_direction: str = data_slice['direction']
self.m_startup = HueLight.Config.Startup(data_slice['startup'])
def __init__(self, id: int, data: dict, parent_bridge_ip: str, parent_bridge_user: str):
self.m_id: int = id
self.m_parent_bridge_ip = parent_bridge_ip
self.m_parent_bridge_user = parent_bridge_user
self.m_state = HueLight.State(data['state'])
self.m_swupdate = HueLight.SwUpdate(data['swupdate'])
self.m_type: str = data['type']
self.m_name: str = data['name']
self.m_modelid: str = data['modelid']
self.m_manufacturername: str = data['manufacturername']
self.m_productname: str = data['productname']
self.m_capabilites = HueLight.Capabilites(data['capabilities'])
self.m_config = HueLight.Config(data['config'])
self.m_uniqueid: str = data['uniqueid']
self.m_swconfigid: str = data['swconfigid']
self.m_productid: str = data['productid']
def __str__(self) -> str:
return self.m_name
def update_state(self):
path: str = "{}/lights/{}".format(self.m_parent_bridge_user, self.m_id)
response = make_request(self.m_parent_bridge_ip, path)
self.m_state = HueLight.State(json.loads(response.text)['state'])
def get_state(self):
return self.m_state
def get_id(self):
return self.m_id
def is_on(self) -> bool:
return self.get_state().is_on()
def is_reachable(self) -> bool:
return self.get_state().is_reachable()
def set_state(self, state: str) -> requests.Response:
path: str = "{}/lights/{}/state".format(self.m_parent_bridge_user, self.m_id)
method: str = "PUT"
response = make_request(self.m_parent_bridge_ip, path, method, state)
self.update_state()
return response
def toggle(self):
if self.is_reachable():
state: str = '{"on":true}'
if self.is_on():
state = '{"on":false}'
self.set_state(state)

47
tinge/HueUtils/__init__.py
Unescape View File

@ -0,0 +1,47 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import requests
from ..UserOrError import UserOrError
def connect(ipaddress: str) -> UserOrError:
user_or_error: UserOrError = UserOrError()
body: str = '{{"devicetype":"{0}"}}'.format("tinge")
path: str = ""
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(user_or_error.UNKNOWNERROR)
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: str = '') -> requests.Response:
rfunct = requests.get
url = "http://{}/api/{}".format(ipaddress, path)
if method == "PUT":
rfunct = requests.put
elif method == "POST":
rfunct = requests.post
elif method == "DELETE":
rfunct = requests.delete
if body and method == "GET":
response = rfunct(url, params=body)
elif body:
response = rfunct(url, data=body)
else:
response = rfunct(url)
return response

4
tinge/UserOrError/__init__.py
Unescape View File

@ -3,10 +3,12 @@
class UserOrError:
def __init__(self):
self.UNKNOWNERROR = 9999
self.muser: str = str()
self.merror: bool = True
self.mcode: int = 0
self.mcode: int = self.UNKNOWNERROR
def get_error_code(self) -> int:
return self.mcode

103
tinge/__init__.py
Unescape View File

@ -1,89 +1,66 @@
#!/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
from .HueBridge import HueBridge
from .HueUtils import connect, is_valid_config
from .UserOrError import UserOrError
class Tinge:
def __init__(self):
self.mbridges: list[HueBridge] = list()
self.mdiscovered: list[str] = list()
self.m_bridges: list[HueBridge] = list()
self.m_discovered: list[str] = list()
self.m_config = os.path.join(os.environ['HOME'], ".config/tinge/config")
self.create_confdir()
self.read_bridges_from_file()
self.discover_new_bridges()
self.write_all_bridges_to_conf()
def create_confdir(self):
if not os.path.exists(os.path.dirname(self.m_config)):
os.makedirs(os.path.dirname(self.m_config))
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:
discovered_devices = upnp.discover()
if not discovered_devices:
print("No devices discovered at this time")
return
for device in discovered_devices:
if device.get_friendly_name().startswith("Philips hue") and device.host not in self.m_discovered:
user_or_error: UserOrError = connect(device.host)
print("Is error: {}".format(str(user_or_error.is_error())))
while user_or_error.is_error():
print("Is error: {}".format(str(user_or_error.get_error_code())))
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)
bridge: HueBridge = HueBridge(device.host, user_or_error.get_user())
self.m_bridges.append(bridge)
self.m_discovered.append(device.host)
return
self.mbridges.append(bridge)
self.mdiscovered.append(device.host)
def get_bridges(self):
return self.m_bridges
def get_bridges_from_file(self):
if is_valid_config(self.config):
with open(self.config, 'r') as configfile:
def read_bridges_from_file(self):
if is_valid_config(self.m_config):
with open(self.m_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)
for key, value in mbridges.items():
if key not in self.m_discovered:
bridge: HueBridge = HueBridge(key, value['user'])
self.m_bridges.append(bridge)
self.m_discovered.append(key)
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))
with open(self.m_config, 'w') as configfile:
for bridge in self.m_bridges:
configfile.write('["{}"]\nuser = "{}"\n'.format(bridge.get_ipaddress(), bridge.get_user()))

48
tinge/__init__.py.bak
Unescape View File

@ -1,48 +0,0 @@
#!/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()
Write Preview
Loading…
Cancel
Save