xiconfjs/pyhOn
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 91 Wiki Activity

Add stricter mypy rules

Browse Source
This commit is contained in:
Andre Basche 2023-07-01 14:59:09 +02:00
parent 0553e6c17d
commit a10ab4423e
4 changed files with 63 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
mypy.ini
View File

@ -2,3 +2,8 @@
check_untyped_defs = True
disallow_any_generics = True
disallow_untyped_defs = True
disallow_any_unimported = True
no_implicit_optional = True
warn_return_any = True
show_error_codes = True
warn_unused_ignores = True

13
pyhon/appliance.py
View File

@ -6,6 +6,7 @@ from pathlib import Path
from typing import Optional, Dict, Any, TYPE_CHECKING, List
from pyhon import diagnose, exceptions
from pyhon.appliances.base import ApplianceBase
from pyhon.attributes import HonAttribute
from pyhon.command_loader import HonCommandLoader
from pyhon.commands import HonCommand
@ -40,7 +41,7 @@ class HonAppliance:
self._default_setting = HonParameter("", {}, "")
try:
self._extra = importlib.import_module(
self._extra: Optional[ApplianceBase] = importlib.import_module(
f"pyhon.appliances.{self.appliance_type.lower()}"
).Appliance(self)
except ModuleNotFoundError:
@ -71,7 +72,8 @@ class HonAppliance:
def _check_name_zone(self, name: str, frontend: bool = True) -> str:
zone = " Z" if frontend else "_z"
if (attribute := self._info.get(name, "")) and self._zone:
attribute: str = self._info.get(name, "")
if attribute and self._zone:
return f"{attribute}{zone}{self._zone}"
return attribute
@ -112,9 +114,10 @@ class HonAppliance:
@property
def code(self) -> str:
if code := self.info.get("code"):
code: str = self.info.get("code", "")
if code:
return code
serial_number = self.info.get("serialNumber", "")
serial_number: str = self.info.get("serialNumber", "")
return serial_number[:8] if len(serial_number) < 18 else serial_number[:11]
@property
@ -198,7 +201,7 @@ class HonAppliance:
@property
def settings(self) -> Dict[str, Parameter]:
result = {}
result: Dict[str, Parameter] = {}
for name, command in self._commands.items():
for key in command.setting_keys:
setting = command.settings.get(key, self._default_setting)

74
pyhon/connection/api.py
View File

@ -75,8 +75,12 @@ class HonAPI:
async def load_appliances(self) -> List[Dict[str, Any]]:
async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp:
if result := await resp.json():
return result.get("payload", {}).get("appliances", {})
result = await resp.json()
if result:
appliances: List[Dict[str, Any]] = result.get("payload", {}).get(
"appliances", {}
)
return appliances
return []
async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -110,9 +114,10 @@ class HonAPI:
)
async with self._hon.get(url) as response:
result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"):
return []
return result["payload"]["history"]
if not result or not result.get("payload"):
return []
command_history: List[Dict[str, Any]] = result["payload"]["history"]
return command_history
async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]:
url: str = (
@ -120,17 +125,20 @@ class HonAPI:
)
async with self._hon.get(url) as response:
result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"):
return []
return result["payload"]["favourites"]
if not result or not result.get("payload"):
return []
favourites: List[Dict[str, Any]] = result["payload"]["favourites"]
return favourites
async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]:
url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params: Dict[str, str] = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response:
result: Dict[str, Any] = await response.json()
if result and (activity := result.get("attributes")):
return activity
if result:
activity: Dict[str, Any] = result.get("attributes", "")
if activity:
return activity
return {}
async def load_appliance_data(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -142,7 +150,10 @@ class HonAPI:
async with self._hon.get(url, params=params) as response:
result: Dict[str, Any] = await response.json()
if result:
return result.get("payload", {}).get("applianceModel", {})
appliance_data: Dict[str, Any] = result.get("payload", {}).get(
"applianceModel", {}
)
return appliance_data
return {}
async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -153,7 +164,8 @@ class HonAPI:
}
url: str = f"{const.API_URL}/commands/v1/context"
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
attributes: Dict[str, Any] = (await response.json()).get("payload", {})
return attributes
async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]:
params: Dict[str, str] = {
@ -162,13 +174,15 @@ class HonAPI:
}
url: str = f"{const.API_URL}/commands/v1/statistics"
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
statistics: Dict[str, Any] = (await response.json()).get("payload", {})
return statistics
async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]:
url = f"{const.API_URL}/commands/v1/maintenance-cycle"
params = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
maintenance: Dict[str, Any] = (await response.json()).get("payload", {})
return maintenance
async def send_command(
self,
@ -207,9 +221,8 @@ class HonAPI:
url: str = f"{const.API_URL}/config/v1/program-list-rules"
async with self._hon_anonymous.get(url) as response:
result: Dict[str, Any] = await response.json()
if result and (data := result.get("payload")):
return data
return {}
data: Dict[str, Any] = result.get("payload", {})
return data
async def app_config(
self, language: str = "en", beta: bool = True
@ -223,17 +236,17 @@ class HonAPI:
}
payload: str = json.dumps(payload_data, separators=(",", ":"))
async with self._hon_anonymous.post(url, data=payload) as response:
if (result := await response.json()) and (data := result.get("payload")):
return data
return {}
result = await response.json()
data: Dict[str, Any] = result.get("payload", {})
return data
async def translation_keys(self, language: str = "en") -> Dict[str, Any]:
config = await self.app_config(language=language)
if url := config.get("language", {}).get("jsonPath"):
async with self._hon_anonymous.get(url) as response:
if result := await response.json():
return result
return {}
if not (url := config.get("language", {}).get("jsonPath")):
return {}
async with self._hon_anonymous.get(url) as response:
result: Dict[str, Any] = await response.json()
return result
async def close(self) -> None:
if self._hon_handler is not None:
@ -250,11 +263,12 @@ class TestAPI(HonAPI):
def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]:
directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower()
if (path := self._path / directory / f"{file}.json").exists():
with open(path, "r", encoding="utf-8") as json_file:
return json.loads(json_file.read())
_LOGGER.warning(f"Can't open {str(path)}")
return {}
if not (path := self._path / directory / f"{file}.json").exists():
_LOGGER.warning("Can't open %s", str(path))
return {}
with open(path, "r", encoding="utf-8") as json_file:
data: Dict[str, Any] = json.loads(json_file.read())
return data
async def load_appliances(self) -> List[Dict[str, Any]]:
result = []

8
pyhon/connection/auth.py
View File

@ -6,7 +6,7 @@ import urllib
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional, Any
from typing import Dict, Optional, Any, List
from urllib import parse
from urllib.parse import quote
@ -115,7 +115,8 @@ class HonAuth:
async with self._request.get(url) as response:
text = await response.text()
self._expires = datetime.utcnow()
if not (login_url := re.findall("url = '(.+?)'", text)):
login_url: List[str] = re.findall("url = '(.+?)'", text)
if not login_url:
if "oauth/done#access_token=" in text:
self._parse_token_data(text)
raise exceptions.HonNoAuthenticationNeeded()
@ -184,7 +185,8 @@ class HonAuth:
if response.status == 200:
with suppress(json.JSONDecodeError, KeyError):
result = await response.json()
return result["events"][0]["attributes"]["values"]["url"]
url: str = result["events"][0]["attributes"]["values"]["url"]
return url
await self._error_logger(response)
return ""