You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
3.2 KiB
100 lines
3.2 KiB
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
import os
|
|
from typing import Union
|
|
|
|
import requests
|
|
|
|
from ..UserOrError import UserOrError
|
|
|
|
|
|
def connect(ipaddress: str) -> UserOrError:
|
|
"""Connect this bridge to tinge and get the username
|
|
|
|
Args:
|
|
ipaddress (str): ip address of the bridge
|
|
|
|
Returns:
|
|
UserOrError: Error if it is not connected otherwise the username
|
|
"""
|
|
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)
|
|
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
|
|
|
|
|
|
def is_valid_config(filename: str) -> bool:
|
|
"""Currently doesn't do much, but we ckan get mor elaborate checks later if we want
|
|
|
|
Args:
|
|
filename (str): The filename to check
|
|
|
|
Returns:
|
|
bool: True if we think it is ok, otherwise False
|
|
"""
|
|
return os.path.exists(filename) and os.path.getsize(filename) > 0
|
|
|
|
|
|
def is_bridge(ipaddress: str):
|
|
try:
|
|
response = make_request(ipaddress, "1234/lights")
|
|
discovered = False
|
|
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:
|
|
# This is kinda ugly but line is too long with and statement
|
|
if resp['error']['description'] == "unauthorized user":
|
|
if resp['error']['address'] == "/lights":
|
|
if resp['error']['type'] == 1:
|
|
discovered = True
|
|
except simplejson.errors.JSONDecodeError:
|
|
pass
|
|
return discovered
|
|
|
|
|
|
def make_request(ipaddress: str, path: str, method: str = "GET",
|
|
body: str = '') -> Union[None, requests.Response]:
|
|
"""Helper function to make an API call to the Hue API on the bridge
|
|
|
|
Args:
|
|
ipaddress (str): ip address of the bridge
|
|
path (str): the API endpoint
|
|
method (str, optional): HTTP method. Defaults to "GET".
|
|
body (str, optional): The body or parameters to the API call. Defaults to ''.
|
|
|
|
Returns:
|
|
requests.Response: The response from the Hue API
|
|
"""
|
|
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:
|
|
try:
|
|
response = rfunct(url)
|
|
except requests.exceptions.ConnectionError:
|
|
response = None
|
|
return response
|