From 22ff7f451166bcfda360552e92d661d0520886ae Mon Sep 17 00:00:00 2001 From: humbertogontijo Date: Mon, 5 Jun 2023 18:44:19 -0300 Subject: [PATCH] fix: merging timer entities --- roborock/containers.py | 70 +++++++++++++++++------------------------- roborock/util.py | 14 +++++++++ 2 files changed, 43 insertions(+), 41 deletions(-) diff --git a/roborock/containers.py b/roborock/containers.py index 154796d..c9488a8 100644 --- a/roborock/containers.py +++ b/roborock/containers.py @@ -1,9 +1,9 @@ from __future__ import annotations +import datetime import logging import re from dataclasses import asdict, dataclass -from datetime import time from enum import Enum from typing import Any, Optional, Type @@ -45,6 +45,7 @@ SENSOR_DIRTY_REPLACE_TIME, SIDE_BRUSH_REPLACE_TIME, ) +from .util import parse_time_to_datetime _LOGGER = logging.getLogger(__name__) @@ -95,6 +96,29 @@ def as_dict(self) -> dict: ) +@dataclass +class RoborockBaseTimer(RoborockBase): + start_hour: Optional[int] = None + start_minute: Optional[int] = None + end_hour: Optional[int] = None + end_minute: Optional[int] = None + enabled: Optional[int] = None + start_time: Optional[datetime.datetime] = None + end_time: Optional[datetime.datetime] = None + + def __post_init__(self) -> None: + self.start_time = ( + parse_time_to_datetime(datetime.time(hour=self.start_hour, minute=self.start_minute)) + if self.start_hour is not None and self.start_minute is not None + else None + ) + self.end_time = ( + parse_time_to_datetime(datetime.time(hour=self.end_hour, minute=self.end_minute)) + if self.end_hour is not None and self.end_minute is not None + else None + ) + + @dataclass class Reference(RoborockBase): r: Optional[str] = None @@ -338,49 +362,13 @@ class S8Status(Status): @dataclass -class DnDTimer(RoborockBase): - start_hour: Optional[int] = None - start_minute: Optional[int] = None - end_hour: Optional[int] = None - end_minute: Optional[int] = None - enabled: Optional[int] = None - start_time: Optional[time] = None - end_time: Optional[time] = None - - def __post_init__(self) -> None: - self.start_time = ( - time(hour=self.start_hour, minute=self.start_minute) - if self.start_hour is not None and self.start_minute is not None - else None - ) - self.end_time = ( - time(hour=self.end_hour, minute=self.end_minute) - if self.end_hour is not None and self.end_minute is not None - else None - ) +class DnDTimer(RoborockBaseTimer): + """DnDTimer""" @dataclass -class ValleyElectricityTimer(RoborockBase): - start_hour: Optional[int] = None - start_minute: Optional[int] = None - end_hour: Optional[int] = None - end_minute: Optional[int] = None - enabled: Optional[int] = None - start_time: Optional[time] = None - end_time: Optional[time] = None - - def __post_init__(self) -> None: - self.start_time = ( - time(hour=self.start_hour, minute=self.start_minute) - if self.start_hour is not None and self.start_minute is not None - else None - ) - self.end_time = ( - time(hour=self.end_hour, minute=self.end_minute) - if self.end_hour is not None and self.end_minute is not None - else None - ) +class ValleyElectricityTimer(RoborockBaseTimer): + """ValleyElectricityTimer""" @dataclass diff --git a/roborock/util.py b/roborock/util.py index b0c9e89..ce7f240 100644 --- a/roborock/util.py +++ b/roborock/util.py @@ -1,11 +1,13 @@ from __future__ import annotations import asyncio +import datetime import functools from asyncio import AbstractEventLoop from typing import Callable, Coroutine, Optional, TypeVar T = TypeVar("T") +DEFAULT_TIME_ZONE: datetime.tzinfo = datetime.timezone.utc def unpack_list(value: list[T], size: int) -> list[Optional[T]]: @@ -21,6 +23,18 @@ def get_running_loop_or_create_one() -> AbstractEventLoop: return loop +def parse_time_to_datetime(initial_time: datetime.time) -> datetime.datetime: + """Help to handle time data.""" + time = datetime.datetime.now(DEFAULT_TIME_ZONE).replace( + hour=initial_time.hour, minute=initial_time.minute, second=0, microsecond=0 + ) + + if time < datetime.datetime.now(DEFAULT_TIME_ZONE): + time += datetime.timedelta(days=1) + + return time + + def run_sync(): loop = get_running_loop_or_create_one()