Fix for Issue#16 (#17)

master
Micke Nordin 3 years ago
parent 61ba783662
commit df5403a519
Signed by: micke
GPG Key ID: 014B273D614BE877

@ -5,7 +5,7 @@ from typing import Union
import wx import wx
import wx.lib.scrolledpanel as scrolled 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): class Hui(wx.Frame):
@ -40,7 +40,6 @@ class Hui(wx.Frame):
self.m_off_icon: str = '' self.m_off_icon: str = ''
self.m_unreachable_icon: str = '' self.m_unreachable_icon: str = ''
self.m_tinge: Tinge = Tinge() self.m_tinge: Tinge = Tinge()
self.m_bridges: list[HueBridge] = self.m_tinge.get_bridges()
self.cur_bridge: Union[None, HueBridge] = None self.cur_bridge: Union[None, HueBridge] = None
self.cur_group: Union[None, HueGroup] = None self.cur_group: Union[None, HueGroup] = None
# create a panel in the frame # create a panel in the frame
@ -56,8 +55,8 @@ class Hui(wx.Frame):
"""Add bridges to sizer, the entry point of the program """Add bridges to sizer, the entry point of the program
""" """
self.SetTitle('Tinge - All Bridges') self.SetTitle('Tinge - All Bridges')
if self.m_bridges: if self.m_tinge.get_bridges():
for bridge in self.m_bridges: for bridge in self.m_tinge.get_bridges():
btn: wx.Button = wx.Button(self.pnl, label=str(bridge)) btn: wx.Button = wx.Button(self.pnl, label=str(bridge))
self.sizer.Add(btn, 0, wx.EXPAND) self.sizer.Add(btn, 0, wx.EXPAND)
self.Bind(wx.EVT_BUTTON, self.Bind(wx.EVT_BUTTON,
@ -217,11 +216,25 @@ class Hui(wx.Frame):
else: else:
self.add_single_light(lightid) 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 """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() found_any: bool = False
self.add_bridges() 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: def get_ok_cancel_answer_from_modal(self, message: str) -> bool:
"""Display a message dialog and return ok or cancel """Display a message dialog and return ok or cancel

@ -1,3 +1,6 @@
toml==0.10.1 toml==0.10.1
UPnPy==1.1.8 UPnPy==1.1.8
requests==2.25.1 requests==2.25.1
wxPython~=4.0.7
simplejson~=3.17.2

@ -27,6 +27,8 @@ class HueBridge:
else: else:
self.m_name = self.m_ipaddress self.m_name = self.m_ipaddress
self.m_lights: list[HueLight] = self.discover_lights() 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() self.m_groups: list[HueGroup] = self.discover_groups()
def __str__(self) -> str: def __str__(self) -> str:
@ -183,6 +185,19 @@ class HueBridge:
""" """
return self.m_lights 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: def get_user(self) -> str:
"""A user, or username, is more like a password and is needed to authenticate with the Hue API """A user, or username, is more like a password and is needed to authenticate with the Hue API

@ -1,6 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import os import os
from typing import Union
import requests import requests
@ -21,12 +22,14 @@ def connect(ipaddress: str) -> UserOrError:
path: str = "" path: str = ""
method: str = "POST" method: str = "POST"
response: requests.Response = make_request(ipaddress, path, method, body) response: requests.Response = make_request(ipaddress, path, method, body)
if response:
data: dict = response.json()[0] data: dict = response.json()[0]
if 'error' in data.keys(): if 'error' in data.keys():
user_or_error.set_error(data['error']['type']) user_or_error.set_error(data['error']['type'])
elif 'success' in data.keys(): elif 'success' in data.keys():
user_or_error.set_user(data['success']['username']) user_or_error.set_user(data['success']['username'])
else:
user_or_error.set_error(user_or_error.UNKNOWNERROR)
else: else:
user_or_error.set_error(user_or_error.UNKNOWNERROR) user_or_error.set_error(user_or_error.UNKNOWNERROR)
return user_or_error 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", 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 """Helper function to make an API call to the Hue API on the bridge
Args: Args:
@ -70,5 +73,8 @@ def make_request(ipaddress: str, path: str, method: str = "GET",
elif body: elif body:
response = rfunct(url, data=body) response = rfunct(url, data=body)
else: else:
response = rfunct(url) try:
response = rfunct(url)
except requests.exceptions.ConnectionError:
response = None
return response return response

@ -1,13 +1,14 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import os import os
import time from typing import Union
import simplejson
import toml import toml
from upnpy import UPnP from upnpy import UPnP
from .HueBridge import HueBridge from .HueBridge import HueBridge
from .HueUtils import connect, is_valid_config from .HueUtils import connect, is_valid_config, make_request
from .UserOrError import UserOrError from .UserOrError import UserOrError
@ -23,8 +24,10 @@ class Tinge:
self.m_config = os.path.join(os.environ['HOME'], ".config/tinge/config") self.m_config = os.path.join(os.environ['HOME'], ".config/tinge/config")
self.create_confdir() self.create_confdir()
self.read_bridges_from_file() 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): def create_confdir(self):
"""Create the config dir if it does not allready exist """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)): if not os.path.exists(os.path.dirname(self.m_config)):
os.makedirs(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 """Use UPnP to discover bridges on the current network
""" """
upnp: UPnP = UPnP() upnp: UPnP = UPnP()
discovered_devices = upnp.discover() discovered_devices = upnp.discover()
discovered_bridges: list[dict] = list()
seen_ips: list[str] = list()
if not discovered_devices: if not discovered_devices:
print("No devices discovered at this time") print("No devices discovered at this time")
return return None
for device in discovered_devices: for device in discovered_devices:
if device.get_friendly_name().startswith("Philips hue") and device.host not in self.m_discovered: discovered: bool = False
user_or_error: UserOrError = connect(device.host) if (device.host not in self.m_discovered) and (device.host not in seen_ips):
print("Is error: {}".format(str(user_or_error.is_error()))) seen_ips.append(device.host)
while user_or_error.is_error(): # Let's check if the device has the default name, if so we assume it's a hue bridge
print("Is error: {}".format(str(user_or_error.get_error_code()))) if device.get_friendly_name().startswith("Philips hue"):
if user_or_error.get_error_code() == 101: discovered = True
print("Please press the button on your Hue Bridge") # If not we try to do a request against the api and see if we get an answer we can understand
time.sleep(5) else:
user_or_error = connect(device.host) try:
bridge: HueBridge = HueBridge(device.host, user_or_error.get_user()) 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) if discovered:
self.m_discovered.append(device.host) bridge = {'ipaddress': device.host, 'name': device.get_friendly_name()}
return if bridge not in discovered_bridges:
discovered_bridges.append(bridge)
return discovered_bridges
def get_bridges(self) -> list[HueBridge]: def get_bridges(self) -> list[HueBridge]:
"""Get the bridges """Get the bridges

Loading…
Cancel
Save