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,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:

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
 

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)
 

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"]