Create common Time module and add time set cli command (#1157)

This commit is contained in:
Steven B.
2024-10-15 08:59:25 +01:00
committed by GitHub
parent 885a04d24f
commit 7fd8c14c1f
18 changed files with 350 additions and 69 deletions

View File

@@ -219,7 +219,7 @@ class IotBulb(IotDevice):
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"))

View File

@@ -20,6 +20,7 @@ import logging
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 @@ class IotDevice(Device):
@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

View File

@@ -60,7 +60,7 @@ class IotPlug(IotDevice):
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"))

View File

@@ -105,7 +105,7 @@ class IotStrip(IotDevice):
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"))

View File

@@ -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 @@ async def get_timezone(index: int) -> tzinfo:
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",

View File

@@ -5,11 +5,12 @@ from __future__ import annotations
from datetime import datetime, timezone, tzinfo
from ...exceptions import KasaException
from ...interfaces import Time as TimeInterface
from ..iotmodule import IotModule, merge
from ..iottimezone import get_timezone
from ..iottimezone import get_timezone, get_timezone_index
class Time(IotModule):
class Time(IotModule, TimeInterface):
"""Implements the timezone settings."""
_timezone: tzinfo = timezone.utc
@@ -57,10 +58,36 @@ class Time(IotModule):
res["hour"],
res["min"],
res["sec"],
tzinfo=self.timezone,
)
except KasaException:
return None
async def set_time(self, dt: datetime) -> dict:
"""Set the device time."""
params = {
"year": dt.year,
"month": dt.month,
"mday": dt.day,
"hour": dt.hour,
"min": dt.minute,
"sec": dt.second,
}
if dt.tzinfo:
index = await get_timezone_index(dt.tzinfo)
current_index = self.data.get("get_timezone", {}).get("index", -1)
if current_index != -1 and current_index != index:
params["index"] = index
method = "set_timezone"
else:
method = "set_time"
else:
method = "set_time"
try:
return await self.call(method, params)
except Exception as ex:
raise KasaException(ex) from ex
async def get_timezone(self):
"""Request timezone information from the device."""
return await self.call("get_timezone")