Websocket example?

[emcniece]

That websocket class just sitting there looks so tempting 😁 I'm having trouble getting it working though. Any suggestions for a demo?

Here's what I've tried so far, inside the sample.py try block:

await account.connect(username=username, password=password, load_robots=True)

print(account.robots[0])
# Name: Litter-Robot 4, Model: Litter-Robot 4, Serial: xxxxx, id: yyyyy

# This instantiates properly and I can inspect methods with ipdb, though it doesn't seem to have all of the class methods from ws_monitor.py
socket = await account.ws_connect(account.robots[0])

# I don't see any other socket-related methods bound to `account`, so let's test the socket object...

# Prints 'False', so I presume it's open!
print(socket.close)

# Error, 'object has no attribute' etc.
print(socket.connected())
await socket.monitor()
await socket.start_monitor()
await socket.new_connection(start_monitor=True)

[natekspencer]

In account.connect, pass subscribe_for_updates=True and it will open the web sockets and monitor those connections for you.

[natekspencer]

Or on the robot, call subscribe() like await account.robots[0].subscribe()

[emcniece]

Good tips, thank you for the help. Feels like I'm missing the asyncio basics, sorry for the hassle. How do we go about getting event data out of this subscription?

        # Connect to the API and load robots.
        await account.connect(username=username, password=password, load_robots=True, subscribe_for_updates=True)

        # Print robots associated with account.
        print(account.robots[0])
        await account.robots[0].subscribe()

        # How do we wait and log messages from the robot?        
        await asyncio.Future()

In this example the subscription succeeds, then Python waits forever processing an indeterminate Future(). I was hoping that by triggering events for the robot through the app that I would see the _log_message method print something out. Nothing happens of course!

Does logging need to be enabled, or are robot events only logged if they're initiated in the Python session and not the app?

[natekspencer]

Here's a snippet of my test file I use to run when I'm developing and trying things out

logging.basicConfig(level=logging.DEBUG)

async def main():
    # Create an account.
    account = Account()
    # Get the token
    try:
        await account.connect(
            username=username,
            password=password,
            load_robots=True,
            subscribe_for_updates=True,
        )
    except (LitterRobotLoginException, LitterRobotException) as ex:
        logging.error(ex)
        await account.disconnect()
        exit()

    await print_robots_info(account.robots)

    b = True
    while b:
        await asyncio.sleep(5)
    await account.disconnect()

async def print_robots_info(robots: list[Robot]) -> None:
    if len(robots) > 0:
        logging.debug(
            "================================== Robots =================================="
        )
        for robot in robots:
            robot.on(
                EVENT_UPDATE,
                lambda robot=robot, *args, **kwargs: logging.debug(
                    "received an update from %s/%s", robot.name, robot.serial
                ),
            )
            logging.debug(
                "============================================================================"
            )
            logging.debug(robot)
            logging.debug(
                "  firmware: %s, setup date: %s, is online: %s, power status: %s, night light mode enabled: %s, panel lock enabled: %s",
                getattr(robot, "firmware", None),
                robot.setup_date,
                robot.is_online,
                robot.power_status,
                robot.night_light_mode_enabled,
                robot.panel_lock_enabled,
            )

            if isinstance(robot, LitterRobot):
                logging.debug(
                    "  cycles after drawer full: %s, status: %s, last seen: %s, cycle capacity: %s, drawer full: %s, waste level: %s",
                    robot.cycles_after_drawer_full,
                    robot.status,
                    robot.last_seen,
                    robot.cycle_capacity,
                    robot.is_waste_drawer_full,
                    robot.waste_drawer_level,
                )
                activities = await robot.get_activity_history()
                for activity in activities:
                    logging.debug(activity)
                insight = await robot.get_insight()
                logging.debug(insight)
            if isinstance(robot, LitterRobot4):
                logging.debug("  litter level: %s", robot.litter_level)
                logging.debug(
                    "  night light brightness: %s, level: %s, mode: %s",
                    robot.night_light_brightness,
                    robot.night_light_level,
                    robot.night_light_mode,
                )
                logging.debug(
                    "  night light brightness: %s", robot.night_light_brightness
                )
                logging.debug("  panel brightness: %s", robot.panel_brightness)
                if await robot.has_firmware_update():
                    logging.debug(
                        "  latest firmware: %s",
                        await robot.get_latest_firmware(),
                    )

            if isinstance(robot, FeederRobot):
                logging.debug(
                    "  food level: %s, meal insert size: %s",
                    robot.food_level,
                    robot.meal_insert_size,
                )

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

[natekspencer]

Does logging need to be enabled, or are robot events only logged if they're initiated in the Python session and not the app?

With await account.connect(username=username, password=password, load_robots=True, subscribe_for_updates=True), you don't need to call await account.robots[0].subscribe() (it does it for you already).

What are you trying to do and maybe I can help direct you to a solution. Enabling logging will just tell you that something has happened. But if you want to react to a change, you have to subscribe to an event change such as:

robot.on(
  EVENT_UPDATE,
  lambda robot=robot, *args, **kwargs: logging.debug("received an update from %s/%s", robot.name, robot.serial))

[emcniece]

Short term: I'd love to document a socket example for the readme, or a wiki if you'd prefer!

Long term: I'd like to package this in Docker and run in my homelab, maybe log and collect metrics. My intent is to send a reset after the "drawer full" indicator triggers, then after a number of times (3 perhaps) notify me, because that drawer is not actually full and the sensor is a bit eager.

I think I can do that with the examples you've provided above, many thanks 😁

[natekspencer]

Short term: I'd love to document a socket example for the readme, or a wiki if you'd prefer!

I kind of built the web sockets as something another developer (ala myself for Home Assistant) would not need to think about. Primarily via the account connection, specifically the subscribe_for_updates=True I referred to earlier. Essentially as a, would you rather refresh on your own, or get updates automatically, depending on the usage scenario.

[emcniece]

Works like a charm, thanks so much for taking the time to share your examples 🎉

Would you be interested in a PR for a containerized implementation?