Update with mypy typing (#427)

pyrainbird/async_client.py

@@ -88,7 +88,7 @@ def _device_busy_retry() -> JitterRetry:
     return JitterRetry(
         attempts=_retry_attempts(),
         start_timeout=_retry_delay(),
-        statuses=[HTTPStatus.SERVICE_UNAVAILABLE.value],
+        statuses=set([HTTPStatus.SERVICE_UNAVAILABLE.value]),
         retry_all_server_errors=False,
     )
 
@@ -114,10 +114,10 @@ def __init__(
         self._password = password
         self._coder = encryption.PayloadCoder(password, _LOGGER)
 
-    def with_retry_options(self, retry_options: RetryOptions) -> "AsyncRainbirdClient":
+    def with_retry_options(self, retry_options: RetryOptions) -> "AsyncRainbirdClient":  # type: ignore[valid-type]
         """Create a new AsyncRainbirdClient with retry options."""
         return AsyncRainbirdClient(
-            RetryClient(client_session=self._websession, retry_options=retry_options),
+            RetryClient(client_session=self._websession, retry_options=retry_options),  # type: ignore[arg-type]
             self._host,
             self._password,
         )
@@ -147,7 +147,7 @@ async def request(
                 "Error communicating with Rain Bird device"
             ) from err
         content = await resp.read()
-        return self._coder.decode_command(content)
+        return self._coder.decode_command(content)  # type: ignore
 
 
 def CreateController(
@@ -165,7 +165,7 @@ class AsyncRainbirdController:
     def __init__(
         self,
         local_client: AsyncRainbirdClient,
-        cloud_client: AsyncRainbirdClient = None,
+        cloud_client: AsyncRainbirdClient | None = None,
     ) -> None:
         """Initialize AsyncRainbirdController."""
         self._local_client = local_client
@@ -418,15 +418,15 @@ async def get_schedule(self) -> Schedule:
             commands.append("%04x" % (0x80 | zone_page))
         _LOGGER.debug("Sending schedule commands: %s", commands)
         # Run command serially to avoid overwhelming the controller
-        schedule_data = {
+        schedule_data: dict[str, Any] = {
             "controllerInfo": {},
             "programInfo": [],
             "programStartInfo": [],
             "durations": [],
         }
         for command in commands:
             result = await self._process_command(
-                None, "RetrieveScheduleRequest", int(command, 16)  # Disable validation
+                None, "RetrieveScheduleRequest", int(command, 16)  # type: ignore
             )
             if not isinstance(result, dict):
                 continue
@@ -509,4 +509,4 @@ async def _cacheable_command(
             return result
         result = await self._process_command(funct, command, *args)
         self._cache[key] = result
-        return result
+        return result  # type: ignore

pyrainbird/data.py

@@ -130,7 +130,7 @@ class States:
     """Rainbird controller response containing a bitmask string e.g. active zones."""
 
     count: int
-    mask: str
+    mask: int
     states: tuple
 
     def __init__(self, mask: str) -> None:
@@ -195,20 +195,42 @@ class WaterBudget:
 class WifiParams(DataClassDictMixin):
     """Wifi parameters for the device."""
 
-    mac_address: Optional[str] = field(metadata=field_options(alias="macAddress"), default=None)
+    mac_address: Optional[str] = field(
+        metadata=field_options(alias="macAddress"), default=None
+    )
     """The mac address for the device, also referred to as the stick id."""
 
-    local_ip_address: Optional[str] = field(metadata=field_options(alias="localIpAddress"), default=None)
-    local_netmask: Optional[str] = field(metadata=field_options(alias="localNetmask"), default=None)
-    local_gateway: Optional[str] = field(metadata=field_options(alias="localGateway"), default=None)
+    local_ip_address: Optional[str] = field(
+        metadata=field_options(alias="localIpAddress"), default=None
+    )
+    local_netmask: Optional[str] = field(
+        metadata=field_options(alias="localNetmask"), default=None
+    )
+    local_gateway: Optional[str] = field(
+        metadata=field_options(alias="localGateway"), default=None
+    )
     rssi: Optional[int] = None
-    wifi_ssid: Optional[str] = field(metadata=field_options(alias="wifiSsid"), default=None)
-    wifi_password: Optional[str] = field(metadata=field_options(alias="wifiPassword"), default=None)
-    wifi_security: Optional[str] = field(metadata=field_options(alias="wifiSecurity"), default=None)
-    ap_timeout_no_lan: Optional[int] = field(metadata=field_options(alias="apTimeoutNoLan"), default=None)
-    ap_timeout_idle: Optional[int] = field(metadata=field_options(alias="apTimeoutIdle"), default=None)
-    ap_security: Optional[str] = field(metadata=field_options(alias="apSecurity"), default=None)
-    sick_version: Optional[str] = field(metadata=field_options(alias="stickVersion"), default=None)
+    wifi_ssid: Optional[str] = field(
+        metadata=field_options(alias="wifiSsid"), default=None
+    )
+    wifi_password: Optional[str] = field(
+        metadata=field_options(alias="wifiPassword"), default=None
+    )
+    wifi_security: Optional[str] = field(
+        metadata=field_options(alias="wifiSecurity"), default=None
+    )
+    ap_timeout_no_lan: Optional[int] = field(
+        metadata=field_options(alias="apTimeoutNoLan"), default=None
+    )
+    ap_timeout_idle: Optional[int] = field(
+        metadata=field_options(alias="apTimeoutIdle"), default=None
+    )
+    ap_security: Optional[str] = field(
+        metadata=field_options(alias="apSecurity"), default=None
+    )
+    sick_version: Optional[str] = field(
+        metadata=field_options(alias="stickVersion"), default=None
+    )
 
 
 class SoilType(IntEnum):
@@ -227,9 +249,15 @@ class ProgramInfo(DataClassDictMixin):
     The values are repeated once for each program.
     """
 
-    soil_types: list[SoilType] = field(default_factory=list, metadata=field_options(alias="SoilTypes"))
-    flow_rates: list[int] = field(default_factory=list, metadata=field_options(alias="FlowRates"))
-    flow_units: list[int] = field(default_factory=list, metadata=field_options(alias="FlowUnits"))
+    soil_types: list[SoilType] = field(
+        default_factory=list, metadata=field_options(alias="SoilTypes")
+    )
+    flow_rates: list[int] = field(
+        default_factory=list, metadata=field_options(alias="FlowRates")
+    )
+    flow_units: list[int] = field(
+        default_factory=list, metadata=field_options(alias="FlowUnits")
+    )
 
     @classmethod
     def __pre_deserialize__(cls, values: dict[Any, Any]) -> dict[Any, Any]:
@@ -253,9 +281,15 @@ class Settings(DataClassDictMixin):
     """Country location of the device."""
 
     # Program information
-    soil_types: list[SoilType] = field(default_factory=list, metadata=field_options(alias="SoilTypes"))
-    flow_rates: list[int] = field(default_factory=list, metadata=field_options(alias="FlowRates"))
-    flow_units: list[int] = field(default_factory=list, metadata=field_options(alias="FlowUnits"))
+    soil_types: list[SoilType] = field(
+        default_factory=list, metadata=field_options(alias="SoilTypes")
+    )
+    flow_rates: list[int] = field(
+        default_factory=list, metadata=field_options(alias="FlowRates")
+    )
+    flow_units: list[int] = field(
+        default_factory=list, metadata=field_options(alias="FlowUnits")
+    )
 
     @classmethod
     def __pre_deserialize__(cls, values: dict[Any, Any]) -> dict[Any, Any]:
@@ -294,7 +328,7 @@ def __init__(self, status: Optional[str], settings: Optional[Settings]) -> None:
     @property
     def status(self) -> str:
         """Return device status."""
-        return self._status
+        return self._status or "unknown"
 
     @property
     def settings(self) -> Optional[Settings]:
@@ -316,7 +350,9 @@ class Controller(DataClassDictMixin):
     available_stations: list[int] = field(
         metadata=field_options(alias="availableStations"), default_factory=list
     )
-    custom_name: Optional[str] = field(metadata=field_options(alias="customName"), default=None)
+    custom_name: Optional[str] = field(
+        metadata=field_options(alias="customName"), default=None
+    )
     custom_program_names: dict[str, str] = field(
         metadata=field_options(alias="customProgramNames"), default_factory=dict
     )
@@ -345,18 +381,30 @@ class Weather(DataClassDictMixin):
     city: Optional[str] = None
     forecast: list[Forecast] = field(default_factory=list)
     location: Optional[str] = None
-    time_zone_id: Optional[str] = field(metadata=field_options(alias="timeZoneId"), default=None)
-    time_zone_raw_offset: Optional[str] = field(metadata=field_options(alias="timeZoneRawOffset"), default=None)
+    time_zone_id: Optional[str] = field(
+        metadata=field_options(alias="timeZoneId"), default=None
+    )
+    time_zone_raw_offset: Optional[str] = field(
+        metadata=field_options(alias="timeZoneRawOffset"), default=None
+    )
 
 
 @dataclass
 class WeatherAndStatus(DataClassDictMixin):
     """Weather and status from the cloud API."""
 
-    stick_id: Optional[str] = field(metadata=field_options(alias="StickId"), default=None)
-    controller: Optional[Controller] = field(metadata=field_options(alias="Controller"), default=None)
-    forecasted_rain: Optional[dict[str, Any]] = field(metadata=field_options(alias="ForecastedRain"), default=None)
-    weather: Optional[Weather] = field(metadata=field_options(alias="Weather"), default=None)
+    stick_id: Optional[str] = field(
+        metadata=field_options(alias="StickId"), default=None
+    )
+    controller: Optional[Controller] = field(
+        metadata=field_options(alias="Controller"), default=None
+    )
+    forecasted_rain: Optional[dict[str, Any]] = field(
+        metadata=field_options(alias="ForecastedRain"), default=None
+    )
+    weather: Optional[Weather] = field(
+        metadata=field_options(alias="Weather"), default=None
+    )
 
 
 @dataclass
@@ -398,6 +446,7 @@ def deserialize(self, values: dict[str, Any]) -> datetime.datetime:
             int(values["second"]),
         )
 
+
 @dataclass
 class ControllerState(DataClassDictMixin):
     """Details about the controller state."""
@@ -417,13 +466,14 @@ class ControllerState(DataClassDictMixin):
     # TODO: Likely need to make this a mask w/ States
     active_station: int = field(metadata=field_options(alias="activeStation"))
 
-    device_time: datetime.datetime = field(metadata=field_options(serialization_strategy=DeviceTime()))
+    device_time: datetime.datetime = field(
+        metadata=field_options(serialization_strategy=DeviceTime())
+    )
 
     @classmethod
     def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]:
         d["device_time"] = {
-            k: d[k]
-            for k in ("year", "month", "day", "hour", "minute", "second")
+            k: d[k] for k in ("year", "month", "day", "hour", "minute", "second")
         }
         return d
 
@@ -459,7 +509,7 @@ def name(self) -> str:
     @classmethod
     def __pre_deserialize__(cls, values: dict[Any, Any]) -> dict[Any, Any]:
         if duration := values.get("duration"):
-            values["duration"] = duration * 60  #datetime.timedelta(minutes=duration)
+            values["duration"] = duration * 60  # datetime.timedelta(minutes=duration)
         return values
 
 
@@ -479,15 +529,13 @@ def deserialize(self, starts: list[int]) -> list[datetime.time]:
         return result
 
 
-
-
 class DayOfWeekSerializationStrategy(SerializationStrategy):
     """Validate different ways the device time parameter is handled."""
 
     def serialize(self, value: Any) -> str:
         raise ValueError("Serialization not implemented")
 
-    def deserialize(self, mask: int) -> list[DayOfWeek]:
+    def deserialize(self, mask: int) -> set[DayOfWeek]:
         """Deserialize the device time fields."""
         _LOGGER.debug("DayOfWeekSerializationStrategy=%s", mask)
         result: set[DayOfWeek] = set()
@@ -512,7 +560,13 @@ class Program(DataClassDictMixin):
     frequency: ProgramFrequency
     """Determines how often the program runs."""
 
-    days_of_week: set[DayOfWeek] = field(metadata=field_options(alias="daysOfWeekMask", serialization_strategy=DayOfWeekSerializationStrategy()), default_factory=set)
+    days_of_week: set[DayOfWeek] = field(
+        metadata=field_options(
+            alias="daysOfWeekMask",
+            serialization_strategy=DayOfWeekSerializationStrategy(),
+        ),
+        default_factory=set,
+    )
     """For a CUSTOM program determines the days of the week."""
 
     period: Optional[int] = None
@@ -521,13 +575,18 @@ class Program(DataClassDictMixin):
     synchro: Optional[int] = None
     """Days from today before starting the first day of the program."""
 
-    starts: list[datetime.time] = field(default_factory=list, metadata=field_options(serialization_strategy=TimeSerializationStrategy()))
+    starts: list[datetime.time] = field(
+        default_factory=list,
+        metadata=field_options(serialization_strategy=TimeSerializationStrategy()),
+    )
     """Time of day the program starts."""
 
     durations: list[ZoneDuration] = field(default_factory=list)
     """Durations for run times for each zone."""
 
-    controller_info: Optional[ControllerInfo] = field(metadata=field_options(alias="controllerInfo"), default=None)
+    controller_info: Optional[ControllerInfo] = field(
+        metadata=field_options(alias="controllerInfo"), default=None
+    )
     """Information about the controller as input into the programs."""
 
     @property
@@ -541,7 +600,7 @@ def timeline(self) -> ProgramTimeline:
         """Return a timeline of events for the program."""
         return self.timeline_tz(datetime.datetime.now().tzinfo)
 
-    def timeline_tz(self, tzinfo: datetime.tzinfo) -> ProgramTimeline:
+    def timeline_tz(self, tzinfo: datetime.tzinfo | None) -> ProgramTimeline:
         """Return a timeline of events for the program."""
         iters: list[Iterable[SortableItem[Timespan, ProgramEvent]]] = []
         now = datetime.datetime.now(tzinfo)
@@ -553,9 +612,9 @@ def timeline_tz(self, tzinfo: datetime.tzinfo) -> ProgramTimeline:
                     self.frequency,
                     dtstart,
                     self.duration,
-                    self.synchro,
+                    self.synchro or 0,
                     self.days_of_week,
-                    self.period,
+                    self.period or 0,
                     delay_days=self.delay_days,
                 ),
             )
@@ -575,9 +634,9 @@ def zone_timeline(self) -> ProgramTimeline:
                         self.frequency,
                         dtstart,
                         zone_duration.duration,
-                        self.synchro,
+                        self.synchro or 0,
                         self.days_of_week,
-                        self.period,
+                        self.period or 0,
                         delay_days=self.delay_days,
                     )
                 )
@@ -604,12 +663,13 @@ def __post_init__(self):
             self.period = None
 
 
-
 @dataclass
 class Schedule(DataClassDictMixin):
     """Details about program schedules."""
 
-    controller_info: Optional[ControllerInfo] = field(metadata=field_options(alias="controllerInfo"))
+    controller_info: Optional[ControllerInfo] = field(
+        metadata=field_options(alias="controllerInfo")
+    )
     """Information about the controller used in the schedule."""
 
     programs: list[Program] = field(metadata=field_options(alias="programInfo"))
@@ -620,7 +680,7 @@ def timeline(self) -> ProgramTimeline:
         """Return a timeline of all programs."""
         return self.timeline_tz(datetime.datetime.now().tzinfo)
 
-    def timeline_tz(self, tzinfo: datetime.tzinfo) -> ProgramTimeline:
+    def timeline_tz(self, tzinfo: datetime.tzinfo | None) -> ProgramTimeline:
         """Return a timeline of all programs."""
         iters: list[Iterable[SortableItem[Timespan, ProgramEvent]]] = []
         now = datetime.datetime.now(tzinfo)
@@ -633,9 +693,9 @@ def timeline_tz(self, tzinfo: datetime.tzinfo) -> ProgramTimeline:
                         program.frequency,
                         dtstart,
                         program.duration,
-                        program.synchro,
+                        program.synchro or 0,
                         program.days_of_week,
-                        program.period,
+                        program.period or 0,
                         delay_days=self.delay_days,
                     )
                 )

pyrainbird/encryption.py

@@ -100,7 +100,7 @@ def encode_command(self, method: str, params: dict[str, Any]) -> str:
             return send_data
         return encrypt(send_data, self._password)
 
-    def decode_command(self, content: bytes) -> str:
+    def decode_command(self, content: bytes) -> str | dict[str, Any]:
         """Decode a response payload."""
         if self._password is not None:
             decrypted_data = (
@@ -112,7 +112,7 @@ def decode_command(self, content: bytes) -> str:
                 .rstrip()
             )
             content = decrypted_data
-        self._logger.debug("Response: %s" % content)
+        self._logger.debug("Response: %r" % content)
         response = json.loads(content)
         if error := response.get("error"):
             msg = ["Error from controller"]