zweckj / lmcloud Goto Github PK

Library to interface with La Marzocco's cloud

License: MIT License

Python 100.00%

lmcloud's Introduction

La Marzocco Client

This is a library to interface with La Marzocco's Home machines. It also has support to get information for the Pico grinder.

Libraries in this project

LaMarzoccoLocalClient calls the new local API the Micra exposes, using the Bearer token from the customer cloud endpoint. However, this API currently only supports getting the config, and some status objects (like shottimer) over websockets, but does not support setting anything (to my knowledge). Local settings appear to only happen through Bluetooth connections.
LaMarzoccoCloudClient interacts with gw-lmz.lamarzocco.com to send commands. lmcloud can be initialized to only issue remote commands, or to initialize an instance of lmlocalapi for getting the current machine settings. This helps to avoid flooding the cloud API and is faster overall.
LaMarzoccoBluetoothClient provides a bluetooth client to send settings to the machine via bluetooth

Setup

LaMarzoccoCloudClient

You need username and password which are the credentials you're using to sign into the La Marzocco Home app.

It is initialized like this

cloud_client = await LaMarzoccoCloudClient(username, password)

LaMarzoccoLocalClient

If you just want to run the local API you need the IP of your machine, the Port it is listening on (8081 by default), the Bearer token (communicationKey) used for local communication. You can obtain that key by inspecting a call to https://cms.lamarzocco.io/api/customer, while connected to mitmproxy (process above), or making a new (authenticated) call to that endpoint.

Then you can init the class with

local_client = LaMarzoccoLocalClient(ip, local_token)

LaMarzoccoBluetoothClient

Some commands, like turning the machine on and off are always sent through bluetooth whenever possible. The available bluetooth characteristics are described in bluetooth_characteristics. The class LaMarzoccoBluetoothClient discovers any bluetooth devices connects to it. Then we can send local bluetooth commands.

To use Bluetooth you can either init LMCloud with

    if bluetooth_devices := LaMarzoccoBluetoothClient.discover_devices():
        print("Found bluetooth device:", bluetooth_devices[0])

    bluetooth_client = LaMarzoccoBluetoothClient(
        username,
        serial_number,
        local_token
        bluetooth_devices[0],
    )

The local_token is the same token you need to initialize the local API, which you need to get from LM's cloud once. The serial number is your machine's serial number and the username is the email of your LaMarzocco account.

Machine

Once you have any or all of the clients, you can initialize a machine object with

machine = Machine.create(model, serial_number, name, cloud_client, local_client, bluetooth_client)

You can then use the machine object to send commands to the machine, or to get the current status of the machine. If you're running in cloud only mode, please be mindful with the requests to not flood the cloud API.

Grinder

The Pico grinder can be initialized with

grinder = LaMarzoccoGrinder.create(model, serial_number, name, cloud_client, local_client, bluetooth_client)

where you can use the same cloud client as for the machine, but you need to initialize new local and bluetooth clients (the same way as for the machine) to use the grinder.

Websockets

The local API initiates a websocket connection to

http://{IP}:8081/api/v1/streaming

The packets which are received on that WebSocket are documented in websockets

If WebSockets are enabled the shot timer becomes available to use, however as long as the library is running in WebSocket mode, the App will no longer be able to connect.

To use WebSockets start the integration with

await machine.websocket_connect(callback)

with an optional callback function that will be called whenever there have been updates for the machine from the websocket.

lmcloud's People

Contributors

Stargazers

Watchers

Forkers

ctbear surengrig rccoleman yaarha

lmcloud's Issues

Check in true_model_name causes errors and breaks HA integration for non-Micra models

In this function:

    @property
    def true_model_name(self) -> str:
        """Return the model name from the cloud, even if it's not one we know about.
        Used for display only."""
        if self.model_name == LaMarzoccoModel.LINEA_MICRA:
            return "Linea Micra"
        if self.model_name in LaMarzoccoModel:
            return self.model_name
        return f"Unsupported Model ({self.model_name})"

This part causes an exception because self.model_name is a string and LaMarzoccoModel is an enum.
if self.model_name in LaMarzoccoModel:

It results in exceptions like the following with a non-Micra model:

