Move request and class to sep. files

Author: fabaff

dingz/__init__.py

@@ -1,77 +1,58 @@
-"""A Python Client to interact with Dingz devices."""
+"""Base details for the dingz Python bindings."""
 import asyncio
-import logging
+import json
+import socket
+from typing import Any, Mapping, Optional
 
 import aiohttp
 import async_timeout
 
-from . import exceptions
-from .constants import TEMPERATURE, MOTION, LIGHT, PUCK, INFO
-
-_LOGGER = logging.getLogger(__name__)
-
-
-class Dingz:
-    """A class for handling the communication with a Dingz device."""
-
-    def __init__(self, device, loop, session):
-        """Initialize the connection."""
-        self._loop = loop
-        self._session = session
-        self.token = None
-        self.device = device
-        self.data = None
-
-    async def get_data(self, endpoint):
-        """Retrieve the data from a given endpoint."""
-        try:
-            with async_timeout.timeout(5, loop=self._loop):
-                response = await self._session.get(f"{endpoint}")
-
-            _LOGGER.debug("Response from Dingz device: %s", response.status)
-            self.data = await response.json()
-            _LOGGER.debug(self.data)
-        except (asyncio.TimeoutError, aiohttp.ClientError):
-            _LOGGER.error("Can not load data from Dingz device")
-            self.data = None
-            raise exceptions.DingzConnectionError()
-
-    @property
-    def info(self):
-        """Get the state of the motion sensor."""
-        data = await self.get_data(INFO)
-        return data['motion']
-
-    @property
-    def puck(self):
-        """Get the state of the hardware."""
-        data = await self.get_data(PUCK)
-        if self.data is not None:
-            return {'firmware': data['fw']['version'], 'hardware': data['hw']['version']}
-
-    @property
-    def temperature(self):
-        """Get the temperature."""
-        data = await self.get_data(TEMPERATURE)
-        return round(data['temperature'], 2)
-
-    @property
-    def light(self):
-        """Get the brightness."""
-        data = await self.get_data(LIGHT)
-        return round(data['intensity'], 2)
-
-    @property
-    def day(self):
-        """Get the state if daytime or not."""
-        data = await self.get_data(LIGHT)
-        return data['day']
-
-    @property
-    def motion(self):
-        """Get the state of the motion sensor."""
-        data = await self.get_data(MOTION)
-        return data['motion']
-
-
-
+from .constants import TIMEOUT, USER_AGENT
+from .exceptions import DingzConnectionError, DingzError
+
+
+async def _request(
+    self,
+    uri: str,
+    method: str = "GET",
+    data: Optional[Any] = None,
+    json_data: Optional[dict] = None,
+    params: Optional[Mapping[str, str]] = None,
+) -> Any:
+    """Handle a request to the dingz unit."""
+    headers = {
+        "User-Agent": USER_AGENT,
+        "Accept": "application/json, text/plain, */*",
+    }
+
+    if self._session is None:
+        self._session = aiohttp.ClientSession()
+        self._close_session = True
+
+    try:
+        with async_timeout.timeout(TIMEOUT):
+            response = await self._session.request(
+                method, uri, data=data, json=json_data, params=params, headers=headers,
+            )
+    except asyncio.TimeoutError as exception:
+        raise DingzConnectionError(
+            "Timeout occurred while connecting to dingz unit."
+        ) from exception
+    except (aiohttp.ClientError, socket.gaierror) as exception:
+        raise DingzConnectionError(
+            "Error occurred while communicating with dingz."
+        ) from exception
+
+    content_type = response.headers.get("Content-Type", "")
+    if (response.status // 100) in [4, 5]:
+        contents = await response.read()
+        response.close()
+
+        if content_type == "application/json":
+            raise DingzError(response.status, json.loads(contents.decode("utf8")))
+        raise DingzError(response.status, {"message": contents.decode("utf8")})
+    if "application/json" in content_type:
+        response_json = await response.json()
+        return response_json
+
+    return response.text

dingz/dingz.py

@@ -0,0 +1,58 @@
+"""A Python Client to interact with Dingz devices."""
+import logging
+
+import aiohttp
+from yarl import URL
+
+from . import _request as request
+from .constants import API, DEVICE_INFO, TEMPERATURE
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Dingz:
+    """A class for handling the communication with a dingz device."""
+
+    def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None:
+        """Initialize the dingz."""
+        self._close_session = False
+        self._host = host
+        self._session = session
+        self._device_details = None
+        self._temperature = None
+        self.uri = URL.build(scheme="http", host=self._host).join(URL(API))
+
+    async def get_device_info(self) -> None:
+        """Get the details from the switch."""
+        url = URL(self.uri).join(URL(DEVICE_INFO))
+        response = await request(self, uri=url)
+        self._device_details = response
+
+    async def get_temperature(self) -> None:
+        """Get the details from the switch."""
+        url = URL(self.uri).join(URL(TEMPERATURE))
+        response = await request(self, uri=url)
+        self._temperature = response["temperature"]
+
+    @property
+    def device_details(self) -> float:
+        """Return the current device details."""
+        return self._device_details
+
+    @property
+    def temperature(self) -> float:
+        """Return the current temperature in celsius."""
+        return round(self._temperature, 1)
+
+    async def close(self) -> None:
+        """Close an open client session."""
+        if self._session and self._close_session:
+            await self._session.close()
+
+    async def __aenter__(self) -> "Dingz":
+        """Async enter."""
+        return self
+
+    async def __aexit__(self, *exc_info) -> None:
+        """Async exit."""
+        await self.close()