| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| """Implementation of time module.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from datetime import datetime, timezone, tzinfo | ||
| from typing import cast | ||
|
|
||
| from zoneinfo import ZoneInfo, ZoneInfoNotFoundError | ||
|
|
||
| from ...cachedzoneinfo import CachedZoneInfo | ||
| from ...feature import Feature | ||
| from ...interfaces import Time as TimeInterface | ||
| from ..smartcameramodule import SmartCameraModule | ||
|
|
||
|
|
||
| class Time(SmartCameraModule, TimeInterface): | ||
| """Implementation of device_local_time.""" | ||
|
|
||
| QUERY_GETTER_NAME = "getTimezone" | ||
| QUERY_MODULE_NAME = "system" | ||
| QUERY_SECTION_NAMES = "basic" | ||
|
|
||
| _timezone: tzinfo = timezone.utc | ||
| _time: datetime | ||
|
|
||
| def _initialize_features(self) -> None: | ||
| """Initialize features after the initial update.""" | ||
| self._add_feature( | ||
| Feature( | ||
| device=self._device, | ||
| id="device_time", | ||
| name="Device time", | ||
| attribute_getter="time", | ||
| container=self, | ||
| category=Feature.Category.Debug, | ||
| type=Feature.Type.Sensor, | ||
| ) | ||
| ) | ||
|
|
||
| def query(self) -> dict: | ||
| """Query to execute during the update cycle.""" | ||
| q = super().query() | ||
| q["getClockStatus"] = {self.QUERY_MODULE_NAME: {"name": "clock_status"}} | ||
|
|
||
| return q | ||
|
|
||
| async def _post_update_hook(self) -> None: | ||
| """Perform actions after a device update.""" | ||
| time_data = self.data["getClockStatus"]["system"]["clock_status"] | ||
| timezone_data = self.data["getTimezone"]["system"]["basic"] | ||
| zone_id = timezone_data["zone_id"] | ||
| timestamp = time_data["seconds_from_1970"] | ||
| try: | ||
| # Zoneinfo will return a DST aware object | ||
| tz: tzinfo = await CachedZoneInfo.get_cached_zone_info(zone_id) | ||
| except ZoneInfoNotFoundError: | ||
| # timezone string like: UTC+10:00 | ||
| timezone_str = timezone_data["timezone"] | ||
| tz = cast(tzinfo, datetime.strptime(timezone_str[-6:], "%z").tzinfo) | ||
|
|
||
| self._timezone = tz | ||
| self._time = datetime.fromtimestamp( | ||
| cast(float, timestamp), | ||
| tz=tz, | ||
| ) | ||
|
|
||
| @property | ||
| def timezone(self) -> tzinfo: | ||
| """Return current timezone.""" | ||
| return self._timezone | ||
|
|
||
| @property | ||
| def time(self) -> datetime: | ||
| """Return device's current datetime.""" | ||
| return self._time | ||
|
|
||
| async def set_time(self, dt: datetime) -> dict: | ||
| """Set device time.""" | ||
| if not dt.tzinfo: | ||
| timestamp = dt.replace(tzinfo=self.timezone).timestamp() | ||
| else: | ||
| timestamp = dt.timestamp() | ||
|
|
||
| lt = datetime.fromtimestamp(timestamp).isoformat().replace("T", " ") | ||
| params = {"seconds_from_1970": int(timestamp), "local_time": lt} | ||
| # Doesn't seem to update the time, perhaps because timing_mode is ntp | ||
| res = await self.call("setTimezone", {"system": {"clock_status": params}}) | ||
| if (zinfo := dt.tzinfo) and isinstance(zinfo, ZoneInfo): | ||
| tz_params = {"zone_id": zinfo.key} | ||
| res = await self.call("setTimezone", {"system": {"basic": tz_params}}) | ||
| return res | ||
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.