pyWorxCloud

This is a PyPI module for communicating with Worx Cloud mowers, primarily developed for use with Home Assistant, but I try to keep it as widely usable as possible.

The module are compatible with cloud enabled devices from these vendors

Documentation

The documentation have been moved to the Wiki
Additional project docs:

• Migration Guide
• Changelog

Async usage

WorxCloud is now async-first.

import asyncio
from pyworxcloud import WorxCloud


async def main() -> None:
    cloud = WorxCloud("user@example.com", "secret", "worx")
    await cloud.authenticate()
    await cloud.connect()
    try:
        for _, device in cloud.devices.items():
            print(device.name, device.online)
    finally:
        await cloud.disconnect()


asyncio.run(main())

You can also use async with:

async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    ...

Async migration cheat sheet

Before (sync):

cloud = WorxCloud("user@example.com", "secret", "worx")
cloud.authenticate()
cloud.connect()
cloud.start("SERIAL")
cloud.disconnect()

After (async):

cloud = WorxCloud("user@example.com", "secret", "worx")
await cloud.authenticate()
await cloud.connect()
await cloud.start("SERIAL")
await cloud.disconnect()

Testing

Run tests locally with:

python -m pip install -e . pytest
bash scripts/prepare_test_fixtures.sh
pytest -q

The fixture prepare script copies JSON sample files from code-ref/data-samples to tests/fixtures/data-samples when available.

Sample validation

Run python scripts/verify_data_samples.py (or rely on tests/test_data_samples.py) to ensure every code-ref/data-samples fixture contains the minimal payload/cfg/dat structure (id, conn, and uuid/mac). This keeps the fixtures aligned with DeviceHandler/EventHandler expectations even as you add new samples.

You can also run python scripts/dump_mapping.py to print the decoded snapshot for each fixture (status, schedules, rain delay, module data, etc.) so you can visually compare the raw JSON against the values DeviceHandler exposes. The script now scans both http.json and mqtt.json fixtures (including multi-document MQTT exports) so you can inspect how either transport influences the decoded output.

Data mapping

DeviceHandler now keeps the raw cfg/dat dictionaries alongside the richer surface model that mirrors what is described in code-ref. Highlights include:

• schedules["slots"] retains every slot that was present in sc.slots or sc.d, so protocol 1 devices with more end-of-day runs can be inspected.

• schedules["auto_schedule"] now exposes a normalized automatic-scheduling view with typed boost, grass_type, soil_type, irrigation, nutrition, and exclusion-scheduler fields when the mower API payload provides them. Matching helpers can update the supported automatic-scheduling settings through the top-level REST payload. Live-observed boost levels are 0, 1, and 2.

• slot-first schedule generation that captures each configured run (legacy d arrays and protocol 1 slots) along with calculated end times, pause-mode awareness, and time-extension handling.

• complete rain-delay state tracking (raw counter, active flag, remaining minutes) plus module status/configuration (ACS, Off Limits shortcuts, etc.).

• real-time updates of lock state, battery/blade statistics, orientation, GPS hooks, and module-specific metadata so MQTT and API consumers stay synchronized.

The fixture-driven tests/test_device_decode.py now asserts that the raw payloads, module data, and rain delay flags survive the round-trip, making regressions obvious as the refactor continues.

Networking helpers

pyworxcloud.utils.requests now exposes async AGET/APOST helpers backed by aiohttp.ClientSession for non-blocking API access. Legacy sync GET/POST helpers are still available for compatibility and utility scripts.

Command timeout configuration

WorxCloud accepts a command_timeout argument (seconds) that controls how long MQTT command calls wait for a matching mower response before raising TimeoutException.

Initial MQTT connection attempts use a separate mqtt_connect_timeout argument. This keeps API-backed startup responsive when the cloud MQTT service is unavailable, while preserving a longer timeout for mower command responses.

The read-only mqtt_connected property reports whether the MQTT client is currently active. It is available on both the WorxCloud instance and the mapped device objects. It is False when startup falls back to API-only mode, after disconnects, and whenever the MQTT client loses its connection.

from pyworxcloud import WorxCloud

cloud = WorxCloud(
    "user@example.com",
    "secret",
    "worx",
    command_timeout=15.0,
    mqtt_connect_timeout=8.0,
)

Schedule CRUD

WorxCloud now exposes a normalized schedule API for both protocol 0 (d/dd) and protocol 1 (slots) mowers.

from pyworxcloud import WorxCloud
from pyworxcloud.utils.schedule_codec import ScheduleEntry


schedule = cloud.get_schedule("SERIAL")

