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 72bef50..5e0debe 100644 --- a/pymystrom/__init__.py +++ b/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 diff --git a/pymystrom/bulb.py b/pymystrom/bulb.py index 3761891..332c1c2 100644 --- a/pymystrom/bulb.py +++ b/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 diff --git a/pymystrom/cli.py b/pymystrom/cli.py index 53558e3..8c06da1 100644 --- a/pymystrom/cli.py +++ b/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) diff --git a/pymystrom/pir.py b/pymystrom/pir.py index 70c6479..cc9b9b2 100644 --- a/pymystrom/pir.py +++ b/pymystrom/pir.py @@ -12,7 +12,10 @@ class MyStromPir: """A class for a myStrom PIR.""" def __init__( - self, host: str, session: aiohttp.client.ClientSession = None + self, + host: str, + token: Optional[str] = None, + session: aiohttp.client.ClientSession = None ) -> None: """Initialize the switch.""" self._close_session = False @@ -32,29 +35,30 @@ def __init__( 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"], @@ -66,7 +70,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 +79,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 56b6368..2c2c4f3 100644 --- a/pymystrom/switch.py +++ b/pymystrom/switch.py @@ -10,7 +10,10 @@ class MyStromSwitch: """A class for a myStrom switch/plug.""" def __init__( - self, host: str, session: aiohttp.client.ClientSession = None + self, + host: str, + token: Optional[str] = None, + session: aiohttp.client.ClientSession = None ) -> None: """Initialize the switch.""" self._close_session = False @@ -23,31 +26,32 @@ def __init__( 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._consumedWs = response["Ws"] self._state = response["relay"] @@ -57,7 +61,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"] @@ -97,7 +101,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: