test_async.py

import asyncio
import datetime
import json
import time
from os import environ

from pyworxcloud import WorxCloud
from pyworxcloud.events import LandroidEvent
from pyworxcloud.utils import DeviceHandler

EMAIL = environ["EMAIL"]
PASS = environ["PASSWORD"]
TYPE = environ["TYPE"]

tz = datetime.datetime.now().astimezone().tzinfo.tzname(None)


async def main():
    loop = asyncio.get_running_loop()
    await async_worx()


async def async_worx():
    # Clear the screen for better visibility when debugging

    # Initialize the class and connect
    cloud = WorxCloud(EMAIL, PASS, TYPE, tz="Europe/Copenhagen")
    cloud.authenticate()
    cloud.connect()
    cloud.set_callback(LandroidEvent.DATA_RECEIVED, receive_data)
    cloud.set_callback(LandroidEvent.API, receive_api_data)

    print("Listening for new data")
    while 1:
        pass

    # Self explanatory - disconnect from the cloud
    cloud.disconnect()


def receive_data(
    name: str, device: DeviceHandler  # pylint: disable=unused-argument
) -> None:
    """Callback function when the MQTT broker sends new data."""
    print("Got data on MQTT from " + name)
    print(name + " last status: " + str(device.last_status))


def receive_api_data(
    name: str, device: DeviceHandler  # pylint: disable=unused-argument
) -> None:
    """Callback function when the API data was updated."""
    print("API data was refreshed")


asyncio.run(main())