From 91675efb16df7428cc8f212ec70e9976db3c63dc Mon Sep 17 00:00:00 2001 From: Micke Nordin Date: Tue, 4 May 2021 16:31:41 +0200 Subject: [PATCH] Add test program, HueLight and HueUtils and fixup HueBridge --- main.py | 11 +++ tinge/HueBridge/__init__.py | 29 ++++++- tinge/HueLight/__init__.py | 143 ++++++++++++++++++++++++++++++++++ tinge/HueUtils/__init__.py | 47 +++++++++++ tinge/UserOrError/__init__.py | 4 +- tinge/__init__.py | 103 ++++++++++-------------- tinge/__init__.py.bak | 48 ------------ 7 files changed, 269 insertions(+), 116 deletions(-) create mode 100644 main.py create mode 100644 tinge/HueLight/__init__.py create mode 100644 tinge/HueUtils/__init__.py delete mode 100644 tinge/__init__.py.bak diff --git a/main.py b/main.py new file mode 100644 index 0000000..34445b2 --- /dev/null +++ b/main.py @@ -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())) diff --git a/tinge/HueBridge/__init__.py b/tinge/HueBridge/__init__.py index 96e6e46..7330f7b 100644 --- a/tinge/HueBridge/__init__.py +++ b/tinge/HueBridge/__init__.py @@ -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 diff --git a/tinge/HueLight/__init__.py b/tinge/HueLight/__init__.py new file mode 100644 index 0000000..96f45ed --- /dev/null +++ b/tinge/HueLight/__init__.py @@ -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) diff --git a/tinge/HueUtils/__init__.py b/tinge/HueUtils/__init__.py new file mode 100644 index 0000000..57b7636 --- /dev/null +++ b/tinge/HueUtils/__init__.py @@ -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 diff --git a/tinge/UserOrError/__init__.py b/tinge/UserOrError/__init__.py index 73a9f96..8d646e1 100644 --- a/tinge/UserOrError/__init__.py +++ b/tinge/UserOrError/__init__.py @@ -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 diff --git a/tinge/__init__.py b/tinge/__init__.py index f59ac89..e4e18ab 100644 --- a/tinge/__init__.py +++ b/tinge/__init__.py @@ -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())) diff --git a/tinge/__init__.py.bak b/tinge/__init__.py.bak deleted file mode 100644 index eca6852..0000000 --- a/tinge/__init__.py.bak +++ /dev/null @@ -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()