Merge pull request #28 from moustic999/dev

Merge dev as new master 0.7.2

README.md

@@ -35,3 +35,45 @@ replace strings with proper values
 
 * run in dir `python3 -m venv .`
 * run `python3 test.py`
+
+# Helper
+Now there is extra command added with this package `bosch_scan`.
+```
+Usage: bosch_scan [OPTIONS] COMMAND [ARGS]...
+
+  A tool to create rawscan of Bosch thermostat.
+
+Options:
+  --ip TEXT                       IP address of gateway  [required]
+  --token TEXT                    Token from sticker without dashes.
+                                  [required]
+  --password TEXT                 Password you set in mobile app.
+  -o, --output TEXT               Path to output file of scan. Default to
+                                  [raw/small]scan_uuid.json
+  --stdout                        Print scan to stdout
+  -d, --debug
+  -s, --smallscan [HC|DHW|SENSORS]
+                                  Scan only single circuit of thermostat.
+  --help                          Show this message and exit.
+
+```
+
+# Examples 
+
+SENSORS:
+```
+bosch_examples sensors --help
+bosch_examples sensors --ip {IP} --token {TOKEN} --password {PASS} -s outdoor_t1
+```
+
+DHW:
+```
+bosch_examples dhw --help
+bosch_examples dhw --ip {IP} --token {TOKEN} --password {PASS} -t --op_modes --setpoints -m
+```
+
+HC:
+```
+bosch_examples hc --help
+bosch_examples hc --ip {IP} --token {TOKEN} --password {PASS} -t --op_modes --setpoints -m
+```

bosch_thermostat_http/__init__.py

@@ -2,5 +2,5 @@
 
 name = "bosch_thermostat_http"
 
-from .errors import *
+from .exceptions import *
 from .gateway import Gateway

bosch_thermostat_http/bosch_examples.py

