diff --git a/pyrainbird/async_client.py b/pyrainbird/async_client.py index 9559450..20af558 100644 --- a/pyrainbird/async_client.py +++ b/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,7 +418,7 @@ 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": [], @@ -426,7 +426,7 @@ async def get_schedule(self) -> Schedule: } 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 diff --git a/pyrainbird/data.py b/pyrainbird/data.py index abadd52..61e00f8 100644 --- a/pyrainbird/data.py +++ b/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, ) ) diff --git a/pyrainbird/encryption.py b/pyrainbird/encryption.py index 749d1b8..cf018dd 100644 --- a/pyrainbird/encryption.py +++ b/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"] diff --git a/pyrainbird/rainbird.py b/pyrainbird/rainbird.py index 9950ac0..90701b3 100644 --- a/pyrainbird/rainbird.py +++ b/pyrainbird/rainbird.py @@ -34,7 +34,7 @@ def decode_template(data: str, cmd_template: dict[str, Any]) -> dict[str, int]: def decode_schedule(data: str, cmd_template: dict[str, Any]) -> dict[str, Any]: """Decode a schedule command.""" subcommand = int(data[4:6], 16) - rest = data[6:] + rest: str | bytes = data[6:] if subcommand == 0: if len(rest) < 8: return {} diff --git a/pyrainbird/resources/__init__.py b/pyrainbird/resources/__init__.py index 92632b2..2203f30 100644 --- a/pyrainbird/resources/__init__.py +++ b/pyrainbird/resources/__init__.py @@ -16,10 +16,10 @@ RESERVED_FIELDS = [COMMAND, TYPE, LENGTH, RESPONSE, DECODER] SIP_COMMANDS = yaml.load( - pkgutil.get_data(__name__, "sipcommands.yaml"), Loader=yaml.FullLoader + pkgutil.get_data(__name__, "sipcommands.yaml") or b"", Loader=yaml.FullLoader ) MODEL_INFO = yaml.load( - pkgutil.get_data(__name__, "models.yaml"), Loader=yaml.FullLoader + pkgutil.get_data(__name__, "models.yaml") or b"", Loader=yaml.FullLoader ) RAINBIRD_MODELS = {info["device_id"]: info for info in MODEL_INFO} diff --git a/pyrainbird/timeline.py b/pyrainbird/timeline.py index a2d42ed..5030052 100644 --- a/pyrainbird/timeline.py +++ b/pyrainbird/timeline.py @@ -61,7 +61,7 @@ class ProgramEvent: program_id: ProgramId start: datetime.datetime - end: datetime.dateetime + end: datetime.datetime rule: rrule.rrule | None = None @property @@ -76,7 +76,7 @@ def rrule_str(self) -> str | None: return parts[1].lstrip("RRULE:") -class ProgramTimeline(SortableItemTimeline[Timespan, ProgramEvent]): +class ProgramTimeline(SortableItemTimeline[ProgramEvent]): """A timeline of events in an irrigation program.""" @@ -135,7 +135,11 @@ def create_recurrence( ) ruleset.rrule(rule) - def adapter(dtstart: datetime.datetime) -> SortableItem[Timespan, ProgramEvent]: + def adapter( + dtstart: datetime.datetime | datetime.date, + ) -> SortableItem[Timespan, ProgramEvent]: + if not isinstance(dtstart, datetime.datetime): + raise ValueError("Expected datetime, got date") dtend = dtstart + duration def build() -> ProgramEvent: