diff --git a/README.md b/README.md index 3ed5a459..8a0ef604 100644 --- a/README.md +++ b/README.md @@ -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. +## 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 The devices can be discovered either by using `kasa discover` or by calling `kasa` without any parameters. diff --git a/kasa/cli.py b/kasa/cli.py index 7201888b..c523e5c2 100755 --- a/kasa/cli.py +++ b/kasa/cli.py @@ -50,7 +50,7 @@ def cli(ctx, host, alias, target, debug, bulb, plug, strip): return 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) if 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) +@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() @click.option("--scrub/--no-scrub", default=True) @click.pass_context @@ -118,7 +149,7 @@ def dump_discover(ctx, scrub): model = dev["system"]["get_sysinfo"]["model"] hw_version = dev["system"]["get_sysinfo"]["hw_ver"] 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: json.dump(dev, f, sort_keys=True, indent=4) f.write("\n") @@ -132,7 +163,7 @@ def dump_discover(ctx, scrub): def discover(ctx, timeout, discover_only, dump_raw): """Discover devices in the network.""" 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( 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.""" host = None click.echo( - "Trying to discover %s using %s attempts of %s seconds" - % (alias, attempts, timeout) + f"Trying to discover {alias} using {attempts} attempts of {timeout} seconds" ) for attempt in range(1, 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(): click.echo(f"{k}: {v}") click.echo(click.style("== Generic information ==", bold=True)) - click.echo("Time: {}".format(asyncio.run(dev.get_time()))) - click.echo("Hardware: {}".format(dev.hw_info["hw_ver"])) - click.echo("Software: {}".format(dev.hw_info["sw_ver"])) + click.echo(f"Time: {asyncio.run(dev.get_time())}") + click.echo(f"Hardware: {dev.hw_info['hw_ver']}") + click.echo(f"Software: {dev.hw_info['sw_ver']}") click.echo(f"MAC (rssi): {dev.mac} ({dev.rssi})") click.echo(f"Location: {dev.location}") @@ -289,9 +319,9 @@ def brightness(dev, brightness): click.echo("This device does not support brightness.") return if brightness is None: - click.echo("Brightness: %s" % dev.brightness) + click.echo(f"Brightness: {dev.brightness}") else: - click.echo("Setting brightness to %s" % brightness) + click.echo(f"Setting brightness to {brightness}") asyncio.run(dev.set_brightness(brightness)) @@ -326,7 +356,7 @@ def temperature(dev: SmartBulb, temperature): def hsv(dev, ctx, h, s, v): """Get or set color in HSV. (Bulb only).""" 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: raise click.BadArgumentUsage("Setting a color requires 3 values.", ctx) else: @@ -340,10 +370,10 @@ def hsv(dev, ctx, h, s, v): def led(dev, state): """Get or set (Plug's) led state.""" 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)) else: - click.echo("LED state: %s" % dev.led) + click.echo(f"LED state: {dev.led}") @cli.command() diff --git a/kasa/smartdevice.py b/kasa/smartdevice.py index e18c8c28..417c6a55 100755 --- a/kasa/smartdevice.py +++ b/kasa/smartdevice.py @@ -14,9 +14,10 @@ http://www.apache.org/licenses/LICENSE-2.0 import functools import inspect import logging +from dataclasses import dataclass from datetime import datetime from enum import Enum -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from kasa.protocol import TPLinkSmartHomeProtocol @@ -33,6 +34,14 @@ class DeviceType(Enum): Unknown = -1 +@dataclass +class WifiNetwork: + """Wifi network container.""" + + ssid: str + key_type: int + + class SmartDeviceException(Exception): """Base exception for device errors.""" @@ -564,6 +573,22 @@ class SmartDevice: """ 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 def device_type(self) -> DeviceType: """Return the device type."""