python-kasa · sdb9696 · Oct 15, 2024 · Oct 8, 2024 · Oct 8, 2024 · Oct 9, 2024
diff --git a/kasa/cli/common.py b/kasa/cli/common.py
@@ -201,6 +201,8 @@
         # Handle exit request from click.
         if isinstance(exc, click.exceptions.Exit):
             sys.exit(exc.exit_code)
+        if isinstance(exc, click.exceptions.Abort):
+            sys.exit(0)
 
         echo(f"Raised error: {exc}")
         if debug:

diff --git a/kasa/cli/time.py b/kasa/cli/time.py
@@ -5,15 +5,18 @@
 from datetime import datetime
 
 import asyncclick as click
+import zoneinfo
 
 from kasa import (
     Device,
     Module,
 )
-from kasa.smart import SmartDevice
+from kasa.iot import IotDevice
+from kasa.iot.iottimezone import get_matching_timezones
 
 from .common import (
     echo,
+    error,
     pass_dev,
 )
 
@@ -31,25 +34,127 @@
 async def time_get(dev: Device):
     """Get the device time."""
     res = dev.time
-    echo(f"Current time: {res}")
+    echo(f"Current time: {dev.time} ({dev.timezone})")
     return res
 
 
 @time.command(name="sync")
+@click.option(
+    "--timezone",
+    type=str,
+    required=False,
+    default=None,
+    help="IANA timezone name, will use current device timezone if not provided.",
+)
+@click.option(
+    "--skip-confirm",
+    type=str,
+    required=False,
+    default=False,
+    is_flag=True,
+    help="Do not ask to confirm the timezone if an exact match is not found.",
+)
 @pass_dev
-async def time_sync(dev: Device):
+async def time_sync(dev: Device, timezone: str | None, skip_confirm: bool):
     """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
+
+    now = datetime.now()
+
+    tzinfo: zoneinfo.ZoneInfo | None = None
+    if timezone:
+        tzinfo = await _get_timezone(dev, timezone, skip_confirm)
+        if tzinfo.utcoffset(now) != now.astimezone().utcoffset():
+            error(
+                f"{timezone} has a different utc offset to local time,"
+                + "syncing will produce unexpected results."
+            )
+        now = now.replace(tzinfo=tzinfo)
+
+    echo(f"Old time: {time.time} ({time.timezone})")
+
+    await time.set_time(now)
+
+    await dev.update()
+    echo(f"New time: {time.time} ({time.timezone})")
+
 
+@time.command(name="set")
+@click.argument("year", type=int)
+@click.argument("month", type=int)
+@click.argument("day", type=int)
+@click.argument("hour", type=int)
+@click.argument("minute", type=int)
+@click.argument("seconds", type=int, required=False, default=0)
+@click.option(
+    "--timezone",
+    type=str,
+    required=False,
+    default=None,
+    help="IANA timezone name, will use current device timezone if not provided.",
+)
+@click.option(
+    "--skip-confirm",
+    type=bool,
+    required=False,
+    default=False,
+    is_flag=True,
+    help="Do not ask to confirm the timezone if an exact match is not found.",
+)
+@pass_dev
+async def time_set(
+    dev: Device,
+    year: int,
+    month: int,
+    day: int,
+    hour: int,
+    seconds: int,
+    timezone: str | None,
+    skip_confirm: bool,
+):
+    """Set the device time to the provided time."""
     if (time := dev.modules.get(Module.Time)) is None:
         echo("Device does not have time module")
         return
 
-    echo("Old time: %s" % time.time)
+    tzinfo: zoneinfo.ZoneInfo | None = None
+    if timezone:
+        tzinfo = await _get_timezone(dev, timezone, skip_confirm)
 
-    local_tz = datetime.now().astimezone().tzinfo
-    await time.set_time(datetime.now(tz=local_tz))
+    echo(f"Old time: {time.time} ({time.timezone})")
+
+    await time.set_time(datetime(year, month, day, hour, minute, seconds, 0, tzinfo))
 
     await dev.update()
-    echo("New time: %s" % time.time)
+    echo(f"New time: {time.time} ({time.timezone})")
+
+
+async def _get_timezone(dev, timezone, skip_confirm) -> zoneinfo.ZoneInfo:
+    """Get the tzinfo from the timezone or return none."""
+    tzinfo: zoneinfo.ZoneInfo | None = None
+
+    if timezone not in zoneinfo.available_timezones():
+        error(f"{timezone} is not a valid IANA timezone.")
+
+    tzinfo = zoneinfo.ZoneInfo(timezone)
+    if skip_confirm is False and isinstance(dev, IotDevice):
+        matches = await get_matching_timezones(tzinfo)
+        if not matches:
+            error(f"Device cannot support {timezone} timezone.")
+        first = matches[0]
+        msg = (
+            f"An exact match for {timezone} could not be found, "
+            + f"timezone will be set to {first}"
+        )
+        if len(matches) == 1:
+            click.confirm(msg, abort=True)
+        else:
+            msg = (
+                f"Supported timezones matching {timezone} are {', '.join(matches)}\n"
+                + msg
+            )
+            click.confirm(msg, abort=True)
+    return tzinfo
diff --git a/kasa/device.py b/kasa/device.py
@@ -51,7 +51,7 @@
 schedule
 usage
 anti_theft
-time
+Time
 cloud
 Led
 

diff --git a/kasa/interfaces/__init__.py b/kasa/interfaces/__init__.py
@@ -6,6 +6,7 @@
 from .light import Light, LightState
 from .lighteffect import LightEffect
 from .lightpreset import LightPreset
+from .time import Time
 
 __all__ = [
     "Fan",
@@ -15,4 +16,5 @@
     "LightEffect",
     "LightState",
     "LightPreset",
+    "Time",
 ]
diff --git a/kasa/interfaces/time.py b/kasa/interfaces/time.py
@@ -0,0 +1,26 @@
+"""Module for time interface."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from datetime import datetime, tzinfo
+
+from ..module import Module
+
+
+class Time(Module, ABC):
+    """Base class for tplink time module."""
+
+    @property
+    @abstractmethod
+    def time(self) -> datetime:
+        """Return timezone aware current device time."""
+
+    @property
+    @abstractmethod
+    def timezone(self) -> tzinfo:
+        """Return current timezone."""
+
+    @abstractmethod
+    async def set_time(self, dt: datetime) -> dict:
+        """Set the device time."""
diff --git a/kasa/iot/iotbulb.py b/kasa/iot/iotbulb.py
@@ -219,7 +219,7 @@ async def _initialize_modules(self):
            self.add_module(
             Module.IotAntitheft, Antitheft(self, "smartlife.iot.common.anti_theft")
         )
