Merge pull request #136 from zweckj/copilot/add-cloud-commands-linea-mini-r

Add brew by weight cloud commands for Linea Mini R

Author: zweckj

pylamarzocco/clients/_cloud.py

@@ -26,6 +26,7 @@
     BASE_URL,
     CUSTOMER_APP_URL,
     CommandStatus,
+    DoseMode,
     PreExtractionMode,
     SmartStandByType,
     SteamTargetLevel,
@@ -571,7 +572,7 @@ async def set_coffee_target_temperature(
         return await self.__execute_command(
             serial_number, "CoffeeMachineSettingCoffeeBoilerTargetTemperature", data
         )
-    
+
     async def set_steam_target_temperature(
         self, serial_number: str, target_temperature: float, boiler_index: int = 1
     ) -> bool:
@@ -649,6 +650,34 @@ async def set_wakeup_schedule(
             serial_number, "CoffeeMachineSetWakeUpSchedule", schedule.to_dict()
         )
 
+    async def change_brew_by_weight_dose_mode(
+        self,
+        serial_number: str,
+        mode: DoseMode,
+    ) -> bool:
+        """Change the brew by weight dose mode (Linea Mini R only)."""
+        data = {"mode": mode.value}
+        return await self.__execute_command(
+            serial_number, "CoffeeMachineBrewByWeightChangeMode", data
+        )
+
+    async def set_brew_by_weight_dose(
+        self,
+        serial_number: str,
+        dose_1: float,
+        dose_2: float,
+    ) -> bool:
+        """Set the brew by weight doses (Linea Mini R only)."""
+        data = {
+            "doses": {
+                "Dose1": round(dose_1, 1),
+                "Dose2": round(dose_2, 1),
+            }
+        }
+        return await self.__execute_command(
+            serial_number, "CoffeeMachineBrewByWeightSettingDoses", data
+        )
+
     async def update_firmware(
         self,
         serial_number: str,

pylamarzocco/devices/_machine.py

@@ -11,6 +11,7 @@
 from pylamarzocco.const import (
     BoilerStatus,
     BoilerType,
+    DoseMode,
     MachineMode,
     MachineState,
     ModelCode,
@@ -21,6 +22,7 @@
 )
 from pylamarzocco.exceptions import BluetoothConnectionFailed
 from pylamarzocco.models import (
+    BrewByWeightDoses,
     CoffeeAndFlushCounter,
     CoffeeAndFlushTrend,
     CoffeeBoiler,
@@ -68,13 +70,13 @@ async def get_schedule(self) -> None:
 
     async def get_model_info_from_bluetooth(self) -> None:
         """Fetch and update model information from Bluetooth.
-        
+
         Retrieves machine capabilities via Bluetooth and updates the dashboard
         with model_name and model_code information.
         """
         if self._bluetooth_client is None:
             raise BluetoothConnectionFailed("Bluetooth client not initialized")
-        
+
         try:
             capabilities = await self._bluetooth_client.get_machine_capabilities()
         except (BleakError, BluetoothConnectionFailed) as exc:
@@ -168,9 +170,13 @@ async def get_dashboard_from_bluetooth(self) -> None:
                         ),
                     )
                     steam_level.enabled = boiler.is_enabled
-                    self.dashboard.config[WidgetType.CM_STEAM_BOILER_LEVEL] = steam_level
+                    self.dashboard.config[WidgetType.CM_STEAM_BOILER_LEVEL] = (
+                        steam_level
+                    )
                     # Remove temperature widget if it exists (not applicable for this model)
-                    self.dashboard.config.pop(WidgetType.CM_STEAM_BOILER_TEMPERATURE, None)
+                    self.dashboard.config.pop(
+                        WidgetType.CM_STEAM_BOILER_TEMPERATURE, None
+                    )
                 else:
                     # Other models (GS3, original Mini) use steam temperature widget
                     steam_temp = cast(
@@ -309,7 +315,7 @@ async def set_coffee_target_temperature(self, temperature: float) -> bool:
             coffee_boiler.target_temperature = float(temperature)
 
         return result
-    
+
     @models_supported((ModelCode.GS3, ModelCode.GS3_AV, ModelCode.GS3_MP))
     async def set_steam_target_temperature(self, temperature: float) -> bool:
         """Set the steam target temperature."""
@@ -401,6 +407,78 @@ async def set_wakeup_schedule(
             self.serial_number, schedule
         )
 
+    @cloud_only
+    @models_supported((ModelCode.LINEA_MINI_R,))
+    async def set_brew_by_weight_dose_mode(self, mode: DoseMode) -> bool:
+        """Set the brew by weight dose mode (Linea Mini R only).
+
+        Args:
+            mode: The dose mode (DoseMode.DOSE_1, DoseMode.DOSE_2, or DoseMode.CONTINUOUS)
+        """
+        assert self._cloud_client
+        result = await self._cloud_client.change_brew_by_weight_dose_mode(
+            self.serial_number, mode
+        )
+
+        # Update dashboard if command succeeded
+        if result and WidgetType.CM_BREW_BY_WEIGHT_DOSES in self.dashboard.config:
+            brew_by_weight = cast(
+                BrewByWeightDoses,
+                self.dashboard.config[WidgetType.CM_BREW_BY_WEIGHT_DOSES],
+            )
+            brew_by_weight.mode = mode
+
+        return result
+
+    @cloud_only
+    @models_supported((ModelCode.LINEA_MINI_R,))
+    async def set_brew_by_weight_dose(self, dose: DoseMode, value: float) -> bool:
+        """Set a brew by weight dose value (Linea Mini R only).
+
+        Args:
+            dose: Which dose to set (must be DoseMode.DOSE_1 or DoseMode.DOSE_2,
+                  CONTINUOUS is not valid for setting dose values)
+            value: The dose value in grams
+
+        Returns:
+            True if the command was successful, False otherwise.
+            Returns False if dose is not DOSE_1 or DOSE_2 or if the brew by
+            weight widget is not available in the dashboard.
+        """
+        assert self._cloud_client
+
+        # Get current doses from dashboard
+        if WidgetType.CM_BREW_BY_WEIGHT_DOSES not in self.dashboard.config:
+            return False
+
+        brew_by_weight = cast(
+            BrewByWeightDoses,
+            self.dashboard.config[WidgetType.CM_BREW_BY_WEIGHT_DOSES],
+        )
+
+        # Set the dose values, keeping the other one unchanged
+        if dose == DoseMode.DOSE_1:
+            dose_1 = value
+            dose_2 = brew_by_weight.doses.dose_2.dose
+        elif dose == DoseMode.DOSE_2:
+            dose_1 = brew_by_weight.doses.dose_1.dose
+            dose_2 = value
+        else:
+            return False
+
+        result = await self._cloud_client.set_brew_by_weight_dose(
+            self.serial_number, dose_1, dose_2
+        )
+
+        # Update dashboard if command succeeded
+        if result:
+            if dose == DoseMode.DOSE_1:
+                brew_by_weight.doses.dose_1.dose = value
+            else:
+                brew_by_weight.doses.dose_2.dose = value
+
+        return result
+
     def to_dict(self) -> dict[Any, Any]:
         """Return self in dict represenation."""
         return {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pylamarzocco"
-version = "2.2.2"
+version = "2.2.3"
 license = { text = "MIT" }
 description = "A Python implementation of the La Marzocco API"
 readme = "README.md"

tests/test_cloud_client.py

@@ -16,6 +16,7 @@
 from pylamarzocco.const import (
     CUSTOMER_APP_URL,
     CommandStatus,
+    DoseMode,
     PreExtractionMode,
     SmartStandByType,
     SteamTargetLevel,
@@ -640,3 +641,48 @@ async def test_start_update(
     client = LaMarzoccoCloudClient("test", "test", MOCK_SECRET_DATA)
     result = await client.update_firmware(serial)
     assert result.to_dict() == snapshot
+
+
+async def test_change_brew_by_weight_dose_mode(
+    mock_aioresponse: aioresponses,
+    serial: str,
+) -> None:
+    """Test changing the brew by weight dose mode."""
+
+    url = f"{CUSTOMER_APP_URL}/things/{serial}/command/CoffeeMachineBrewByWeightChangeMode"
+    mock_aioresponse.post(
+        url=url,
+        status=200,
+        payload=MOCK_COMMAND_RESPONSE,
+    )
+
+    client = LaMarzoccoCloudClient("test", "test", MOCK_SECRET_DATA)
+    result = await client.change_brew_by_weight_dose_mode(serial, DoseMode.DOSE_1)
+    call = mock_aioresponse.requests[(HTTPMethod.POST, URL(url))][0]
+    assert call.kwargs["json"] == {"mode": "Dose1"}
+    assert result is True
+
+
+async def test_set_brew_by_weight_dose(
+    mock_aioresponse: aioresponses,
+    serial: str,
+) -> None:
+    """Test setting the brew by weight doses."""
+
+    url = f"{CUSTOMER_APP_URL}/things/{serial}/command/CoffeeMachineBrewByWeightSettingDoses"
+    mock_aioresponse.post(
+        url=url,
+        status=200,
+        payload=MOCK_COMMAND_RESPONSE,
+    )
+
+    client = LaMarzoccoCloudClient("test", "test", MOCK_SECRET_DATA)
+    result = await client.set_brew_by_weight_dose(serial, 32.56, 45.67)
+    call = mock_aioresponse.requests[(HTTPMethod.POST, URL(url))][0]
+    assert call.kwargs["json"] == {
+        "doses": {
+            "Dose1": 32.6,
+            "Dose2": 45.7,
+        }
+    }
+    assert result is True

tests/test_machine.py

@@ -9,9 +9,21 @@
     LaMarzoccoCloudClient,
     LaMarzoccoMachine,
 )
-from pylamarzocco.const import BoilerType, SteamTargetLevel, SmartStandByType, ModelCode
+from pylamarzocco.const import (
+    BoilerType,
+    DoseMode,
+    ModelCode,
+    SmartStandByType,
+    SteamTargetLevel,
+    WidgetType,
+)
 from pylamarzocco.exceptions import BluetoothConnectionFailed
-from pylamarzocco.models import BluetoothCommandStatus
+from pylamarzocco.models import (
+    BaseDoseSettings,
+    BluetoothCommandStatus,
+    BrewByWeightDoses,
+    BrewByWeightDoseSettings,
+)
 
 
 @pytest.fixture(name="mock_bluetooth_client")
@@ -210,3 +222,46 @@ async def test_failing_command(
     )
     mock_cloud_client.set_power.return_value = False
     assert not await mock_machine.set_power(True)
+
+
+async def test_set_brew_by_weight_dose_mode(
+    mock_machine: LaMarzoccoMachine,
+    mock_cloud_client: MagicMock,
+) -> None:
+    """Test the set_brew_by_weight_dose_mode method."""
+    mock_machine.dashboard.model_code = ModelCode.LINEA_MINI_R
+    mock_cloud_client.change_brew_by_weight_dose_mode.return_value = True
+
+    assert await mock_machine.set_brew_by_weight_dose_mode(DoseMode.DOSE_1)
+    mock_cloud_client.change_brew_by_weight_dose_mode.assert_called_once_with(
+        "MR123456", DoseMode.DOSE_1
+    )
+
+
+async def test_set_brew_by_weight_dose(
+    mock_machine: LaMarzoccoMachine,
+    mock_cloud_client: MagicMock,
+) -> None:
+    """Test the set_brew_by_weight_dose method."""
+    mock_machine.dashboard.model_code = ModelCode.LINEA_MINI_R
+    mock_cloud_client.set_brew_by_weight_dose.return_value = True
+
+    # Set up the dashboard with brew by weight widget
+    mock_machine.dashboard.config[WidgetType.CM_BREW_BY_WEIGHT_DOSES] = (
+        BrewByWeightDoses(
+            mode=DoseMode.DOSE_1,
+            doses=BrewByWeightDoseSettings(
+                dose_1=BaseDoseSettings(
+                    dose=32.0, dose_min=5, dose_max=100, dose_step=1
+                ),
+                dose_2=BaseDoseSettings(
+                    dose=34.0, dose_min=5, dose_max=100, dose_step=1
+                ),
+            ),
+        )
+    )
+
+    assert await mock_machine.set_brew_by_weight_dose(DoseMode.DOSE_1, 36.5)
+    mock_cloud_client.set_brew_by_weight_dose.assert_called_once_with(
+        "MR123456", 36.5, 34.0
+    )