micke/tinge
1
0
Fork 0
Code Issues 9 Pull requests Projects Releases 2 Packages Wiki Activity Actions

Fix for Issue#16

Browse source 
This commit fixes upnp discovery for bridges that does not have the
default name that starts with "Philips hue"
This commit is contained in:
Micke Nordin 2021-05-22 14:42:42 +02:00
parent 707f764a65
commit 03de059384
Signed by: micke
GPG key ID: 014B273D614BE877
5 changed files with 88 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
main.py
View file

@ -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

3
requirements.txt
View file

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

15
tinge/HueBridge/__init__.py
View file

@ -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

22
tinge/HueUtils/__init__.py
View file

@ -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

55
tinge/__init__.py
View file

@ -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