v/wideq
1
0
Fork 0
mirror of https://github.com/no2chem/wideq.git synced 2025-06-02 15:30:14 -07:00
Code Issues Projects Releases Wiki Activity

Run Black on example.py

Browse Source
This commit is contained in:
Adrian Sampson 2020-11-28 17:05:17 -05:00
parent 6c2232a7db
commit 8319b2ef97
1 changed files with 94 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
example.py
View File

@ -10,7 +10,7 @@ import os.path
import logging import logging
from typing import List from typing import List
STATE_FILE = 'wideq_state.json' STATE_FILE = "wideq_state.json"
LOGGER = logging.getLogger("wideq.example") LOGGER = logging.getLogger("wideq.example")
@ -20,9 +20,9 @@ def authenticate(gateway):
""" """
login_url = gateway.oauth_url() login_url = gateway.oauth_url()
print('Log in here:') print("Log in here:")
print(login_url) print(login_url)
print('Then paste the URL where the browser is redirected:') print("Then paste the URL where the browser is redirected:")
callback_url = input() callback_url = input()
return wideq.Auth.from_url(gateway, callback_url) return wideq.Auth.from_url(gateway, callback_url)
@ -31,7 +31,7 @@ def ls(client):
"""List the user's devices.""" """List the user's devices."""
for device in client.devices: for device in client.devices:
print('{0.id}: {0.name} ({0.type.name} {0.model_id})'.format(device)) print("{0.id}: {0.name} ({0.type.name} {0.model_id})".format(device))
def gen_mon(client, device_id): def gen_mon(client, device_id):
@ -46,27 +46,33 @@ def gen_mon(client, device_id):
try: try:
while True: while True:
time.sleep(1) time.sleep(1)
print('Polling...') print("Polling...")
data = mon.poll() data = mon.poll()
if data: if data:
try: try:
res = model.decode_monitor(data) res = model.decode_monitor(data)
except ValueError: except ValueError:
print('status data: {!r}'.format(data)) print("status data: {!r}".format(data))
else: else:
for key, value in res.items(): for key, value in res.items():
try: try:
desc = model.value(key) desc = model.value(key)
except KeyError: except KeyError:
print('- {}: {}'.format(key, value)) print("- {}: {}".format(key, value))
if isinstance(desc, wideq.EnumValue): if isinstance(desc, wideq.EnumValue):
print('- {}: {}'.format( print(
"- {}: {}".format(
key, desc.options.get(value, value) key, desc.options.get(value, value)
)) )
)
elif isinstance(desc, wideq.RangeValue): elif isinstance(desc, wideq.RangeValue):
print('- {0}: {1} ({2.min}-{2.max})'.format( print(
key, value, desc, "- {0}: {1} ({2.min}-{2.max})".format(
)) key,
value,
desc,
)
)
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
@ -80,7 +86,7 @@ def ac_mon(ac):
try: try:
ac.monitor_start() ac.monitor_start()
except wideq.core.NotConnectedError: except wideq.core.NotConnectedError:
print('Device not available.') print("Device not available.")
return return
try: try:
@ -89,18 +95,16 @@ def ac_mon(ac):
state = ac.poll() state = ac.poll()
if state: if state:
print( print(
'{1}; ' "{1}; "
'{0.mode.name}; ' "{0.mode.name}; "
'cur {0.temp_cur_f}°F; ' "cur {0.temp_cur_f}°F; "
'cfg {0.temp_cfg_f}°F; ' "cfg {0.temp_cfg_f}°F; "
'fan speed {0.fan_speed.name}' "fan speed {0.fan_speed.name}".format(
.format( state, "on" if state.is_on else "off"
state,
'on' if state.is_on else 'off'
) )
) )
else: else:
print('no state. Wait 1 more second.') print("no state. Wait 1 more second.")
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
@ -121,8 +125,8 @@ def mon(client, device_id):
class UserError(Exception): class UserError(Exception):
"""A user-visible command-line error. """A user-visible command-line error."""
"""
def __init__(self, msg): def __init__(self, msg):
self.msg = msg self.msg = msg
@ -147,11 +151,13 @@ def set_temp(client, device_id, temp):
ac.set_fahrenheit(int(temp)) ac.set_fahrenheit(int(temp))
elif device.type == wideq.client.DeviceType.REFRIGERATOR: elif device.type == wideq.client.DeviceType.REFRIGERATOR:
refrigerator = wideq.RefrigeratorDevice( refrigerator = wideq.RefrigeratorDevice(
client, _force_device(client, device_id)) client, _force_device(client, device_id)
)
refrigerator.set_temp_refrigerator_c(int(temp)) refrigerator.set_temp_refrigerator_c(int(temp))
else: else:
raise UserError( raise UserError(
'set-temp only suported for AC or refrigerator devices') "set-temp only suported for AC or refrigerator devices"
)
def set_temp_freezer(client, device_id, temp): def set_temp_freezer(client, device_id, temp):
@ -161,18 +167,20 @@ def set_temp_freezer(client, device_id, temp):
if device.type == wideq.client.DeviceType.REFRIGERATOR: if device.type == wideq.client.DeviceType.REFRIGERATOR:
refrigerator = wideq.RefrigeratorDevice( refrigerator = wideq.RefrigeratorDevice(
client, _force_device(client, device_id)) client, _force_device(client, device_id)
)
refrigerator.set_temp_freezer_c(int(temp)) refrigerator.set_temp_freezer_c(int(temp))
else: else:
raise UserError( raise UserError(
'set-temp-freezer only suported for refrigerator devices') "set-temp-freezer only suported for refrigerator devices"
)
def turn(client, device_id, on_off): def turn(client, device_id, on_off):
"""Turn on/off an AC device.""" """Turn on/off an AC device."""
ac = wideq.ACDevice(client, _force_device(client, device_id)) ac = wideq.ACDevice(client, _force_device(client, device_id))
ac.set_on(on_off == 'on') ac.set_on(on_off == "on")
def ac_config(client, device_id): def ac_config(client, device_id):
@ -190,26 +198,30 @@ def ac_config(client, device_id):
EXAMPLE_COMMANDS = { EXAMPLE_COMMANDS = {
'ls': ls, "ls": ls,
'mon': mon, "mon": mon,
'set-temp': set_temp, "set-temp": set_temp,
'set-temp-freezer': set_temp_freezer, "set-temp-freezer": set_temp_freezer,
'turn': turn, "turn": turn,
'ac-config': ac_config, "ac-config": ac_config,
} }
def example_command(client, cmd, args): def example_command(client, cmd, args):
func = EXAMPLE_COMMANDS.get(cmd) func = EXAMPLE_COMMANDS.get(cmd)
if not func: if not func:
LOGGER.error("Invalid command: '%s'.\n" LOGGER.error(
"Use one of: %s", cmd, ', '.join(EXAMPLE_COMMANDS)) "Invalid command: '%s'.\n" "Use one of: %s",
cmd,
", ".join(EXAMPLE_COMMANDS),
)
return return
func(client, *args) func(client, *args)
def example(country: str, language: str, verbose: bool, def example(
cmd: str, args: List[str]) -> None: country: str, language: str, verbose: bool, cmd: str, args: List[str]
) -> None:
if verbose: if verbose:
wideq.set_log_level(logging.DEBUG) wideq.set_log_level(logging.DEBUG)
@ -220,8 +232,9 @@ def example(country: str, language: str, verbose: bool,
state = json.load(f) state = json.load(f)
except IOError: except IOError:
state = {} state = {}
LOGGER.debug("No state file found (tried: '%s')", LOGGER.debug(
os.path.abspath(STATE_FILE)) "No state file found (tried: '%s')", os.path.abspath(STATE_FILE)
)
client = wideq.Client.load(state) client = wideq.Client.load(state)
if country: if country:
@ -240,7 +253,7 @@ def example(country: str, language: str, verbose: bool,
break break
except wideq.NotLoggedInError: except wideq.NotLoggedInError:
LOGGER.info('Session expired.') LOGGER.info("Session expired.")
client.refresh() client.refresh()
except UserError as exc: except UserError as exc:
@ -249,54 +262,67 @@ def example(country: str, language: str, verbose: bool,
# Save the updated state. # Save the updated state.
state = client.dump() state = client.dump()
with open(STATE_FILE, 'w') as f: with open(STATE_FILE, "w") as f:
json.dump(state, f) json.dump(state, f)
LOGGER.debug("Wrote state file '%s'", os.path.abspath(STATE_FILE)) LOGGER.debug("Wrote state file '%s'", os.path.abspath(STATE_FILE))
def main() -> None: def main() -> None:
"""The main command-line entry point. """The main command-line entry point."""
"""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description='Interact with the LG SmartThinQ API.' description="Interact with the LG SmartThinQ API."
)
parser.add_argument(
"cmd",
metavar="CMD",
nargs="?",
default="ls",
help=f'one of: {", ".join(EXAMPLE_COMMANDS)}',
)
parser.add_argument(
"args", metavar="ARGS", nargs="*", help="subcommand arguments"
) )
parser.add_argument('cmd', metavar='CMD', nargs='?', default='ls',
help=f'one of: {", ".join(EXAMPLE_COMMANDS)}')
parser.add_argument('args', metavar='ARGS', nargs='*',
help='subcommand arguments')
parser.add_argument( parser.add_argument(
'--country', '-c', "--country",
help=f'country code for account (default: {wideq.DEFAULT_COUNTRY})', "-c",
default=wideq.DEFAULT_COUNTRY help=f"country code for account (default: {wideq.DEFAULT_COUNTRY})",
default=wideq.DEFAULT_COUNTRY,
) )
parser.add_argument( parser.add_argument(
'--language', '-l', "--language",
help=f'language code for the API (default: {wideq.DEFAULT_LANGUAGE})', "-l",
default=wideq.DEFAULT_LANGUAGE help=f"language code for the API (default: {wideq.DEFAULT_LANGUAGE})",
default=wideq.DEFAULT_LANGUAGE,
) )
parser.add_argument( parser.add_argument(
'--verbose', '-v', "--verbose",
help='verbose mode to help debugging', "-v",
action='store_true', default=False help="verbose mode to help debugging",
action="store_true",
default=False,
) )
args = parser.parse_args() args = parser.parse_args()
country_regex = re.compile(r"^[A-Z]{2,3}$") country_regex = re.compile(r"^[A-Z]{2,3}$")
if not country_regex.match(args.country): if not country_regex.match(args.country):
LOGGER.error("Country must be two or three letters" LOGGER.error(
"Country must be two or three letters"
" all upper case (e.g. US, NO, KR) got: '%s'", " all upper case (e.g. US, NO, KR) got: '%s'",
args.country) args.country,
)
exit(1) exit(1)
language_regex = re.compile(r"^[a-z]{2,3}-[A-Z]{2,3}$") language_regex = re.compile(r"^[a-z]{2,3}-[A-Z]{2,3}$")
if not language_regex.match(args.language): if not language_regex.match(args.language):
LOGGER.error("Language must be a combination of language" LOGGER.error(
"Language must be a combination of language"
" and country (e.g. en-US, no-NO, kr-KR)" " and country (e.g. en-US, no-NO, kr-KR)"
" got: '%s'", " got: '%s'",
args.language) args.language,
)
exit(1) exit(1)
example(args.country, args.language, args.verbose, args.cmd, args.args) example(args.country, args.language, args.verbose, args.cmd, args.args)
if __name__ == '__main__': if __name__ == "__main__":
main() main()