1
0
mirror of https://github.com/mjg59/python-broadlink.git synced 2024-09-21 04:20:36 +02:00

Move constants to const.py (#584)

This commit is contained in:
Felipe Martins Diel 2021-04-29 18:59:58 -03:00 committed by GitHub
parent e1f3b83efd
commit d48d1347a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 27 additions and 18 deletions

View File

@ -4,6 +4,7 @@ import socket
import typing as t import typing as t
from . import exceptions as e from . import exceptions as e
from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_TIMEOUT
from .alarm import S1C from .alarm import S1C
from .climate import hysen from .climate import hysen
from .cover import dooya from .cover import dooya
@ -143,8 +144,8 @@ def gendevice(
def hello( def hello(
host: str, host: str,
port: int = 80, port: int = DEFAULT_PORT,
timeout: int = 10, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str = None,
) -> Device: ) -> Device:
"""Direct device discovery. """Direct device discovery.
@ -162,10 +163,10 @@ def hello(
def discover( def discover(
timeout: int = 10, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = "255.255.255.255", discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = 80, discover_ip_port: int = DEFAULT_PORT,
) -> t.List[Device]: ) -> t.List[Device]:
"""Discover devices connected to the local network.""" """Discover devices connected to the local network."""
responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port) responses = scan(timeout, local_ip_address, discover_ip_address, discover_ip_port)
@ -173,10 +174,10 @@ def discover(
def xdiscover( def xdiscover(
timeout: int = 10, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = "255.255.255.255", discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = 80, discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[Device, None, None]: ) -> t.Generator[Device, None, None]:
"""Discover devices connected to the local network. """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 = 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_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 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() sock.close()

5
broadlink/const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants."""
DEFAULT_BCAST_ADDR = "255.255.255.255"
DEFAULT_PORT = 80
DEFAULT_RETRY_INTVL = 1
DEFAULT_TIMEOUT = 10

View File

@ -9,16 +9,17 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from . import exceptions as e from . import exceptions as e
from .const import DEFAULT_BCAST_ADDR, DEFAULT_PORT, DEFAULT_RETRY_INTVL, DEFAULT_TIMEOUT
from .protocol import Datetime from .protocol import Datetime
HelloResponse = t.Tuple[int, t.Tuple[str, int], str, str, bool] HelloResponse = t.Tuple[int, t.Tuple[str, int], str, str, bool]
def scan( def scan(
timeout: int = 10, timeout: int = DEFAULT_TIMEOUT,
local_ip_address: str = None, local_ip_address: str = None,
discover_ip_address: str = "255.255.255.255", discover_ip_address: str = DEFAULT_BCAST_ADDR,
discover_ip_port: int = 80, discover_ip_port: int = DEFAULT_PORT,
) -> t.Generator[HelloResponse, None, None]: ) -> t.Generator[HelloResponse, None, None]:
"""Broadcast a hello message and yield responses.""" """Broadcast a hello message and yield responses."""
conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@ -70,7 +71,7 @@ def scan(
conn.close() 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. """Send a ping packet to an address.
This packet feeds the watchdog timer of firmwares >= v53. This packet feeds the watchdog timer of firmwares >= v53.
@ -97,7 +98,7 @@ class Device:
host: t.Tuple[str, int], host: t.Tuple[str, int],
mac: t.Union[bytes, str], mac: t.Union[bytes, str],
devtype: int, devtype: int,
timeout: int = 10, timeout: int = DEFAULT_TIMEOUT,
name: str = "", name: str = "",
model: str = "", model: str = "",
manufacturer: str = "", manufacturer: str = "",
@ -292,7 +293,7 @@ class Device:
while True: while True:
time_left = timeout - (time.time() - start_time) 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) conn.sendto(packet, self.host)
try: try:

View File

@ -5,6 +5,7 @@ import codecs
import time import time
import broadlink import broadlink
from broadlink.const import DEFAULT_PORT
from broadlink.exceptions import ReadError, StorageError from broadlink.exceptions import ReadError, StorageError
TICK = 32.84 TICK = 32.84
@ -102,7 +103,7 @@ elif args.mac:
mac = bytearray.fromhex(args.mac) mac = bytearray.fromhex(args.mac)
if args.host or args.device: if args.host or args.device:
dev = broadlink.gendevice(devtype, (host, 80), mac) dev = broadlink.gendevice(devtype, (host, DEFAULT_PORT), mac)
dev.auth() dev.auth()
if args.joinwifi: if args.joinwifi:

View File

@ -2,12 +2,13 @@
import argparse import argparse
import broadlink import broadlink
from broadlink.const import DEFAULT_BCAST_ADDR, DEFAULT_TIMEOUT
from broadlink.exceptions import StorageError from broadlink.exceptions import StorageError
parser = argparse.ArgumentParser(fromfile_prefix_chars='@') 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("--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() args = parser.parse_args()
print("Discovering...") print("Discovering...")