-        self.add_module(Module.IotTime, Time(self, "smartlife.iot.common.timesetting"))
+        self.add_module(Module.Time, Time(self, "smartlife.iot.common.timesetting"))
         self.add_module(Module.Energy, Emeter(self, self.emeter_type))
         self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
         self.add_module(Module.IotCloud, Cloud(self, "smartlife.iot.common.cloud"))

diff --git a/kasa/iot/iotdevice.py b/kasa/iot/iotdevice.py
@@ -20,6 +20,7 @@
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timedelta, tzinfo
 from typing import TYPE_CHECKING, Any, cast
+from warnings import warn
 
 from ..device import Device, WifiNetwork
 from ..deviceconfig import DeviceConfig
@@ -460,27 +461,27 @@
     @requires_update
     def time(self) -> datetime:
         """Return current time from the device."""
-        return self.modules[Module.IotTime].time
+        return self.modules[Module.Time].time
 
     @property
     @requires_update
     def timezone(self) -> tzinfo:
         """Return the current timezone."""
-        return self.modules[Module.IotTime].timezone
+        return self.modules[Module.Time].timezone
 
-    async def get_time(self) -> datetime | None:
+    async def get_time(self) -> datetime:
         """Return current time from the device, if available."""
-        _LOGGER.warning(
-            "Use `time` property instead, this call will be removed in the future."
-        )
-        return await self.modules[Module.IotTime].get_time()
+        msg = "Use `time` property instead, this call will be removed in the future."
+        warn(msg, DeprecationWarning, stacklevel=1)
+        return self.time
 
-    async def get_timezone(self) -> dict:
+    async def get_timezone(self) -> tzinfo:
         """Return timezone information."""
