Add commands to control the wifi settings (#45)

* Add commands to control the wifi settings

Enables initial provisioning and changing the wifi network later on without the official app

* new api to smartdevice: wifi_scan() and wifi_join(ssid, password, keytype)
* cli: new subcommand 'wifi' with two commands: scan and join

* update readme to initial setup

* improvements based on code review, f-strings++
This commit is contained in:
Teemu R 2020-04-20 18:57:33 +02:00 committed by GitHub
parent b73c0d222e
commit 7f625cd1c2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 75 additions and 14 deletions

View File

@ -44,7 +44,13 @@ which you can find by adding `--help` after the command, e.g. `kasa emeter --hel
If no command is given, the `state` command will be executed to query the device state. If no command is given, the `state` command will be executed to query the device state.
## Initial Setup
You can provision your device without any extra apps by using the `kasa wifi` command:
1. If the device is unprovisioned, connect to its open network
2. Use `kasa discover` (or check the routes) to locate the IP address of the device (likely 192.168.0.1)
3. Scan for available networks using `kasa wifi scan`
4. Join/change the network using `kasa wifi join` command, see `--help` for details.
## Discovering devices ## Discovering devices
The devices can be discovered either by using `kasa discover` or by calling `kasa` without any parameters. The devices can be discovered either by using `kasa discover` or by calling `kasa` without any parameters.

View File

@ -50,7 +50,7 @@ def cli(ctx, host, alias, target, debug, bulb, plug, strip):
return return
if alias is not None and host is None: if alias is not None and host is None:
click.echo("Alias is given, using discovery to find host %s" % alias) click.echo(f"Alias is given, using discovery to find host {alias}")
host = find_host_from_alias(alias=alias, target=target) host = find_host_from_alias(alias=alias, target=target)
if host: if host:
click.echo(f"Found hostname is {host}") click.echo(f"Found hostname is {host}")
@ -81,6 +81,37 @@ def cli(ctx, host, alias, target, debug, bulb, plug, strip):
ctx.invoke(state) ctx.invoke(state)
@cli.group()
@pass_dev
def wifi(dev):
"""Commands to control wifi settings."""
@wifi.command()
@pass_dev
def scan(dev):
"""Scan for available wifi networks."""
click.echo("Scanning for wifi networks, wait a second..")
devs = asyncio.run(dev.wifi_scan())
click.echo(f"Found {len(devs)} wifi networks!")
for dev in devs:
click.echo(f"\t {dev}")
@wifi.command()
@click.argument("ssid")
@click.option("--password", prompt=True, hide_input=True)
@click.option("--keytype", default=3)
@pass_dev
def join(dev: SmartDevice, ssid, password, keytype):
"""Join the given wifi network."""
click.echo("Asking the device to connect to {ssid}.." % (ssid))
res = asyncio.run(dev.wifi_join(ssid, password, keytype=keytype))
click.echo(
f"Response: {res} - if the device is not able to join the network, it will revert back to its previous state."
)
@cli.command() @cli.command()
@click.option("--scrub/--no-scrub", default=True) @click.option("--scrub/--no-scrub", default=True)
@click.pass_context @click.pass_context
@ -118,7 +149,7 @@ def dump_discover(ctx, scrub):
model = dev["system"]["get_sysinfo"]["model"] model = dev["system"]["get_sysinfo"]["model"]
hw_version = dev["system"]["get_sysinfo"]["hw_ver"] hw_version = dev["system"]["get_sysinfo"]["hw_ver"]
save_to = f"{model}_{hw_version}.json" save_to = f"{model}_{hw_version}.json"
click.echo("Saving info to %s" % save_to) click.echo(f"Saving info to {save_to}")
with open(save_to, "w") as f: with open(save_to, "w") as f:
json.dump(dev, f, sort_keys=True, indent=4) json.dump(dev, f, sort_keys=True, indent=4)
f.write("\n") f.write("\n")
@ -132,7 +163,7 @@ def dump_discover(ctx, scrub):
def discover(ctx, timeout, discover_only, dump_raw): def discover(ctx, timeout, discover_only, dump_raw):
"""Discover devices in the network.""" """Discover devices in the network."""
target = ctx.parent.params["target"] target = ctx.parent.params["target"]
click.echo("Discovering devices for %s seconds" % timeout) click.echo(f"Discovering devices for {timeout} seconds")
found_devs = asyncio.run( found_devs = asyncio.run(
Discover.discover(target=target, timeout=timeout, return_raw=dump_raw) Discover.discover(target=target, timeout=timeout, return_raw=dump_raw)
) )
@ -153,8 +184,7 @@ def find_host_from_alias(alias, target="255.255.255.255", timeout=1, attempts=3)
"""Discover a device identified by its alias.""" """Discover a device identified by its alias."""
host = None host = None
click.echo( click.echo(
"Trying to discover %s using %s attempts of %s seconds" f"Trying to discover {alias} using {attempts} attempts of {timeout} seconds"
% (alias, attempts, timeout)
) )
for attempt in range(1, attempts): for attempt in range(1, attempts):
click.echo(f"Attempt {attempt} of {attempts}") click.echo(f"Attempt {attempt} of {attempts}")
@ -205,9 +235,9 @@ def state(ctx, dev: SmartDevice):
for k, v in dev.state_information.items(): for k, v in dev.state_information.items():
click.echo(f"{k}: {v}") click.echo(f"{k}: {v}")
click.echo(click.style("== Generic information ==", bold=True)) click.echo(click.style("== Generic information ==", bold=True))
click.echo("Time: {}".format(asyncio.run(dev.get_time()))) click.echo(f"Time: {asyncio.run(dev.get_time())}")
click.echo("Hardware: {}".format(dev.hw_info["hw_ver"])) click.echo(f"Hardware: {dev.hw_info['hw_ver']}")
click.echo("Software: {}".format(dev.hw_info["sw_ver"])) click.echo(f"Software: {dev.hw_info['sw_ver']}")
click.echo(f"MAC (rssi): {dev.mac} ({dev.rssi})") click.echo(f"MAC (rssi): {dev.mac} ({dev.rssi})")
click.echo(f"Location: {dev.location}") click.echo(f"Location: {dev.location}")
@ -289,9 +319,9 @@ def brightness(dev, brightness):
click.echo("This device does not support brightness.") click.echo("This device does not support brightness.")
return return
if brightness is None: if brightness is None:
click.echo("Brightness: %s" % dev.brightness) click.echo(f"Brightness: {dev.brightness}")
else: else:
click.echo("Setting brightness to %s" % brightness) click.echo(f"Setting brightness to {brightness}")
asyncio.run(dev.set_brightness(brightness)) asyncio.run(dev.set_brightness(brightness))
@ -326,7 +356,7 @@ def temperature(dev: SmartBulb, temperature):
def hsv(dev, ctx, h, s, v): def hsv(dev, ctx, h, s, v):
"""Get or set color in HSV. (Bulb only).""" """Get or set color in HSV. (Bulb only)."""
if h is None or s is None or v is None: if h is None or s is None or v is None:
click.echo("Current HSV: %s %s %s" % dev.hsv) click.echo(f"Current HSV: {dev.hsv}")
elif s is None or v is None: elif s is None or v is None:
raise click.BadArgumentUsage("Setting a color requires 3 values.", ctx) raise click.BadArgumentUsage("Setting a color requires 3 values.", ctx)
else: else:
@ -340,10 +370,10 @@ def hsv(dev, ctx, h, s, v):
def led(dev, state): def led(dev, state):
"""Get or set (Plug's) led state.""" """Get or set (Plug's) led state."""
if state is not None: if state is not None:
click.echo("Turning led to %s" % state) click.echo(f"Turning led to {state}")
asyncio.run(dev.set_led(state)) asyncio.run(dev.set_led(state))
else: else:
click.echo("LED state: %s" % dev.led) click.echo(f"LED state: {dev.led}")
@cli.command() @cli.command()

View File

@ -14,9 +14,10 @@ http://www.apache.org/licenses/LICENSE-2.0
import functools import functools
import inspect import inspect
import logging import logging
from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from typing import Any, Dict, Optional from typing import Any, Dict, List, Optional
from kasa.protocol import TPLinkSmartHomeProtocol from kasa.protocol import TPLinkSmartHomeProtocol
@ -33,6 +34,14 @@ class DeviceType(Enum):
Unknown = -1 Unknown = -1
@dataclass
class WifiNetwork:
"""Wifi network container."""
ssid: str
key_type: int
class SmartDeviceException(Exception): class SmartDeviceException(Exception):
"""Base exception for device errors.""" """Base exception for device errors."""
@ -564,6 +573,22 @@ class SmartDevice:
""" """
return self.mac return self.mac
async def wifi_scan(self) -> List[WifiNetwork]:
"""Scan for available wifi networks."""
info = await self._query_helper("netif", "get_scaninfo", {"refresh": 1})
if "ap_list" not in info:
raise SmartDeviceException("Invalid response for wifi scan: %s" % info)
return [WifiNetwork(**x) for x in info["ap_list"]]
async def wifi_join(self, ssid, password, keytype=3):
"""Join the given wifi network.
If joining the network fails, the device will return to AP mode after a while.
"""
payload = {"ssid": ssid, "password": password, "key_type": keytype}
return await self._query_helper("netif", "set_stainfo", payload)
@property @property
def device_type(self) -> DeviceType: def device_type(self) -> DeviceType:
"""Return the device type.""" """Return the device type."""