From 95d0da48753ef13045839739b194e9ea13962404 Mon Sep 17 00:00:00 2001 From: Georg Ostertag Date: Mon, 22 Dec 2025 13:20:15 +0100 Subject: [PATCH 1/2] add optional token functionality --- examples/example-switch-with-token.py | 39 ++++++++++++++ pymystrom/__init__.py | 8 ++- pymystrom/bulb.py | 16 +++--- pymystrom/cli.py | 76 ++++++++++++++++++++------- pymystrom/pir.py | 17 +++--- pymystrom/switch.py | 17 +++--- 6 files changed, 129 insertions(+), 44 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..a3b9fb8 --- /dev/null +++ b/examples/example-switch-with-token.py @@ -0,0 +1,39 @@ +"""Example code for communicating with a myStrom plug/switch.""" + +import asyncio + +from pymystrom.switch import MyStromSwitch + +IP_ADDRESS = "192.168.0.40" +TOKEN = "secret" + + +async def main(): + """Sample code to work with a myStrom switch.""" + async with MyStromSwitch(IP_ADDRESS, token=TOKEN) as switch: + # Collect the data of the current state + await switch.get_state() + + print("Device type:", switch.device_type) + print("Power consumption:", switch.consumption) + print("Energy consumed:", switch.consumedWs) + 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 e81c6a8..ddc965b 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -20,12 +20,15 @@ async def _request( data: Optional[Any] = None, json_data: Optional[dict] = None, params: Optional[Mapping[str, str]] = None, + token: Optional[str] = None, ) -> Any: """Handle a request to the myStrom device.""" headers = { "User-Agent": USER_AGENT, "Accept": "application/json, text/plain, */*", } + if token: + headers["Token"] = token if self._session is None: self._session = aiohttp.ClientSession() @@ -53,7 +56,10 @@ async def _request( ) from exception content_type = response.headers.get("Content-Type", "") - if (response.status // 100) in [4, 5]: + if response.status == 404: + raise MyStromConnectionError("Error occurred while authenticating with myStrom device.") + + elif (response.status // 100) in [4, 5]: response.close() if "application/json" in content_type: diff --git a/pymystrom/bulb.py b/pymystrom/bulb.py index adfa70f..c2de001 100644 --- a/pymystrom/bulb.py +++ b/pymystrom/bulb.py @@ -21,6 +21,7 @@ def __init__( self, host: str, mac: str, + token: Optional[str] = None, session: aiohttp.client.ClientSession = None, ) -> None: """Initialize the bulb.""" @@ -38,10 +39,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) -> None: """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"] @@ -93,7 +95,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 @@ -109,7 +111,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): @@ -122,7 +124,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): @@ -144,7 +146,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): @@ -159,14 +161,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 b76d034..2270fb5 100644 --- a/pymystrom/cli.py +++ b/pymystrom/cli.py @@ -47,11 +47,16 @@ def config(): 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) + headers = {"Token": token} if token else {} + request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device") @@ -71,6 +76,9 @@ def button(): 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", @@ -87,7 +95,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 = { @@ -97,8 +105,9 @@ def write_config(ip, mac, single, double, long, touch): "touch": touch, } try: + headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers ) if request.status_code == 200: @@ -116,6 +125,9 @@ def write_config(ip, mac, single, double, long, touch): 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", @@ -133,7 +145,7 @@ def write_config(ip, mac, single, double, long, touch): 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) @@ -145,8 +157,9 @@ def write_ha_config(ip, mac, hass, port, id): "touch": action.format("touch", hass, port, id), } try: + headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers ) if request.status_code == 200: @@ -170,7 +183,10 @@ def write_ha_config(ip, mac, hass, port, id): 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 = { @@ -180,8 +196,9 @@ def reset_config(ip, mac): "touch": "", } try: + headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT + "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers ) if request.status_code == 200: @@ -201,11 +218,15 @@ def reset_config(ip, 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) + headers = {"Token": token} if token else {} + request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device. No action performed") @@ -222,9 +243,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=ip, mac=mac, token=token) as bulb: await bulb.set_color_hex("00FFFFFF") @@ -234,6 +258,9 @@ async def on(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( "--hue", prompt="Set the hue of the bulb", help="Set the hue of the bulb." ) @@ -247,9 +274,9 @@ async def on(ip, mac): 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=ip, mac=mac, token=token) as bulb: await bulb.set_color_hsv(hue, saturation, value) @@ -259,9 +286,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=ip, mac=mac, token=token) as bulb: await bulb.set_off() @@ -271,15 +301,18 @@ 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=ip, mac=mac, token=token) as bulb: await bulb.set_flashing(time, [100, 50, 30], [200, 0, 71]) @@ -289,15 +322,18 @@ 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=ip, mac=mac, token=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 dc68441..2af2f22 100644 --- a/pymystrom/pir.py +++ b/pymystrom/pir.py @@ -13,10 +13,11 @@ class MyStromPir: """A class for a myStrom PIR.""" - def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> None: + def __init__(self, host: str, session: aiohttp.client.ClientSession = None, token: Optional[str] = None) -> None: """Initialize the switch.""" self._close_session = False self._host = host + self._token = token self._session = session self._intensity = None self._day = None @@ -36,25 +37,25 @@ def __init__(self, host: str, session: aiohttp.client.ClientSession = None) -> N 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"], @@ -66,7 +67,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) @@ -75,13 +76,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 bda3373..808c6c8 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -12,10 +12,11 @@ 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, session: aiohttp.client.ClientSession = None, token: Optional[str] = None) -> None: """Initialize the switch.""" self._close_session = False self._host = host + self._token = token self._session = session self._consumption = 0 self._consumedWs = 0 @@ -33,26 +34,26 @@ 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) try: self._consumption = response["power"] except KeyError: @@ -81,11 +82,11 @@ async def get_state(self) -> None: # Try the new API (Devices with newer firmware) url = URL(self.uri).join(URL("api/v1/info")) - response = await request(self, uri=url) + response = await request(self, uri=url, token=self._token) 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) + response = await request(self, uri=url, token=self._token) # Tolerate missing keys on legacy firmware (e.g., v1 devices) self._firmware = response.get("version") @@ -161,7 +162,7 @@ def temperature(self) -> Optional[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 274e095539d4ae47d178e6b37cd54cdd3b7c3045 Mon Sep 17 00:00:00 2001 From: Georg Ostertag Date: Thu, 25 Dec 2025 14:00:49 +0100 Subject: [PATCH 2/2] streamline formatting --- pymystrom/__init__.py | 6 ++-- pymystrom/bulb.py | 18 +++++++++--- pymystrom/cli.py | 64 +++++++++++++++++++------------------------ pymystrom/pir.py | 7 ++++- pymystrom/switch.py | 7 ++++- 5 files changed, 58 insertions(+), 44 deletions(-) diff --git a/pymystrom/__init__.py b/pymystrom/__init__.py index ddc965b..d0958ba 100644 --- a/pymystrom/__init__.py +++ b/pymystrom/__init__.py @@ -57,8 +57,10 @@ async def _request( content_type = response.headers.get("Content-Type", "") if response.status == 404: - raise MyStromConnectionError("Error occurred while authenticating with myStrom device.") - + raise MyStromConnectionError( + "Error occurred while authenticating with myStrom device." + ) + elif (response.status // 100) in [4, 5]: response.close() diff --git a/pymystrom/bulb.py b/pymystrom/bulb.py index c2de001..8637528 100644 --- a/pymystrom/bulb.py +++ b/pymystrom/bulb.py @@ -111,7 +111,9 @@ async def set_color_hex(self, value): "action": "on", "color": value, } - response = await request(self, uri=self.uri, method="POST", data=data, token=self.token) + 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): @@ -124,7 +126,9 @@ 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, token=self.token) + response = await request( + self, uri=self.uri, method="POST", data=data, token=self.token + ) return response async def set_white(self): @@ -146,7 +150,9 @@ 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, token=self.token) + 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): @@ -161,7 +167,11 @@ 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))}, token=self.token + self, + uri=self.uri, + method="POST", + data={"ramp": int(round(value))}, + token=self.token, ) return response diff --git a/pymystrom/cli.py b/pymystrom/cli.py index 2270fb5..b952303 100644 --- a/pymystrom/cli.py +++ b/pymystrom/cli.py @@ -47,16 +47,15 @@ def config(): prompt="MAC address of the device", help="MAC address of the device.", ) -@click.option( - "--token", prompt="Token of the device", help="Token of the device." -) - +@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: headers = {"Token": token} if token else {} - request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers) + request = requests.get( + "http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers + ) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device") @@ -76,9 +75,7 @@ def button(): 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("--token", prompt="Token of the device", help="Token of the device.") @click.option( "--single", prompt="URL for a single tap", @@ -107,7 +104,10 @@ def write_config(ip, mac, token, single, double, long, touch): try: headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers + "http://{}/{}/{}/".format(ip, URI, mac), + data=data, + timeout=TIMEOUT, + headers=headers, ) if request.status_code == 200: @@ -125,9 +125,7 @@ def write_config(ip, mac, token, single, double, long, touch): 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("--token", prompt="Token of the device", help="Token of the device.") @click.option( "--hass", prompt="IP address of the Home Assistant instance", @@ -159,7 +157,10 @@ def write_ha_config(ip, mac, token, hass, port, id): try: headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers + "http://{}/{}/{}/".format(ip, URI, mac), + data=data, + timeout=TIMEOUT, + headers=headers, ) if request.status_code == 200: @@ -183,9 +184,7 @@ def write_ha_config(ip, mac, token, hass, port, id): prompt="MAC address of the button", help="MAC address of the Wifi Button.", ) -@click.option( - "--token", prompt="Token of the device", help="Token of the device." -) +@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) @@ -198,7 +197,10 @@ def reset_config(ip, mac, token): try: headers = {"Token": token} if token else {} request = requests.post( - "http://{}/{}/{}/".format(ip, URI, mac), data=data, timeout=TIMEOUT, headers=headers + "http://{}/{}/{}/".format(ip, URI, mac), + data=data, + timeout=TIMEOUT, + headers=headers, ) if request.status_code == 200: @@ -218,15 +220,15 @@ def reset_config(ip, mac, token): prompt="MAC address of the button", help="MAC address of the Wifi Button.", ) -@click.option( - "--token", prompt="Token of the device", help="Token of the device." -) +@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: headers = {"Token": token} if token else {} - request = requests.get("http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers) + request = requests.get( + "http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers=headers + ) click.echo(request.json()) except requests.exceptions.ConnectionError: click.echo("Communication issue with the device. No action performed") @@ -243,9 +245,7 @@ def bulb(): @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("--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=ip, mac=mac, token=token) as bulb: @@ -258,9 +258,7 @@ async def on(ip, mac, token): @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("--token", prompt="Token of the device", help="Token of the device.") @click.option( "--hue", prompt="Set the hue of the bulb", help="Set the hue of the bulb." ) @@ -286,9 +284,7 @@ async def color(ip, mac, token, hue, saturation, value): @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("--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=ip, mac=mac, token=token) as bulb: @@ -301,9 +297,7 @@ async def off(ip, mac, token): @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("--token", prompt="Token of the device", help="Token of the device.") @click.option( "--time", prompt="Time to flash", @@ -322,9 +316,7 @@ async def flash(ip, mac, token, 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("--token", prompt="Token of the device", help="Token of the device.") @click.option( "--time", prompt="Time for the complete rainbow", diff --git a/pymystrom/pir.py b/pymystrom/pir.py index 2af2f22..9098df3 100644 --- a/pymystrom/pir.py +++ b/pymystrom/pir.py @@ -13,7 +13,12 @@ class MyStromPir: """A class for a myStrom PIR.""" - def __init__(self, host: str, session: aiohttp.client.ClientSession = None, token: Optional[str] = None) -> None: + def __init__( + self, + host: str, + session: aiohttp.client.ClientSession = None, + token: Optional[str] = None, + ) -> None: """Initialize the switch.""" self._close_session = False self._host = host diff --git a/pymystrom/switch.py b/pymystrom/switch.py index 808c6c8..0cde0e7 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -12,7 +12,12 @@ class MyStromSwitch: """A class for a myStrom switch/plug.""" - def __init__(self, host: str, session: aiohttp.client.ClientSession = None, token: Optional[str] = None) -> None: + def __init__( + self, + host: str, + session: aiohttp.client.ClientSession = None, + token: Optional[str] = None, + ) -> None: """Initialize the switch.""" self._close_session = False self._host = host