2023-12-29 13:03:08.499 ERROR (MainThread) [homeassistant] Error doing job: Task exception was never retrieved
Traceback (most recent call last):
  File "/usr/src/homeassistant/homeassistant/helpers/entity_platform.py", line 495, in async_add_entities
    tasks = [
            ^
  File "/usr/src/homeassistant/homeassistant/helpers/entity_platform.py", line 495, in <listcomp>
    tasks = [
            ^
  File "/config/custom_components/lamarzocco/binary_sensor.py", line 61, in <genexpr>
    LaMarzoccoBinarySensorEntity(coordinator, hass, description)
  File "/config/custom_components/lamarzocco/entity.py", line 53, in __init__
    model=self._lm_client.true_model_name,
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/lmcloud/lmcloud.py", line 123, in true_model_name
    if self.model_name in LaMarzoccoModel:
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/enum.py", line 740, in __contains__
    raise TypeError(
TypeError: unsupported operand type(s) for 'in': 'str' and 'EnumType'
2023-12-29 13:03:08.505 ERROR (MainThread) [homeassistant] Error doing job: Task exception was never retrieved
Traceback (most recent call last):
  File "/usr/src/homeassistant/homeassistant/helpers/entity_platform.py", line 495, in async_add_entities
    tasks = [
            ^
  File "/usr/src/homeassistant/homeassistant/helpers/entity_platform.py", line 495, in <listcomp>
    tasks = [
            ^
  File "/config/custom_components/lamarzocco/button.py", line 44, in <genexpr>
    LaMarzoccoButtonEntity(coordinator, hass, description)
  File "/config/custom_components/lamarzocco/entity.py", line 53, in __init__
    model=self._lm_client.true_model_name,
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/lmcloud/lmcloud.py", line 123, in true_model_name
    if self.model_name in LaMarzoccoModel:
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/enum.py", line 740, in __contains__
    raise TypeError(
TypeError: unsupported operand type(s) for 'in': 'str' and 'EnumType'

These exceptions completely break the HA integration with v0.13.2b2 of the integration and HA 2024.1.0b2. Looks like it came in with a7d4c69.

One of these tests would fix it:

if self.model_name in LaMarzoccoModel._value2member_map_:
  return self.model_name

try:
  return LaMarzoccoModel(self.model_name)
except:
  return f"Unsupported Model ({self.model_name})"

I'm using this and it's working for me:

    @property                            
    def true_model_name(self) -> str:     
        """Return the model name from the cloud, even if it's not one we know about.
        Used for display only."""                                         
        if self.model_name == LaMarzoccoModel.LINEA_MICRA:                    
            return "Linea Micra"                      
        try:                                       
            return LaMarzoccoModel(self.model_name)         
        except:                                         
            return f"Unsupported Model ({self.model_name})"

New /api/v1/logs endpoint in GW 3.1-rc6

The new version 3.1-rc6 of the gw fw seems to support a new endpoint for logs that is missing in the former v2.2-rc0.

strings home-gateway_v3.1-rc6.bin | grep /v1/           
/api/v1/config
/api/v1/command/*
/api/v1/streaming
/api/v1/logs

Sending a GET request to that url causes the connection to be closed (instead of a 404 for an invalid uri).
I am not sure if that endpoint is of any use but maybe we can find out :-)

/api/v1/command/* also looks kind of promising.

Setting prebew/preinfusion times leads to type conflict

This function has the on/off times passed in as float, but expects an int, resulting in TypeError being raised:

    async def configure_prebrew(
        self, on_time=5000, off_time=5000, key: int = 1
    ) -> bool:
        """Set Pre-Brew details. Also used for preinfusion (prebrewOnTime=0, prebrewOnTime=ms)."""

        if not isinstance(on_time, int) or not isinstance(off_time, int):
            msg = "Prebrew times must be in ms (integer)"
            _logger.debug(msg)
            raise TypeError(msg)

From the HA integration:

    async def set_prebrew_times(
        self, key: int, seconds_on: float, seconds_off: float
    ) -> None:
        """Set the prebrew times of the machine."""
        await self.configure_prebrew(
            on_time=seconds_on * 1000, off_time=seconds_off * 1000, key=key
        )

Failed to call service lamarzocco/set_preinfusion_time. Service call encountered error: Prebrew times must be in ms (integer)

Is it possible to get a ping when shot extraction is started and ended from the Micra?

Thanks for your work on this! Is it possible to get a ping when shot extraction is started and ended from the Micra with this API?
This would be great to integrate it with a scale like the Acaia

[Question] Brew start/stop sent over only localapi?

Hello,

I have recently purchased a micra (still shipping) and have some resistance on having to keep an app on standby to access settings on the machine. I have been looking into building some basic controls via an ESP32 setup over bluetooth/micropython and found all this great work via the HA forums :).

I would like to build set/get temp controls which I can follow from their source and yours, however, I would also like to build an automatic shot timer. I see the localapi websocket seems to expose this notification: Would you know if this is also sent over bluetooth? I would like to keep the machine completely off of wifi if possible (or at least their cloud) and not sure if I can also get a local connection without also phoning LM directy.