💧 Regenmaschine: A Simple Python Library for RainMachine™

[![CI][ci-badge]][ci] [![PyPI][pypi-badge]][pypi] [![Version][version-badge]][version] [![License][license-badge]][license] [![Code Coverage][codecov-badge]][codecov] [![Maintainability][maintainability-badge]][maintainability]

regenmaschine (German for "rain machine") is a simple, clean, well-tested Python library for interacting with [RainMachine™ smart sprinkler controllers][regenmaschine]. It gives developers an easy API to manage their controllers over their local LAN or remotely via the RainMachine™ cloud.

Remote Access Announcement
Python Versions
Installation
Usage
Loading Controllers Multiple Times
Contributing

Remote Access Announcement (2022-06-26)

On June 2, 2022, RainMachine announced a [Premium Services][rainmachine-premium] addition; under this new model, remote access is only available to subscribers of these Premium Services.

I do not currently intend to subscribe to Premium Services; as such, the remote access abilities of regenmaschine will remain as-is from here on out unless spurred on by others. They may stop working at any time. PRs from subscribing users are always welcome.

Python Versions

regenmaschine is currently supported on:

Python 3.10
Python 3.11
Python 3.12

Installation

pip install regenmaschine

Usage

Creating a regenmaschine Client might be the easiest thing you do all day:

import asyncio

from aiohttp import ClientSession

from regenmaschine import Client


async def main() -> None:
    """Run!"""
    client = Client()

    # ...


asyncio.run(main())

By default, the library creates a new connection to the sprinkler controller with each coroutine. If you are calling a large number of coroutines (or merely want to squeeze out every second of runtime savings possible), an [aiohttp][aiohttp] ClientSession can be used for connection pooling:

See the module docstrings throughout the library for full info on all parameters, return types, etc.

import asyncio

from aiohttp import ClientSession

from regenmaschine import Client


async def main() -> None:
    """Run!"""
    async with ClientSession() as session:
        client = Client(session=session)


asyncio.run(main())

Loading Local (Accessible Over the LAN) Controllers

Once you have a client, you can load a local controller (i.e., one that is accessible over the LAN) very easily:

import asyncio

from aiohttp import ClientSession

from regenmaschine import Client


async def main() -> None:
    """Run!"""
    async with ClientSession() as session:
        client = Client(session=session)

        await client.load_local("192.168.1.101", "my_password", port=8080, use_ssl=True)

        controllers = client.controllers
        # >>> {'ab:cd:ef:12:34:56': <LocalController>}


asyncio.run(main())

Loading Remote (Accessible Over the RainMachine Cloud) Controllers

If you have 1, 2 or 100 other local controllers, you can load them in the same way – client.controllers will keep your controllers all organized.

What if you have controllers around the world and can't access them all over the same local network? No problem! regenmaschine allows you to load remote controllers very easily, as well:

import asyncio

from aiohttp import ClientSession

from regenmaschine import Client


async def main() -> None:
    """Run!"""
    async with ClientSession() as session:
        client = Client(session=session)

        await client.load_remote("rainmachine_email@host.com", "my_password")

        controllers = client.controllers
        # >>> {'xx:xx:xx:xx:xx:xx': <RemoteController>, ...}


asyncio.run(main())

Bonus tip: client.load_remote will load all controllers owned by that email address.

Using the Controller

Regardless of the type of controller you have loaded (local or remote), the same properties and methods are available to each:

import asyncio
import datetime

from aiohttp import ClientSession

from regenmaschine import Client


async def main() -> None:
    """Run!"""
    async with ClientSession() as session:
        client = Client(session=session)

        # Load a local controller:
        await client.load_local("192.168.1.101", "my_password", port=8080, use_ssl=True)

        # Load all remote controllers associated with an account:
        await client.load_remote("rainmachine_email@host.com", "my_password")

        # They all act the same! The only difference is that remote API calls
        # will pass through the RainMachine™ cloud:
        for mac_address, controller in client.controllers:
            # Print some client properties:
            print(f"Name: {controller.name}")
            print(f"Host: {controller.host}")
            print(f"MAC Address: {controller.mac}")
            print(f"API Version: {controller.api_version}")
            print(f"Software Version: {controller.software_version}")
            print(f"Hardware Version: {controller.hardware_version}")

            # Get all diagnostic information:
            diagnostics = await controller.diagnostics.current()

            # Get all weather parsers:
            parsers = await controller.parsers.current()

            # Get all programs:
            programs = await controller.programs.all()

            # Include inactive programs:
            programs = await controller.programs.all(include_inactive=True)

            # Get a specific program:
            program_1 = await controller.programs.get(1)

            # Enable or disable a specific program:
            await controller.programs.enable(1)
            await controller.programs.disable(1)

            # Get the next run time for all programs:
            runs = await controller.programs.next()

            # Get all running programs:
            programs = await controller.programs.running()

            # Start and stop a program:
            await controller.programs.start(1)
            await controller.programs.stop(1)

            # Get basic details about all zones:
            zones = await controller.zones.all()

            # Get advanced details about all zones:
            zones = await controller.zones.all(details=True)

            # Include inactive zones:
            zones = await controller.zones.all(include_inactive=True)

            # Get basic details about a specific zone:
            zone_1 = await controller.zones.get(1)

            # Get advanced details about a specific zone:
            zone_1 = await controller.zones.get(1, details=True)

            # Enable or disable a specific zone:
            await controller.zones.enable(1)
            await controller.zones.disable(1)

            # Start a zone for 60 seconds:
            await controller.zones.start(1, 60)

            # ...and stop it:
            await controller.zones.stop(1)

            # Get all running zones:
            programs = await controller.zones.running()

            # Get the device name:
            name = await controller.provisioning.device_name

            # Get all provisioning settings:
            settings = await controller.provisioning.settings()

            # Get all networking info related to the device:
            wifi = await controller.provisioning.wifi()

            # Get various types of active watering restrictions:
            current = await controller.restrictions.current()
            universal = await controller.restrictions.universal()
            hourly = await controller.restrictions.hourly()
            raindelay = await controller.restrictions.raindelay()

            # Set universal restrictions – note that the payload is the same structure
            # as returned by controller.restrictions.universal():
            await controller.restrictions.set_universal(
                {
                    "hotDaysExtraWatering": False,
                    "freezeProtectEnabled": True,
                }
            )

            # Get watering stats:
            today = await controller.stats.on_date(datetime.date.today())
            upcoming_days = await controller.stats.upcoming(details=True)

            # Get info on various watering activities not already covered:
            log = await controller.watering.log(datetime.date.today(), 2)
            queue = await controller.watering.queue()
            runs = await controller.watering.runs(datetime.date.today())

            # Pause all watering activities for 30 seconds:
            await controller.watering.pause_all(30)

            # Unpause all watering activities:
            await controller.watering.unpause_all()

            # Stop all watering activities:
            await controller.watering.stop_all()

            # See if a firmware update is available:
            update_data = await controller.machine.get_firmware_update_status()
            # ...and request the update:
            update_data = await controller.machine.update_firmware()

            # Reboot the controller:
            update_data = await controller.machine.reboot()

            # Return the current flow meter data:
            flowmeter = await controller.watering.flowmeter()

            # Add values to flowmeter counters from an external smart water meter
            # not wired directly to the controller.
            # Units can be "clicks", "gal", "m3" and "litre".
            await controller.watering.post_flowmeter({"value": 2000, "units": "clicks"})


asyncio.run(main())

Check out example.py, the tests, and the source files themselves for method signatures and more exa

Regenmaschine

Install / Use

README