Compare commits

...

2 Commits

Author SHA1 Message Date
46e6a85e84 add diagnose property for devices 2023-04-11 22:14:36 +02:00
8c832b44cd Fix token refresh problems 2023-04-11 17:09:02 +02:00
7 changed files with 112 additions and 71 deletions

83
pyhon/__main__.py Executable file → Normal file
View File

@ -10,7 +10,7 @@ from pathlib import Path
if __name__ == "__main__": if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent))
from pyhon import Hon, HonAPI from pyhon import Hon, HonAPI, helper
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -34,61 +34,6 @@ def get_arguments():
return vars(parser.parse_args()) return vars(parser.parse_args())
# yaml.dump() would be done the same, but needs an additional dependency...
def pretty_print(data, key="", intend=0, is_list=False):
if type(data) is list:
if key:
print(f"{' ' * intend}{'- ' if is_list else ''}{key}:")
intend += 1
for i, value in enumerate(data):
pretty_print(value, intend=intend, is_list=True)
elif type(data) is dict:
if key:
print(f"{' ' * intend}{'- ' if is_list else ''}{key}:")
intend += 1
for i, (key, value) in enumerate(sorted(data.items())):
if is_list and not i:
pretty_print(value, key=key, intend=intend, is_list=True)
elif is_list:
pretty_print(value, key=key, intend=intend + 1)
else:
pretty_print(value, key=key, intend=intend)
else:
print(
f"{' ' * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}"
)
def key_print(data, key="", start=True):
if type(data) is list:
for i, value in enumerate(data):
key_print(value, key=f"{key}.{i}", start=False)
elif type(data) is dict:
for k, value in sorted(data.items()):
key_print(value, key=k if start else f"{key}.{k}", start=False)
else:
print(f"{key}: {data}")
def create_command(commands, concat=False):
result = {}
for name, command in commands.items():
if not concat:
result[name] = {}
for parameter, data in command.parameters.items():
if data.typology == "enum":
value = data.values
elif data.typology == "range":
value = {"min": data.min, "max": data.max, "step": data.step}
else:
continue
if not concat:
result[name][parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result
async def translate(language, json_output=False): async def translate(language, json_output=False):
async with HonAPI(anonymous=True) as hon: async with HonAPI(anonymous=True) as hon:
keys = await hon.translation_keys(language) keys = await hon.translation_keys(language)
@ -102,7 +47,7 @@ async def translate(language, json_output=False):
.replace("\\r", "") .replace("\\r", "")
) )
keys = json.loads(clean_keys) keys = json.loads(clean_keys)
pretty_print(keys) print(helper.pretty_print(keys))
async def main(): async def main():
@ -120,13 +65,25 @@ async def main():
if args.get("keys"): if args.get("keys"):
data = device.data.copy() data = device.data.copy()
attr = "get" if args.get("all") else "pop" attr = "get" if args.get("all") else "pop"
key_print(data["attributes"].__getattribute__(attr)("parameters")) print(
key_print(data.__getattribute__(attr)("appliance")) helper.key_print(
key_print(data) data["attributes"].__getattribute__(attr)("parameters")
pretty_print(create_command(device.commands, concat=True)) )
)
print(helper.key_print(data.__getattribute__(attr)("appliance")))
print(helper.key_print(data))
print(
helper.pretty_print(
helper.create_command(device.commands, concat=True)
)
)
else: else:
pretty_print({"data": device.data}) print(helper.pretty_print({"data": device.data}))
pretty_print({"settings": create_command(device.commands)}) print(
helper.pretty_print(
{"settings": helper.create_command(device.commands)}
)
)
def start(): def start():

View File

@ -1,6 +1,7 @@
import importlib import importlib
from contextlib import suppress from contextlib import suppress
from pyhon import helper
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter import HonParameterFixed from pyhon.parameter import HonParameterFixed
@ -172,3 +173,15 @@ class HonAppliance:
if self._extra: if self._extra:
return self._extra.data(result) return self._extra.data(result)
return result return result
@property
def diagnose(self):
data = self.data.copy()
for sensible in ["PK", "SK", "serialNumber", "code"]:
data["appliance"].pop(sensible, None)
result = helper.pretty_print({"data": self.data}, whitespace="\u200B \u200B ")
result += helper.pretty_print(
{"commands": helper.create_command(self.commands)},
whitespace="\u200B \u200B ",
)
return result.replace(self.mac_address, "12-34-56-78-90-ab")

View File

