diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 244f3e6..b40bcfb 100644 --- a/.devcontainer/devcontainer.json +++ b/.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" -} +} \ No newline at end of file diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..e0ea542 --- /dev/null +++ b/.flake8 @@ -0,0 +1,3 @@ +[flake8] +max-line-length = 88 +extend-ignore = E203 \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..a0e76ff --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,6 @@ +{ + "[python]": { + "editor.defaultFormatter": "ms-python.black-formatter", + "editor.formatOnSave": true + } +} \ No newline at end of file diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index 80cc054..fbae3f5 100644 --- a/pymystrom/__init__.py +++ b/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() diff --git a/pymystrom/bulb.py b/pymystrom/bulb.py index 93aab20..28d357c 100644 --- a/pymystrom/bulb.py +++ b/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() diff --git a/pymystrom/device.py b/pymystrom/device.py new file mode 100644 index 0000000..b57cdaa --- /dev/null +++ b/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() diff --git a/pymystrom/pir.py b/pymystrom/pir.py index de66b29..84c3723 100644 --- a/pymystrom/pir.py +++ b/pymystrom/pir.py @@ -1,21 +1,20 @@ """Support for communicating with myStrom PIRs.""" 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 -URI_PIR = URL("api/v1/") +API_PREFIX = "api/v1" -class MyStromPir: +class MyStromPir(MyStromDevice): """A class for a myStrom PIR.""" def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None: - """Initialize the switch.""" - self._close_session = False - self._host = host - self._session = session + """Initialize the PIR.""" + super().__init__(host, session) self._intensity = None self._day = None self._light_raw = None @@ -29,29 +28,28 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N self._pir = None self._actions = None - self.uri = URL.build(scheme="http", host=self._host).join(URI_PIR) async def get_settings(self) -> None: """Get the current settings from the PIR.""" - url = URL(self.uri).join(URL("settings")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/settings")) response = await request(self, uri=url) self._settings = response async def get_actions(self) -> None: """Get the current action settings from the PIR.""" - url = URL(self.uri).join(URL("action")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/action")) response = await request(self, uri=url) self._actions = response async def get_pir(self) -> None: """Get the current PIR settings.""" - url = URL(self.uri).join(URL("settings/pir")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/settings/pir")) response = await request(self, uri=url) self._pir = response async def get_sensors_state(self) -> None: """Get the state of the sensors from the PIR.""" - url = URL(self.uri).join(URL("sensors")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/sensors")) response = await request(self, uri=url) # The return data has the be re-written as the temperature is not rounded self._sensors = { @@ -72,13 +70,13 @@ async def get_temperatures(self) -> None: async def get_motion(self) -> None: """Get the state of the motion sensor from the PIR.""" - url = URL(self.uri).join(URL("motion")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/motion")) response = await request(self, uri=url) self._motion = response["motion"] async def get_light(self) -> None: """Get the state of the light sensor from the PIR.""" - url = URL(self.uri).join(URL("light")) + url = URL(self.uri).join(URL(f"{API_PREFIX}/light")) response = await request(self, uri=url) self._intensity = response["intensity"] self._day = response["day"] @@ -147,15 +145,5 @@ def light_raw(self) -> Optional[str]: "infrared": self._light_raw["adc1"], } - 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) -> "MyStromPir": - """Async enter.""" return self - - async def __aexit__(self, *exc_info) -> None: - """Async exit.""" - await self.close() diff --git a/pymystrom/switch.py b/pymystrom/switch.py index a8d80db..31edc2d 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -1,26 +1,23 @@ """Support for communicating with myStrom plugs/switches.""" import aiohttp from yarl import URL -from typing import Any, Dict, Iterable, List, Optional, Union -from . import _request as request +from .device import _request as request +from .device import MyStromDevice -class MyStromSwitch: +class MyStromSwitch(MyStromDevice): """A class for a myStrom switch/plug.""" def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None: """Initialize the switch.""" - self._close_session = False - self._host = host - self._session = session + super().__init__(host, session) self._consumption = 0 self._consumedWs = 0 self._state = None self._temperature = None self._firmware = None self._mac = None - self.uri = URL.build(scheme="http", host=self._host) async def turn_on(self) -> None: """Turn the relay on.""" @@ -71,6 +68,12 @@ async def get_state(self) -> None: self._firmware = response["version"] self._mac = response["mac"] + async def get_temperature_full(self) -> str: + """Get current temperature in celsius.""" + url = URL(self.uri).join(URL("temp")) + response = await request(self, uri=url) + return response + @property def relay(self) -> bool: """Return the relay state.""" @@ -110,21 +113,5 @@ def temperature(self) -> float: return self._temperature - async def get_temperature_full(self) -> str: - """Get current temperature in celsius.""" - url = URL(self.uri).join(URL("temp")) - 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) -> "MyStromSwitch": - """Async enter.""" return self - - async def __aexit__(self, *exc_info) -> None: - """Async exit.""" - await self.close() diff --git a/setup.py b/setup.py index 4f87602..d989a27 100644 --- a/setup.py +++ b/setup.py @@ -1,7 +1,6 @@ """Set up the Python API for myStrom devices.""" import os -import sys from setuptools import setup, find_packages