Added functionality for tokens

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())

pymystrom/__init__.py

@@ -22,12 +22,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()
@@ -56,6 +59,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

pymystrom/bulb.py

@@ -20,6 +20,7 @@ def __init__(
         self,
         host: str,
         mac: str,
+        token: Optional[str] = None,
         session: aiohttp.client.ClientSession = None,
     ):
         """Initialize the bulb."""
@@ -39,10 +40,11 @@ def __init__(
         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"]
@@ -94,7 +96,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
 
@@ -110,7 +112,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):
@@ -123,7 +125,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):
@@ -145,7 +147,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):
@@ -160,14 +162,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
 

pymystrom/cli.py

@@ -46,12 +46,15 @@ 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
+            "http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers={'Token': token}
         )
         click.echo(request.json())
     except requests.exceptions.ConnectionError:
@@ -72,11 +75,15 @@ def button():
     prompt="MAC address of the button",
     help="MAC address of the button.",
 )
+
 @click.option(
     "--single",
     prompt="URL for a single tap",
     default="",
-    help="URL for a single tap.",
+    help="URL for a single tap."
+)
+@click.option(
+    "--token", prompt="Token of the device", help="Token of the device."
 )
 @click.option(
     "--double",
@@ -90,7 +97,7 @@ def button():
 @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 = {
@@ -101,7 +108,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:
@@ -119,6 +126,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",
@@ -136,7 +146,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)
 
@@ -149,7 +159,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:
@@ -173,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 = {
@@ -184,7 +197,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:
@@ -204,12 +217,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
+            "http://{}/{}/{}/".format(ip, URI, mac), timeout=TIMEOUT, headers={'Token': token}
         )
         click.echo(request.json())
     except requests.exceptions.ConnectionError:
@@ -229,9 +245,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")
 
 
@@ -246,6 +265,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",
@@ -256,9 +278,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, mac, token) as bulb:
         await bulb.set_color_hsv(hue, saturation, value)
 
 
@@ -270,9 +292,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()
 
 
@@ -284,15 +309,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, mac, token) as bulb:
         await bulb.set_flashing(time, [100, 50, 30], [200, 0, 71])
 
 
@@ -304,15 +332,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, mac, token) as bulb:
         await bulb.set_rainbow(time)
         await bulb.set_transition_time(1000)