-        _LOGGER.warning(
+        msg = (
             "Use `timezone` property instead, this call will be removed in the future."
         )
-        return await self.modules[Module.IotTime].get_timezone()
+        warn(msg, DeprecationWarning, stacklevel=1)
+        return self.timezone
 
     @property  # type: ignore
     @requires_update

diff --git a/kasa/iot/iotplug.py b/kasa/iot/iotplug.py
@@ -60,7 +60,7 @@ async def _initialize_modules(self):
         self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
         self.add_module(Module.IotUsage, Usage(self, "schedule"))
         self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
-        self.add_module(Module.IotTime, Time(self, "time"))
+        self.add_module(Module.Time, Time(self, "time"))
         self.add_module(Module.IotCloud, Cloud(self, "cnCloud"))
         self.add_module(Module.Led, Led(self, "system"))
 

diff --git a/kasa/iot/iotstrip.py b/kasa/iot/iotstrip.py
@@ -105,7 +105,7 @@ async def _initialize_modules(self):
         self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
         self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
         self.add_module(Module.IotUsage, Usage(self, "schedule"))
-        self.add_module(Module.IotTime, Time(self, "time"))
+        self.add_module(Module.Time, Time(self, "time"))
         self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
         self.add_module(Module.Led, Led(self, "system"))
         self.add_module(Module.IotCloud, Cloud(self, "cnCloud"))

diff --git a/kasa/iot/iottimezone.py b/kasa/iot/iottimezone.py
@@ -3,7 +3,10 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime, tzinfo
+from datetime import datetime, timedelta, tzinfo
+from typing import cast
+
+from zoneinfo import ZoneInfo
 
 from ..cachedzoneinfo import CachedZoneInfo
 
@@ -22,26 +25,53 @@
     return await CachedZoneInfo.get_cached_zone_info(name)
 
 
-async def get_timezone_index(name: str) -> int:
+async def get_timezone_index(tzone: tzinfo) -> int:
     """Return the iot firmware index for a valid IANA timezone key."""
-    rev = {val: key for key, val in TIMEZONE_INDEX.items()}
-    if name in rev:
-        return rev[name]
+    if isinstance(tzone, ZoneInfo):
+        name = tzone.key
+        rev = {val: key for key, val in TIMEZONE_INDEX.items()}
+        if name in rev:
+            return rev[name]
 
-    # Try to find a supported timezone matching dst true/false
-    zone = await CachedZoneInfo.get_cached_zone_info(name)
-    now = datetime.now()
-    winter = datetime(now.year, 1, 1, 12)
-    summer = datetime(now.year, 7, 1, 12)
     for i in range(110):
-        configured_zone = await get_timezone(i)
-        if zone.utcoffset(winter) == configured_zone.utcoffset(
-            winter
-        ) and zone.utcoffset(summer) == configured_zone.utcoffset(summer):
+        if _is_same_timezone(tzone, await get_timezone(i)):
             return i
     raise ValueError("Device does not support timezone %s", name)
 
 
+async def get_matching_timezones(tzone: tzinfo) -> list[str]:
+    """Return the iot firmware index for a valid IANA timezone key."""
+    matches = []
+    if isinstance(tzone, ZoneInfo):
+        name = tzone.key
+        vals = {val for val in TIMEZONE_INDEX.values()}
+        if name in vals:
+            matches.append(name)
+
+    for i in range(110):
+        fw_tz = await get_timezone(i)
+        if _is_same_timezone(tzone, fw_tz):
+            match_key = cast(ZoneInfo, fw_tz).key
+            if match_key not in matches:
+                matches.append(match_key)
+    return matches
+
+
+def _is_same_timezone(tzone1: tzinfo, tzone2: tzinfo) -> bool:
+    """Return true if the timezones have the same utcffset and dst offset.
+
+    Iot devices only support a limited static list of IANA timezones; this is used to
+    check if a static timezone matches the same utc offset and dst settings.
+    """
+    now = datetime.now()
+    start_day = datetime(now.year, 1, 1, 12)
+    for i in range(365):
+        the_day = start_day + timedelta(days=i)
+        if tzone1.utcoffset(the_day) != tzone2.utcoffset(the_day):
+            return False
+    return True
+
+
 TIMEZONE_INDEX = {
     0: "Etc/GMT+12",
     1: "Pacific/Samoa",
-Original file line number
+Diff line change
@@ Expand Up / @@ -51,7 +51,7 @@ @@
     schedule
     usage
     anti_theft
-    time
+    Time
     cloud
     Led
@@ Expand Down @@