Initial release 0.0.3 (#1)

* Initial setup

* Add publish script

* Fix publish script

* It links

* It works

* All test pass, 2 warnings

* All test pass, no warnings

* Release 0.0.3

ccm15/CCM15Device.py

@@ -0,0 +1,41 @@
+import httpx
+import xmltodict
+from .CCM15DeviceState import CCM15DeviceState
+from .CCM15SlaveDevice import CCM15SlaveDevice
+
+BASE_URL = "http://{0}:{1}/{2}"
+CONF_URL_STATUS = "status.xml"
+DEFAULT_TIMEOUT = 10
+
+class CCM15Device:
+    def __init__(self, host: str, port: int, timeout = DEFAULT_TIMEOUT):
+        self.host = host
+        self.port = port
+        self.timeout = timeout
+
+    async def _fetch_xml_data(self) -> str:
+        url = BASE_URL.format(self.host, self.port, CONF_URL_STATUS)
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, self.timeout)
+        return response.text
+
+    async def _fetch_data(self) -> CCM15DeviceState:
+        """Get the current status of all AC devices."""
+        str_data = await self._fetch_xml_data()
+        doc = xmltodict.parse(str_data)
+        data = doc["response"]
+        ac_data = CCM15DeviceState(devices={})
+        ac_index = 0
+        for ac_name, ac_binary in data.items():
+            if ac_binary == "-":
+                break
+            bytesarr = bytes.fromhex(ac_binary.strip(","))
+            ac_slave = CCM15SlaveDevice(bytesarr)
+            ac_data.devices[ac_index] = ac_slave
+            ac_index += 1
+        return ac_data
+
+    async def get_status_async(self) -> CCM15DeviceState:
+        return await self._fetch_data()
+
+

ccm15/CCM15DeviceState.py

@@ -0,0 +1,9 @@
+"""Data model to represent state of a CCM15 device."""
+from dataclasses import dataclass
+from . import CCM15SlaveDevice
+
+@dataclass
+class CCM15DeviceState:
+    """Data retrieved from a CCM15 device."""
+
+    devices: dict[int, CCM15SlaveDevice]

ccm15/CCM15SlaveDevice.py

@@ -0,0 +1,54 @@
+"""Data model to represent state of a CCM15 device."""
+from dataclasses import dataclass
+from enum import Enum
+
+@dataclass
+class TemperatureUnit(Enum):
+  CELSIUS = 1
+  FAHRENHEIT = 2
+
+@dataclass
+class CCM15SlaveDevice:
+    """Data retrieved from a CCM15 slave device."""
+
+    def __init__(self, bytesarr: bytes) -> None:
+        """Initialize the slave device."""
+        self.unit = TemperatureUnit.CELSIUS
+        buf = bytesarr[0]
+        if (buf >> 0) & 1:
+            self.unit = TemperatureUnit.FAHRENHEIT
+        self.locked_cool_temperature: int = (buf >> 3) & 0x1F
+
+        buf = bytesarr[1]
+        self.locked_heat_temperature: int = (buf >> 0) & 0x1F
+        self.locked_wind: int = (buf >> 5) & 7
+
+        buf = bytesarr[2]
+        self.locked_ac_mode: int = (buf >> 0) & 3
+        self.error_code: int = (buf >> 2) & 0x3F
+
+        buf = bytesarr[3]
+        self.ac_mode: int = (buf >> 2) & 7
+        self.fan_mode: int = (buf >> 5) & 7
+
+        buf = (buf >> 1) & 1
+        self.is_ac_mode_locked: bool = buf != 0
+
+        buf = bytesarr[4]
+        self.temperature_setpoint: int = (buf >> 3) & 0x1F
+        if self.unit == TemperatureUnit.FAHRENHEIT:
+            self.temperature_setpoint += 62
+            self.locked_cool_temperature += 62
+            self.locked_heat_temperature += 62
+        self.is_swing_on: bool = (buf >> 1) & 1 != 0
+
+        buf = bytesarr[5]
+        if ((buf >> 3) & 1) == 0:
+            self.locked_cool_temperature = 0
+        if ((buf >> 4) & 1) == 0:
+            self.locked_heat_temperature = 0
+        self.fan_locked: bool = buf >> 5 & 1 != 0
+        self.is_remote_locked: bool = ((buf >> 6) & 1) != 0
+
+        buf = bytesarr[6]
+        self.temperature: int = buf if buf < 128 else buf - 256