@@ -0,0 +1,150 @@
+import click
+import logging
+import aiohttp
+import bosch_thermostat_http as bosch
+from bosch_thermostat_http.db import open_json, MAINPATH
+from bosch_thermostat_http.const import HC, SENSORS, DHW
+import os
+import asyncio
+
+from functools import wraps
+
+_LOGGER = logging.getLogger(__name__)
+
+pass_bosch = click.make_pass_decorator(dict, ensure=True)
+
+rc300 = open_json(os.path.join(MAINPATH, "rc300.json"))
+default = open_json(os.path.join(MAINPATH, "default.json"))
+
+sensors_dict = next(iter(default.values()))[SENSORS]
+sensors_dict.update(
+    next(iter(rc300.values()))[SENSORS]
+)
+sensor_list = ' , '.join(sensors_dict.keys())
+
+
+def coro(f):
+    @wraps(f)
+    def wrapper(*args, **kwargs):
+        return asyncio.run(f(*args, **kwargs))
+
+    return wrapper
+
+
+@click.group(invoke_without_command=True)
+@click.pass_context
+@coro
+async def cli(ctx):
+    """A tool to create rawscan of Bosch thermostat."""
+    pass
+
+
+@cli.command()
+@click.option("--ip", envvar="BOSCH_IP", type=str, required=True, help="IP address of gateway")
+@click.option("--token", envvar="BOSCH_ACCESS_TOKEN", type=str, required=True, help="Token from sticker without dashes.")
+@click.option("--password", envvar="BOSCH_PASSWORD", type=str, required=False, help="Password you set in mobile app.")
+@click.option("-d", "--debug", default=False, count=True)
+@click.option('--sensor', '-s', multiple=True, help="You can use multiple sensors. Possible values: %s" % sensor_list)
+@click.pass_context
+@coro
+async def sensors(ctx, ip: str, token: str, password: str, debug: int, sensor):
+    if debug:
+        logging.basicConfig(level=logging.DEBUG)
+        _LOGGER.info("Debug mode active")
+        _LOGGER.debug(f"Lib version is {bosch.version.__version__}")
+    else:
+        logging.basicConfig(level=logging.INFO)
+    async with aiohttp.ClientSession() as session:
+        gateway = bosch.Gateway(
+            session=session, host=ip, access_key=token, password=password
+        )
+        _LOGGER.debug("Trying to connect to gateway.")
+        if await gateway.check_connection():
+            _LOGGER.info("Successfully connected to gateway. Found UUID: %s", gateway.uuid)
+            sensors = gateway.initialize_sensors(list(sensor))
+            for sensor_obj in sensors:
+                await sensor_obj.update()
+                print(sensor_obj.name, ":", sensor_obj.get_property(sensor_obj.attr_id))
+        else:
+            _LOGGER.error("Couldn't connect to gateway!")
+        await session.close()
+
+@cli.command()
+@click.option("--ip", envvar="BOSCH_IP", type=str, required=True, help="IP address of gateway")
+@click.option("--token", envvar="BOSCH_ACCESS_TOKEN", type=str, required=True, help="Token from sticker without dashes.")
+@click.option("--password", envvar="BOSCH_PASSWORD", type=str, required=False, help="Password you set in mobile app.")
+@click.option("-d", "--debug", default=False, count=True)
+@click.option("-t", "--target_temp", default=False, count=True, help="Get target temperature")
+@click.option("-m", "--op_mode", default=False, count=True, help="Print current mode")
+@click.option("--op_modes", default=False, count=True, help="Print available operation modes")
+@click.option("--setpoints", default=False, count=True, help="Print setpoints from schedule")
+@click.pass_context
+@coro
+async def hc(ctx, ip: str, token: str, password: str, debug: int, target_temp: int, op_mode: int, op_modes: int, setpoints: int):
+    if debug:
+        logging.basicConfig(level=logging.DEBUG)
+        _LOGGER.info("Debug mode active")
+        _LOGGER.debug(f"Lib version is {bosch.version.__version__}")
+    else:
+        logging.basicConfig(level=logging.INFO)
+    async with aiohttp.ClientSession() as session:
+        gateway = bosch.Gateway(
+            session=session, host=ip, access_key=token, password=password
+        )
+        _LOGGER.debug("Trying to connect to gateway.")
+        if await gateway.check_connection():
+            _LOGGER.info("Successfully connected to gateway. Found UUID: %s", gateway.uuid)
+            await circuit_fetch(gateway, HC, target_temp, op_mode, op_modes, setpoints)
+        else:
+            _LOGGER.error("Couldn't connect to gateway!")
+        await session.close()
+
+@cli.command()
+@click.option("--ip", envvar="BOSCH_IP", type=str, required=True, help="IP address of gateway")
+@click.option("--token", envvar="BOSCH_ACCESS_TOKEN", type=str, required=True, help="Token from sticker without dashes.")
+@click.option("--password", envvar="BOSCH_PASSWORD", type=str, required=False, help="Password you set in mobile app.")
+@click.option("-d", "--debug", default=False, count=True)
+@click.option("-t", "--target_temp", default=False, count=True, help="Get target temperature")
+@click.option("-m", "--op_mode", default=False, count=True, help="Print current mode")
+@click.option("--op_modes", default=False, count=True, help="Print available operation modes")
+@click.option("--setpoints", default=False, count=True, help="Print setpoints from schedule")
+@click.pass_context
+@coro
+async def dhw(ctx, ip: str, token: str, password: str, debug: int, target_temp: int, op_mode: int, op_modes: int, setpoints: int):
+    if debug:
+        logging.basicConfig(level=logging.DEBUG)
+        _LOGGER.info("Debug mode active")
+        _LOGGER.debug(f"Lib version is {bosch.version.__version__}")
+    else:
+        logging.basicConfig(level=logging.INFO)
+    async with aiohttp.ClientSession() as session:
+        gateway = bosch.Gateway(
+            session=session, host=ip, access_key=token, password=password
+        )
+        _LOGGER.debug("Trying to connect to gateway.")
+        if await gateway.check_connection():
+            _LOGGER.info("Successfully connected to gateway. Found UUID: %s", gateway.uuid)
+            await circuit_fetch(gateway, DHW, target_temp, op_mode, op_modes, setpoints)
+        else:
+            _LOGGER.error("Couldn't connect to gateway!")
+        await session.close()
+
+
+async def circuit_fetch(gateway, circuit_type, target_temp, op_mode, op_modes, setpoints):
+    await gateway.initialize_circuits(circuit_type)
+    circuits = gateway.get_circuits(circuit_type)
+    for circuit in circuits:
+        _LOGGER.debug("Fetching data from circuit.")
+        await circuit.update()
+        if target_temp:
+            print(f"Target temp of {circuit.name} is: {circuit.target_temperature}")
+        if op_mode:
+            print(f"Operation mode of {circuit.name} is: {circuit.current_mode}")
+        if op_modes:
+            print(f"Available operation modes of {circuit.name} are: {circuit.available_operation_modes}")
+        if setpoints:
+            print(f"Available setpoints of {circuit.name} are: {circuit.schedule.setpoints}")
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(cli())

bosch_thermostat_http/bosch_rawscan_cli.py

@@ -0,0 +1,65 @@
+import click
+import logging
+import aiohttp
+import bosch_thermostat_http as bosch
+import json
+import asyncio
+from functools import wraps
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def coro(f):
+    @wraps(f)
+    def wrapper(*args, **kwargs):
+        return asyncio.run(f(*args, **kwargs))
+
+    return wrapper
+
+
+@click.group(invoke_without_command=True)
+@click.option("--ip", envvar="BOSCH_IP", type=str, required=True, help="IP address of gateway")
+@click.option("--token", envvar="BOSCH_ACCESS_TOKEN", type=str, required=True, help="Token from sticker without dashes.")
+@click.option("--password", envvar="BOSCH_PASSWORD", type=str, required=False, help="Password you set in mobile app.")
+@click.option("-o", "--output", type=str, required=False, help="Path to output file of scan. Default to [raw/small]scan_uuid.json")
+@click.option("--stdout", default=False, count=True, help="Print scan to stdout")
+@click.option("-d", "--debug", default=False, count=True)
+@click.option("-s", "--smallscan", type=click.Choice(['HC', 'DHW', 'SENSORS'], case_sensitive=False), help="Scan only single circuit of thermostat.")
+@click.pass_context
+@coro
+async def cli(ctx, ip: str, token: str, password: str, output: str, stdout: int, debug: int, smallscan: str):
+    """A tool to create rawscan of Bosch thermostat."""
+    if debug:
+        logging.basicConfig(level=logging.DEBUG)
+        _LOGGER.info("Debug mode active")
+        _LOGGER.debug(f"Lib version is {bosch.version.__version__}")
+    else:
+        logging.basicConfig(level=logging.INFO)
+    async with aiohttp.ClientSession() as session:
+        _LOGGER.debug("Connecting to %s with token '%s' and password '%s'", ip, token, password)
+        gateway = bosch.Gateway(
+            session=session, host=ip, access_key=token, password=password
+        )
+        _LOGGER.debug("Trying to connect to gateway.")
+        if await gateway.check_connection():
+            _LOGGER.info("Successfully connected to gateway. Found UUID: %s", gateway.uuid)
+            if smallscan:
+                result = await gateway.smallscan(smallscan)
+                out_file = output if output else f"smallscan_{gateway.uuid}.json"
+            else:
+                result = await gateway.rawscan()
+                out_file = output if output else f"rawscan_{gateway.uuid}.json"
+            if stdout:
+                print(json.dumps(result, indent=4))
+            else:
+                with open(out_file, "w") as logfile:
+                    json.dump(result, logfile, indent=4)
+                    _LOGGER.info("Successfully saved result to file: %s", out_file)
+                _LOGGER.debug("Job done.")
+        else:
+            _LOGGER.error("Couldn't connect to gateway!")
+        await session.close()
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(cli())