await cloud.add_schedule_entry(
    "SERIAL",
    ScheduleEntry(
        entry_id="",
        day="monday",
        start="09:00",
        duration=60,
        boundary=False,
        source="slot",
        secondary=False,
    ),
)

Notes:

• get_schedule() returns a normalized ScheduleModel.
• set_schedule(), add_schedule_entry(), update_schedule_entry(), and delete_schedule_entry() automatically serialize back to the correct protocol payload.
• Protocol 0 deletion promotes same-day secondary schedules into primary when needed.
• Protocol 1 updates preserve extra slot metadata such as zone lists when the current payload contains them.
• set_time_extension() is only supported for protocol 0 mowers.

Automatic scheduling

WorxCloud now supports both reading and updating the observed automatic-scheduling fields exposed on product-items.

from pyworxcloud import WorxCloud


async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    serial = "SERIAL"

    await cloud.toggle_auto_schedule(serial, True)
    await cloud.set_auto_schedule_boost(serial, 1)
    await cloud.set_auto_schedule_grass_type(serial, "festuca_arundinacea")
    await cloud.set_auto_schedule_soil_type(serial, "clay")
    await cloud.set_auto_schedule_irrigation(serial, True)
    await cloud.set_auto_schedule_nutrition(serial, 10, 20, 5)
    await cloud.set_auto_schedule_exclude_nights(serial, True)
    await cloud.set_auto_schedule_exclusion_day(serial, 2, True)
    await cloud.set_auto_schedule_exclusion_slots(
        serial,
        4,
        [{"start_time": 900, "duration": 60, "reason": "irrigation"}],
    )

Notes:

• device.schedules["auto_schedule"] exposes the normalized read model with enabled and settings.
• Supported write helpers cover auto_schedule, boost, grass_type, soil_type, irrigation, nutrition, exclude_nights, and per-day exclusion slots.
• Use clear_auto_schedule_nutrition() to send nutrition = null when nutrition should be disabled.
• Exclusion slot writes currently send a full seven-day exclusion_scheduler.days payload, with the selected day replaced by the provided slots.

Firmware auto-upgrade

WorxCloud can now update the observed top-level firmware_auto_upgrade field on product-items.

from pyworxcloud import WorxCloud


async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    serial = "SERIAL"

    await cloud.set_firmware_auto_upgrade(serial, True)

Notes:

• device.firmware["auto_upgrade"] reflects the current auto-upgrade setting.
• This helper only toggles the top-level auto-upgrade flag. Firmware availability and OTA-trigger flows are handled separately.

Firmware availability

WorxCloud can fetch the app-observed firmware-upgrade metadata for a mower.

from pyworxcloud import WorxCloud


async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    serial = "SERIAL"
    firmware = await cloud.get_firmware_upgrade_info(serial)

    print(firmware["current_version"])
    print(firmware["latest_version"])
    print(firmware["update_available"])

Notes:

• The helper reads GET /api/v2/product-items/{serial}/firmware-upgrade.
• The returned dictionary includes current_version, latest_version, update_available, mandatory, ota_supported, auto_upgrade, upgrade_failed, and normalized product / head entries when present.
• Firmware changelogs are exposed both as the raw changelog payload and as a Markdown-friendly changelog_markdown variant for UIs that render rich text.
• The normalized result is cached on both mower["firmware_upgrade"] and device.firmware["upgrade"].

Firmware OTA trigger

WorxCloud can queue an OTA firmware upgrade when the mower supports it and the backend exposes an available upgrade.

from pyworxcloud import WorxCloud


async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    serial = "SERIAL"
    await cloud.start_firmware_upgrade(serial)

Notes:

• The helper queues POST /api/v2/product-items/{serial}/firmware-upgrade.
• NoFirmwareOtaError is raised when OTA updates are known unsupported for the mower.
• NoFirmwareAvailableError is raised when the backend reports that no OTA firmware is currently available.
• On success the helper refreshes mower state and marks the cached firmware upgrade payload as command_queued = True when it already exists locally.

Lawn fields

WorxCloud can now read and write top-level REST lawn fields on product-items: lawn_size (m²) and lawn_perimeter (m).

from pyworxcloud import WorxCloud


async with WorxCloud("user@example.com", "secret", "worx") as cloud:
    serial = "SERIAL"

    # Update fields individually.
    await cloud.set_lawn_size(serial, 250)
    await cloud.set_lawn_perimeter(serial, 115)

    # Or update both in one request.
    await cloud.set_lawn(serial, size=250, perimeter=115)

Deprecations

The test suite reports overdue deprecations as warnings so maintainers can plan their removal without blocking compatible patch releases.