feat(cloud): add support for Japanese Toshiba IOLife devices

midealocal/cloud.py

@@ -71,6 +71,12 @@
         "app_key": "434a209a5ce141c3b726de067835d7f0",
         "api_url": "https://mapp.appsmb.com",  # codespell:ignore
     },
+    "Toshiba Iolife": {
+        "class_name": "ToshibaIOLife",
+        "app_id": "1203",
+        "app_key": "09c4d09f0da1513bb62dc7b6b0af9c11",
+        "api_url": "https://app.iolife.toshiba-lifestyle.com",  # codespell:ignore
+    },
 }
 
 DEFAULT_KEYS = {
@@ -768,7 +774,7 @@ async def download_lua(
         data = self._make_general_data()
         data.update(
             {
-                "applianceMFCode": manufacturer_code,
+                "applianceMFCode": "0008",
                 "applianceType": hex(device_type),
                 "applianceSn": self._security.aes_encrypt_with_fixed_key(
                     sn.encode("ascii"),
@@ -983,6 +989,106 @@ async def list_appliances(
         return None
 
 
+class ToshibaIOLife(MideaAirCloud):
+    """ Toshiba IOLife """
+    async def list_appliance_types(
+        self,
+    ) -> dict[int, dict[str, Any]] | None:
+        """ List Toshiba IOLife device types """
+        data = self._make_general_data()
+        data.update({"applianceType": "0xFF"})
+        if response := await self._api_request(
+            endpoint="/v1/appliance/type/list/get",
+            data=data,
+        ):
+            return(data)
+        return None
+
+    async def list_appliances(
+        self,
+    ) -> dict[int, dict[str, Any]] | None:
+        """ Get Toshiba IOLife devices."""
+        data = self._make_general_data()
+        if response := await self._api_request(
+            endpoint="/v2/appliance/user/list/get",
+            data=data,
+        ):
+            appliances = {}
+            for appliance in response["list"]:
+                try:
+                    model_number = int(appliance.get("modelNumber", 0))
+                except ValueError:
+                    model_number = 0
+
+                # Skip virtual batch devices:
+                if appliance.get("type") == "0x_BATCH_AC":
+                    continue
+                if appliance.get("id") == "virtual_ag_0xAC":
+                    continue
+                device_info = {
+                    "name": appliance.get("name"),
+                    "type": int(appliance.get("type"), 16),
+                    "sn": appliance.get("sn"),
+                    "sn8": "",
+                    "model_number": model_number,
+                    "manufacturer_code": appliance.get("enterpriseCode", "0000"),
+                    "model": "",
+                    "online": appliance.get("onlineStatus") == "1",
+                }
+                serial_num = device_info.get("sn")
+                device_info["sn8"] = (
+                    serial_num[9:17]
+                    if (serial_num and len(serial_num) > SN8_MIN_SERIAL_LENGTH)
+                    else ""
+                )
+                device_info["model"] = device_info.get("sn8")
+                appliances[int(appliance["id"])] = device_info
+            return appliances
+        return None
+
+    # FIXME: this isn't working:
+    async def download_lua(
+        self,
+        path: str,
+        device_type: int,
+        sn: str,
+        model_number: str | None = None,
+        manufacturer_code: str = "0008",
+    ) -> str | None:
+        """Download lua integration."""
+        data = self._make_general_data()
+        data.update(
+            {
+            "appId": self._app_id,
+            "appKey": self._app_key,
+            "applianceMFCode": manufacturer_code,
+            "applianceType": hex(device_type),
+            "modelNumber": "",
+            "applianceSn": sn,
+            "version": "0",
+        })
+        if model_number is not None:
+            data["modelNumber"] = model_number
+        fnm = None
+        if response := await self._api_request(
+            endpoint="/v2/open/sdk/product/encrptedLuaGet", # FIXME: Wrong URL?
+            data=data,
+        ):
+            res = await self._session.get(response["url"])
+            if res.status == HTTPStatus.OK:
+                lua = await res.text()
+                if lua:
+                    stream = (
+                        'local bit = require "bit"\n'
+                        + self._security.aes_decrypt_with_fixed_key(lua)
+                    )
+                    stream = stream.replace("\r\n", "\n")
+                    fnm = f"{path}/{response['fileName']}"
+                    async with aiofiles.open(fnm, "w") as fp:
+                        await fp.write(stream)
+        return str(fnm) if fnm else None
+
+
 def get_midea_cloud(
     cloud_name: str,
     session: ClientSession,

midealocal/decrypt.py

@@ -0,0 +1,53 @@
+from Crypto.Cipher import AES
+import binascii
+import hashlib
+import sys
+
+def decrypt_file(input_file, key):
+    # Generate MD5 hash and take first 16 chars
+    key_bytes = hashlib.md5(key.encode()).hexdigest()[:16].encode()
+
+    # Print the MD5 hash used as key
+    print(f"Using key: {hashlib.md5(key.encode()).hexdigest()[:16]}")
+
+    # Read encrypted data
+    with open(input_file, 'r') as f:
+        hex_str = f.read().strip()
+        # Convert hex string to bytes
+        encrypted_data = bytes([int(hex_str[i:i+2], 16) for i in range(0, len(hex_str), 2)])
+
+    # Create cipher object and decrypt the data
+    cipher = AES.new(key_bytes, AES.MODE_ECB)
+
+    # Add PKCS5 padding if needed
+    block_size = 16
+    pad_length = block_size - (len(encrypted_data) % block_size)
+    if pad_length < block_size:
+        encrypted_data += bytes([pad_length] * pad_length)
+
+    # Decrypt data
+    decrypted_data = cipher.decrypt(encrypted_data)
+
+    # Remove padding
+    padding_length = decrypted_data[-1]
+    decrypted_data = decrypted_data[:-padding_length]
+
+    # Convert to string
+    decrypted_text = decrypted_data.decode('utf-8')
+
+    # Write decrypted data
+    output_file = input_file + '.decrypted'
+    with open(output_file, 'w') as f:
+        f.write(decrypted_text)
+
+    print(f"Decrypted file saved as: {output_file}")
+
+# Decrypt the file
+app_key = "09c4d09f0da1513bb62dc7b6b0af9c11"  # IoLife app key
+
+if len(sys.argv) < 2:
+    print("Usage: python decrypt.py <input_file>")
+    sys.exit(1)
+
+input_file = sys.argv[1]
+decrypt_file(input_file, app_key)

midealocal/device.py

@@ -15,6 +15,7 @@
 from .message import (
     MessageApplianceResponse,
     MessageQueryAppliance,
+    MessageQueryToshibaIolife,
     MessageQuestCustom,
     MessageRequest,
     MessageType,

midealocal/devices/__init__.py

@@ -22,7 +22,9 @@ def device_selector(
 ) -> MideaDevice:
     """Select and load device."""
     try:
-        if device_type < DeviceType.A0:
+        if model == '00000008':
+            device_path = f".{f'x0008{device_type:02x}'}"
+        elif device_type < DeviceType.A0:
             device_path = f".{f'x{device_type:02x}'}"
         else:
             device_path = f".{f'{device_type:02x}'}"

midealocal/devices/x0008db/__init__.py

@@ -0,0 +1,107 @@
+"""Midea local DB device."""
+
+import json
+import logging
+from enum import StrEnum
+from typing import Any
+
+from midealocal.const import DeviceType, ProtocolVersion
+from midealocal.device import MideaDevice
+from midealocal.devices.x0008db.lua_converter import LuaConverter
+from midealocal.exceptions import ValueWrongType
+from .message import MessageQuery
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class DeviceAttributes(StrEnum):
+    """Midea DB device attributes."""
+
+    machine_status = "machine_status"
+    mode = "mode"
+    door_open = "door_open"
+    detergent_remain = "detergent_remain"
+    softner_remain = "softner_remain"
+    program = "program"
+    dry_filter_clean = "dry_filter_clean"
+    drain_filter_clean = "drain_filter_clean"
+    over_capacity = "over_capacity"
+    fungus_protect = "fungus_protect"
+    remain_time = "remain_time"
+
+class MideaDBDevice(MideaDevice):
+    """Midea DB device."""
+
+    def __init__(
+        self,
+        name: str,
+        device_id: int,
+        ip_address: str,
+        port: int,
+        token: str,
+        key: str,
+        device_protocol: ProtocolVersion,
+        model: str,
+        subtype: int,
+        customize: str,  # noqa: ARG002
+    ) -> None:
+        """Initialize Midea DB device."""
+        super().__init__(
+            name=name,
+            device_id=device_id,
+            device_type=DeviceType.DB,
+            ip_address=ip_address,
+            port=port,
+            token=token,
+            key=key,
+            device_protocol=device_protocol,
+            model=model,
+            subtype=subtype,
+            attributes={
+                DeviceAttributes.machine_status: "power_off", # power_off | power_on | running | pause | finish | pop_up
+                DeviceAttributes.mode: "none", # none | wash_dry | wash | dry | clean_care | care
+                DeviceAttributes.door_open: 0, # 0: closed, 1: open
+                DeviceAttributes.detergent_remain: None,
+                DeviceAttributes.softner_remain: None,
+                DeviceAttributes.program: "none", # none | standard | tub_clean | fast | careful | sixty_wash | blanket | delicate | tub_clean_dry | memory | sterilization | mute | soft | delicate_dryer | wool | quick_dry | quick_dry_delicate | quick_dry_blanket | quick_dry_delicate_blanket | quick_dry_sterilization | quick_dry_mute | quick_dry_soft | quick_dry_delicate | quick_dry_blanket_delicate | quick_dry_sterilization_blanket | quick_dry_sterilization_blanket_delicate
+                DeviceAttributes.dry_filter_clean: 0, # 0: clean, 1: dirty
+                DeviceAttributes.drain_filter_clean: 0, # 0: clean, 1: dirty
+                DeviceAttributes.over_capacity: 0, # 0: normal, 1: over capacity
+                DeviceAttributes.fungus_protect: "off", # off | on
+                DeviceAttributes.remain_time: 0,
+            },
+        )
+        self._appliance_query = False
+
+    def build_query(self) -> list[MessageQuery]:
+        """Midea DB device build query."""
+        return [MessageQuery()]
+
+    def process_message(self, msg: bytes) -> dict[str, Any]:
+        """Midea DB device process message."""
+        raw_data = bytearray(msg).hex()
+        _LOGGER.debug("[%s] Received raw data: %s", self.device_id, raw_data)
+        lua = LuaConverter()
+        json_string = lua.data_to_json(raw_data)
+        message = json.loads(json_string)
+        _LOGGER.debug("[%s] Received: %s", self.device_id, message)
+        data = message.get("status", {})
+        new_status = {}
+
+        for field in self._attributes:
+            field_str = str(field)
+            if field_str in data:
+                self._attributes[field] = data[field]
+                new_status[field] = self._attributes[field]
+            else:
+                _LOGGER.warning("[%s] Field %s not found in data", self.device_id, field)
+                _LOGGER.debug("[%s] Available data fields: %s", self.device_id, list(data.keys()))
+
+        return new_status
+
+    def set_attribute(self, attr: str, value: bool | int | str) -> None:
+        """Midea DB device set attribute."""
+        # not supported yet
+
+class MideaAppliance(MideaDBDevice):
+    """Midea DB appliance."""