pyrainbird.async_client

  1"""An asyncio based client library for Rain Bird.
  2
  3For asyncio usage, prefer creating an `AsyncRainbirdController` via
  4`await create_controller(...)`, which accepts the hostname/IP address and
  5password of the Rain Bird controller and performs local HTTP/HTTPS discovery.
  6
  7`CreateController` is a legacy factory that does not perform discovery and may
  8not work with HTTPS-only controllers.
  9
 10Most API calls are fairly low level with thin response wrappers that are data classes,
 11though some static data about the device may have the underlying calls cached.
 12
 13Note that in general the Rain Bird device can only communicate with one client at
 14a time and may raise exceptions when the device is busy. Keep this in mind when polling
 15and querying the device.
 16"""
 17
 18import datetime
 19import logging
 20import math
 21import ssl
 22from collections.abc import Callable
 23from http import HTTPStatus
 24from typing import Any, TypeVar, Union
 25
 26import aiohttp
 27from aiohttp.client_exceptions import (
 28    ClientConnectorCertificateError,
 29    ClientConnectorError,
 30    ClientConnectorSSLError,
 31    ClientError,
 32    ClientResponseError,
 33)
 34from aiohttp_retry import RetryClient, RetryOptions, JitterRetry
 35
 36
 37from . import encryption, rainbird
 38from .data import (
 39    AvailableStations,
 40    ControllerFirmwareVersion,
 41    ControllerState,
 42    ModelAndVersion,
 43    NetworkStatus,
 44    ProgramInfo,
 45    Schedule,
 46    ScheduleAndSettings,
 47    ServerMode,
 48    Settings,
 49    States,
 50    WaterBudget,
 51    WeatherAdjustmentMask,
 52    WeatherAndStatus,
 53    WifiParams,
 54    ZipCode,
 55)
 56from .exceptions import (
 57    RainbirdApiException,
 58    RainbirdAuthException,
 59    RainbirdCertificateError,
 60    RainbirdConnectionError,
 61    RainbirdDeviceBusyException,
 62)
 63from .resources import LENGTH, RAINBIRD_COMMANDS, RESPONSE
 64
 65__all__ = [
 66    "CreateController",
 67    "create_controller",
 68    "AsyncRainbirdController",
 69]
 70
 71_LOGGER = logging.getLogger(__name__)
 72T = TypeVar("T")
 73
 74
 75HEAD = {
 76    "Accept-Language": "en",
 77    "Accept-Encoding": "gzip, deflate",
 78    "User-Agent": "RainBird/2.0 CFNetwork/811.5.4 Darwin/16.7.0",
 79    "Accept": "*/*",
 80    "Connection": "keep-alive",
 81    "Content-Type": "application/octet-stream",
 82}
 83DATA = "data"
 84CLOUD_API_URL = "http://rdz-rbcloud.rainbird.com/phone-api"
 85
 86# In general, these devices can handle only one in flight request at a time
 87# otherwise return a 503. The caller is expected to follow that, however ESP
 88# ME devices also seem to return 503s more regularly than other devices so we
 89# include retry behavior for them. We only retry the specific device busy error.
 90START_TIMEOUT = 1.0
 91ATTEMPTS = 3
 92
 93
 94def _retry_delay() -> float:
 95    return START_TIMEOUT
 96
 97
 98def _retry_attempts() -> int:
 99    return ATTEMPTS
