Manage Shutter

myFox2Mqtt/business/__init__.py

@@ -70,6 +70,7 @@ def ha_devices_config(
         other_devices = api.get_devices_other(site_id=site_id)
         camera_devices = api.get_devices_camera(site_id=site_id)
         scenarios = api.get_scenarios(site_id=site_id)
+        shutter_devices = api.get_devices_shutter(site_id=site_id)
 
         for device in my_devices:
             LOGGER.info(f"Configuring Device: {device.label}")
@@ -211,6 +212,24 @@ def ha_devices_config(
                         retain=True,
                     )
 
+            # Shutter
+            for shutter_device in shutter_devices:
+                if shutter_device.get("deviceId") == device.device_id:
+                    LOGGER.info(f"Found Shutter for {device.device_id}: {shutter_device.get('label')}")
+                    shutter = ha_discovery_devices(
+                        site_id=site_id,
+                        device=device,
+                        mqtt_config=mqtt_config,
+                        sensor_name="shutter",
+                    )
+                    mqtt_publish(
+                        mqtt_client=mqtt_client,
+                        topic=shutter.get("topic"),
+                        payload=shutter.get("config"),
+                        retain=True,
+                    )
+                    mqtt_client.client.subscribe(shutter.get("config").get("command_topic"))
+
             # Works with Websockets
             if "Télécommande 4 boutons" in device.device_definition.get("device_definition_label"):
                 LOGGER.info(f"Found {device.device_definition.get('device_definition_label')}")

myFox2Mqtt/business/mqtt.py

@@ -85,6 +85,13 @@ def consume_mqtt_message(msg, mqtt_config: dict, api: MyFoxApi, mqtt_client: cli
                 site_id=site_id,
             )
 
+        # Manage Shutter
+        elif text_payload in ["open", "close", "my"]:
+            site_id = msg.topic.split("/")[1]
+            device_id = msg.topic.split("/")[2]
+            LOGGER.info(f"{text_payload} Shutter on {site_id} / {device_id}")
+            api.shutter_action_device(site_id=site_id, device_id=device_id, action=text_payload)
+
         # Manage Siren
         elif text_payload == "panic":
             site_id = msg.topic.split("/")[1]

myFox2Mqtt/docker-compose.yml

@@ -2,6 +2,6 @@ version: "3"
 services:
   myfox2mqtt:
     build: .
-    image: myfox2mqtt:v2023.8.0
+    image: myfox2mqtt:v2023.9.0
     volumes:
       - ./config:/config

myFox2Mqtt/homeassistant/ha_discovery.py

@@ -24,6 +24,10 @@
         "type": "number",
         "config": {"min": 0, "max": 100, "step": 1},
     },
+    "shutter": {
+        "type": "cover",
+        "config": {"pl_open": "open", "pl_close": "close", "pl_stop": "my", "optimistic": "true"},
+    },
     "image_detection_enabled": {
         "type": "switch",
         "config": {
@@ -738,6 +742,10 @@ def ha_discovery_devices(
             )
     if device_type in ("switch", "number", "select", "button"):
         device_config["config"]["command_topic"] = command_topic
+    if device_type in ("cover"):
+        device_config["config"]["command_topic"] = command_topic
+        device_config["config"].pop("state_topic")
+        device_config["config"].pop("value_template")
     if sensor_name == "snapshot":
         device_config["config"].pop("value_template")
     if sensor_name == "presence":

myFox2Mqtt/main.py

@@ -15,7 +15,7 @@
 from myfox.api import MyFoxApi
 from myfox.websocket import MyFoxWebsocket
 
-VERSION = "2023.8.0"
+VERSION = "2023.9.0"
 
 
 def myfox_loop(config, mqtt_client, api):

myFox2Mqtt/myfox/api/__init__.py

@@ -398,7 +398,7 @@ def get_scenarios(
         LOGGER.info(f"Scenarios: {response.json()}")
         return response.json()
 
-    def get_devices_temperature(self, site_id: str, category: Optional[Category] = None):
+    def get_devices_temperature(self, site_id: str):
         """List Devices from a Site ID
 
         Args:
@@ -430,7 +430,7 @@ def get_device_temperature(self, site_id: str, device_id: str):
         LOGGER.info(f"Device: {response.json()}")
         return response.json()
 
-    def get_devices_state(self, site_id: str, category: Optional[Category] = None):
+    def get_devices_state(self, site_id: str):
         """List Devices from a Site ID
 
         Args:
@@ -440,7 +440,6 @@ def get_devices_state(self, site_id: str, category: Optional[Category] = None):
         Returns:
             List[Device]: List of Device object
         """
-        devices = []  # type: List[Device]
         response = self.get(f"/v2/site/{site_id}/device/data/state/items")
         try:
             content = response.json()
@@ -464,7 +463,7 @@ def get_device_state(self, site_id: str, device_id: str):
         LOGGER.info(f"Device: {response.json()}")
         return response.json()
 
-    def get_devices_light(self, site_id: str, category: Optional[Category] = None):
+    def get_devices_light(self, site_id: str):
         """List Devices from a Site ID
 
         Args:
@@ -497,7 +496,10 @@ def get_device_light(self, site_id: str, device_id: str):
         LOGGER.info(f"Device: {response.json()}")
         return response.json()
 
-    def get_devices_other(self, site_id: str, category: Optional[Category] = None):
+    def get_devices_other(
+        self,
+        site_id: str,
+    ):
         """List Devices from a Site ID
 
         Args:
@@ -515,7 +517,10 @@ def get_devices_other(self, site_id: str, category: Optional[Category] = None):
         LOGGER.info(f"Devices Other: {response.json()}")
         return content.get("payload").get("items")
 
-    def get_devices_camera(self, site_id: str, category: Optional[Category] = None):
+    def get_devices_camera(
+        self,
+        site_id: str,
+    ):
         """List Devices from a Site ID
 
         Args:
@@ -525,7 +530,6 @@ def get_devices_camera(self, site_id: str, category: Optional[Category] = None):
         Returns:
             List[Device]: List of Device object
         """
-        devices = []  # type: List[Device]
         response = self.get(f"/v2/site/{site_id}/device/camera/items")
         try:
             content = response.json()
@@ -550,3 +554,46 @@ def get_site_history(self, site_id: str):
             response.raise_for_status()
         LOGGER.info(f"Site History: {response.json()}")
         return content.get("payload").get("items")
+
+    def get_devices_shutter(self, site_id: str):
+        """Get Devices Shutter from a Site ID
+
+        Args:
+            site_id (Optional[str], optional): Site ID. Defaults to None.
+
+        Returns:
+            List[Device]: List of Device object
+        """
+        response = self.get(f"/v2/site/{site_id}/device/shutter/items")
+        try:
+            content = response.json()
+        except JSONDecodeError:
+            response.raise_for_status()
+        LOGGER.info(f"Devices Shutter: {response.json()}")
+        return content.get("payload").get("items")
+
+    def shutter_action_device(
+        self,
+        site_id: str,
+        device_id: str,
+        action: str,
+    ) -> Dict:
+        """Make an action on a Shutter
+
+        Args:
+            site_id (str): Site ID
+            device_id (str): Device ID
+            action (str): Action
+
+        Returns:
+            str: Task ID
+        """
+        if action not in ["open", "close", "my"]:
+            raise ValueError(f"Unknown action {action}")
+
+        response = self.post(
+            f"/v2/site/{site_id}/device/{device_id}/shutter/{action}",
+            json={},
+        )
+        response.raise_for_status()
+        return response.json()