@ -194,7 +194,9 @@ class HonAuth:
return False return False
if not await self._get_token(url): if not await self._get_token(url):
return False return False
return await self._api_auth()
async def _api_auth(self):
post_headers = {"id-token": self._id_token} post_headers = {"id-token": self._id_token}
data = self._device.get() data = self._device.get()
async with self._session.post( async with self._session.post(
@ -225,4 +227,4 @@ class HonAuth:
data = await response.json() data = await response.json()
self._id_token = data["id_token"] self._id_token = data["id_token"]
self._access_token = data["access_token"] self._access_token = data["access_token"]
return True return await self._api_auth()

View File

@ -75,17 +75,20 @@ class HonConnectionHandler(HonBaseConnectionHandler):
self._request_headers["id-token"] = self._auth.id_token self._request_headers["id-token"] = self._auth.id_token
else: else:
raise HonAuthenticationError("Can't login") raise HonAuthenticationError("Can't login")
return {h: v for h, v in self._request_headers.items() if h not in headers} return headers | self._request_headers
@asynccontextmanager @asynccontextmanager
async def _intercept(self, method, *args, loop=0, **kwargs): async def _intercept(self, method, *args, loop=0, **kwargs):
kwargs["headers"] = await self._check_headers(kwargs.get("headers", {})) kwargs["headers"] = await self._check_headers(kwargs.get("headers", {}))
async with method(*args, **kwargs) as response: async with method(*args, **kwargs) as response:
if response.status == 403 and not loop: if response.status in [401, 403] and loop == 0:
_LOGGER.info("Try refreshing token...") _LOGGER.info("Try refreshing token...")
await self._auth.refresh() await self._auth.refresh()
yield await self._intercept(method, *args, loop=loop + 1, **kwargs) async with self._intercept(
elif response.status == 403 and loop < 2: method, *args, loop=loop + 1, **kwargs
) as result:
yield result
elif response.status in [401, 403] and loop == 1:
_LOGGER.warning( _LOGGER.warning(
"%s - Error %s - %s", "%s - Error %s - %s",
response.request_info.url, response.request_info.url,
@ -93,7 +96,10 @@ class HonConnectionHandler(HonBaseConnectionHandler):
await response.text(), await response.text(),
) )
await self.create() await self.create()
yield await self._intercept(method, *args, loop=loop + 1, **kwargs) async with self._intercept(
method, *args, loop=loop + 1, **kwargs
) as result:
yield result
elif loop >= 2: elif loop >= 2:
_LOGGER.error( _LOGGER.error(
"%s - Error %s - %s", "%s - Error %s - %s",

63
pyhon/helper.py Normal file
View File

@ -0,0 +1,63 @@
def key_print(data, key="", start=True):
result = ""
if isinstance(data, list):
for i, value in enumerate(data):
result += key_print(value, key=f"{key}.{i}", start=False)
elif isinstance(data, dict):
for k, value in sorted(data.items()):
result += key_print(value, key=k if start else f"{key}.{k}", start=False)
else:
result += f"{key}: {data}\n"
return result
# yaml.dump() would be done the same, but needs an additional dependency...
def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "):
result = ""
if isinstance(data, list):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, value in enumerate(data):
result += pretty_print(
value, intend=intend, is_list=True, whitespace=whitespace
)
elif isinstance(data, dict):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, (key, value) in enumerate(sorted(data.items())):
if is_list and not i:
result += pretty_print(
value, key=key, intend=intend, is_list=True, whitespace=whitespace
)
elif is_list:
result += pretty_print(
value, key=key, intend=intend + 1, whitespace=whitespace
)
else:
result += pretty_print(
value, key=key, intend=intend, whitespace=whitespace
)
else:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
return result
def create_command(commands, concat=False):
result = {}
for name, command in commands.items():
if not concat:
result[name] = {}
for parameter, data in command.parameters.items():
if data.typology == "enum":
value = data.values
elif data.typology == "range":
value = {"min": data.min, "max": data.max, "step": data.step}
else:
continue
if not concat:
result[name][parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result

View File

@ -5,7 +5,7 @@ def str_to_float(string):
try: try:
return int(string) return int(string)
except ValueError: except ValueError:
return float(str(string.replace(",", "."))) return float(str(string).replace(",", "."))
class HonParameter: class HonParameter:

View File

@ -7,7 +7,7 @@ with open("README.md", "r") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.6.4", version="0.7.0",
author="Andre Basche", author="Andre Basche",
description="Control hOn devices with python", description="Control hOn devices with python",
long_description=long_description, long_description=long_description,