feat: add cloud apis related to the Fog platform

src/libdeye/cloud_api.py

@@ -16,7 +16,9 @@
 from .types import (
     DeyeApiResponseDeviceInfo,
     DeyeApiResponseEnvelope,
-    DeyeApiResponseMqttInfo,
+    DeyeApiResponseDeyePlatformMqttInfo,
+    DeyeApiResponseFogPlatformMqttInfo,
+    DeyeApiResponseFogPlatformDeviceProperties,
 )
 
 
@@ -141,8 +143,8 @@ async def get_device_list(self) -> list[DeyeApiResponseDeviceInfo]:
         ensure_valid_response_code(result)
         return cast(list[DeyeApiResponseDeviceInfo], result["data"])
 
-    async def get_mqtt_info(self) -> DeyeApiResponseMqttInfo:
-        """Get MQTT server info / credentials for current user"""
+    async def get_deye_platform_mqtt_info(self) -> DeyeApiResponseDeyePlatformMqttInfo:
+        """Get MQTT server info / credentials for current user (Deye platform)"""
         await self.refresh_token_if_near_expiry()
 
         try:
@@ -155,7 +157,62 @@ async def get_mqtt_info(self) -> DeyeApiResponseMqttInfo:
             raise DeyeCloudApiCannotConnectError from err
 
         ensure_valid_response_code(result)
-        return cast(DeyeApiResponseMqttInfo, result["data"])
+        return cast(DeyeApiResponseDeyePlatformMqttInfo, result["data"])
+
+    async def get_fog_platform_mqtt_info(self) -> DeyeApiResponseFogPlatformMqttInfo:
+        """Get MQTT server info / credentials for current user (Fog platform)"""
+        await self.refresh_token_if_near_expiry()
+
+        try:
+            response = await self._session.get(
+                f"{DEYE_API_END_USER_ENDPOINT}/fogmqttinfo/",
+                headers={"Authorization": f"JWT {self.auth_token}"},
+            )
+            result: DeyeApiResponseEnvelope = await response.json()
+        except ClientError as err:
+            raise DeyeCloudApiCannotConnectError from err
+
+        ensure_valid_response_code(result)
+        return cast(DeyeApiResponseFogPlatformMqttInfo, result["data"])
+
+    async def get_fog_platform_device_properties(
+        self, device_id: str
+    ) -> DeyeApiResponseFogPlatformDeviceProperties:
+        """Get properties for a device on the Fog platform"""
+        await self.refresh_token_if_near_expiry()
+
+        try:
+            response = await self._session.get(
+                f"{DEYE_API_END_USER_ENDPOINT}/get/properties/?device_id={device_id}",
+                headers={"Authorization": f"JWT {self.auth_token}"},
+            )
+            result: DeyeApiResponseEnvelope = await response.json()
+        except ClientError as err:
+            raise DeyeCloudApiCannotConnectError from err
+
+        ensure_valid_response_code(result)
+        return cast(
+            DeyeApiResponseFogPlatformDeviceProperties, result["data"]["properties"]
+        )
+
+    async def poll_fog_platform_device_properties(self, device_id: str) -> None:
+        """Poll properties for a device on the Fog platform"""
+        await self.refresh_token_if_near_expiry()
+
+        try:
+            response = await self._session.post(
+                f"{DEYE_API_END_USER_ENDPOINT}/set/properties/",
+                headers={"Authorization": f"JWT {self.auth_token}"},
+                data={
+                    "device_id": device_id,
+                    "params": {"RealData": 1},
+                },
+            )
+            result: DeyeApiResponseEnvelope = await response.json()
+        except ClientError as err:
+            raise DeyeCloudApiCannotConnectError from err
+
+        ensure_valid_response_code(result)
 
 
 def ensure_valid_response_code(result: DeyeApiResponseEnvelope) -> None:

src/libdeye/types.py

@@ -97,8 +97,8 @@ class DeyeApiResponseEnvelope(TypedDict):
     data: Any
 
 
-class DeyeApiResponseMqttInfo(TypedDict):
-    """MQTT information returned by the API"""
+class DeyeApiResponseDeyePlatformMqttInfo(TypedDict):
+    """MQTT information for the Deye platform returned by the API"""
 
     password: str
     loginname: str
@@ -109,6 +109,62 @@ class DeyeApiResponseMqttInfo(TypedDict):
     sslport: int
 
 
+class DeyeApiResponseFogPlatformMqttTopics(TypedDict):
+    """MQTT topics returned by the API"""
+
+    all: list[str]
+    pub: list[str]
+    sub: list[str]
+
+
+class DeyeApiResponseFogPlatformMqttInfo(TypedDict):
+    """MQTT information for the Fog platform returned by the API"""
+
+    username: str
+    clientid: str
+    password: str
+    mqtt_host: str
+    ws_port: str
+    ssl_port: str
+    topic: DeyeApiResponseFogPlatformMqttTopics
+    expire: int
+
+
+class DeyeApiResponseFogPlatformDeviceProperties(TypedDict):
+    """Device properties for the Fog platform returned by the API"""
+
+    CompressorStatus: int
+    CurrentAmbientTemperature: int
+    CurrentCoilTemperature: int
+    CurrentEnvironmentalHumidity: int
+    CurrentExhaustTemperature: int
+    Demisting: int
+    EnvironmentalRating: int
+    Fan: int
+    KeyLock: int
+    Mode: int
+    NegativeIon: int
+    Power: int
+    ProtocolVersion: int
+    SetHumidity: int
+    SolenoidValve: int
+    SwingingWind: int
+    TimedOff: int
+    TimedOn: int
+    TimedShutdownHourSetting: int
+    TimedShutdownMinuteSettingTime: int
+    TimedShutdownTimeRemainingHours: int
+    TimedShutdownTimeRemainingMinutes: int
+    TimedStartupHoursSetTime: int
+    TimedStartupMinuteSettingTime: int
+    TimedStartupTimeRemainingHours: int
+    TimedStartupTimeRemainingMinutes: int
+    WaterPump: int
+    WaterTank: int
+    WindSpeed: int
+    fault: dict[str, int]
+
+
 class DeyeApiResponseDeviceInfo(TypedDict):
     """Device information returned by the API"""