Configurable timezone for Datetime device using localtime (#1663)

* Configurable timezone for `Datetime` device using `localtime`

* simplify

Author: farmio

docs/changelog.md

@@ -6,6 +6,10 @@ nav_order: 2
 
 # Changelog
 
+### Devices
+
+- Datetime: Accept `datetime.tzinfo` for `timezone` argument to send time information for specific timezone. Boolean works like before: If `True` use system localtime. If `False` an arbitrary time can be sent.
+
 ### DPT
 
 - Add `DPTBase.dpt_number_str` and `DPTBase.dpt_name` classmethods for human readable DPT number (eg. "9.001") and class name (eg. "DPTTemperature (9.001)").

docs/time.md

@@ -28,7 +28,7 @@ await xknx.devices['TimeTest'].sync()
 * `xknx` is the XKNX object.
 * `name` is the name of the object.
 * `group_address` is the KNX group address of the sensor device.
-* `localtime` If set `True` sync() and GroupValueRead requests always return the current local time and it is also sent every 60 minutes. On `False` the set value will be sent. Default: `True`
+* `localtime` If set `True` sync() and GroupValueRead requests always return the current systems local time and it is also sent every 60 minutes. Same if set to a `datetime.tzinfo` object, but the time for that timezone information will be used. On `False` the set value will be sent, no automatic sending will be scheduled. Default: `True`
 * `device_updated_cb` Callback for each update.
 
 ## [](#header-2)Local time

test/devices_tests/datetime_test.py

@@ -2,6 +2,7 @@
 
 import datetime as dt
 from unittest.mock import AsyncMock, patch
+from zoneinfo import ZoneInfo
 
 from freezegun import freeze_time
 import pytest
@@ -104,10 +105,21 @@ async def test_sync_time_custom(self) -> None:
         assert telegram.destination_address == GroupAddress("1/2/4")
         assert isinstance(telegram.payload, GroupValueRead)
 
-    async def test_process_read_localtime(self) -> None:
+    @pytest.mark.parametrize(
+        ("expected", "localtime"),
+        [
+            ((0xC9, 0xD, 0xE), True),
+            ((0xCA, 0xD, 0xE), ZoneInfo("Europe/Vienna")),
+        ],
+    )
+    async def test_process_read_localtime_time(
+        self, expected: tuple[int, int, int], localtime: bool | dt.tzinfo
+    ) -> None:
         """Test test process a read telegram from KNX bus."""
         xknx = XKNX()
-        test_device = TimeDevice(xknx, "TestDateTime", group_address="1/2/3")
+        test_device = TimeDevice(
+            xknx, "TestTime", group_address="1/2/3", localtime=localtime
+        )
 
         telegram_read = Telegram(
             destination_address=GroupAddress("1/2/3"), payload=GroupValueRead()
@@ -118,7 +130,41 @@ async def test_process_read_localtime(self) -> None:
         telegram = xknx.telegrams.get_nowait()
         assert telegram == Telegram(
             destination_address=GroupAddress("1/2/3"),
-            payload=GroupValueResponse(DPTArray((0xC9, 0xD, 0xE))),
+            payload=GroupValueResponse(DPTArray(expected)),
+        )
+
+    @pytest.mark.parametrize(
+        ("expected", "localtime"),
+        [
+            (
+                (0x75, 0x07, 0x0B, 0x49, 0x0D, 0x0E, 0x20, 0xC0),
+                True,
+            ),
+            (
+                (0x75, 0x07, 0x0B, 0x4B, 0x0D, 0x0E, 0x21, 0xC0),
+                ZoneInfo("Europe/Vienna"),
+            ),
+        ],
+    )
+    async def test_process_read_localtime_datetime(
+        self, expected: tuple[int, int, int], localtime: bool | dt.tzinfo
+    ) -> None:
+        """Test test process a read telegram from KNX bus."""
+        xknx = XKNX()
+        test_device = DateTimeDevice(
+            xknx, "TestDateTime", group_address="1/2/3", localtime=localtime
+        )
+
+        telegram_read = Telegram(
+            destination_address=GroupAddress("1/2/3"), payload=GroupValueRead()
+        )
+        with freeze_time("2017-07-11 09:13:14"):  # summer time
+            test_device.process(telegram_read)
+
+        telegram = xknx.telegrams.get_nowait()
+        assert telegram == Telegram(
+            destination_address=GroupAddress("1/2/3"),
+            payload=GroupValueResponse(DPTArray(expected)),
         )
 
     async def test_process_read_custom_time(self) -> None:

xknx/devices/datetime.py

@@ -51,7 +51,7 @@ def __init__(
         self,
         xknx: XKNX,
         name: str,
-        localtime: bool = True,
+        localtime: bool | datetime.tzinfo = True,
         group_address: GroupAddressesType = None,
         group_address_state: GroupAddressesType = None,
         respond_to_read: bool = False,
@@ -60,7 +60,10 @@ def __init__(
     ) -> None:
         """Initialize DateTime class."""
         super().__init__(xknx, name, device_updated_cb)
-        self.localtime = localtime
+        self.localtime = bool(localtime)
+        self._localtime_zone = (
+            localtime if isinstance(localtime, datetime.tzinfo) else None
+        )
         if localtime and group_address_state is not None:
             logger.warning(
                 "State address invalid in %s device when using `localtime=True`. Ignoring `group_address_state=%s` argument.",
@@ -169,7 +172,7 @@ async def set(self, value: KNXTime | datetime.time) -> None:
 
     def broadcast_localtime(self, response: bool = False) -> None:
         """Broadcast the local time to KNX bus."""
-        now = datetime.datetime.now()
+        now = datetime.datetime.now(self._localtime_zone)
         knx_time = KNXTime.from_time(now.time())
         knx_time.day = KNXDay(now.weekday() + 1)
         self.remote_value.set(knx_time, response=response)
@@ -193,7 +196,7 @@ async def set(self, value: KNXDate | datetime.date) -> None:
 
     def broadcast_localtime(self, response: bool = False) -> None:
         """Broadcast the local date to KNX bus."""
-        now = datetime.datetime.now()
+        now = datetime.datetime.now(self._localtime_zone)
         self.remote_value.set(KNXDate.from_date(now.date()), response=response)
 
 
@@ -219,11 +222,15 @@ async def set(self, value: KNXDateTime | datetime.datetime) -> None:
 
     def broadcast_localtime(self, response: bool = False) -> None:
         """Broadcast the local date/time to KNX bus."""
-        time_now = time.localtime()
-        now = datetime.datetime.now()
+        now = datetime.datetime.now(self._localtime_zone)
         knx_datetime = KNXDateTime.from_datetime(now)
         knx_datetime.day_of_week = KNXDayOfWeek(now.weekday() + 1)
-        knx_datetime.dst = time_now.tm_isdst > 0
+        if self._localtime_zone is not None:
+            dst = self._localtime_zone.dst(now)
+            knx_datetime.dst = dst > datetime.timedelta(0) if dst is not None else False
+        else:
+            time_now = time.localtime()
+            knx_datetime.dst = time_now.tm_isdst > 0
         knx_datetime.external_sync = True
         knx_datetime.source_reliable = True
         self.remote_value.set(knx_datetime, response=response)