Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2024-12-22 19:23:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/master' into feat/hub_pairing

Browse Source
This commit is contained in:
sdb9696 2024-06-17 10:05:36 +01:00
commit fd68468f2e
4 changed files with 107 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
docs/source/cli.rst
View File

@ -58,6 +58,11 @@ As with all other commands, you can also pass ``--help`` to both ``join`` and ``
However, note that communications with devices provisioned using this method will stop working
when connected to the cloud.
.. note::
Some commands do not work if the device time is out-of-sync.
You can use ``kasa time sync`` command to set the device time from the system where the command is run.
.. warning::
At least some devices (e.g., Tapo lights L530 and L900) are known to have a watchdog that reboots them every 10 minutes if they are unable to connect to the cloud.

80
kasa/cli.py
View File

@ -9,6 +9,7 @@ import logging
import re
import sys
from contextlib import asynccontextmanager
from datetime import datetime
from functools import singledispatch, wraps
from pprint import pformat as pf
from typing import Any, cast
@ -70,6 +71,12 @@ except ImportError:
echo = _do_echo
def error(msg: str):
"""Print an error and exit."""
echo(f"[bold red]{msg}[/bold red]")
sys.exit(1)
TYPE_TO_CLASS = {
"plug": IotPlug,
"switch": IotWallSwitch,
@ -366,6 +373,9 @@ async def cli(
credentials = None
if host is None:
if ctx.invoked_subcommand and ctx.invoked_subcommand != "discover":
error("Only discover is available without --host or --alias")
echo("No host name given, trying discovery..")
return await ctx.invoke(discover)
@ -763,7 +773,7 @@ async def emeter(dev: Device, index: int, name: str, year, month, erase):
"""
if index is not None or name is not None:
if not dev.is_strip:
echo("Index and name are only for power strips!")
error("Index and name are only for power strips!")
return
if index is not None:
@ -773,11 +783,11 @@ async def emeter(dev: Device, index: int, name: str, year, month, erase):
echo("[bold]== Emeter ==[/bold]")
if not dev.has_emeter:
echo("Device has no emeter")
error("Device has no emeter")
return
if (year or month or erase) and not isinstance(dev, IotDevice):
echo("Device has no historical statistics")
error("Device has no historical statistics")
return
else:
dev = cast(IotDevice, dev)
@ -864,7 +874,7 @@ async def usage(dev: Device, year, month, erase):
async def brightness(dev: Device, brightness: int, transition: int):
"""Get or set brightness."""
if not (light := dev.modules.get(Module.Light)) or not light.is_dimmable:
echo("This device does not support brightness.")
error("This device does not support brightness.")
return
if brightness is None:
@ -884,7 +894,7 @@ async def brightness(dev: Device, brightness: int, transition: int):
async def temperature(dev: Device, temperature: int, transition: int):
"""Get or set color temperature."""
if not (light := dev.modules.get(Module.Light)) or not light.is_variable_color_temp:
echo("Device does not support color temperature")
error("Device does not support color temperature")
return
if temperature is None:
@ -910,7 +920,7 @@ async def temperature(dev: Device, temperature: int, transition: int):
async def effect(dev: Device, ctx, effect):
"""Set an effect."""
if not (light_effect := dev.modules.get(Module.LightEffect)):
echo("Device does not support effects")
error("Device does not support effects")
return
if effect is None:
echo(
@ -938,7 +948,7 @@ async def effect(dev: Device, ctx, effect):
async def hsv(dev: Device, ctx, h, s, v, transition):
"""Get or set color in HSV."""
if not (light := dev.modules.get(Module.Light)) or not light.is_color:
echo("Device does not support colors")
error("Device does not support colors")
return
if h is None and s is None and v is None:
@ -957,7 +967,7 @@ async def hsv(dev: Device, ctx, h, s, v, transition):
async def led(dev: Device, state):
"""Get or set (Plug's) led state."""
if not (led := dev.modules.get(Module.Led)):
echo("Device does not support led.")
error("Device does not support led.")
return
if state is not None:
echo(f"Turning led to {state}")
@ -967,15 +977,43 @@ async def led(dev: Device, state):
return led.led
@cli.command()
@cli.group(invoke_without_command=True)
@click.pass_context
async def time(ctx: click.Context):
"""Get and set time."""
if ctx.invoked_subcommand is None:
await ctx.invoke(time_get)
@time.command(name="get")
@pass_dev
async def time(dev):
async def time_get(dev: Device):
"""Get the device time."""
res = dev.time
echo(f"Current time: {res}")
return res
@time.command(name="sync")
@pass_dev
async def time_sync(dev: SmartDevice):
"""Set the device time to current time."""
if not isinstance(dev, SmartDevice):
raise NotImplementedError("setting time currently only implemented on smart")
if (time := dev.modules.get(Module.Time)) is None:
echo("Device does not have time module")
return
echo("Old time: %s" % time.time)
local_tz = datetime.now().astimezone().tzinfo
await time.set_time(datetime.now(tz=local_tz))
await dev.update()
echo("New time: %s" % time.time)
@cli.command()
@click.option("--index", type=int, required=False)
@click.option("--name", type=str, required=False)
@ -985,7 +1023,7 @@ async def on(dev: Device, index: int, name: str, transition: int):
"""Turn the device on."""
if index is not None or name is not None:
if not dev.children:
echo("Index and name are only for devices with children.")
error("Index and name are only for devices with children.")
return
if index is not None:
@ -1006,7 +1044,7 @@ async def off(dev: Device, index: int, name: str, transition: int):
"""Turn the device off."""
if index is not None or name is not None:
if not dev.children:
echo("Index and name are only for devices with children.")
error("Index and name are only for devices with children.")
return
if index is not None:
@ -1027,7 +1065,7 @@ async def toggle(dev: Device, index: int, name: str, transition: int):
"""Toggle the device on/off."""
if index is not None or name is not None:
if not dev.children:
echo("Index and name are only for devices with children.")
error("Index and name are only for devices with children.")
return
if index is not None:
@ -1067,7 +1105,7 @@ def _schedule_list(dev, type):
for rule in sched.rules:
print(rule)
else:
echo(f"No rules of type {type}")
error(f"No rules of type {type}")
return sched.rules
@ -1083,7 +1121,7 @@ async def delete_rule(dev, id):
echo(f"Deleting rule id {id}")
return await schedule.delete_rule(rule_to_delete)
else:
echo(f"No rule with id {id} was found")
error(f"No rule with id {id} was found")
@cli.group(invoke_without_command=True)
@ -1099,7 +1137,7 @@ async def presets(ctx):
def presets_list(dev: IotBulb):
"""List presets."""
if not dev.is_bulb or not isinstance(dev, IotBulb):
echo("Presets only supported on iot bulbs")
error("Presets only supported on iot bulbs")
return
for preset in dev.presets:
@ -1121,7 +1159,7 @@ async def presets_modify(dev: IotBulb, index, brightness, hue, saturation, tempe
if preset.index == index:
break
else:
echo(f"No preset found for index {index}")
error(f"No preset found for index {index}")
return
if brightness is not None:
@ -1146,7 +1184,7 @@ async def presets_modify(dev: IotBulb, index, brightness, hue, saturation, tempe
async def turn_on_behavior(dev: IotBulb, type, last, preset):
"""Modify bulb turn-on behavior."""
if not dev.is_bulb or not isinstance(dev, IotBulb):
echo("Presets only supported on iot bulbs")
error("Presets only supported on iot bulbs")
return
settings = await dev.get_turn_on_behavior()
echo(f"Current turn on behavior: {settings}")
@ -1183,9 +1221,7 @@ async def turn_on_behavior(dev: IotBulb, type, last, preset):
async def update_credentials(dev, username, password):
"""Update device credentials for authenticated devices."""
if not isinstance(dev, SmartDevice):
raise NotImplementedError(
"Credentials can only be updated on authenticated devices."
)
error("Credentials can only be updated on authenticated devices.")
click.confirm("Do you really want to replace the existing credentials?", abort=True)
@ -1242,7 +1278,7 @@ async def feature(dev: Device, child: str, name: str, value):
return
if name not in dev.features:
echo(f"No feature by name '{name}'")
error(f"No feature by name '{name}'")
return
feat = dev.features[name]

8
kasa/smart/modules/time.py
View File

@ -51,7 +51,13 @@ class Time(SmartModule):
async def set_time(self, dt: datetime):
"""Set device time."""
unixtime = mktime(dt.timetuple())
offset = cast(timedelta, dt.utcoffset())
diff = offset / timedelta(minutes=1)
return await self.call(
"set_device_time",
{"timestamp": unixtime, "time_diff": dt.utcoffset(), "region": dt.tzname()},
{
"timestamp": int(unixtime),
"time_diff": int(diff),
"region": dt.tzname(),
},
)

42
kasa/tests/test_cli.py
View File

@ -33,6 +33,7 @@ from kasa.cli import (
state,
sysinfo,
temperature,
time,
toggle,
update_credentials,
wifi,
@ -302,6 +303,37 @@ async def test_update_credentials(dev, runner):
)
async def test_time_get(dev, runner):
"""Test time get command."""
res = await runner.invoke(
time,
obj=dev,
)
assert res.exit_code == 0
assert "Current time: " in res.output
@device_smart
async def test_time_sync(dev, mocker, runner):
"""Test time sync command.
Currently implemented only for SMART.
"""
update = mocker.patch.object(dev, "update")
set_time_mock = mocker.spy(dev.modules[Module.Time], "set_time")
res = await runner.invoke(
time,
["sync"],
obj=dev,
)
set_time_mock.assert_called()
update.assert_called()
assert res.exit_code == 0
assert "Old time: " in res.output
assert "New time: " in res.output
async def test_emeter(dev: Device, mocker, runner):
res = await runner.invoke(emeter, obj=dev)
if not dev.has_emeter:
@ -471,12 +503,12 @@ async def test_led(dev: Device, runner: CliRunner):
async def test_json_output(dev: Device, mocker, runner):
"""Test that the json output produces correct output."""
mocker.patch("kasa.Discover.discover", return_value={"127.0.0.1": dev})
# These will mock the features to avoid accessing non-existing
mocker.patch("kasa.Discover.discover_single", return_value=dev)
# These will mock the features to avoid accessing non-existing ones
mocker.patch("kasa.device.Device.features", return_value={})
mocker.patch("kasa.iot.iotdevice.IotDevice.features", return_value={})
res = await runner.invoke(cli, ["--json", "state"], obj=dev)
res = await runner.invoke(cli, ["--host", "127.0.0.1", "--json", "state"], obj=dev)
assert res.exit_code == 0
assert json.loads(res.output) == dev.internal_state
@ -799,7 +831,7 @@ async def test_errors(mocker, runner):
)
assert res.exit_code == 1
assert (
"Raised error: Managed to invoke callback without a context object of type 'Device' existing."
"Only discover is available without --host or --alias"
in res.output.replace("\n", "") # Remove newlines from rich formatting
)
assert isinstance(res.exception, SystemExit)
@ -870,7 +902,7 @@ async def test_feature_missing(mocker, runner):
)
assert "No feature by name 'missing'" in res.output
assert "== Features ==" not in res.output
assert res.exit_code == 0
assert res.exit_code == 1
async def test_feature_set(mocker, runner):