From dd26cd6f65d981ad3dc1481ba72d3717e67040df Mon Sep 17 00:00:00 2001 From: Oscar Calvo <2091582+ocalvo@users.noreply.github.com> Date: Thu, 10 Aug 2023 19:31:32 -0600 Subject: [PATCH] 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 | 41 +++++++++++++++++++++++++++++ ccm15/CCM15DeviceState.py | 9 +++++++ ccm15/CCM15SlaveDevice.py | 54 +++++++++++++++++++++++++++++++++++++++ ccm15/__init__.py | 8 ++++++ publish.ps1 | 5 ++++ setup.py | 25 ++++++++++++++++++ tests/test_ccm15.py | 29 +++++++++++++++++++++ tests/test_slave.py | 36 ++++++++++++++++++++++++++ 8 files changed, 207 insertions(+) create mode 100644 ccm15/CCM15Device.py create mode 100644 ccm15/CCM15DeviceState.py create mode 100644 ccm15/CCM15SlaveDevice.py create mode 100644 ccm15/__init__.py create mode 100644 publish.ps1 create mode 100644 setup.py create mode 100644 tests/test_ccm15.py create mode 100644 tests/test_slave.py diff --git a/ccm15/CCM15Device.py b/ccm15/CCM15Device.py new file mode 100644 index 0000000..e78d26f --- /dev/null +++ b/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() + + diff --git a/ccm15/CCM15DeviceState.py b/ccm15/CCM15DeviceState.py new file mode 100644 index 0000000..feac9e8 --- /dev/null +++ b/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] diff --git a/ccm15/CCM15SlaveDevice.py b/ccm15/CCM15SlaveDevice.py new file mode 100644 index 0000000..305fcb1 --- /dev/null +++ b/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 diff --git a/ccm15/__init__.py b/ccm15/__init__.py new file mode 100644 index 0000000..25071be --- /dev/null +++ b/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'] + diff --git a/publish.ps1 b/publish.ps1 new file mode 100644 index 0000000..10f9362 --- /dev/null +++ b/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/* diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..401dfd3 --- /dev/null +++ b/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="oscar@calvonet.com", + 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' + ] +) diff --git a/tests/test_ccm15.py b/tests/test_ccm15.py new file mode 100644 index 0000000..a0712de --- /dev/null +++ b/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 = """ + + 00000001020304 + 00000005060708 + + """ + 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() diff --git a/tests/test_slave.py b/tests/test_slave.py new file mode 100644 index 0000000..b203cbd --- /dev/null +++ b/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()