diff --git a/ccm15/CCM15Device.py b/ccm15/CCM15Device.py index e78d26f..5b6ddc3 100644 --- a/ccm15/CCM15Device.py +++ b/ccm15/CCM15Device.py @@ -5,6 +5,7 @@ BASE_URL = "http://{0}:{1}/{2}" CONF_URL_STATUS = "status.xml" +CONF_URL_CTRL = "ctrl.xml" DEFAULT_TIMEOUT = 10 class CCM15Device: @@ -38,4 +39,41 @@ async def _fetch_data(self) -> CCM15DeviceState: async def get_status_async(self) -> CCM15DeviceState: return await self._fetch_data() + async def async_test_connection(self): # pragma: no cover + """Test the connection to the CCM15 device.""" + url = f"http://{self._host}:{self._port}/{CONF_URL_STATUS}" + try: + async with aiohttp.ClientSession() as session, session.get( + url, self.timeout + ) as response: + if response.status == 200: + return True + return False + except (aiohttp.ClientError, asyncio.TimeoutError): + _LOGGER.debug("Test connection: Timeout") + return False + + async def async_send_state(self, url: str) -> bool: # pragma: no cover + """Send the url to set state to the ccm15 slave.""" + async with httpx.AsyncClient() as client: + response = await client.get(url, self.timeout) + return response.status_code in (httpx.codes.OK, httpx.codes.FOUND) + + async def async_set_state(self, ac_index: int, state: str, value: int) -> bool: + """Set new target states.""" + ac_id: int = 2**ac_index + url = BASE_URL.format( + self.host, + self.port, + CONF_URL_CTRL + + "?ac0=" + + str(ac_id) + + "&ac1=0" + + "&" + + state + + "=" + + str(value), + ) + + return await self.async_send_state(url) diff --git a/setup.py b/setup.py index 401dfd3..1f46663 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="py-ccm15", - version="0.0.3", + version="0.0.4", author="Oscar Calvo", author_email="oscar@calvonet.com", description="A package to control Midea CCM15 data converter modules", diff --git a/tests/test_ccm15.py b/tests/test_ccm15.py index a0712de..b0d7850 100644 --- a/tests/test_ccm15.py +++ b/tests/test_ccm15.py @@ -25,5 +25,16 @@ async def test_get_status_async(self, mock_get) -> None: self.assertIsInstance(state.devices[0], CCM15SlaveDevice) self.assertIsInstance(state.devices[1], CCM15SlaveDevice) + @patch("httpx.AsyncClient.get") + async def test_async_set_state(self, mock_get): + # Set up mock response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_get.return_value = mock_response + + # Call method and check result + result = await self.ccm.async_set_state(0, "state", 1) + self.assertTrue(result) + if __name__ == "__main__": unittest.main()