### Async Migration Example for WorxCloud Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Provides a comparison between synchronous and asynchronous usage patterns for the WorxCloud class. It highlights the necessary changes in method calls, specifically the addition of `await` for asynchronous operations. ```python # 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() ``` -------------------------------- ### Exception Handling Example Source: https://context7.com/mtrab/pyworxcloud/llms.txt Demonstrates how to use try-except blocks to catch and handle specific exceptions during authentication, connection, and mower command execution. ```APIDOC ## Handling Exceptions Comprehensive error handling for all library operations including authentication, connectivity, and device-specific errors. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import ( AuthorizationError, TooManyRequestsError, OfflineError, NoConnectionError, TimeoutException, NoPauseModeError, NoOneTimeScheduleError, NoOfflimitsError, NoACSModuleError, NoCuttingHeightError, ZoneNotDefined, ZoneNoProbability, MowerNotFoundError, ) async def with_error_handling(): try: cloud = WorxCloud("user@example.com", "password", "worx") await cloud.authenticate() await cloud.connect() serial = "WX123456789" try: await cloud.start(serial) except OfflineError: print("Mower is offline") except TimeoutException: print("Command timed out - no response from mower") except MowerNotFoundError: print("Mower not found with that serial number") except AuthorizationError: print("Invalid username or password") except TooManyRequestsError: print("Rate limited - wait before retrying") except NoConnectionError: print("Cannot connect to cloud service") finally: if 'cloud' in locals(): await cloud.disconnect() asyncio.run(with_error_handling()) ``` ### Description This example showcases how to wrap asynchronous operations with `try-except` blocks to gracefully handle potential errors. It covers authentication, connection, and specific mower commands, providing user-friendly messages for each exception. ### Method N/A (Illustrative Python code) ### Endpoint N/A (Illustrative Python code) ### Parameters N/A ### Request Example N/A ### Response N/A (Prints messages to console based on exceptions) ### Error Handling Details - **AuthorizationError**: Raised when authentication credentials are invalid. - **TooManyRequestsError**: Raised when the API rate limit is exceeded. - **OfflineError**: Raised when the mower is detected as offline. - **NoConnectionError**: Raised when a connection to the Worx cloud service cannot be established. - **TimeoutException**: Raised when a command sent to the mower times out without a response. - **MowerNotFoundError**: Raised when a mower with the specified serial number is not found. ``` -------------------------------- ### Initiate Zone Training with PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt This Python code starts the zone training process for a Worx Landroid mower. The mower will then trace its boundary to map the lawn area. This function requires the pyworxcloud library. ```python import asyncio from pyworxcloud import WorxCloud async def zone_training(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" await cloud.zonetraining(serial) print("Zone training started") asyncio.run(zone_training()) ``` -------------------------------- ### Running Local Tests for pyWorxCloud Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Instructions for setting up and running local tests for the pyWorxCloud module. This involves installing development dependencies, preparing test fixtures, and executing pytest. ```bash python -m pip install -e . pytest bash scripts/prepare_test_fixtures.sh pytest -q ``` -------------------------------- ### Async Usage of WorxCloud in Python Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Demonstrates how to use the WorxCloud class asynchronously to connect to Worx Cloud mowers. It shows authentication, connection, iterating through devices, and proper disconnection. This example uses the asyncio library for non-blocking operations. ```python 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()) ``` -------------------------------- ### GET /mower/error Source: https://github.com/mtrab/pyworxcloud/blob/master/Files/Status and error codes.txt Retrieves the current error state of the mower. ```APIDOC ## GET /mower/error ### Description Returns the current error code indicating any active faults on the mower. ### Method GET ### Endpoint /mower/error ### Response #### Success Response (200) - **error_code** (integer) - The current error state (e.g., 0 for no error, 1 for trapped). #### Response Example { "error_code": 0 } ``` -------------------------------- ### Initiate One-Time Mowing Schedule Source: https://context7.com/mtrab/pyworxcloud/llms.txt Starts a one-time mowing task for a Worx Landroid mower with a specified runtime. Optionally, the operation can begin with an edge-cutting phase. This function includes error handling for devices that do not support one-time schedules or are offline. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import NoOneTimeScheduleError, OfflineError async def run_ots(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" try: # Start OTS: 60 minutes, with boundary cut first await cloud.ots(serial, boundary=True, runtime="60") print("One-time schedule started: 60 min with edge cut") # Start OTS: 45 minutes, without boundary cut await cloud.ots(serial, boundary=False, runtime="45") print("One-time schedule started: 45 min without edge cut") except NoOneTimeScheduleError: print("This device does not support one-time schedules") except OfflineError: print("Device is offline") asyncio.run(run_ots()) ``` -------------------------------- ### GET /mower/status Source: https://github.com/mtrab/pyworxcloud/blob/master/Files/Status and error codes.txt Retrieves the current operational status of the mower. ```APIDOC ## GET /mower/status ### Description Returns the current operational status of the mower as an integer code. ### Method GET ### Endpoint /mower/status ### Response #### Success Response (200) - **status_code** (integer) - The current state of the mower (e.g., 1 for home, 7 for cutting grass). #### Response Example { "status_code": 7 } ``` -------------------------------- ### Mower Control API Source: https://github.com/mtrab/pyworxcloud/wiki/Methods Endpoints for controlling mower operations such as starting, pausing, and sending commands. ```APIDOC ## POST /start ### Description Starts the mowing operation for a specific device. ### Method POST ### Endpoint /start ### Parameters #### Request Body - **serial_number** (str) - Required - The unique identifier of the mower. ### Request Example { "serial_number": "123456789" } ### Response #### Success Response (200) - **status** (string) - Success confirmation. ``` ```APIDOC ## POST /ots ### Description Starts a one-time-schedule for the specified mower. ### Method POST ### Endpoint /ots ### Parameters #### Request Body - **serial_number** (str) - Required - The unique identifier of the mower. - **boundary** (bool) - Required - Whether to include boundary cutting. - **runtime** (str/int) - Required - Duration for the schedule. ### Request Example { "serial_number": "123456789", "boundary": true, "runtime": "60" } ### Response #### Success Response (200) - **status** (string) - Success confirmation. ``` ```APIDOC ## POST /send ### Description Sends raw JSON data to the mower for advanced configuration. ### Method POST ### Endpoint /send ### Parameters #### Request Body - **serial_number** (str) - Required - The unique identifier of the mower. - **data** (str) - Required - Raw JSON string to be sent to the device. ### Request Example { "serial_number": "123456789", "data": "{\"key\": \"value\"}" } ### Response #### Success Response (200) - **status** (string) - Success confirmation. ``` -------------------------------- ### Get Mower Schedule with PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt This Python script retrieves the normalized schedule for a Worx Landroid mower. It works with devices using either protocol 0 or protocol 1. The output includes schedule status, protocol type, time extension, and detailed entries for each scheduled mowing session. Requires the pyworxcloud library. ```python import asyncio from pyworxcloud import WorxCloud async def get_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Get normalized schedule model schedule = cloud.get_schedule(serial) print(f"Schedule enabled: {schedule.enabled}") print(f"Protocol: {schedule.protocol}") print(f"Time extension: {schedule.time_extension}%") print(f"Entries: {len(schedule.entries)}") for entry in schedule.entries: print(f" {entry.day}: {entry.start} for {entry.duration} min") print(f" Boundary: {entry.boundary}, Source: {entry.source}") print(f" Entry ID: {entry.entry_id}") asyncio.run(get_schedule()) ``` -------------------------------- ### Control Mower Operations Source: https://context7.com/mtrab/pyworxcloud/llms.txt Illustrates how to send operational commands like start, pause, home, and safehome to a specific mower using its serial number. It includes error handling for offline device states. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import OfflineError async def control_mower(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" try: await cloud.start(serial) print("Mower started") await cloud.pause(serial) print("Mower paused") await cloud.home(serial) print("Mower returning home") await cloud.safehome(serial) print("Mower safely returning home with blades off") except OfflineError: print("Cannot send command - mower is offline") asyncio.run(control_mower()) ``` -------------------------------- ### GET /cfg-payload-mapping Source: https://github.com/mtrab/pyworxcloud/blob/master/DATA_MAPPING.md Describes the mapping of the 'cfg' payload fields which represent mower configuration, schedules, and module settings. ```APIDOC ## GET /cfg-payload-mapping ### Description This documentation details how the `cfg` JSON payload is mapped to the `schedules` and `capabilities` objects within the system. ### Method GET ### Parameters #### Request Body - **rd** (integer) - Required - Rain delay configured minutes. - **mz/mzv** (array) - Required - Multi-zone start distances and index map. - **sc** (object) - Required - Schedule definition including slots and time extensions. - **modules** (object) - Required - Module configuration for capabilities like ACS or Off-Limits. ### Response #### Success Response (200) - **schedules** (object) - Parsed schedule slots and metadata. - **capabilities** (array) - List of enabled device features. ``` -------------------------------- ### GET /dat-payload-mapping Source: https://github.com/mtrab/pyworxcloud/blob/master/DATA_MAPPING.md Describes the mapping of the 'dat' payload fields which represent real-time mower telemetry and status. ```APIDOC ## GET /dat-payload-mapping ### Description This endpoint documentation outlines how the `dat` JSON payload from the mower API is parsed into the internal `device` object. ### Method GET ### Parameters #### Request Body - **uuid/mac** (string) - Required - Unique mower identifier. - **ls** (integer) - Required - Current status/state code. - **le** (integer) - Required - Error code. - **bt** (object) - Required - Battery telemetry data. - **st** (object) - Required - Runtime statistics. ### Response #### Success Response (200) - **device** (object) - The updated internal device state object. ``` -------------------------------- ### Update Schedule Entry with PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt This Python script updates an existing schedule entry for a Worx Landroid mower using its entry ID. The changes are then persisted to the device. It requires the pyworxcloud and pyworxcloud.utils.schedule_codec libraries. The example first retrieves the current schedule to find an entry ID. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils.schedule_codec import ScheduleEntry async def update_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Get current schedule to find entry IDs schedule = cloud.get_schedule(serial) if schedule.entries: entry_id = schedule.entries[0].entry_id # Create updated entry updated_entry = ScheduleEntry( entry_id=entry_id, day="monday", start="10:30", # Changed time duration=90, # Changed duration boundary=False, source="primary", secondary=False ) await cloud.update_schedule_entry(serial, entry_id, updated_entry) print(f"Updated entry {entry_id}") asyncio.run(update_schedule()) ``` -------------------------------- ### Manage Mower Lock Status Source: https://context7.com/mtrab/pyworxcloud/llms.txt Controls the lock status of a Worx Landroid mower. The mower can be locked to prevent unauthorized operation or unlocked to allow manual starting. The code also checks and prints the current lock status of the device. ```python import asyncio from pyworxcloud import WorxCloud async def manage_lock(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Lock the device await cloud.set_lock(serial, state=True) print("Mower locked") # Unlock the device await cloud.set_lock(serial, state=False) print("Mower unlocked") # Check lock status device = cloud.get_mower(serial, device=True) print(f"Lock status: {'Locked' if device.locked else 'Unlocked'}") asyncio.run(manage_lock()) ``` -------------------------------- ### Initialize and Authenticate WorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Demonstrates how to instantiate the WorxCloud client, authenticate, and connect to retrieve device information. It includes error handling for authorization and rate-limiting scenarios. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.clouds import CloudType from pyworxcloud.exceptions import AuthorizationError, TooManyRequestsError async def main(): cloud = WorxCloud( username="user@example.com", password="your_password", cloud=CloudType.WORX, verify_ssl=True, tz="Europe/Berlin", command_timeout=30.0 ) try: await cloud.authenticate() await cloud.connect() for name, device in cloud.devices.items(): print(f"Device: {device.name}") print(f" Model: {device.model}") print(f" Serial: {device.serial_number}") print(f" Online: {device.online}") print(f" Battery: {device.battery.percent}%") except AuthorizationError: print("Invalid credentials") except TooManyRequestsError: print("Rate limited, try again later") finally: await cloud.disconnect() asyncio.run(main()) ``` -------------------------------- ### Pyworxcloud: Establish Basic Connection and Authenticate Source: https://github.com/mtrab/pyworxcloud/wiki/Examples This snippet demonstrates how to initialize the WorxCloud client, authenticate with provided credentials, and establish a connection to a device. It includes error handling for authorization failures and proper disconnection. ```python from pyworxcloud import WorxCloud cloud = WorxCloud("your@email", "password", "worx") # Initialize connection try: cloud.authenticate() except AuthorizationError: # If invalid credentials are used, or something happend during # authorize, then exit exit(0) # Connect to device with index 0 (devices are enumerated 0, 1, 2 ...) # and do not verify SSL (False) cloud.connect() # Do your stuff # Disconnect from the API cloud.disconnect() ``` -------------------------------- ### Manage Connection with Async Context Manager Source: https://context7.com/mtrab/pyworxcloud/llms.txt Shows the usage of the async context manager pattern to automatically handle connection establishment and cleanup of the WorxCloud client. ```python import asyncio from pyworxcloud import WorxCloud async def main(): async with WorxCloud("user@example.com", "password", "worx") as cloud: for name, device in cloud.devices.items(): print(f"{device.name}: {'Online' if device.online else 'Offline'}") print(f" Status: {device.status.description}") print(f" Error: {device.error.description if device.error.id else 'None'}") asyncio.run(main()) ``` -------------------------------- ### Migrating to Async WorxCloud Methods Source: https://github.com/mtrab/pyworxcloud/blob/master/MIGRATION.md Demonstrates the transition from synchronous to asynchronous method calls for WorxCloud operations, requiring the use of the await keyword for I/O-bound tasks. ```python # Before cloud.authenticate() cloud.connect() cloud.update("SERIAL") cloud.disconnect() # After await cloud.authenticate() await cloud.connect() await cloud.update("SERIAL") await cloud.disconnect() ``` -------------------------------- ### Async Context Manager for WorxCloud Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Illustrates using the WorxCloud class as an asynchronous context manager (`async with`). This simplifies resource management by automatically handling connection and disconnection. ```python async with WorxCloud("user@example.com", "secret", "worx") as cloud: ... ``` -------------------------------- ### Register Event Callbacks for MQTT Updates Source: https://context7.com/mtrab/pyworxcloud/llms.txt Demonstrates how to register callback functions to handle real-time data updates, MQTT connection status changes, and API refreshes from the mower. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.events import LandroidEvent def on_data_received(name, device): print(f"Data received from {name}") print(f" Status: {device.status.description}") print(f" Battery: {device.battery.percent}%") def on_mqtt_connection(state): print(f"MQTT connected: {state}") async def on_api_update(api_data=None, name=None, device=None): if name: print(f"API refreshed for {name}") async def with_callbacks(): cloud = WorxCloud("user@example.com", "password", "worx") cloud.set_callback(LandroidEvent.DATA_RECEIVED, on_data_received) cloud.set_callback(LandroidEvent.MQTT_CONNECTION, on_mqtt_connection) cloud.set_callback(LandroidEvent.API, on_api_update) await cloud.authenticate() await cloud.connect() try: await asyncio.sleep(3600) finally: await cloud.disconnect() asyncio.run(with_callbacks()) ``` -------------------------------- ### POST /schedule/full Source: https://context7.com/mtrab/pyworxcloud/llms.txt Replaces the entire mowing schedule with a new ScheduleModel configuration. ```APIDOC ## POST /schedule/full ### Description Replaces the entire schedule with a new schedule model. This is useful for bulk schedule operations. ### Method POST ### Endpoint /schedule/full ### Parameters #### Request Body - **serial** (string) - Required - The device serial number. - **schedule** (ScheduleModel) - Required - The full schedule object containing entries and configuration. ### Request Example cloud.set_schedule(serial, schedule) ### Response #### Success Response (200) - **status** (string) - Confirmation of schedule update. ``` -------------------------------- ### Pyworxcloud: Print Latest Device State (Option 1 - Context Manager) Source: https://github.com/mtrab/pyworxcloud/wiki/Examples This method uses a context manager (`with` statement) to automatically handle connection and disconnection. It iterates through devices, updates their state, and prints their attributes using `pprint`. ```python from pyworxcloud import WorxCloud from pprint import pprint with WorxCloud("your@email","password","worx") as cloud: for _, device in cloud.devices.items(): cloud.update(device.serial_number) pprint(vars(device)) ``` -------------------------------- ### Accessing Raw Device Data in Python Source: https://github.com/mtrab/pyworxcloud/blob/master/DATA_MAPPING.md This snippet demonstrates how to access raw configuration and data payloads from a device using `device.raw_cfg` and `device.raw_dat`. This is useful for inspecting all numeric values before or after transformation. ```python raw_config = device.raw_cfg raw_data = device.raw_dat ``` -------------------------------- ### Configuring Command Timeout for WorxCloud Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Demonstrates how to configure a command timeout for MQTT command calls when initializing the WorxCloud class. This setting specifies how long the library will wait for a mower response before raising a TimeoutException. ```python from pyworxcloud import WorxCloud cloud = WorxCloud("user@example.com", "secret", "worx", command_timeout=15.0) ``` -------------------------------- ### Manage WorxCloud Schedules with Python Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md This Python snippet demonstrates how to interact with the WorxCloud schedule API. It shows how to retrieve an existing schedule and add a new schedule entry using the `pyworxcloud` library. The `ScheduleEntry` utility helps in formatting schedule data. ```python 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, ), ) ``` -------------------------------- ### Manage Mower Zones with PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt This code snippet demonstrates how to view and set the next mowing zone for a Worx Landroid mower. It retrieves current zone information and attempts to set a new zone, handling potential errors like 'ZoneNotDefined' or 'ZoneNoProbability'. Requires the pyworxcloud library. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import ZoneNotDefined, ZoneNoProbability async def manage_zones(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) # View current zone configuration print(f"Current zone index: {device.zone.index}") print(f"Zone starting points: {device.zone.starting_point}") print(f"Zone indices: {device.zone.indicies}") try: # Set next zone to zone 2 (0-indexed) await cloud.setzone(serial, zone=2) print("Next zone set to zone 2") except ZoneNotDefined: print("Zone 2 is not configured") except ZoneNoProbability: print("Zone 2 has no probability set") asyncio.run(manage_zones()) ``` -------------------------------- ### Processing MQTT Payloads with Python Source: https://github.com/mtrab/pyworxcloud/blob/master/DATA_MAPPING.md This section describes the processing of MQTT fixtures (`mqtt.json`) which are now sequential JSON payloads from the mower. The `dump_mapping.py` script and device decoders iterate through each document to align decoded slots and status with the live MQTT stream. ```python # Example usage of dump_mapping.py and device decoders # Assuming 'mqtt.json' contains sequential JSON payloads # and 'device_decoders' are available. # Placeholder for the actual script logic # dump_mapping.py and device decoders process each document # to ensure alignment with live MQTT stream. ``` -------------------------------- ### Pyworxcloud: Print Latest Device State (Option 2 - Manual Connection) Source: https://github.com/mtrab/pyworxcloud/wiki/Examples This option manually handles authentication, connection, and disconnection. It retrieves the latest states for each device by calling `cloud.update()` and then prints the device's attributes using `pprint`. ```python from pyworxcloud import WorxCloud from pprint import pprint cloud = WorxCloud("your@email", "password", "worx") # Initialize connection try: cloud.authenticate() except AuthorizationError: # If invalid credentials are used, or something happend during # authorize, then exit exit(0) # Connect to device with index 0 (devices are enumerated 0, 1, 2 ...) # and do not verify SSL (False) cloud.connect() # Read latest states received from the device for _, device in cloud.devices.items(): cloud.update(device.serial_number) # Print all vars and attributes of the cloud object pprint(vars(device)) # Disconnect from the API cloud.disconnect() ``` -------------------------------- ### POST /schedule/toggle Source: https://context7.com/mtrab/pyworxcloud/llms.txt Enables or disables the entire mowing schedule. ```APIDOC ## POST /schedule/toggle ### Description Enable or disable the entire mowing schedule without modifying individual entries. ### Method POST ### Endpoint /schedule/toggle ### Parameters #### Request Body - **serial** (string) - Required - The device serial number. - **enable** (boolean) - Required - True to enable, False to disable. ### Request Example cloud.toggle_schedule(serial, enable=True) ### Response #### Success Response (200) - **status** (string) - Confirmation of toggle state. ``` -------------------------------- ### Set Full Schedule in PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Replaces the entire mower schedule with a new ScheduleModel. This is ideal for bulk updates and requires defining entries with specific days and durations. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils.schedule_codec import ScheduleEntry, ScheduleModel async def set_full_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Get device protocol mower = cloud.get_mower(serial) protocol = mower["protocol"] # Create complete schedule model schedule = ScheduleModel( enabled=True, time_extension=0 if protocol == 0 else None, protocol=protocol, entries=[ ScheduleEntry( entry_id="", day="monday", start="08:00", duration=120, boundary=True, source="primary" if protocol == 0 else "slot", secondary=False ), ScheduleEntry( entry_id="", day="wednesday", start="08:00", duration=120, boundary=False, source="primary" if protocol == 0 else "slot", secondary=False ), ScheduleEntry( entry_id="", day="friday", start="08:00", duration=120, boundary=True, source="primary" if protocol == 0 else "slot", secondary=False ), ] ) await cloud.set_schedule(serial, schedule) print("Full schedule set for Mon/Wed/Fri") asyncio.run(set_full_schedule()) ``` -------------------------------- ### Set Time Extension in PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Adjusts the mowing time extension percentage for protocol 0 devices. Values must be provided in steps of 10, where positive values increase duration and negative values decrease it. ```python import asyncio from pyworxcloud import WorxCloud async def set_time_extension(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Set time extension to +20% (must be in steps of 10) await cloud.set_time_extension(serial, time_extension=20) print("Time extension set to +20%") # Set time extension to -30% await cloud.set_time_extension(serial, time_extension=-30) print("Time extension set to -30%") asyncio.run(set_time_extension()) ``` -------------------------------- ### Add Schedule Entry with PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt This Python code adds a new schedule entry to a Worx Landroid mower. The PyWorxCloud library automatically handles the protocol-specific serialization. It requires the pyworxcloud and pyworxcloud.utils.schedule_codec libraries. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils.schedule_codec import ScheduleEntry async def add_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Create a new schedule entry new_entry = ScheduleEntry( entry_id="", # Auto-generated day="monday", start="09:00", duration=120, # 2 hours boundary=True, # Start with edge cut source="primary", # 'primary', 'secondary', or 'slot' secondary=False ) await cloud.add_schedule_entry(serial, new_entry) print("Schedule entry added for Monday at 09:00") asyncio.run(add_schedule()) ``` -------------------------------- ### Access Comprehensive Device Properties Source: https://context7.com/mtrab/pyworxcloud/llms.txt Iterates through connected devices to retrieve and print detailed status information, including battery metrics, blade runtime, statistics, and hardware capabilities. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability async def device_properties(): async with WorxCloud("user@example.com", "password", "worx") as cloud: for name, device in cloud.devices.items(): print(f"=== {device.name} ===") print(f"Model: {device.model}") print(f"Battery: {device.battery.percent}%") for cap in DeviceCapability: if device.capabilities.check(cap): print(f" - {cap.name}") asyncio.run(device_properties()) ``` -------------------------------- ### Validating Sample Data with Python Scripts Source: https://github.com/mtrab/pyworxcloud/blob/master/README.md Details on how to use Python scripts to validate the integrity and structure of data samples used in testing. This ensures that fixtures contain the necessary payload components and that decoded data matches expectations. ```python # Run to ensure fixtures have minimal payload structure: python scripts/verify_data_samples.py # Run to inspect decoded snapshot for each fixture: python scripts/dump_mapping.py ``` -------------------------------- ### Updating Schedule Progress in Python Source: https://github.com/mtrab/pyworxcloud/blob/master/DATA_MAPPING.md This function, `schedules.update_progress_and_next()`, is used to synchronize `daily_progress` and `next_schedule_start` with the current timezone. `daily_progress` is set to `None` for days without schedule slots. ```python schedules.update_progress_and_next() ``` -------------------------------- ### POST /device/time-extension Source: https://context7.com/mtrab/pyworxcloud/llms.txt Adjusts the schedule time extension percentage for protocol 0 devices. ```APIDOC ## POST /device/time-extension ### Description Adjust the schedule time extension percentage (protocol 0 devices only). Positive values extend mowing time, negative values reduce it. ### Method POST ### Endpoint /device/time-extension ### Parameters #### Request Body - **serial** (string) - Required - The device serial number. - **time_extension** (integer) - Required - Percentage value in steps of 10. ### Request Example cloud.set_time_extension(serial, time_extension=20) ### Response #### Success Response (200) - **status** (string) - Confirmation of time extension update. ``` -------------------------------- ### Toggle Schedule State in PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Enables or disables the entire mowing schedule globally without needing to modify individual schedule entries. ```python import asyncio from pyworxcloud import WorxCloud async def toggle_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Disable schedule await cloud.toggle_schedule(serial, enable=False) print("Schedule disabled") # Enable schedule await cloud.toggle_schedule(serial, enable=True) print("Schedule enabled") asyncio.run(toggle_schedule()) ``` -------------------------------- ### Configure Rain Delay for Mower Source: https://context7.com/mtrab/pyworxcloud/llms.txt Sets the rain delay for a Worx Landroid mower, pausing operations after rain detection. The delay is specified in minutes. It also retrieves and prints the current rain sensor status, including configured delay, trigger status, and remaining delay time. ```python import asyncio from pyworxcloud import WorxCloud async def set_rain_delay(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Set rain delay to 180 minutes (3 hours) await cloud.raindelay(serial, "180") # Check current rain sensor status device = cloud.devices[list(cloud.devices.keys())[0]] print(f"Rain delay configured: {device.rainsensor.delay} minutes") print(f"Rain sensor triggered: {device.rainsensor.triggered}") print(f"Remaining delay: {device.rainsensor.remaining} minutes") asyncio.run(set_rain_delay()) ``` -------------------------------- ### Request Manual State Refresh Source: https://context7.com/mtrab/pyworxcloud/llms.txt Shows how to manually trigger a state update for a specific mower using its serial number. This is typically unnecessary as devices push updates automatically. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import NoConnectionError async def request_update(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" try: await cloud.update(serial) print("State refresh requested") except NoConnectionError: print("Cannot reach device") asyncio.run(request_update()) ``` -------------------------------- ### Implement Exception Handling for WorxCloud Operations Source: https://context7.com/mtrab/pyworxcloud/llms.txt This snippet demonstrates how to wrap pyWorxCloud operations in try-except blocks to catch specific library exceptions like AuthorizationError, OfflineError, and TimeoutException. It ensures that resources are properly cleaned up using a finally block. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import ( AuthorizationError, TooManyRequestsError, OfflineError, NoConnectionError, TimeoutException, MowerNotFoundError, ) async def with_error_handling(): try: cloud = WorxCloud("user@example.com", "password", "worx") await cloud.authenticate() await cloud.connect() serial = "WX123456789" try: await cloud.start(serial) except OfflineError: print("Mower is offline") except TimeoutException: print("Command timed out - no response from mower") except MowerNotFoundError: print("Mower not found with that serial number") except AuthorizationError: print("Invalid username or password") except TooManyRequestsError: print("Rate limited - wait before retrying") except NoConnectionError: print("Cannot connect to cloud service") finally: if 'cloud' in locals(): await cloud.disconnect() asyncio.run(with_error_handling()) ``` -------------------------------- ### Manage Off-Limits Module Source: https://context7.com/mtrab/pyworxcloud/llms.txt Enables or disables the off-limits module and its shortcut feature. It verifies device capability before attempting to modify the state. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability from pyworxcloud.exceptions import NoOfflimitsError async def manage_offlimits(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) if device.capabilities.check(DeviceCapability.OFF_LIMITS): try: await cloud.set_offlimits(serial, state=True) await cloud.set_offlimits_shortcut(serial, state=True) except NoOfflimitsError: print("Off-limits module not available") asyncio.run(manage_offlimits()) ``` -------------------------------- ### Restart Device Source: https://context7.com/mtrab/pyworxcloud/llms.txt Triggers a remote reboot of the mower's baseboard. ```python import asyncio from pyworxcloud import WorxCloud async def restart_device(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" await cloud.restart(serial) asyncio.run(restart_device()) ``` -------------------------------- ### Send Raw JSON Command Source: https://context7.com/mtrab/pyworxcloud/llms.txt Sends custom JSON payloads directly to the mower for advanced or unsupported functionality. ```python import asyncio import json from pyworxcloud import WorxCloud async def send_raw(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" custom_command = json.dumps({"cmd": 3}) await cloud.send(serial, custom_command) asyncio.run(send_raw()) ``` -------------------------------- ### Set Wheel Torque in PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Configures the wheel torque for the mower to improve performance on difficult terrain. This operation checks for device capability before applying the setting within the range of -50 to +50. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability async def set_torque(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) if device.capabilities.check(DeviceCapability.TORQUE): # Set torque to +25% (range: -50 to +50) await cloud.set_torque(serial, torque=25) print("Wheel torque set to +25%") else: print("Device does not support torque adjustment") asyncio.run(set_torque()) ``` -------------------------------- ### DELETE /schedule/entry Source: https://context7.com/mtrab/pyworxcloud/llms.txt Deletes a specific schedule entry by its ID. For protocol 0 devices, removing a primary entry promotes the secondary entry. ```APIDOC ## DELETE /schedule/entry ### Description Deletes a schedule entry by its unique entry ID. Note that for protocol 0 devices, deleting a primary entry will promote the secondary entry. ### Method DELETE ### Endpoint /schedule/entry ### Parameters #### Path Parameters - **serial** (string) - Required - The device serial number. - **entry_id** (string) - Required - The unique identifier for the schedule entry. ### Request Example cloud.delete_schedule_entry(serial, entry_id) ### Response #### Success Response (200) - **status** (string) - Confirmation of deletion. ``` -------------------------------- ### Delete Schedule Entry in PyWorxCloud Source: https://context7.com/mtrab/pyworxcloud/llms.txt Deletes a specific schedule entry by its ID. For protocol 0 devices, removing a primary entry automatically promotes the secondary entry. ```python import asyncio from pyworxcloud import WorxCloud async def delete_schedule(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" # Get current schedule schedule = cloud.get_schedule(serial) if schedule.entries: entry_id = schedule.entries[0].entry_id await cloud.delete_schedule_entry(serial, entry_id) print(f"Deleted schedule entry {entry_id}") asyncio.run(delete_schedule()) ``` -------------------------------- ### POST /device/torque Source: https://context7.com/mtrab/pyworxcloud/llms.txt Adjusts the wheel torque for better performance on slopes or difficult terrain. ```APIDOC ## POST /device/torque ### Description Adjust the wheel torque for better performance on slopes or difficult terrain. Supported only on devices with torque capability. ### Method POST ### Endpoint /device/torque ### Parameters #### Request Body - **serial** (string) - Required - The device serial number. - **torque** (integer) - Required - Torque value (range: -50 to +50). ### Request Example cloud.set_torque(serial, torque=25) ### Response #### Success Response (200) - **status** (string) - Confirmation of torque update. ``` -------------------------------- ### Control ACS Module Source: https://context7.com/mtrab/pyworxcloud/llms.txt Toggles the Anti-Collision System (ACS) ultrasonic sensors on supported mower models. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability from pyworxcloud.exceptions import NoACSModuleError async def manage_acs(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) if device.capabilities.check(DeviceCapability.ACS): try: await cloud.set_acs(serial, state=True) await cloud.set_acs(serial, state=False) except NoACSModuleError: print("ACS module not installed") asyncio.run(manage_acs()) ``` -------------------------------- ### Adjust Cutting Height Source: https://context7.com/mtrab/pyworxcloud/llms.txt Retrieves and updates the cutting height for Vision models equipped with the EA module. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability from pyworxcloud.exceptions import NoCuttingHeightError async def set_cutting_height(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) if device.capabilities.check(DeviceCapability.CUTTING_HEIGHT): try: current_height = cloud.get_cutting_height(serial) await cloud.set_cutting_height(serial, height=50) except NoCuttingHeightError: print("Cutting height not supported") asyncio.run(set_cutting_height()) ``` -------------------------------- ### Reset Maintenance Counters Source: https://context7.com/mtrab/pyworxcloud/llms.txt Resets the battery charge cycle counter and blade usage counter after performing maintenance. ```python import asyncio from pyworxcloud import WorxCloud async def reset_counters(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" await cloud.reset_charge_cycle_counter(serial) await cloud.reset_blade_counter(serial) asyncio.run(reset_counters()) ``` -------------------------------- ### Control Mower Pause Mode Source: https://context7.com/mtrab/pyworxcloud/llms.txt Enables or disables pause mode (formerly Party Mode) for a Worx Landroid mower. Pause mode temporarily suspends the mowing schedule without altering its configuration. The code includes error handling for devices that do not support this feature. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.exceptions import NoPauseModeError async def manage_pause_mode(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" try: # Enable pause mode (schedule disabled) await cloud.set_pause_mode(serial, state=True) print("Pause mode enabled - schedule suspended") # Disable pause mode (schedule resumed) await cloud.set_pause_mode(serial, state=False) print("Pause mode disabled - schedule active") except NoPauseModeError: print("This device does not support pause mode") asyncio.run(manage_pause_mode()) ``` -------------------------------- ### Trigger Edge Cutting Operation Source: https://context7.com/mtrab/pyworxcloud/llms.txt Triggers an edge cutting operation for a Worx Landroid mower, where the mower follows the boundary wire to cut grass along the edges. The code first checks if the device supports the edge cut capability before initiating the operation. ```python import asyncio from pyworxcloud import WorxCloud from pyworxcloud.utils import DeviceCapability async def edge_cut(): async with WorxCloud("user@example.com", "password", "worx") as cloud: serial = "WX123456789" device = cloud.get_mower(serial, device=True) # Check if device supports edge cut if device.capabilities.check(DeviceCapability.EDGE_CUT): await cloud.edgecut(serial) print("Edge cut started") else: print("Device does not support edge cut") asyncio.run(edge_cut()) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.