From 867b7b88309c6ccf39d661aae92afb25c487b24f Mon Sep 17 00:00:00 2001 From: Teemu R Date: Mon, 17 Jun 2024 10:37:08 +0200 Subject: [PATCH] Add time sync command (#951) Allows setting the device time (on SMART devices) to the current time. Fixes also setting the time which was previously broken. --- docs/source/cli.rst | 5 +++++ kasa/cli.py | 33 +++++++++++++++++++++++++++++++-- kasa/smart/modules/time.py | 8 +++++++- kasa/tests/test_cli.py | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 75 insertions(+), 3 deletions(-) diff --git a/docs/source/cli.rst b/docs/source/cli.rst index dad754d2..7d4eb080 100644 --- a/docs/source/cli.rst +++ b/docs/source/cli.rst @@ -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. diff --git a/kasa/cli.py b/kasa/cli.py index 39f6636f..a8d8b6ec 100755 --- a/kasa/cli.py +++ b/kasa/cli.py @@ -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 @@ -967,15 +968,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) diff --git a/kasa/smart/modules/time.py b/kasa/smart/modules/time.py index 958cf9e2..3c2b96af 100644 --- a/kasa/smart/modules/time.py +++ b/kasa/smart/modules/time.py @@ -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(), + }, ) diff --git a/kasa/tests/test_cli.py b/kasa/tests/test_cli.py index 2104de05..41b1e1ad 100644 --- a/kasa/tests/test_cli.py +++ b/kasa/tests/test_cli.py @@ -31,6 +31,7 @@ from kasa.cli import ( state, sysinfo, temperature, + time, toggle, update_credentials, wifi, @@ -260,6 +261,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: