Inheritance

.devcontainer/devcontainer.json

@@ -4,19 +4,22 @@
 	"name": "Python 3",
 	// Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
 	"image": "mcr.microsoft.com/devcontainers/python:1-3.12-bullseye",
-
 	// Features to add to the dev container. More info: https://containers.dev/features.
 	// "features": {},
-
 	// Use 'forwardPorts' to make a list of ports inside the container available locally.
 	// "forwardPorts": [],
-
 	// Use 'postCreateCommand' to run commands after the container is created.
-	"postCreateCommand": "pip3 install -e ."
-
-	// Configure tool-specific properties.
-	// "customizations": {},
-
+	"postCreateCommand": "pip3 install -e .",
+	"customizations": {
+		"vscode": {
+			"extensions": [
+				"ms-python.black-formatter",
+				"ms-python.flake8",
+				"ms-python.vscode-pylance",
+				"ms-python.python"
+			]
+		}
+	}
 	// Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
 	// "remoteUser": "root"
-}
+}

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+max-line-length = 88
+extend-ignore = E203

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+    "[python]": {
+        "editor.defaultFormatter": "ms-python.black-formatter",
+        "editor.formatOnSave": true
+    }
+}

pymystrom/__init__.py

@@ -1,107 +1 @@
 """Base details for the myStrom Python bindings."""
-import asyncio
-import aiohttp
-import async_timeout
-from yarl import URL
-from typing import Any, Mapping, Optional
-import socket
-from .exceptions import MyStromConnectionError
-
-import pkg_resources
-
-__version__ = pkg_resources.get_distribution("setuptools").version
-
-TIMEOUT = 10
-USER_AGENT = f"PythonMyStrom/{__version__}"
-
-
-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 myStrom device."""
-    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 MyStromConnectionError(
-            "Timeout occurred while connecting to myStrom device."
-        ) from exception
-    except (aiohttp.ClientError, socket.gaierror) as exception:
-        raise MyStromConnectionError(
-            "Error occurred while communicating with myStrom device."
-        ) from exception
-
-    content_type = response.headers.get("Content-Type", "")
-    if (response.status // 100) in [4, 5]:
-        response.close()
-
-    if "application/json" in content_type:
-        response_json = await response.json()
-        return response_json
-
-    return response.text
-
-
-class MyStromDevice:
-    """A class for a myStrom device."""
-
-    def __init__(
-        self,
-        host,
-        session: aiohttp.client.ClientSession = None,
-    ):
-        """Initialize the device."""
-        self._close_session = False
-        self._host = host
-        self._session = session
-        self.uri = URL.build(scheme="http", host=self._host)
-
-    async def get_device_info(self) -> dict:
-        """Get the device info of a myStrom device."""
-        url = URL(self.uri).join(URL("api/v1/info"))
-        response = await _request(self, uri=url)
-        if not isinstance(response, dict):
-            # Fall back to the old API version if the device runs with old firmware
-            url = URL(self.uri).join(URL("info.json"))
-            response = await _request(self, uri=url)
-        return response
-
-    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) -> "MyStromDevice":
-        """Async enter."""
-        return self
-
-    async def __aexit__(self, *exc_info) -> None:
-        """Async exit."""
-        await self.close()
-
-
-async def get_device_info(host: str) -> dict:
-    """Get the device info of a myStrom device."""
-    async with MyStromDevice(host) as device:
-        return await device.get_device_info()

pymystrom/bulb.py

@@ -4,16 +4,17 @@
 
 import aiohttp
 from yarl import URL
-from typing import Any, Dict, Iterable, List, Optional, Union
+from typing import Optional
 
-from . import _request as request
+from .device import _request as request
+from .device import MyStromDevice
 
 _LOGGER = logging.getLogger(__name__)
 
-URI_BULB = URL("api/v1/device")
+API_PREFIX = URL("api/v1/device")
 
 
-class MyStromBulb:
+class MyStromBulb(MyStromDevice):
     """A class for a myStrom bulb."""
 
     def __init__(
@@ -23,10 +24,8 @@ def __init__(
         session: aiohttp.client.ClientSession = None,
     ):
         """Initialize the bulb."""
-        self._close_session = False
-        self._host = host
+        super().__init__(host, session)
         self._mac = mac
-        self._session = session
         self.brightness = 0
         self._color = None
         self._consumption = 0
@@ -36,7 +35,9 @@ def __init__(
         self._bulb_type = None
         self._state = None
         self._transition_time = 0
-        self.uri = URL.build(scheme="http", host=self._host).join(URI_BULB) / self._mac
+        # self.uri = URL.build(
+        #   scheme="http", host=self._host
+        # ).join(URI_BULB) / self._mac
 
     async def get_state(self) -> object:
         """Get the state of the bulb."""
@@ -91,9 +92,8 @@ def state(self) -> Optional[str]:
 
     async def set_on(self):
         """Turn the bulb on with the previous settings."""
-        response = await request(
-            self, uri=self.uri, method="POST", data={"action": "on"}
-        )
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
+        response = await request(self, url, method="POST", data={"action": "on"})
         return response
 
     async def set_color_hex(self, value):
@@ -104,11 +104,12 @@ async def set_color_hex(self, value):
         green: 0000FF00
         blue:  000000FF
         """
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
         data = {
             "action": "on",
             "color": value,
         }
-        response = await request(self, uri=self.uri, method="POST", data=data)
+        response = await request(self, url, method="POST", data=data)
         return response
 
     async def set_color_hsv(self, hue, saturation, value):
@@ -120,8 +121,9 @@ async def set_color_hsv(self, hue, saturation, value):
         #     'action': 'on',
         #     'color': f"{hue};{saturation};{value}",
         # }
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
         data = "action=on&color={};{};{}".format(hue, saturation, value)
-        response = await request(self, uri=self.uri, method="POST", data=data)
+        response = await request(self, url, method="POST", data=data)
         return response
 
     async def set_white(self):
@@ -139,11 +141,12 @@ async def set_sunrise(self, duration):
 
         The brightness is from 0 till 100.
         """
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
         max_brightness = 100
         await self.set_transition_time((duration / max_brightness))
         for i in range(0, duration):
             data = "action=on&color=3;{}".format(i)
-            await request(self, uri=self.uri, method="POST", data=data)
+            await request(self, url, method="POST", data=data)
             await asyncio.sleep(duration / max_brightness)
 
     async def set_flashing(self, duration, hsv1, hsv2):
@@ -157,27 +160,17 @@ async def set_flashing(self, duration, hsv1, hsv2):
 
     async def set_transition_time(self, value):
         """Set the transition time in ms."""
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
         response = await request(
-            self, uri=self.uri, method="POST", data={"ramp": int(round(value))}
+            self, url, method="POST", data={"ramp": int(round(value))}
         )
         return response
 
     async def set_off(self):
         """Turn the bulb off."""
-        response = await request(
-            self, uri=self.uri, method="POST", data={"action": "off"}
-        )
+        url = URL(self.uri).join(URL(f"{API_PREFIX}/{self.mac}"))
+        response = await request(self, url, method="POST", data={"action": "off"})
         return response
 
-    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) -> "MyStromBulb":
-        """Async enter."""
         return self
-
-    async def __aexit__(self, *exc_info) -> None:
-        """Async exit."""
-        await self.close()

pymystrom/device.py

@@ -0,0 +1,107 @@
+"""Base device class for all myStrom devices."""
+from typing import Any, Mapping, Optional
+import asyncio
+import aiohttp
+import pkg_resources
+import socket
+import async_timeout
+from yarl import URL
+from .exceptions import MyStromConnectionError
+
+__version__ = pkg_resources.get_distribution("setuptools").version
+
+TIMEOUT = 10
+USER_AGENT = f"PythonMyStrom/{__version__}"
+
+
+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 myStrom device."""
+    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 MyStromConnectionError(
+            "Timeout occurred while connecting to myStrom device."
+        ) from exception
+    except (aiohttp.ClientError, socket.gaierror) as exception:
+        raise MyStromConnectionError(
+            "Error occurred while communicating with myStrom device."
+        ) from exception
+
+    content_type = response.headers.get("Content-Type", "")
+    if (response.status // 100) in [4, 5]:
+        response.close()
+
+    if "application/json" in content_type:
+        response_json = await response.json()
+        return response_json
+
+    return response.text
+
+
+class MyStromDevice:
+    """A class for a myStrom device."""
+
+    def __init__(
+        self,
+        host,
+        session: aiohttp.client.ClientSession = None,
+    ):
+        """Initialize the device."""
+        self._close_session = False
+        self._host = host
+        self._session = session
+        self.uri = URL.build(scheme="http", host=self._host)
+
+    async def get_device_info(self) -> dict:
+        """Get the device info of a myStrom device."""
+        url = URL(self.uri).join(URL("api/v1/info"))
+        print(url)
+        response = await _request(self, uri=url)
+        if not isinstance(response, dict):
+            # Fall back to the old API version if the device runs with old firmware
+            url = URL(self.uri).join(URL("info.json"))
+            response = await _request(self, uri=url)
+        return response
+
+    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) -> "MyStromDevice":
+        """Async enter."""
+        return self
+
+    async def __aexit__(self, *exc_info) -> None:
+        """Async exit."""
+        await self.close()
+
+
+async def get_device_info(host: str) -> dict:
+    """Get the device info of a myStrom device."""
+    async with MyStromDevice(host) as device:
+        return await device.get_device_info()