From 55d4cdc527c52dd8b1166b16d64e3692bb2ca2a3 Mon Sep 17 00:00:00 2001 From: Micke Nordin Date: Mon, 24 May 2021 13:45:42 +0000 Subject: [PATCH] Fix for Issue#16 (#17) --- main.py | 27 +++++++++++++----- requirements.txt | 3 ++ tinge/HueBridge/__init__.py | 15 ++++++++++ tinge/HueUtils/__init__.py | 22 +++++++++------ tinge/__init__.py | 55 ++++++++++++++++++++++++------------- 5 files changed, 88 insertions(+), 34 deletions(-) diff --git a/main.py b/main.py index c9baac9..ca838f6 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,7 @@ from typing import Union import wx import wx.lib.scrolledpanel as scrolled -from tinge import Tinge, HueBridge, HueGroup, HueLight +from tinge import Tinge, HueBridge, HueGroup, HueLight, HueUtils class Hui(wx.Frame): @@ -40,7 +40,6 @@ class Hui(wx.Frame): self.m_off_icon: str = '☾' self.m_unreachable_icon: str = '⚠' self.m_tinge: Tinge = Tinge() - self.m_bridges: list[HueBridge] = self.m_tinge.get_bridges() self.cur_bridge: Union[None, HueBridge] = None self.cur_group: Union[None, HueGroup] = None # create a panel in the frame @@ -56,8 +55,8 @@ class Hui(wx.Frame): """Add bridges to sizer, the entry point of the program """ self.SetTitle('Tinge - All Bridges') - if self.m_bridges: - for bridge in self.m_bridges: + if self.m_tinge.get_bridges(): + for bridge in self.m_tinge.get_bridges(): btn: wx.Button = wx.Button(self.pnl, label=str(bridge)) self.sizer.Add(btn, 0, wx.EXPAND) self.Bind(wx.EVT_BUTTON, @@ -217,11 +216,25 @@ class Hui(wx.Frame): else: self.add_single_light(lightid) - def discover_new_bridges(self): + def discover_new_bridges(self) -> bool: """Call back for button that is displayed if no bridges were found + + Returns: + bool: True if we found any bridge, False otherwise """ - self.m_tinge.discover_new_bridges() - self.add_bridges() + found_any: bool = False + found_bridges: list[dict] = self.m_tinge.discover_new_bridges() + if found_bridges: + for bridge in found_bridges: + user_or_error = HueUtils.connect(bridge['ipaddress']) + while user_or_error.is_error(): + user_or_error = HueUtils.connect(bridge['ipaddress']) + self.m_tinge.append_bridge(HueBridge(bridge['ipaddress'], user_or_error.get_user(), bridge['name'])) + found_any = True + self.m_tinge.write_all_bridges_to_conf() + self.add_bridges() + return found_any + def get_ok_cancel_answer_from_modal(self, message: str) -> bool: """Display a message dialog and return ok or cancel diff --git a/requirements.txt b/requirements.txt index c9a4973..4b24588 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,6 @@ toml==0.10.1 UPnPy==1.1.8 requests==2.25.1 + +wxPython~=4.0.7 +simplejson~=3.17.2 diff --git a/tinge/HueBridge/__init__.py b/tinge/HueBridge/__init__.py index 8ed397c..48e5212 100644 --- a/tinge/HueBridge/__init__.py +++ b/tinge/HueBridge/__init__.py @@ -27,6 +27,8 @@ class HueBridge: else: self.m_name = self.m_ipaddress self.m_lights: list[HueLight] = self.discover_lights() + self.discover_new_lights() + self.m_new_lights: list[HueLight] = self.get_new_lights() self.m_groups: list[HueGroup] = self.discover_groups() def __str__(self) -> str: @@ -183,6 +185,19 @@ class HueBridge: """ return self.m_lights + def get_new_lights(self) -> list[HueLight]: + path: str = "{}/lights/new".format(self.m_username) + response = make_request(self.m_ipaddress, path) + newlights: list[HueLight] = list() + for lightid, nameobj in response.json().items(): + if lightid != "lastscan": + print(lightid) + if not self.get_light_by_id(int(lightid)): + lightpath: str = "{}/lights/{}".format(self.m_username, int(lightid)) + lightresponse = make_request(self.m_ipaddress, lightpath) + newlights.append(HueLight(int(lightid), lightresponse.json(), self.get_ipaddress(),self.get_user())) + return newlights + def get_user(self) -> str: """A user, or username, is more like a password and is needed to authenticate with the Hue API diff --git a/tinge/HueUtils/__init__.py b/tinge/HueUtils/__init__.py index fe2077c..7822e45 100644 --- a/tinge/HueUtils/__init__.py +++ b/tinge/HueUtils/__init__.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- import os +from typing import Union import requests @@ -21,12 +22,14 @@ def connect(ipaddress: str) -> UserOrError: 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']) + if response: + 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) else: user_or_error.set_error(user_or_error.UNKNOWNERROR) return user_or_error @@ -45,7 +48,7 @@ def is_valid_config(filename: str) -> bool: def make_request(ipaddress: str, path: str, method: str = "GET", - body: str = '') -> requests.Response: + body: str = '') -> Union[None, requests.Response]: """Helper function to make an API call to the Hue API on the bridge Args: @@ -70,5 +73,8 @@ def make_request(ipaddress: str, path: str, method: str = "GET", elif body: response = rfunct(url, data=body) else: - response = rfunct(url) + try: + response = rfunct(url) + except requests.exceptions.ConnectionError: + response = None return response diff --git a/tinge/__init__.py b/tinge/__init__.py index adf0745..0ab95ee 100644 --- a/tinge/__init__.py +++ b/tinge/__init__.py @@ -1,13 +1,14 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- import os -import time +from typing import Union +import simplejson import toml from upnpy import UPnP from .HueBridge import HueBridge -from .HueUtils import connect, is_valid_config +from .HueUtils import connect, is_valid_config, make_request from .UserOrError import UserOrError @@ -23,8 +24,10 @@ class Tinge: 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 append_bridge(self, bridge: HueBridge): + self.m_bridges.append(bridge) + self.m_discovered.append(bridge.get_ipaddress()) def create_confdir(self): """Create the config dir if it does not allready exist @@ -32,29 +35,43 @@ class Tinge: if not os.path.exists(os.path.dirname(self.m_config)): os.makedirs(os.path.dirname(self.m_config)) - def discover_new_bridges(self): + def discover_new_bridges(self) -> Union[None, list[dict]]: """Use UPnP to discover bridges on the current network """ upnp: UPnP = UPnP() discovered_devices = upnp.discover() + discovered_bridges: list[dict] = list() + seen_ips: list[str] = list() if not discovered_devices: print("No devices discovered at this time") - return + return None 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()) + discovered: bool = False + if (device.host not in self.m_discovered) and (device.host not in seen_ips): + seen_ips.append(device.host) + # Let's check if the device has the default name, if so we assume it's a hue bridge + if device.get_friendly_name().startswith("Philips hue"): + discovered = True + # If not we try to do a request against the api and see if we get an answer we can understand + else: + try: + response = make_request(device.host, "1234/lights") + if response: + resp = response.json()[0] + if 'error' in resp.keys(): + m_keys = resp['error'].keys() + if 'description' in m_keys and 'address' in m_keys and 'type' in m_keys: + if resp['error']['description'] == "unauthorized user" and resp['error'][ + 'address'] == "/lights" and resp['error']['type'] == 1: + discovered = True + except simplejson.errors.JSONDecodeError: + pass - self.m_bridges.append(bridge) - self.m_discovered.append(device.host) - return + if discovered: + bridge = {'ipaddress': device.host, 'name': device.get_friendly_name()} + if bridge not in discovered_bridges: + discovered_bridges.append(bridge) + return discovered_bridges def get_bridges(self) -> list[HueBridge]: """Get the bridges