Python Aquatlantis Ori

An asynchronous Python client for the Aquatlantis Ori Smart Controller

Installation

You can install this package using your preferred package manager. For example, using pip:

pip install aquatlantis-ori

Usage

To use the Aquatlantis Ori client, you can import it in your Python scripts and start interacting with your Ori controller. Here is a simple example:

import asyncio
import logging

from aquatlantis_ori import AvailabilityType, AquatlantisOriClient, LightOptions, PowerType

logging.basicConfig(level=logging.WARNING)
logging.getLogger("aquatlantis_ori").setLevel(logging.INFO)


async def main() -> None:
    async with AquatlantisOriClient("email", "password") as client:
        await client.connect()
        device = client.get_devices()[0]

        if device.availability_state is not AvailabilityType.AVAILABLE:
            raise RuntimeError("Device is not currently available")

        scenarios: list[dict] = [
            {"power": PowerType.ON, "options": LightOptions(intensity=100, red=100, green=0, blue=0, white=0)},
            {"power": PowerType.ON, "options": LightOptions(intensity=100, red=0, green=100, blue=0, white=0)},
            {"power": PowerType.ON, "options": LightOptions(intensity=100, red=0, green=0, blue=100, white=0)},
            {"power": PowerType.ON, "options": LightOptions(intensity=100, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.ON, "options": LightOptions(intensity=100, red=100, green=100, blue=100, white=100)},
            {"power": PowerType.OFF},
            {"power": PowerType.ON, "options": LightOptions(intensity=80, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.ON, "options": LightOptions(intensity=60, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.ON, "options": LightOptions(intensity=40, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.ON, "options": LightOptions(intensity=20, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.ON, "options": LightOptions(intensity=1, red=0, green=0, blue=0, white=100)},
            {"power": PowerType.OFF},
        ]

        for scenario in scenarios:
            device.set_light(
                power=scenario.get("power"),
                options=scenario.get("options"),
            )
            await asyncio.sleep(2)


if __name__ == "__main__":
    asyncio.run(main())

device.status remains the raw vendor-reported protocol value for diagnostics. Use device.availability_state as the canonical runtime liveness signal:

• AvailabilityType.UNKNOWN: No MQTT liveness evidence has been observed yet.
• AvailabilityType.AVAILABLE: Recent MQTT telemetry or an explicit online status proves the device is alive.
• AvailabilityType.UNAVAILABLE: A fresh explicit offline status was received, or MQTT activity has gone stale.

MQTT activity is considered stale after 6 minutes. Because not every device publishes telemetry on its own, the client requests the state of every device every 2 minutes while MQTT is connected, so a responsive device stays available. This keepalive starts on connect() and stops on close().

Contributing

Contributions are welcome! Please familiarize yourself with the contribution guidelines. This document will also help you set up your development environment.

Thank you for your interest in the Python Aquatlantis Ori client! If you have any questions or need further assistance, feel free to open an issue or submit a pull request.