From 7788144d4358cdf9416d7c742155023a28f1e60a Mon Sep 17 00:00:00 2001 From: Lenny5156 Date: Tue, 11 May 2021 21:12:10 +0200 Subject: [PATCH 1/4] Added functionality for Tokens. Example in example-switch-with-token.py Only the mystrom switch is tested. --- examples/example-switch-with-token.py | 37 ++++++++++++++ pymystrom/__init__.py | 2 + pymystrom/bulb.py | 17 ++++--- pymystrom/cli.py | 70 +++++++++++++++++++-------- pymystrom/pir.py | 17 ++++--- pymystrom/switch.py | 15 +++--- 6 files changed, 115 insertions(+), 43 deletions(-) create mode 100644 examples/example-switch-with-token.py diff --git a/examples/example-switch-with-token.py b/examples/example-switch-with-token.py new file mode 100644 index 0000000..831fabf --- /dev/null +++ b/examples/example-switch-with-token.py @@ -0,0 +1,37 @@ +"""Example code for communicating with a myStrom plug/switch.""" +import asyncio + +from pymystrom.switch import MyStromSwitch + +IP_ADDRESS = "192.168.0.40" +TOKEN = '' + + +async def main(): + """Sample code to work with a myStrom switch.""" + async with MyStromSwitch(IP_ADDRESS, TOKEN) as switch: + + # Collect the data of the current state + await switch.get_state() + + print("Power consumption:", switch.consumption) + print("Relay state:", switch.relay) + print("Temperature:", switch.temperature) + print("Firmware:", switch.firmware) + print("MAC address:", switch.mac) + + print("Turn on the switch") + if not switch.relay: + await switch.turn_on() + + # print("Toggle the switch") + # await switch.toggle() + + # Switch relay off if it was off + if switch.relay: + await switch.turn_off() + + +if __name__ == "__main__": + loop = asyncio.get_event_loop() + loop.run_until_complete(main()) diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index 6df2e0a..06e6a8b 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -22,11 +22,13 @@ async def _request( data: Optional[Any] = None, json_data: Optional[dict] = None, params: Optional[Mapping[str, str]] = None, + token: str = '' ) -> Any: """Handle a request to the myStrom device.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json, text/plain, */*", + "Token": token } if self._session is None: diff --git a/pymystrom/bulb.py b/pymystrom/bulb.py index 0fa9fc0..a21bee3 100644 --- a/pymystrom/bulb.py +++ b/pymystrom/bulb.py @@ -17,7 +17,7 @@ class MyStromBulb: """A class for a myStrom bulb.""" def __init__( - self, host: str, mac: str, session: aiohttp.client.ClientSession = None, + self, host: str, mac: str, token='', session: aiohttp.client.ClientSession = None, ): """Initialize the bulb.""" self._close_session = False @@ -34,10 +34,11 @@ def __init__( self._state = None self._transition_time = 0 self.uri = URL.build(scheme="http", host=self._host).join(URI_BULB) / self._mac + self.token = token async def get_state(self) -> object: """Get the state of the bulb.""" - response = await request(self, uri=self.uri) + response = await request(self, uri=self.uri, token=self.token) self._consumption = response[self._mac]["power"] self._firmware = response[self._mac]["fw_version"] self._color = response[self._mac]["color"] @@ -89,7 +90,7 @@ 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"} + self, uri=self.uri, method="POST", data={"action": "on"}, token=self.token ) return response @@ -105,7 +106,7 @@ async def set_color_hex(self, value): "action": "on", "color": value, } - response = await request(self, uri=self.uri, method="POST", data=data) + response = await request(self, uri=self.uri, method="POST", data=data, token=self.token) return response async def set_color_hsv(self, hue, saturation, value): @@ -118,7 +119,7 @@ async def set_color_hsv(self, hue, saturation, value): # 'color': f"{hue};{saturation};{value}", # } data = "action=on&color={};{};{}".format(hue, saturation, value) - response = await request(self, uri=self.uri, method="POST", data=data) + response = await request(self, uri=self.uri, method="POST", data=data, token=self.token) return response async def set_white(self): @@ -140,7 +141,7 @@ async def set_sunrise(self, duration): 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, uri=self.uri, method="POST", data=data, token=self.token) await asyncio.sleep(duration / max_brightness) async def set_flashing(self, duration, hsv1, hsv2): @@ -155,14 +156,14 @@ async def set_flashing(self, duration, hsv1, hsv2): async def set_transition_time(self, value): """Set the transition time in ms.""" response = await request( - self, uri=self.uri, method="POST", data={"ramp": int(round(value))} + self, uri=self.uri, method="POST", data={"ramp": int(round(value))}, token=self.token ) return response async def set_off(self): """Turn the bulb off.""" response = await request( - self, uri=self.uri, method="POST", data={"action": "off"} + self, uri=self.uri, method="POST", data={"action": "off"}, token=self.token ) return response diff --git a/pymystrom/cli.py b/pymystrom/cli.py index b46dbb9..28e09bd 100644 --- a/pymystrom/cli.py +++ b/pymystrom/cli.py @@ -44,11 +44,14 @@ def config(): @click.option( "--mac", prompt="MAC address of the device", help="MAC address of the device." ) -def read_config(ip, mac): +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) +def read_config(ip, mac, token): """Read the current configuration of a myStrom device.""" click.echo("Read configuration from %s" % ip) try: - request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT) + request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers={'Token': token}) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device") @@ -66,6 +69,9 @@ def button(): @click.option( "--mac", prompt="MAC address of the button", help="MAC address of the button." ) +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) @click.option( "--single", prompt="URL for a single tap", default="", help="URL for a single tap." ) @@ -76,7 +82,7 @@ def button(): "--long", prompt="URL for a long tab", default="", help="URL for a long tab." ) @click.option("--touch", prompt="URL for a touch", default="", help="URL for a touch.") -def write_config(ip, mac, single, double, long, touch): +def write_config(ip, mac, token, single, double, long, touch): """Write the current configuration of a myStrom button.""" click.echo("Write configuration to device %s" % ip) data = { @@ -87,7 +93,7 @@ def write_config(ip, mac, single, double, long, touch): } try: request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers={'Token': token} ) if request.status_code == 200: @@ -103,6 +109,9 @@ def write_config(ip, mac, single, double, long, touch): @click.option( "--mac", prompt="MAC address of the button", help="MAC address of the button." ) +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) @click.option( "--hass", prompt="IP address of the Home Assistant instance", @@ -117,7 +126,7 @@ def write_config(ip, mac, single, double, long, touch): @click.option( "--id", prompt="ID of the button", default="", help="ID of the myStrom button." ) -def write_ha_config(ip, mac, hass, port, id): +def write_ha_config(ip, mac, token, hass, port, id): """Write the configuration for Home Assistant to a myStrom button.""" click.echo("Write configuration for Home Assistant to device %s..." % ip) @@ -130,7 +139,7 @@ def write_ha_config(ip, mac, hass, port, id): } try: request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers={'Token': token} ) if request.status_code == 200: @@ -150,7 +159,10 @@ def write_ha_config(ip, mac, hass, port, id): @click.option( "--mac", prompt="MAC address of the button", help="MAC address of the Wifi Button." ) -def reset_config(ip, mac): +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) +def reset_config(ip, mac, token): """Reset the current configuration of a myStrom WiFi Button.""" click.echo("Reset configuration of button %s..." % ip) data = { @@ -161,7 +173,7 @@ def reset_config(ip, mac): } try: request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers={'Token': token} ) if request.status_code == 200: @@ -177,11 +189,14 @@ def reset_config(ip, mac): @click.option( "--mac", prompt="MAC address of the button", help="MAC address of the Wifi Button." ) -def read_config(ip, mac): +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) +def read_config(ip, mac, token): """Read the current configuration of a myStrom WiFi Button.""" click.echo("Read the configuration of button %s..." % ip) try: - request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT) + request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers={'Token': token}) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device. No action performed") @@ -198,9 +213,12 @@ def bulb(): @click.option( "--mac", prompt="MAC address of the bulb", help="MAC address of the bulb." ) -async def on(ip, mac): +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) +async def on(ip, mac, token): """Switch the bulb on.""" - async with MyStromBulb(ip, mac) as bulb: + async with MyStromBulb(ip, mac, token) as bulb: await bulb.set_color_hex("00FFFFFF") @@ -213,6 +231,9 @@ async def on(ip, mac): @click.option( "--hue", prompt="Set the hue of the bulb", help="Set the hue of the bulb." ) +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) @click.option( "--saturation", prompt="Set the saturation of the bulb", @@ -221,9 +242,9 @@ async def on(ip, mac): @click.option( "--value", prompt="Set the value of the bulb", help="Set the value of the bulb." ) -async def color(ip, mac, hue, saturation, value): +async def color(ip, mac, token, hue, saturation, value): """Switch the bulb on with the given color.""" - async with MyStromBulb(ip, mac) as bulb: + async with MyStromBulb(ip, mac, token) as bulb: await bulb.set_color_hsv(hue, saturation, value) @@ -233,9 +254,12 @@ async def color(ip, mac, hue, saturation, value): @click.option( "--mac", prompt="MAC address of the bulb", help="MAC address of the bulb." ) -async def off(ip, mac): +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) +async def off(ip, mac, token): """Switch the bulb off.""" - async with MyStromBulb(ip, mac) as bulb: + async with MyStromBulb(ip, mac, token) as bulb: await bulb.set_off() @@ -245,12 +269,15 @@ async def off(ip, mac): @click.option( "--mac", prompt="MAC address of the bulb", help="MAC address of the bulb." ) +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) @click.option( "--time", prompt="Time to flash", help="Time to flash the bulb in seconds.", default=10, ) -async def flash(ip, mac, time): +async def flash(ip, mac, token, time): """Flash the bulb off.""" - async with MyStromBulb(ip, mac) as bulb: + async with MyStromBulb(ip, mac, token) as bulb: await bulb.set_flashing(time, [100, 50, 30], [200, 0, 71]) @@ -260,12 +287,15 @@ async def flash(ip, mac, time): @click.option( "--mac", prompt="MAC address of the bulb", help="MAC address of the bulb." ) +@click.option( + "--token", prompt="Token of the device", help="Token of the device." +) @click.option( "--time", prompt="Time for the complete rainbow", help="Time to perform the rainbow in seconds.", default=30, ) -async def rainbow(ip, mac, time): +async def rainbow(ip, mac, token, time): """Let the buld change the color and show a rainbow.""" - async with MyStromBulb(ip, mac) as bulb: + async with MyStromBulb(ip, mac, token) as bulb: await bulb.set_rainbow(time) await bulb.set_transition_time(1000) diff --git a/pymystrom/pir.py b/pymystrom/pir.py index 23e3976..d2e8071 100644 --- a/pymystrom/pir.py +++ b/pymystrom/pir.py @@ -11,7 +11,7 @@ class MyStromPir: """A class for a myStrom PIR.""" - def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None: + def __init__(self, host: str, token='', session: aiohttp.client.ClientSession = None) -> None: """Initialize the switch.""" self._close_session = False self._host = host @@ -30,29 +30,30 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N self._actions = None self.uri = URL.build(scheme="http", host=self._host).join(URI_PIR) + self.token = token async def get_settings(self) -> None: """Get the current settings from the PIR.""" url = URL(self.uri).join(URL("settings")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._settings = response async def get_actions(self) -> None: """Get the current action settings from the PIR.""" url = URL(self.uri).join(URL("action")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._actions = response async def get_pir(self) -> None: """Get the current PIR settings.""" url = URL(self.uri).join(URL("settings/pir")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) 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")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) # The return data has the be re-written as the temperature is not rounded self._sensors = { "motion": response["motion"], @@ -64,7 +65,7 @@ async def get_temperatures(self) -> None: """Get the temperatures from the PIR.""" # There is a different URL for the temp endpoint url = URL.build(scheme="http", host=self._host) / "temp" - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._temperature_raw = response self._temperature_measured = round(response["measured"], 2) self._temperature_compensated = round(response["compensated"], 2) @@ -73,13 +74,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")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) 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")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._intensity = response["intensity"] self._day = response["day"] self._light_raw = response["raw"] diff --git a/pymystrom/switch.py b/pymystrom/switch.py index b156ae5..f89c895 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -9,7 +9,7 @@ class MyStromSwitch: """A class for a myStrom switch/plug.""" - def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None: + def __init__(self, host: str, token='', session: aiohttp.client.ClientSession = None) -> None: """Initialize the switch.""" self._close_session = False self._host = host @@ -20,31 +20,32 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N self._firmware = None self._mac = None self.uri = URL.build(scheme="http", host=self._host) + self.token = token async def turn_on(self) -> None: """Turn the relay on.""" parameters = {"state": "1"} url = URL(self.uri).join(URL("relay")) - await request(self, uri=url, params=parameters) + await request(self, uri=url, params=parameters, token=self.token) await self.get_state() async def turn_off(self) -> None: """Turn the relay off.""" parameters = {"state": "0"} url = URL(self.uri).join(URL("relay")) - await request(self, uri=url, params=parameters) + await request(self, uri=url, params=parameters, token=self.token) await self.get_state() async def toggle(self) -> None: """Toggle the relay.""" url = URL(self.uri).join(URL("toggle")) - await request(self, uri=url) + await request(self, uri=url, token=self.token) await self.get_state() async def get_state(self) -> None: """Get the details from the switch/plug.""" url = URL(self.uri).join(URL("report")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._consumption = response["power"] self._state = response["relay"] try: @@ -53,7 +54,7 @@ async def get_state(self) -> None: self._temperature = None url = URL(self.uri).join(URL("info.json")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self.token) self._firmware = response["version"] self._mac = response["mac"] @@ -88,7 +89,7 @@ def temperature(self) -> float: 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) + response = await request(self, uri=url, token=self.token) return response async def close(self) -> None: From e83d2aba80d548e1196bd9393d8f2a69cda75541 Mon Sep 17 00:00:00 2001 From: Lenny5156 Date: Wed, 19 May 2021 20:11:26 +0200 Subject: [PATCH 2/4] Fix token is now optional string. Fix only send token as header if set. --- pymystrom/__init__.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index 06e6a8b..aae5f3d 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -22,14 +22,15 @@ async def _request( data: Optional[Any] = None, json_data: Optional[dict] = None, params: Optional[Mapping[str, str]] = None, - token: str = '' + token: Optional[str] = None ) -> Any: """Handle a request to the myStrom device.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json, text/plain, */*", - "Token": token } + if token: + headers["Token"] = token if self._session is None: self._session = aiohttp.ClientSession() From 11a0f8b7fff8aa8ae4c0ebfaddca830e41017c9c Mon Sep 17 00:00:00 2001 From: Lenny5156 Date: Wed, 19 May 2021 20:11:26 +0200 Subject: [PATCH 3/4] Fix token is now optional string. Fix only send token as header if set. --- pymystrom/__init__.py | 5 +++-- pymystrom/switch.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index 06e6a8b..aae5f3d 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -22,14 +22,15 @@ async def _request( data: Optional[Any] = None, json_data: Optional[dict] = None, params: Optional[Mapping[str, str]] = None, - token: str = '' + token: Optional[str] = None ) -> Any: """Handle a request to the myStrom device.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json, text/plain, */*", - "Token": token } + if token: + headers["Token"] = token if self._session is None: self._session = aiohttp.ClientSession() diff --git a/pymystrom/switch.py b/pymystrom/switch.py index f89c895..5cfdd73 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -9,7 +9,7 @@ class MyStromSwitch: """A class for a myStrom switch/plug.""" - def __init__(self, host: str, token='', session: aiohttp.client.ClientSession = None) -> None: + def __init__(self, host: str, token: Optional[str] = None, session: aiohttp.client.ClientSession = None) -> None: """Initialize the switch.""" self._close_session = False self._host = host From 4f4c92514f55a5ae59409246fddf917f413edfb7 Mon Sep 17 00:00:00 2001 From: ljo106770 Date: Thu, 16 Sep 2021 13:30:41 +0200 Subject: [PATCH 4/4] fix raise exception if required token not provided --- pymystrom/__init__.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index aae5f3d..0b96c99 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -54,6 +54,11 @@ async def _request( if (response.status // 100) in [4, 5]: response.close() + if (response.status == 404): + raise MyStromConnectionError( + "Error occurred while communicating with myStrom device." + ) + if "application/json" in content_type: response_json = await response.json() return response_json