ccm15/__init__.py

@@ -0,0 +1,8 @@
+""" Init file """
+
+from .CCM15Device import CCM15Device
+from .CCM15DeviceState import CCM15DeviceState
+from .CCM15SlaveDevice import CCM15SlaveDevice, TemperatureUnit
+
+__all__ = ['CCM15Device', 'CCM15DeviceState', 'CCM15SlaveDevice', 'TemperatureUnit']
+

publish.ps1

@@ -0,0 +1,5 @@
+param($repo = "testpypi")
+
+git clean -dfx .
+python setup.py sdist bdist_wheel
+python -m twine upload --repository $repo dist/*

setup.py

@@ -0,0 +1,25 @@
+from setuptools import setup, find_packages
+
+with open("README.md", "r") as f:
+    long_description = f.read()
+
+setup(
+    name="py-ccm15",
+    version="0.0.3",
+    author="Oscar Calvo",
+    author_email="[email protected]",
+    description="A package to control Midea CCM15 data converter modules",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/ocalvo/py-ccm15",
+    packages=find_packages(),
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+    ],
+    install_requires=[
+        'httpx>=0.24.1',
+        'xmltodict>=0.13.0'
+    ]
+)

tests/test_ccm15.py

@@ -0,0 +1,29 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from ccm15 import CCM15Device, CCM15DeviceState, CCM15SlaveDevice
+
+class TestCCM15(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.ccm = CCM15Device("localhost", 8000)
+
+    @patch("httpx.AsyncClient.get")
+    async def test_get_status_async(self, mock_get) -> None:
+        # Set up mock response
+        mock_response = MagicMock()
+        mock_response.text = """
+        <response>
+            <ac1>00000001020304</ac1>
+            <ac2>00000005060708</ac2>
+        </response>
+        """
+        mock_get.return_value = mock_response
+
+        # Call method and check result
+        state = await self.ccm.get_status_async()
+        self.assertIsInstance(state, CCM15DeviceState)
+        self.assertEqual(len(state.devices), 2)
+        self.assertIsInstance(state.devices[0], CCM15SlaveDevice)
+        self.assertIsInstance(state.devices[1], CCM15SlaveDevice)
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_slave.py

@@ -0,0 +1,36 @@
+import unittest
+from ccm15 import CCM15SlaveDevice, TemperatureUnit
+
+class TestCCM15SlaveDevice(unittest.TestCase):
+    def test_swing_mode_on(self) -> None:
+        """Test that the swing mode is on."""
+        data = bytes.fromhex("00000041d2001a")
+        device = CCM15SlaveDevice(data)
+        self.assertTrue(device.is_swing_on)
+
+    def test_swing_mode_off(self) -> None:
+        """Test that the swing mode is off."""
+        data = bytes.fromhex("00000041d0001a")
+        device = CCM15SlaveDevice(data)
+        self.assertFalse(device.is_swing_on)
+
+    def test_temp_fan_mode(self) -> None:
+        """Test that the swing mode is on."""
+        data = bytes.fromhex("00000041d2001a")
+        device = CCM15SlaveDevice(data)
+        self.assertEqual(26, device.temperature)
+        self.assertEqual(2, device.fan_mode)
+        self.assertEqual(0, device.ac_mode)
+
+    def test_fahrenheit(self) -> None:
+        """Test that farenheith bit."""
+
+        data = bytearray.fromhex("81000041d2001a")
+        device = CCM15SlaveDevice(data)
+        self.assertEqual(TemperatureUnit.FAHRENHEIT, device.unit)
+        self.assertEqual(88, device.temperature_setpoint)
+        self.assertEqual(0, device.locked_cool_temperature)
+        self.assertEqual(0, device.locked_heat_temperature)
+
+if __name__ == '__main__':
+    unittest.main()