100
101
102def _device_busy_retry() -> JitterRetry:
103    return JitterRetry(
104        attempts=_retry_attempts(),
105        start_timeout=_retry_delay(),
106        statuses=set([HTTPStatus.SERVICE_UNAVAILABLE.value]),
107        retry_all_server_errors=False,
108    )
109
110
111class AsyncRainbirdClient:
112    """An asyncio rainbird client.
113
114    This is used by the controller and not expected to be used directly.
115    """
116
117    def __init__(
118        self,
119        websession: aiohttp.ClientSession,
120        url: str,
121        password: Union[str, None],
122        *,
123        ssl_context: ssl.SSLContext | bool | None = None,
124    ) -> None:
125        self._websession = websession
126        self._url = url
127        self._ssl_context = ssl_context
128        _LOGGER.debug("Using Rain Bird API endpoint: %s", self._url)
129        self._password = password
130        self._coder = encryption.PayloadCoder(password, _LOGGER)
131
132    def with_retry_options(self, retry_options: RetryOptions) -> "AsyncRainbirdClient":  # type: ignore[valid-type]
133        """Create a new AsyncRainbirdClient with retry options."""
134        return AsyncRainbirdClient(
135            RetryClient(client_session=self._websession, retry_options=retry_options),  # type: ignore[arg-type]
136            self._url,
137            self._password,
138            ssl_context=self._ssl_context,
139        )
140
141    async def request(
142        self, method: str, params: Union[dict[str, Any], None] = None
143    ) -> dict[str, Any]:
144        """Send a request for any command."""
145        payload = self._coder.encode_command(method, params or {})
146        try:
147            request_kwargs: dict[str, Any] = {}
148            if self._ssl_context is not None:
149                request_kwargs["ssl"] = self._ssl_context
150            resp = await self._websession.request(
151                "post", self._url, data=payload, headers=HEAD, **request_kwargs
152            )
153            resp.raise_for_status()
154        except ClientConnectorCertificateError as err:
155            _LOGGER.debug("Certificate verification failed: %s", err)
156            raise RainbirdCertificateError(
157                "TLS certificate verification error communicating with Rain Bird device"
158            ) from err
159        except (ClientConnectorError, ClientConnectorSSLError) as err:
160            _LOGGER.debug(
161                "Connection error communicating with Rain Bird device: %s", err
162            )
163            raise RainbirdConnectionError(
164                "Connection error communicating with Rain Bird device"
165            ) from err
166        except ClientResponseError as err:
167            _LOGGER.debug("Error response from Rain Bird device: %s", err)
168            if err.status == HTTPStatus.SERVICE_UNAVAILABLE:
169                raise RainbirdDeviceBusyException(
170                    "Rain Bird device is busy; Wait and try again"
171                ) from err
172            if err.status == HTTPStatus.FORBIDDEN:
173                raise RainbirdAuthException(
174                    "Rain Bird device denied authentication; Incorrect Password?"
175                ) from err
176            raise RainbirdApiException("Rain Bird responded with an error")
177        except ClientError as err:
178            _LOGGER.debug("Error communicating with Rain Bird device: %s", err)
179            raise RainbirdApiException(
180                "Error communicating with Rain Bird device"
181            ) from err
182        content = await resp.read()
183        return self._coder.decode_command(content)  # type: ignore
184
185
186def CreateController(
187    websession: aiohttp.ClientSession, host: str, password: str
188) -> "AsyncRainbirdController":
189    """Create an AsyncRainbirdController."""
190    local_url = f"http://{host}/stick"
191    local_client = AsyncRainbirdClient(websession, local_url, password)
192    cloud_client = AsyncRainbirdClient(websession, CLOUD_API_URL, None)
193    return AsyncRainbirdController(local_client, cloud_client)
194
195
196async def create_controller(
197    websession: aiohttp.ClientSession, host: str, password: str
198) -> "AsyncRainbirdController":
199    """Create an AsyncRainbirdController with local HTTP/HTTPS discovery.
200
201    The local Rain Bird controller API appears to vary by firmware. Some devices
202    accept HTTP on port 80, while others require HTTPS (commonly with a
203    self-signed certificate). This factory probes the controller to determine
204    the correct scheme while keeping the public API hostname-based.
205
206    Notes:
207    - The cloud client keeps its default behavior (no TLS relaxation).
208    - The local client is probed for HTTPS first using relaxed certificate
209      validation, then falls back to HTTP on transport-level errors (e.g.,
210      connection refused / TLS handshake).
211    """
212    host = host.strip()
213    host = host.rstrip("/")
214    cloud_client = AsyncRainbirdClient(websession, CLOUD_API_URL, None)
215
216    async def _probe(
217        *, url: str, ssl_context: ssl.SSLContext | bool | None
218    ) -> AsyncRainbirdController:
219        local_client = AsyncRainbirdClient(
220            websession,
221            url,
222            password,
223            ssl_context=ssl_context,
224        )
225        controller = AsyncRainbirdController(local_client, cloud_client)
226        await controller.get_model_and_version()
227        return controller
228
229    https_url = f"https://{host}/stick"
230    http_url = f"http://{host}/stick"
231    try:
232        return await _probe(url=https_url, ssl_context=False)
233    except RainbirdConnectionError:
234        # Likely wrong scheme (device doesn't speak TLS); fall back to HTTP.
235        return await _probe(url=http_url, ssl_context=None)
236
237
238class AsyncRainbirdController:
239    """Rainbird controller that uses asyncio."""
240
241    def __init__(
242        self,
243        local_client: AsyncRainbirdClient,
244        cloud_client: AsyncRainbirdClient | None = None,
245    ) -> None:
246        """Initialize AsyncRainbirdController."""
247        self._local_client = local_client
248        self._cloud_client = cloud_client
249        self._cache: dict[str, Any] = {}
250        self._model: ModelAndVersion | None = None
251
252    async def get_model_and_version(self) -> ModelAndVersion:
253        """Return the model and version."""
254        response = await self._cacheable_command(
255            lambda response: ModelAndVersion(
256                response["modelID"],
257                response["protocolRevisionMajor"],
258                response["protocolRevisionMinor"],
259            ),
260            "ModelAndVersionRequest",
261        )
262        if self._model is None:
263            self._model = response
264            if self._model.model_info.retries:
265                self._local_client = self._local_client.with_retry_options(
266                    _device_busy_retry()
267                )
268        return response
269
270    async def get_available_stations(self) -> AvailableStations:
271        """Get the available stations."""
272        mask = (
273            "%%0%dX"
274            % RAINBIRD_COMMANDS["AvailableStationsResponse"]["setStations"][LENGTH]
275        )
276        return await self._cacheable_command(
277            lambda resp: AvailableStations(mask % resp["setStations"]),
278            "AvailableStationsRequest",
279            0,
280        )
281
282    async def get_serial_number(self) -> str:
283        """Get the device serial number."""
284        try:
285            return await self._cacheable_command(
286                lambda resp: resp["serialNumber"], "SerialNumberRequest"
287            )
288        except RainbirdApiException as err:
289            _LOGGER.debug("Error while fetching serial number: %s", err)
290            raise
291
292    async def get_current_time(self) -> datetime.time:
293        """Get the device current time."""
294        return await self._process_command(
295            lambda resp: datetime.time(resp["hour"], resp["minute"], resp["second"]),
296            "CurrentTimeRequest",
297        )
298
299    async def set_current_time(self, value: datetime.time) -> None:
300        """Set the device current time."""
301        await self._process_command(
302            lambda resp: True,
303            "SetCurrentTimeRequest",
304            value.hour,
305            value.minute,
306            value.second,
307        )
308
309    async def get_current_date(self) -> datetime.date:
310        """Get the device current date."""
311        return await self._process_command(
312            lambda resp: datetime.date(resp["year"], resp["month"], resp["day"]),
313            "CurrentDateRequest",
314        )
315
316    async def set_current_date(self, value: datetime.date) -> None:
317        """Set the device current date."""
318        await self._process_command(
319            lambda resp: True,
320            "SetCurrentDateRequest",
321            value.day,
322            value.month,
323            value.year,
324        )
325
326    async def get_wifi_params(self) -> WifiParams:
327        """Return wifi parameters and other settings."""
328        try:
329            result = await self._local_client.request("getWifiParams")
330        except RainbirdApiException as err:
331            _LOGGER.debug("Error while fetching get_wifi_params: %s", err)
332            raise
333        return WifiParams.from_dict(result)
334
335    async def get_settings(self) -> Settings:
336        """Return a combined set of device settings."""
337        result = await self._local_client.request("getSettings")
338        return Settings.from_dict(result)
339
340    async def get_weather_adjustment_mask(self) -> WeatherAdjustmentMask:
341        """Return the weather adjustment mask, subset of the settings."""
342        result = await self._local_client.request("getWeatherAdjustmentMask")
343        return WeatherAdjustmentMask.from_dict(result)
344
345    async def get_zip_code(self) -> ZipCode:
346        """Return zip code and location, a subset of the settings."""
347        result = await self._local_client.request("getZipCode")
348        return ZipCode.from_dict(result)
349
350    async def get_program_info(self) -> ProgramInfo:
351        """Return program information, a subset of the settings."""
352        result = await self._local_client.request("getProgramInfo")
353        return ProgramInfo.from_dict(result)
354
355    async def get_network_status(self) -> NetworkStatus:
356        """Return the device network status."""
357        result = await self._local_client.request("getNetworkStatus")
358        return NetworkStatus.from_dict(result)
359
360    async def get_server_mode(self) -> ServerMode:
361        """Return details about the device server setup."""
362        result = await self._local_client.request("getServerMode")
363        return ServerMode.from_dict(result)
364
365    async def water_budget(self, budget) -> WaterBudget:
366        """Return the water budget."""
367        return await self._process_command(
368            lambda resp: WaterBudget(resp["programCode"], resp["seasonalAdjust"]),
369            "WaterBudgetRequest",
370            budget,
371        )
372
373    async def get_rain_sensor_state(self) -> bool:
374        """Get the current state for the rain sensor."""
375        return await self._process_command(
376            lambda resp: bool(resp["sensorState"]),
377            "CurrentRainSensorStateRequest",
378        )
379
380    async def get_zone_states(self) -> States:
381        """Return the current state of all zones."""
382        mask = (
383            "%%0%dX"
384            % RAINBIRD_COMMANDS["CurrentStationsActiveResponse"]["activeStations"][
385                LENGTH
386            ]
387        )
388        return await self._process_command(
389            lambda resp: States((mask % resp["activeStations"])),
390            "CurrentStationsActiveRequest",
391            0,
392        )
393
394    async def get_zone_state(self, zone: int) -> bool:
395        """Return the current state of the zone."""
396        states = await self.get_zone_states()
397        return states.active(zone)
398
399    async def set_program(self, program: int) -> None:
400        """Start a program."""
401        await self._process_command(
402            lambda resp: True, "ManuallyRunProgramRequest", program
403        )
404
405    async def test_zone(self, zone: int) -> None:
406        """Test a zone."""
407        await self._process_command(lambda resp: True, "TestStationsRequest", zone)
408
409    async def irrigate_zone(self, zone: int, minutes: int) -> None:
410        """Send the irrigate command."""
411        await self._process_command(
412            lambda resp: True, "ManuallyRunStationRequest", zone, minutes
413        )
414
415    async def stop_irrigation(self) -> None:
416        """Send the stop command."""
417        await self._process_command(lambda resp: True, "StopIrrigationRequest")
418
419    async def get_rain_delay(self) -> int:
420        """Return the current rain delay value."""
421        return await self._process_command(
422            lambda resp: resp["delaySetting"], "RainDelayGetRequest"
423        )
424
425    async def set_rain_delay(self, days: int) -> None:
426        """Set the rain delay value in days."""
427        await self._process_command(lambda resp: True, "RainDelaySetRequest", days)
428
429    async def advance_zone(self, param: int) -> None:
430        """Advance to the specified zone."""
431        await self._process_command(lambda resp: True, "AdvanceStationRequest", param)
432
433    async def get_current_irrigation(self) -> bool:
434        """Return True if the irrigation state is on."""
435        return await self._process_command(
436            lambda resp: bool(resp["irrigationState"]),
437            "CurrentIrrigationStateRequest",
438        )
439
440    async def get_schedule_and_settings(self, stick_id: str) -> ScheduleAndSettings:
441        """Request the schedule and settings from the cloud."""
442        if not self._cloud_client:
443            raise ValueError("Cloud client not configured")
444        result = await self._cloud_client.request(
445            "requestScheduleAndSettings", {"StickId": stick_id}
446        )
447        return ScheduleAndSettings.from_dict(result)
448
449    async def get_weather_and_status(
450        self, stick_id: str, country: str, zip_code: str
451    ) -> WeatherAndStatus:
452        """Request the weather and status of the device.
453
454        The results include things like custom station names, program names, etc.
455        """
456        if not self._cloud_client:
457            raise ValueError("Cloud client not configured")
458        result = await self._cloud_client.request(
459            "requestWeatherAndStatus",
460            {
461                "Country": country,
462                "StickId": stick_id,
463                "ZipCode": zip_code,
464            },
465        )
466        return WeatherAndStatus.from_dict(result)
467
468    async def get_combined_controller_state(self) -> ControllerState:
469        """Return the combined controller state."""
470        return await self._process_command(
471            lambda resp: ControllerState.from_dict(resp),
472            "CombinedControllerStateRequest",
473        )
474
475    async def get_controller_firmware_version(self) -> ControllerFirmwareVersion:
476        """Return the controller firmware version."""
477        return await self._process_command(
478            lambda resp: ControllerFirmwareVersion(
479                resp["major"], resp["minor"], resp["patch"]
480            ),
481            "ControllerFirmwareVersionRequest",
482        )
483
484    async def get_schedule(self) -> Schedule:
485        """Return the device schedule."""
486
487        model = await self.get_model_and_version()
488        max_programs = model.model_info.max_programs
489        stations = await self.get_available_stations()
490        max_stations = model.model_info.max_stations
491        if not max_stations:
492            # Fallback for unknown models
493            max_stations = min(stations.stations.count, 22)
494
495        commands = ["00"]
496        # Program details
497        for program in range(0, max_programs):
498            commands.append("%04x" % (0x10 | program))
499        # Start times
500        for program in range(0, max_programs):
501            commands.append("%04x" % (0x60 | program))
502        # Run times per zone
503        _LOGGER.debug("Loading schedule for %d zones", max_stations)
504        for zone_page in range(0, math.ceil(max_stations / 2)):
505            commands.append("%04x" % (0x80 | zone_page))
506        _LOGGER.debug("Sending schedule commands: %s", commands)
507        # Run command serially to avoid overwhelming the controller
508        schedule_data: dict[str, Any] = {
509            "controllerInfo": {},
510            "programInfo": [],
511            "programStartInfo": [],
512            "durations": [],
513        }
514        for command in commands:
515            result = await self._process_command(
516                lambda resp: resp,
517                "RetrieveScheduleRequest",
518                int(command, 16),
519            )
520            if not isinstance(result, dict):
521                continue
522            for key in schedule_data:
523                if (value := result.get(key)) is not None:
524                    if key == "durations":
525                        for entry in value:
526                            if (
527                                entry.get("zone", 0) + 1
528                            ) not in stations.stations.active_set:
529                                continue
530                            schedule_data[key].append(entry)
531                    elif key == "controllerInfo":
532                        schedule_data[key].update(value)
533                    else:
534                        schedule_data[key].append(value)
535        return Schedule.from_dict(schedule_data)
536
537    async def get_schedule_command(self, command_code: str) -> dict[str, Any]:
538        """Run the schedule command for the specified raw command code."""
539        return await self._process_command(
540            lambda resp: resp,
541            "RetrieveScheduleRequest",
542            command_code,
543        )
544
545    async def test_command_support(self, command_id: int) -> bool:
546        """Debugging command to test if the device supports the specified command."""
547        return await self._process_command(
548            lambda resp: bool(resp["support"]), "CommandSupportRequest", command_id
549        )
550
551    async def test_rpc_support(self, rpc: str) -> dict[str, Any]:
552        """Debugging command to test device support for a json RPC method."""
553        return await self._local_client.request(rpc)
554
555    async def _tunnelSip(self, data: str, length: int) -> str:
556        """Send a tunnelSip request."""
557        result = await self._local_client.request(
558            "tunnelSip", {DATA: data, LENGTH: length}
559        )
560        if DATA not in result:
561            _LOGGER.error(
562                "Rain Bird device reply missing required 'data' field in tunnelSip"
563            )
564            raise RainbirdApiException("Unexpected response from Rain Bird device")
565        return result[DATA]
566
567    async def _process_command(
568        self, funct: Callable[[dict[str, Any]], T], command: str, *args
569    ) -> T:
570        data = rainbird.encode(command, *args)
571        _LOGGER.debug("Request (%s): %s", command, str(data))
572        command_data = RAINBIRD_COMMANDS[command]
573        decrypted_data = await self._tunnelSip(
574            data,
575            command_data[LENGTH],
576        )
577        _LOGGER.debug("Response from line: " + str(decrypted_data))
578        decoded = rainbird.decode(decrypted_data)
579        _LOGGER.debug("Response: %s" % decoded)
580        response_code = decrypted_data[:2]
581        allowed = set([command_data[RESPONSE]])
582        if funct is None:
583            allowed.add("00")  # Allow NACK
584        if response_code not in allowed:
585            _LOGGER.error(
586                "Request (%s) failed with wrong response! Requested (%s), got %s:\n%s"
587                % (command, allowed, response_code, decoded)
588            )
589            raise RainbirdApiException("Unexpected response from Rain Bird device")
590        return funct(decoded)
591
592    async def _cacheable_command(
593        self, funct: Callable[[dict[str, Any]], T], command: str, *args
594    ) -> T:
595        key = f"{command}-{args}"
596        if result := self._cache.get(key):
597            _LOGGER.debug("Returned cached result for key '%s'", key)
598            return result
599        result = await self._process_command(funct, command, *args)
600        self._cache[key] = result
601        return result  # type: ignore