mirror of
https://github.com/python-kasa/python-kasa.git
synced 2024-12-23 03:33:35 +00:00
Improve typing for protocol class (#289)
This commit is contained in:
parent
b4036e55ac
commit
bcb9fe18ab
@ -15,7 +15,7 @@ import json
|
|||||||
import logging
|
import logging
|
||||||
import struct
|
import struct
|
||||||
from pprint import pformat as pf
|
from pprint import pformat as pf
|
||||||
from typing import Dict, Optional, Union
|
from typing import Dict, Generator, Optional, Union
|
||||||
|
|
||||||
from .exceptions import SmartDeviceException
|
from .exceptions import SmartDeviceException
|
||||||
|
|
||||||
@ -107,7 +107,7 @@ class TPLinkSmartHomeProtocol:
|
|||||||
|
|
||||||
return json_payload
|
return json_payload
|
||||||
|
|
||||||
async def close(self):
|
async def close(self) -> None:
|
||||||
"""Close the connection."""
|
"""Close the connection."""
|
||||||
writer = self.writer
|
writer = self.writer
|
||||||
self.reader = self.writer = None
|
self.reader = self.writer = None
|
||||||
@ -116,7 +116,7 @@ class TPLinkSmartHomeProtocol:
|
|||||||
with contextlib.suppress(Exception):
|
with contextlib.suppress(Exception):
|
||||||
await writer.wait_closed()
|
await writer.wait_closed()
|
||||||
|
|
||||||
def _reset(self):
|
def _reset(self) -> None:
|
||||||
"""Clear any varibles that should not survive between loops."""
|
"""Clear any varibles that should not survive between loops."""
|
||||||
self.reader = self.writer = self.loop = self.query_lock = None
|
self.reader = self.writer = self.loop = self.query_lock = None
|
||||||
|
|
||||||
@ -154,13 +154,13 @@ class TPLinkSmartHomeProtocol:
|
|||||||
await self.close()
|
await self.close()
|
||||||
raise SmartDeviceException("Query reached somehow to unreachable")
|
raise SmartDeviceException("Query reached somehow to unreachable")
|
||||||
|
|
||||||
def __del__(self):
|
def __del__(self) -> None:
|
||||||
if self.writer and self.loop and self.loop.is_running():
|
if self.writer and self.loop and self.loop.is_running():
|
||||||
self.writer.close()
|
self.writer.close()
|
||||||
self._reset()
|
self._reset()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _xor_payload(unencrypted):
|
def _xor_payload(unencrypted: bytes) -> Generator[int, None, None]:
|
||||||
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
|
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
|
||||||
for unencryptedbyte in unencrypted:
|
for unencryptedbyte in unencrypted:
|
||||||
key = key ^ unencryptedbyte
|
key = key ^ unencryptedbyte
|
||||||
@ -179,7 +179,7 @@ class TPLinkSmartHomeProtocol:
|
|||||||
)
|
)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _xor_encrypted_payload(ciphertext):
|
def _xor_encrypted_payload(ciphertext: bytes) -> Generator[int, None, None]:
|
||||||
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
|
key = TPLinkSmartHomeProtocol.INITIALIZATION_VECTOR
|
||||||
for cipherbyte in ciphertext:
|
for cipherbyte in ciphertext:
|
||||||
plainbyte = key ^ cipherbyte
|
plainbyte = key ^ cipherbyte
|
||||||
|
Loading…
Reference in New Issue
Block a user