diff --git a/broadlink/__init__.py b/broadlink/__init__.py index 73ff30b..693820d 100644 --- a/broadlink/__init__.py +++ b/broadlink/__init__.py @@ -4,6 +4,7 @@ import socket import typing as t from . import exceptions as e +from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_TIMEOUT from .alarm import S1C from .climate import hysen from .cover import dooya @@ -143,8 +144,8 @@ def gendevice( def hello( host: str, - port: int = 80, - timeout: int = 10, + port: int = DEFAULT_PORT, + timeout: int = DEFAULT_TIMEOUT, local_ip_address: str = None, ) -> Device: """Direct device discovery. @@ -162,10 +163,10 @@ def hello( def discover( - timeout: int = 10, + timeout: int = DEFAULT_TIMEOUT, local_ip_address: str = None, - discover_ip_address: str = "255.255.255.255", - discover_ip_port: int = 80, + discover_ip_address: str = DEFAULT_BCAST_ADDR, + discover_ip_port: int = DEFAULT_PORT, ) -> t.List[Device]: """Discover devices connected to the local network.""" responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port) @@ -173,10 +174,10 @@ def discover( def xdiscover( - timeout: int = 10, + timeout: int = DEFAULT_TIMEOUT, local_ip_address: str = None, - discover_ip_address: str = "255.255.255.255", - discover_ip_port: int = 80, + discover_ip_address: str = DEFAULT_BCAST_ADDR, + discover_ip_port: int = DEFAULT_PORT, ) -> t.Generator[Device, None, None]: """Discover devices connected to the local network. @@ -218,5 +219,5 @@ def setup(ssid: str, password: str, security_mode: int) -> None: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Internet # UDP sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - sock.sendto(payload, ("255.255.255.255", 80)) + sock.sendto(payload, (DEFAULT_BCAST_ADDR, DEFAULT_PORT)) sock.close() diff --git a/broadlink/const.py b/broadlink/const.py new file mode 100644 index 0000000..19c37f5 --- /dev/null +++ b/broadlink/const.py @@ -0,0 +1,5 @@ +"""Constants.""" +DEFAULT_BCAST_ADDR = "255.255.255.255" +DEFAULT_PORT = 80 +DEFAULT_RETRY_INTVL = 1 +DEFAULT_TIMEOUT = 10 diff --git a/broadlink/device.py b/broadlink/device.py index 1399628..c1f8fa8 100644 --- a/broadlink/device.py +++ b/broadlink/device.py @@ -9,16 +9,17 @@ from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from . import exceptions as e +from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_RETRY_INTVL, DEFAULT_TIMEOUT from .protocol import Datetime HelloResponse = t.Tuple[int, t.Tuple[str, int], str, str, bool] def scan( - timeout: int = 10, + timeout: int = DEFAULT_TIMEOUT, local_ip_address: str = None, - discover_ip_address: str = "255.255.255.255", - discover_ip_port: int = 80, + discover_ip_address: str = DEFAULT_BCAST_ADDR, + discover_ip_port: int = DEFAULT_PORT, ) -> t.Generator[HelloResponse, None, None]: """Broadcast a hello message and yield responses.""" conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) @@ -70,7 +71,7 @@ def scan( conn.close() -def ping(address: str, port: int = 80) -> None: +def ping(address: str, port: int = DEFAULT_PORT) -> None: """Send a ping packet to an address. This packet feeds the watchdog timer of firmwares >= v53. @@ -97,7 +98,7 @@ class Device: host: t.Tuple[str, int], mac: t.Union[bytes, str], devtype: int, - timeout: int = 10, + timeout: int = DEFAULT_TIMEOUT, name: str = "", model: str = "", manufacturer: str = "", @@ -292,7 +293,7 @@ class Device: while True: time_left = timeout - (time.time() - start_time) - conn.settimeout(min(1, time_left)) + conn.settimeout(min(DEFAULT_RETRY_INTVL, time_left)) conn.sendto(packet, self.host) try: diff --git a/cli/broadlink_cli b/cli/broadlink_cli index 36a83e1..f7a24ad 100755 --- a/cli/broadlink_cli +++ b/cli/broadlink_cli @@ -5,6 +5,7 @@ import codecs import time import broadlink +from broadlink.const import DEFAULT_PORT from broadlink.exceptions import ReadError, StorageError TICK = 32.84 @@ -102,7 +103,7 @@ elif args.mac: mac = bytearray.fromhex(args.mac) if args.host or args.device: - dev = broadlink.gendevice(devtype, (host, 80), mac) + dev = broadlink.gendevice(devtype, (host, DEFAULT_PORT), mac) dev.auth() if args.joinwifi: diff --git a/cli/broadlink_discovery b/cli/broadlink_discovery index c098c91..477e1bd 100755 --- a/cli/broadlink_discovery +++ b/cli/broadlink_discovery @@ -2,12 +2,13 @@ import argparse import broadlink +from broadlink.const import DEFAULT_BCAST_ADDR, DEFAULT_TIMEOUT from broadlink.exceptions import StorageError parser = argparse.ArgumentParser(fromfile_prefix_chars='@') -parser.add_argument("--timeout", type=int, default=5, help="timeout to wait for receiving discovery responses") +parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="timeout to wait for receiving discovery responses") parser.add_argument("--ip", default=None, help="ip address to use in the discovery") -parser.add_argument("--dst-ip", default="255.255.255.255", help="destination ip address to use in the discovery") +parser.add_argument("--dst-ip", default=DEFAULT_BCAST_ADDR, help="destination ip address to use in the discovery") args = parser.parse_args() print("Discovering...")