diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..7b55066 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,5 @@ +[run] +source = src +omit = + */__init__.py + */debugger.py \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4efc283..f6057b7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,11 +35,11 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} restore-keys: | - ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ hashFiles('requirements.txt') }}-${{ hashFiles('requirements_test.txt') }}- - ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ hashFiles('requirements.txt') }} + ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ hashFiles('requirements_all.txt') }}-${{ hashFiles('requirements_test_all.txt') }}- + ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ hashFiles('requirements_all.txt') }} ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}- - name: Create Python virtual environment if: steps.cache-venv.outputs.cache-hit != 'true' @@ -47,7 +47,7 @@ jobs: python -m venv venv . venv/bin/activate pip install -U "pip<20.3" setuptools - pip install -r requirements.txt -r requirements_test.txt + pip install -r requirements_all.txt -r requirements_test_all.txt - name: Restore pre-commit environment from cache id: cache-precommit uses: actions/cache@v2.1.4 @@ -83,8 +83,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -127,8 +127,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -171,8 +171,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -218,8 +218,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -265,8 +265,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -312,8 +312,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -356,8 +356,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -403,8 +403,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | @@ -447,8 +447,8 @@ jobs: key: >- ${{ env.CACHE_VERSION}}-${{ runner.os }}-base-venv-${{ steps.python.outputs.python-version }}-${{ - hashFiles('requirements.txt') }}-${{ - hashFiles('requirements_test.txt') }} + hashFiles('requirements_all.txt') }}-${{ + hashFiles('requirements_test_all.txt') }} - name: Fail job if Python cache restore failed if: steps.cache-venv.outputs.cache-hit != 'true' run: | diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index b8e9aed..34c4ece 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -23,7 +23,7 @@ jobs: run: | python -m pip install --upgrade pip python -m pip install flake8 - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + if [ -f requirements_all.txt ]; then pip install -r requirements_all.txt; fi - name: Lint with flake8 run: | # stop the build if there are Python syntax errors. diff --git a/.gitignore b/.gitignore index 80496ef..65ee091 100644 --- a/.gitignore +++ b/.gitignore @@ -10,11 +10,11 @@ __pycache__/ build dist pyhiveapi.egg-info +*.coverage # due to using tox and pytest .tox .cache -test* custom_tests/* # virtual environment folder diff --git a/.vscode/settings.json b/.vscode/settings.json index 7688f82..46987f9 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,5 +1,9 @@ { - "python.testing.pytestArgs": [], + "python.testing.pytestArgs": ["--disable-warnings"], "python.testing.unittestEnabled": false, - "python.testing.pytestEnabled": true + "python.testing.pytestEnabled": true, + "[python]": { + "editor.showUnused": false, + }, + "python.formatting.provider": "black", } \ No newline at end of file diff --git a/.vscode/tasks.json b/.vscode/tasks.json new file mode 100644 index 0000000..4e6f625 --- /dev/null +++ b/.vscode/tasks.json @@ -0,0 +1,115 @@ +{ + "version": "2.0.0", + "tasks": [ + { + "label": "Pytest", + "detail": "Execute pytest tests for pyhiveapi/apyhiveapi", + "type": "shell", + "command": "pytest --timeout=10 tests", + "dependsOn": ["Install all Test Requirements", "Install Package Locally"], + "group": { + "kind": "test", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Flake8", + "detail": "Run flake8 checks on the libary", + "type": "shell", + "command": "pre-commit run flake8 --all-files", + "group": { + "kind": "test", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Pylint", + "detail": "Run pylint checks on the libary", + "type": "shell", + "command": "pylint src", + "dependsOn": ["Install all Requirements"], + "group": { + "kind": "test", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Code Coverage", + "detail": "Generate local code coverage report.", + "type": "shell", + "command": "pytest --cov-report html --cov=apyhiveapi tests/apyhiveapi && pytest --cov-report html --cov=pyhiveapi --cov-append tests/pyhiveapi", + "dependsOn": ["Install all Test Requirements", "Install Package Locally"], + "group": { + "kind": "test", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Install all Requirements", + "detail": "Install all requirements needed to run pyhiveapi/apyihiveapi ", + "type": "shell", + "command": "pip3 install --use-deprecated=legacy-resolver -r requirements_all.txt", + "group": { + "kind": "build", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Install all Test Requirements", + "detail": "Install all test requirement for pytest.", + "type": "shell", + "command": "pip3 install --use-deprecated=legacy-resolver -r requirements_test_all.txt", + "group": { + "kind": "build", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + { + "label": "Install Package Locally", + "detail": "Install pyhiveapi/apyhiveapi locally.", + "type": "shell", + "command": "pip3 install --upgrade --use-deprecated=legacy-resolver .", + "group": { + "kind": "build", + "isDefault": true + }, + "presentation": { + "reveal": "always", + "panel": "new" + }, + "problemMatcher": [] + }, + ], + "inputs": [] + } + \ No newline at end of file diff --git a/MANIFEST.in b/MANIFEST.in index 776f831..d04c8c0 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,4 +1,4 @@ -recursive-include pyhiveapi * -include requirements.txt -include requirements_test.txt +recursive-include src * +include requirements_all.txt +include requirements_test_all.txt recursive-include data * \ No newline at end of file diff --git a/pyhiveapi/apyhiveapi/data/camera.json b/pyhiveapi/apyhiveapi/data/camera.json deleted file mode 100644 index 604ffbe..0000000 --- a/pyhiveapi/apyhiveapi/data/camera.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "cameraImage": { - "parsed": { - "events": [ - { - "thumbnailUrls": [ - "https://test.com/image" - ], - "hasRecording": true - } - ] - } - }, - "camaeraRecording": { - "parsed": "https://test.com/video" - } -} \ No newline at end of file diff --git a/pyhiveapi/apyhiveapi/heating.py b/pyhiveapi/apyhiveapi/heating.py deleted file mode 100644 index 95b67c0..0000000 --- a/pyhiveapi/apyhiveapi/heating.py +++ /dev/null @@ -1,519 +0,0 @@ -"""Hive Heating Module.""" -# pylint: skip-file -from .helper.const import HIVETOHA - - -class HiveHeating: - """Hive Heating Code. - - Returns: - object: heating - """ - - heatingType = "Heating" - - async def getMinTemperature(self, device: dict): - """Get heating minimum target temperature. - - Args: - device (dict): Device to get min temp for. - - Returns: - int: Minimum temperature - """ - if device["hiveType"] == "nathermostat": - return self.session.data.products[device["hiveID"]]["props"]["minHeat"] - return 5 - - async def getMaxTemperature(self, device: dict): - """Get heating maximum target temperature. - - Args: - device (dict): Device to get max temp for. - - Returns: - int: Maximum temperature - """ - if device["hiveType"] == "nathermostat": - return self.session.data.products[device["hiveID"]]["props"]["maxHeat"] - return 32 - - async def getCurrentTemperature(self, device: dict): - """Get heating current temperature. - - Args: - device (dict): Device to get current temperature for. - - Returns: - float: current temperature - """ - from datetime import datetime - - f_state = None - state = None - final = None - - try: - data = self.session.data.products[device["hiveID"]] - state = data["props"]["temperature"] - - if device["hiveID"] in self.session.data.minMax: - if self.session.data.minMax[device["hiveID"]]["TodayDate"] == str( - datetime.date(datetime.now()) - ): - if state < self.session.data.minMax[device["hiveID"]]["TodayMin"]: - self.session.data.minMax[device["hiveID"]]["TodayMin"] = state - - if state > self.session.data.minMax[device["hiveID"]]["TodayMax"]: - self.session.data.minMax[device["hiveID"]]["TodayMax"] = state - else: - data = { - "TodayMin": state, - "TodayMax": state, - "TodayDate": str(datetime.date(datetime.now())), - } - self.session.data.minMax[device["hiveID"]].update(data) - - if state < self.session.data.minMax[device["hiveID"]]["RestartMin"]: - self.session.data.minMax[device["hiveID"]]["RestartMin"] = state - - if state > self.session.data.minMax[device["hiveID"]]["RestartMax"]: - self.session.data.minMax[device["hiveID"]]["RestartMax"] = state - else: - data = { - "TodayMin": state, - "TodayMax": state, - "TodayDate": str(datetime.date(datetime.now())), - "RestartMin": state, - "RestartMax": state, - } - self.session.data.minMax[device["hiveID"]] = data - - f_state = round(float(state), 1) - final = f_state - except KeyError as e: - await self.session.log.error(e) - - return final - - async def getTargetTemperature(self, device: dict): - """Get heating target temperature. - - Args: - device (dict): Device to get target temperature for. - - Returns: - str: Target temperature. - """ - state = None - - try: - data = self.session.data.products[device["hiveID"]] - state = float(data["state"].get("target", None)) - state = float(data["state"].get("heat", state)) - except (KeyError, TypeError) as e: - await self.session.log.error(e) - - return state - - async def getMode(self, device: dict): - """Get heating current mode. - - Args: - device (dict): Device to get current mode for. - - Returns: - str: Current Mode - """ - state = None - final = None - - try: - data = self.session.data.products[device["hiveID"]] - state = data["state"]["mode"] - if state == "BOOST": - state = data["props"]["previous"]["mode"] - final = HIVETOHA[self.heatingType].get(state, state) - except KeyError as e: - await self.session.log.error(e) - - return final - - async def getState(self, device: dict): - """Get heating current state. - - Args: - device (dict): Device to get state for. - - Returns: - str: Current state. - """ - state = None - final = None - - try: - current_temp = await self.getCurrentTemperature(device) - target_temp = await self.getTargetTemperature(device) - if current_temp < target_temp: - state = "ON" - else: - state = "OFF" - final = HIVETOHA[self.heatingType].get(state, state) - except KeyError as e: - await self.session.log.error(e) - - return final - - async def getCurrentOperation(self, device: dict): - """Get heating current operation. - - Args: - device (dict): Device to get current operation for. - - Returns: - str: Current operation. - """ - state = None - - try: - data = self.session.data.products[device["hiveID"]] - state = data["props"]["working"] - except KeyError as e: - await self.session.log.error(e) - - return state - - async def getBoostStatus(self, device: dict): - """Get heating boost current status. - - Args: - device (dict): Device to get boost status for. - - Returns: - str: Boost status. - """ - state = None - - try: - data = self.session.data.products[device["hiveID"]] - state = HIVETOHA["Boost"].get(data["state"].get("boost", False), "ON") - except KeyError as e: - await self.session.log.error(e) - - return state - - async def getBoostTime(self, device: dict): - """Get heating boost time remaining. - - Args: - device (dict): device to get boost time for. - - Returns: - str: Boost time. - """ - if await self.getBoostStatus(device) == "ON": - state = None - - try: - data = self.session.data.products[device["hiveID"]] - state = data["state"]["boost"] - except KeyError as e: - await self.session.log.error(e) - - return state - return None - - async def getHeatOnDemand(self, device): - """Get heat on demand status. - - Args: - device ([dictionary]): [Get Heat on Demand status for Thermostat device.] - - Returns: - str: [Return True or False for the Heat on Demand status.] - """ - state = None - - try: - data = self.session.data.products[device["hiveID"]] - state = data["props"]["autoBoost"]["active"] - except KeyError as e: - await self.session.log.error(e) - - return state - - @staticmethod - async def getOperationModes(): - """Get heating list of possible modes. - - Returns: - list: Operation modes. - """ - return ["SCHEDULE", "MANUAL", "OFF"] - - async def setTargetTemperature(self, device: dict, new_temp: str): - """Set heating target temperature. - - Args: - device (dict): Device to set target temperature for. - new_temp (str): New temperature. - - Returns: - boolean: True/False if successful - """ - await self.session.hiveRefreshTokens() - final = False - - if ( - device["hiveID"] in self.session.data.products - and device["deviceData"]["online"] - ): - await self.session.hiveRefreshTokens() - data = self.session.data.products[device["hiveID"]] - resp = await self.session.api.setState( - data["type"], device["hiveID"], target=new_temp - ) - - if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) - final = True - - return final - - async def setMode(self, device: dict, new_mode: str): - """Set heating mode. - - Args: - device (dict): Device to set mode for. - new_mode (str): New mode to be set. - - Returns: - boolean: True/False if successful - """ - await self.session.hiveRefreshTokens() - final = False - - if ( - device["hiveID"] in self.session.data.products - and device["deviceData"]["online"] - ): - data = self.session.data.products[device["hiveID"]] - resp = await self.session.api.setState( - data["type"], device["hiveID"], mode=new_mode - ) - - if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) - final = True - - return final - - async def setBoostOn(self, device: dict, mins: str, temp: float): - """Turn heating boost on. - - Args: - device (dict): Device to boost. - mins (str): Number of minutes to boost for. - temp (float): Temperature to boost to. - - Returns: - boolean: True/False if successful - """ - if int(mins) > 0 and int(temp) >= await self.getMinTemperature(device): - if int(temp) <= await self.getMaxTemperature(device): - await self.session.hiveRefreshTokens() - final = False - - if ( - device["hiveID"] in self.session.data.products - and device["deviceData"]["online"] - ): - data = self.session.data.products[device["hiveID"]] - resp = await self.session.api.setState( - data["type"], - device["hiveID"], - mode="BOOST", - boost=mins, - target=temp, - ) - - if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) - final = True - - return final - return None - - async def setBoostOff(self, device: dict): - """Turn heating boost off. - - Args: - device (dict): Device to update boost for. - - Returns: - boolean: True/False if successful - """ - final = False - - if ( - device["hiveID"] in self.session.data.products - and device["deviceData"]["online"] - ): - await self.session.hiveRefreshTokens() - data = self.session.data.products[device["hiveID"]] - await self.session.getDevices(device["hiveID"]) - if await self.getBoostStatus(device) == "ON": - prev_mode = data["props"]["previous"]["mode"] - if prev_mode == "MANUAL" or prev_mode == "OFF": - pre_temp = data["props"]["previous"].get("target", 7) - resp = await self.session.api.setState( - data["type"], - device["hiveID"], - mode=prev_mode, - target=pre_temp, - ) - else: - resp = await self.session.api.setState( - data["type"], device["hiveID"], mode=prev_mode - ) - if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) - final = True - - return final - - async def setHeatOnDemand(self, device: dict, state: str): - """Enable or disable Heat on Demand for a Thermostat. - - Args: - device ([dictionary]): [This is the Thermostat device you want to update.] - state ([str]): [This is the state you want to set. (Either "ENABLED" or "DISABLED")] - - Returns: - [boolean]: [Return True or False if the Heat on Demand was set successfully.] - """ - final = False - - if ( - device["hiveID"] in self.session.data.products - and device["deviceData"]["online"] - ): - data = self.session.data.products[device["hiveID"]] - await self.session.hiveRefreshTokens() - resp = await self.session.api.setState( - data["type"], device["hiveID"], autoBoost=state - ) - - if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) - final = True - - return final - - -class Climate(HiveHeating): - """Climate class for Home Assistant. - - Args: - Heating (object): Heating class - """ - - def __init__(self, session: object = None): - """Initialise heating. - - Args: - session (object, optional): Used to interact with hive account. Defaults to None. - """ - self.session = session - - async def getClimate(self, device: dict): - """Get heating data. - - Args: - device (dict): Device to update. - - Returns: - dict: Updated device. - """ - device["deviceData"].update( - {"online": await self.session.attr.onlineOffline(device["device_id"])} - ) - - if device["deviceData"]["online"]: - dev_data = {} - self.session.helper.deviceRecovered(device["device_id"]) - data = self.session.data.devices[device["device_id"]] - dev_data = { - "hiveID": device["hiveID"], - "hiveName": device["hiveName"], - "hiveType": device["hiveType"], - "haName": device["haName"], - "haType": device["haType"], - "device_id": device["device_id"], - "device_name": device["device_name"], - "temperatureunit": device["temperatureunit"], - "min_temp": await self.getMinTemperature(device), - "max_temp": await self.getMaxTemperature(device), - "status": { - "current_temperature": await self.getCurrentTemperature(device), - "target_temperature": await self.getTargetTemperature(device), - "action": await self.getCurrentOperation(device), - "mode": await self.getMode(device), - "boost": await self.getBoostStatus(device), - }, - "deviceData": data.get("props", None), - "parentDevice": data.get("parent", None), - "custom": device.get("custom", None), - "attributes": await self.session.attr.stateAttributes( - device["device_id"], device["hiveType"] - ), - } - self.session.devices.update({device["hiveID"]: dev_data}) - return self.session.devices[device["hiveID"]] - else: - await self.session.log.errorCheck( - device["device_id"], "ERROR", device["deviceData"]["online"] - ) - return device - - async def getScheduleNowNextLater(self, device: dict): - """Hive get heating schedule now, next and later. - - Args: - device (dict): Device to get schedule for. - - Returns: - dict: Schedule now, next and later - """ - online = await self.session.attr.onlineOffline(device["device_id"]) - current_mode = await self.getMode(device) - state = None - - try: - if online and current_mode == "SCHEDULE": - data = self.session.data.products[device["hiveID"]] - state = self.session.helper.getScheduleNNL(data["state"]["schedule"]) - except KeyError as e: - await self.session.log.error(e) - - return state - - async def minmaxTemperature(self, device: dict): - """Min/Max Temp. - - Args: - device (dict): device to get min/max temperature for. - - Returns: - dict: Shows min/max temp for the day. - """ - state = None - final = None - - try: - state = self.session.data.minMax[device["hiveID"]] - final = state - except KeyError as e: - await self.session.log.error(e) - - return final diff --git a/pyhiveapi/apyhiveapi/helper/hive_helper.py b/pyhiveapi/apyhiveapi/helper/hive_helper.py deleted file mode 100644 index 2da30d0..0000000 --- a/pyhiveapi/apyhiveapi/helper/hive_helper.py +++ /dev/null @@ -1,209 +0,0 @@ -"""Helper class for pyhiveapi.""" -# pylint: skip-file -import datetime -import operator - -from .const import HIVE_TYPES - - -class HiveHelper: - """Hive helper class.""" - - def __init__(self, session: object = None): - """Hive Helper. - - Args: - session (object, optional): Interact with hive account. Defaults to None. - """ - self.session = session - - def getDeviceName(self, n_id: str): - """Resolve a id into a name. - - Args: - n_id (str): ID of a device. - - Returns: - str: Name of device. - """ - try: - product_name = self.session.data.products[n_id]["state"]["name"] - except KeyError: - product_name = False - - try: - device_name = self.session.data.devices[n_id]["state"]["name"] - except KeyError: - device_name = False - - if product_name: - return product_name - elif device_name: - return device_name - elif n_id == "No_ID": - return "Hive" - else: - return n_id - - def deviceRecovered(self, n_id: str): - """Register that a device has recovered from being offline. - - Args: - n_id (str): ID of the device. - """ - # name = HiveHelper.getDeviceName(n_id) - if n_id in self.session.config.errorList: - self.session.config.errorList.pop(n_id) - - def getDeviceFromID(self, n_id: str): - """Get product/device data from ID. - - Args: - n_id (str): ID of the device. - - Returns: - dict: Device data. - """ - data = False - try: - data = self.session.devices[n_id] - except KeyError: - pass - - return data - - def getDeviceData(self, product: dict): - """Get device from product data. - - Args: - product (dict): Product data. - - Returns: - [type]: Device data. - """ - device = product - type = product["type"] - if type in ("heating", "hotwater"): - for aDevice in self.session.data.devices: - if self.session.data.devices[aDevice]["type"] in HIVE_TYPES["Thermo"]: - try: - if ( - product["props"]["zone"] - == self.session.data.devices[aDevice]["props"]["zone"] - ): - device = self.session.data.devices[aDevice] - except KeyError: - pass - elif type == "trvcontrol": - device = self.session.data.devices[product["props"]["trvs"][0]] - elif type == "warmwhitelight" and product["props"]["model"] == "SIREN001": - device = self.session.data.devices[product["parent"]] - elif type == "sense": - device = self.session.data.devices[product["parent"]] - else: - device = self.session.data.devices[product["id"]] - - return device - - def convertMinutesToTime(self, minutes_to_convert: str): - """Convert minutes string to datetime. - - Args: - minutes_to_convert (str): minutes in string value. - - Returns: - timedelta: time object of the minutes. - """ - hours_converted, minutes_converted = divmod(minutes_to_convert, 60) - converted_time = datetime.datetime.strptime( - str(hours_converted) + ":" + str(minutes_converted), "%H:%M" - ) - converted_time_string = converted_time.strftime("%H:%M") - return converted_time_string - - def getScheduleNNL(self, hive_api_schedule: list): - """Get the schedule now, next and later of a given nodes schedule. - - Args: - hive_api_schedule (list): Schedule to parse. - - Returns: - dict: Now, Next and later values. - """ - schedule_now_and_next = {} - date_time_now = datetime.datetime.now() - date_time_now_day_int = date_time_now.today().weekday() - - days_t = ( - "monday", - "tuesday", - "wednesday", - "thursday", - "friday", - "saturday", - "sunday", - ) - - days_rolling_list = list(days_t[date_time_now_day_int:] + days_t)[:7] - - full_schedule_list = [] - - for day_index in range(0, len(days_rolling_list)): - current_day_schedule = hive_api_schedule[days_rolling_list[day_index]] - current_day_schedule_sorted = sorted( - current_day_schedule, - key=operator.itemgetter("start"), - reverse=False, - ) - - for current_slot in range(0, len(current_day_schedule_sorted)): - current_slot_custom = current_day_schedule_sorted[current_slot] - - slot_date = datetime.datetime.now() + datetime.timedelta(days=day_index) - slot_time = self.convertMinutesToTime(current_slot_custom["start"]) - slot_time_date_s = slot_date.strftime("%d-%m-%Y") + " " + slot_time - slot_time_date_dt = datetime.datetime.strptime( - slot_time_date_s, "%d-%m-%Y %H:%M" - ) - if slot_time_date_dt <= date_time_now: - slot_time_date_dt = slot_time_date_dt + datetime.timedelta(days=7) - - current_slot_custom["Start_DateTime"] = slot_time_date_dt - full_schedule_list.append(current_slot_custom) - - fsl_sorted = sorted( - full_schedule_list, - key=operator.itemgetter("Start_DateTime"), - reverse=False, - ) - - schedule_now = fsl_sorted[-1] - schedule_next = fsl_sorted[0] - schedule_later = fsl_sorted[1] - - schedule_now["Start_DateTime"] = schedule_now[ - "Start_DateTime" - ] - datetime.timedelta(days=7) - - schedule_now["End_DateTime"] = schedule_next["Start_DateTime"] - schedule_next["End_DateTime"] = schedule_later["Start_DateTime"] - schedule_later["End_DateTime"] = fsl_sorted[2]["Start_DateTime"] - - schedule_now_and_next["now"] = schedule_now - schedule_now_and_next["next"] = schedule_next - schedule_now_and_next["later"] = schedule_later - - return schedule_now_and_next - - def getHeatOnDemandDevice(self, device: dict): - """Use TRV device to get the linked thermostat device. - - Args: - device ([dictionary]): [The TRV device to lookup.] - - Returns: - [dictionary]: [Gets the thermostat device linked to TRV.] - """ - trv = self.session.data.products.get(device["HiveID"]) - thermostat = self.session.data.products.get(trv["state"]["zone"]) - return thermostat diff --git a/pyhiveapi/apyhiveapi/session.py b/pyhiveapi/apyhiveapi/session.py deleted file mode 100644 index 25ac508..0000000 --- a/pyhiveapi/apyhiveapi/session.py +++ /dev/null @@ -1,579 +0,0 @@ -"""Hive Session Module.""" -# pylint: skip-file -import asyncio -import copy -import json -import operator -import os -import time -from datetime import datetime, timedelta - -from aiohttp.web import HTTPException -from apyhiveapi import API, Auth - -from .device_attributes import HiveAttributes -from .helper.const import ACTIONS, DEVICES, HIVE_TYPES, PRODUCTS -from .helper.hive_exceptions import ( - HiveApiError, - HiveFailedToRefreshTokens, - HiveInvalid2FACode, - HiveInvalidDeviceAuthentication, - HiveInvalidPassword, - HiveInvalidUsername, - HiveReauthRequired, - HiveUnknownConfiguration, - NoApiToken, -) -from .helper.hive_helper import HiveHelper -from .helper.logger import Logger -from .helper.map import Map - - -class HiveSession: - """Hive Session Code. - - Raises: - HiveUnknownConfiguration: Unknown configuration. - HTTPException: HTTP error has occurred. - HiveApiError: Hive has retuend an error code. - HiveReauthRequired: Tokens have expired and reauthentiction is required. - - Returns: - object: Session object. - """ - - sessionType = "Session" - - def __init__( - self, - username: str = None, - password: str = None, - websession: object = None, - ): - """Initialise the base variable values. - - Args: - username (str, optional): Hive username. Defaults to None. - password (str, optional): Hive Password. Defaults to None. - websession (object, optional): Websession for api calls. Defaults to None. - """ - self.auth = Auth( - username=username, - password=password, - ) - self.api = API(hiveSession=self, websession=websession) - self.helper = HiveHelper(self) - self.attr = HiveAttributes(self) - self.log = Logger(self) - self.updateLock = asyncio.Lock() - self.tokens = Map( - { - "tokenData": {}, - "tokenCreated": datetime.now() - timedelta(seconds=4000), - "tokenExpiry": timedelta(seconds=3600), - } - ) - self.config = Map( - { - "alarm": False, - "battery": [], - "camera": False, - "errorList": {}, - "file": False, - "homeID": None, - "lastUpdated": datetime.now(), - "mode": [], - "scanInterval": timedelta(seconds=120), - "userID": None, - "username": username, - } - ) - self.data = Map( - { - "products": {}, - "devices": {}, - "actions": {}, - "user": {}, - "minMax": {}, - "alarm": {}, - "camera": {}, - } - ) - self.devices = {} - self.deviceList = {} - - def openFile(self, file: str): - """Open a file. - - Args: - file (str): File location - - Returns: - dict: Data from the chosen file. - """ - path = os.path.dirname(os.path.realpath(__file__)) + "/data/" + file - path = path.replace("/pyhiveapi/", "/apyhiveapi/") - with open(path) as j: - data = json.loads(j.read()) - - return data - - def addList(self, entityType: str, data: dict, **kwargs: dict): - """Add entity to the list. - - Args: - type (str): Type of entity - data (dict): Information to create entity. - - Returns: - dict: Entity. - """ - device = self.helper.getDeviceData(data) - device_name = ( - device["state"]["name"] - if device["state"]["name"] != "Receiver" - else "Heating" - ) - formatted_data = {} - - try: - formatted_data = { - "hiveID": data.get("id", ""), - "hiveName": device_name, - "hiveType": data.get("type", ""), - "haType": entityType, - "deviceData": device.get("props", data.get("props", {})), - "parentDevice": data.get("parent", None), - "isGroup": data.get("isGroup", False), - "device_id": device["id"], - "device_name": device_name, - } - - if kwargs.get("haName", "FALSE")[0] == " ": - kwargs["haName"] = device_name + kwargs["haName"] - else: - formatted_data["haName"] = device_name - formatted_data.update(kwargs) - except KeyError as error: - self.logger.error(error) - - self.deviceList[entityType].append(formatted_data) - return formatted_data - - async def updateInterval(self, new_interval: timedelta): - """Update the scan interval. - - Args: - new_interval (int): New interval for polling. - """ - if isinstance(new_interval, int): - new_interval = timedelta(seconds=new_interval) - - interval = new_interval - if interval < timedelta(seconds=15): - interval = timedelta(seconds=15) - self.config.scanInterval = interval - - async def useFile(self, username: str = None): - """Update to check if file is being used. - - Args: - username (str, optional): Looks for use@file.com. Defaults to None. - """ - using_file = True if username == "use@file.com" else False - if using_file: - self.config.file = True - - async def updateTokens(self, tokens: dict, update_expiry_time: bool = True): - """Update session tokens. - - Args: - tokens (dict): Tokens from API response. - refresh_interval (Boolean): Should the refresh internval be updated - - Returns: - dict: Parsed dictionary of tokens - """ - data = {} - if "AuthenticationResult" in tokens: - data = tokens.get("AuthenticationResult") - self.tokens.tokenData.update({"token": data["IdToken"]}) - if "RefreshToken" in data: - self.tokens.tokenData.update({"refreshToken": data["RefreshToken"]}) - self.tokens.tokenData.update({"accessToken": data["AccessToken"]}) - if update_expiry_time: - self.tokens.tokenCreated = datetime.now() - elif "token" in tokens: - data = tokens - self.tokens.tokenData.update({"token": data["token"]}) - self.tokens.tokenData.update({"refreshToken": data["refreshToken"]}) - self.tokens.tokenData.update({"accessToken": data["accessToken"]}) - - if "ExpiresIn" in data: - self.tokens.tokenExpiry = timedelta(seconds=data["ExpiresIn"]) - - return self.tokens - - async def login(self): - """Login to hive account. - - Raises: - HiveUnknownConfiguration: Login information is unknown. - - Returns: - dict: result of the authentication request. - """ - result = None - if not self.auth: - raise HiveUnknownConfiguration - - try: - result = await self.auth.login() - except HiveInvalidUsername: - print("invalid_username") - except HiveInvalidPassword: - print("invalid_password") - except HiveApiError: - print("no_internet_available") - - if "AuthenticationResult" in result: - await self.updateTokens(result) - return result - - async def sms2fa(self, code, session): - """Login to hive account with 2 factor authentication. - - Raises: - HiveUnknownConfiguration: Login information is unknown. - - Returns: - dict: result of the authentication request. - """ - result = None - if not self.auth: - raise HiveUnknownConfiguration - - try: - result = await self.auth.sms_2fa(code, session) - except HiveInvalid2FACode: - print("invalid_code") - except HiveApiError: - print("no_internet_available") - - if "AuthenticationResult" in result: - await self.updateTokens(result) - return result - - async def deviceLogin(self): - """Login to hive account using device authentication. - - Raises: - HiveUnknownConfiguration: Login information is unknown. - HiveInvalidDeviceAuthentication: Device information is unknown. - - Returns: - dict: result of the authentication request. - """ - result = None - if not self.auth: - raise HiveUnknownConfiguration - - try: - result = await self.auth.device_login() - except HiveInvalidDeviceAuthentication: - raise HiveInvalidDeviceAuthentication - - if "AuthenticationResult" in result: - await self.updateTokens(result) - self.tokens.tokenExpiry = timedelta(seconds=0) - return result - - async def hiveRefreshTokens(self): - """Refresh Hive tokens. - - Returns: - boolean: True/False if update was successful - """ - result = None - - if self.config.file: - return None - else: - expiry_time = self.tokens.tokenCreated + self.tokens.tokenExpiry - if datetime.now() >= expiry_time: - result = await self.auth.refresh_token( - self.tokens.tokenData["refreshToken"] - ) - - if "AuthenticationResult" in result: - await self.updateTokens(result) - else: - raise HiveFailedToRefreshTokens - - return result - - async def updateData(self, device: dict): - """Get latest data for Hive nodes - rate limiting. - - Args: - device (dict): Device requesting the update. - - Returns: - boolean: True/False if update was successful - """ - updated = False - ep = self.config.lastUpdate + self.config.scanInterval - if datetime.now() >= ep and not self.updateLock.locked(): - try: - await self.updateLock.acquire() - await self.getDevices(device["hiveID"]) - if len(self.deviceList["camera"]) > 0: - for camera in self.data.camera: - await self.getCamera(self.devices[camera]) - updated = True - finally: - self.updateLock.release() - - return updated - - async def getAlarm(self): - """Get alarm data. - - Raises: - HTTPException: HTTP error has occurred updating the devices. - HiveApiError: An API error code has been returned. - """ - if self.config.file: - api_resp_d = self.openFile("alarm.json") - elif self.tokens is not None: - api_resp_d = await self.api.getAlarm() - if operator.contains(str(api_resp_d["original"]), "20") is False: - raise HTTPException - elif api_resp_d["parsed"] is None: - raise HiveApiError - - self.data.alarm = api_resp_d["parsed"] - - async def getCamera(self, device): - """Get camera data. - - Raises: - HTTPException: HTTP error has occurred updating the devices. - HiveApiError: An API error code has been returned. - """ - cameraImage = None - cameraRecording = None - hasCameraImage = False - hasCameraRecording = False - - if self.config.file: - cameraImage = self.openFile("camera.json") - cameraRecording = self.openFile("camera.json") - elif self.tokens is not None: - cameraImage = await self.api.getCameraImage(device) - hasCameraRecording = bool( - cameraImage["parsed"]["events"][0]["hasRecording"] - ) - if hasCameraRecording: - cameraRecording = await self.api.getCameraRecording( - device, cameraImage["parsed"]["events"][0]["eventId"] - ) - - if operator.contains(str(cameraImage["original"]), "20") is False: - raise HTTPException - elif cameraImage["parsed"] is None: - raise HiveApiError - else: - raise NoApiToken - - hasCameraImage = bool(cameraImage["parsed"]["events"][0]) - - self.data.camera[device["id"]] = {} - self.data.camera[device["id"]]["cameraImage"] = None - self.data.camera[device["id"]]["cameraRecording"] = None - - if cameraImage is not None and hasCameraImage: - self.data.camera[device["id"]] = {} - self.data.camera[device["id"]]["cameraImage"] = cameraImage["parsed"][ - "events" - ][0] - if cameraRecording is not None and hasCameraRecording: - self.data.camera[device["id"]]["cameraRecording"] = cameraRecording[ - "parsed" - ] - - async def getDevices(self, n_id: str): - """Get latest data for Hive nodes. - - Args: - n_id (str): ID of the device requesting data. - - Raises: - HTTPException: HTTP error has occurred updating the devices. - HiveApiError: An API error code has been returned. - - Returns: - boolean: True/False if update was successful. - """ - get_nodes_successful = False - api_resp_d = None - - try: - if self.config.file: - api_resp_d = self.openFile("data.json") - elif self.tokens is not None: - await self.hiveRefreshTokens() - api_resp_d = await self.api.getAll() - if operator.contains(str(api_resp_d["original"]), "20") is False: - raise HTTPException - elif api_resp_d["parsed"] is None: - raise HiveApiError - - api_resp_p = api_resp_d["parsed"] - tmpProducts = {} - tmpDevices = {} - tmpActions = {} - - for hiveType in api_resp_p: - if hiveType == "user": - self.data.user = api_resp_p[hiveType] - self.config.userID = api_resp_p[hiveType]["id"] - if hiveType == "products": - for aProduct in api_resp_p[hiveType]: - tmpProducts.update({aProduct["id"]: aProduct}) - if hiveType == "devices": - for aDevice in api_resp_p[hiveType]: - tmpDevices.update({aDevice["id"]: aDevice}) - if aDevice["type"] == "siren": - self.config.alarm = True - # if aDevice["type"] == "hivecamera": - # await self.getCamera(aDevice) - if hiveType == "actions": - for aAction in api_resp_p[hiveType]: - tmpActions.update({aAction["id"]: aAction}) - if hiveType == "homes": - self.config.homeID = api_resp_p[hiveType]["homes"][0]["id"] - - if len(tmpProducts) > 0: - self.data.products = copy.deepcopy(tmpProducts) - if len(tmpDevices) > 0: - self.data.devices = copy.deepcopy(tmpDevices) - self.data.actions = copy.deepcopy(tmpActions) - if self.config.alarm: - await self.getAlarm() - self.config.lastUpdate = datetime.now() - get_nodes_successful = True - except (OSError, RuntimeError, HiveApiError, ConnectionError, HTTPException): - get_nodes_successful = False - - return get_nodes_successful - - async def startSession(self, config: dict = {}): - """Setup the Hive platform. - - Args: - config (dict, optional): Configuration for Home Assistant to use. Defaults to {}. - - Raises: - HiveUnknownConfiguration: Unknown configuration identifed. - HiveReauthRequired: Tokens have expired and reauthentication is required. - - Returns: - list: List of devices - """ - await self.useFile(config.get("username", self.config.username)) - await self.updateInterval( - config.get("options", {}).get("scan_interval", self.config.scanInterval) - ) - - if config != {}: - if "tokens" in config and not self.config.file: - await self.updateTokens(config["tokens"], False) - - if "device_data" in config and not self.config.file: - self.auth.device_group_key = config["device_data"][0] - self.auth.device_key = config["device_data"][1] - self.auth.device_password = config["device_data"][2] - - if not self.config.file and "tokens" not in config: - raise HiveUnknownConfiguration - - try: - await self.getDevices("No_ID") - except HTTPException: - return HTTPException - - if self.data.devices == {} or self.data.products == {}: - raise HiveReauthRequired - - return await self.createDevices() - - async def createDevices(self): - """Create list of devices. - - Returns: - list: List of devices - """ - self.deviceList["alarm_control_panel"] = [] - self.deviceList["binary_sensor"] = [] - self.deviceList["camera"] = [] - self.deviceList["climate"] = [] - self.deviceList["light"] = [] - self.deviceList["sensor"] = [] - self.deviceList["switch"] = [] - self.deviceList["water_heater"] = [] - - hive_type = HIVE_TYPES["Heating"] + HIVE_TYPES["Switch"] + HIVE_TYPES["Light"] - for aProduct in self.data.products: - p = self.data.products[aProduct] - if "error" in p: - continue - # Only consider single items or heating groups - if ( - p.get("isGroup", False) - and self.data.products[aProduct]["type"] not in HIVE_TYPES["Heating"] - ): - continue - product_list = PRODUCTS.get(self.data.products[aProduct]["type"], []) - for code in product_list: - eval("self." + code) - - if self.data.products[aProduct]["type"] in hive_type: - self.config.mode.append(p["id"]) - - hive_type = HIVE_TYPES["Thermo"] + HIVE_TYPES["Sensor"] - for aDevice in self.data["devices"]: - d = self.data.devices[aDevice] - device_list = DEVICES.get(self.data.devices[aDevice]["type"], []) - for code in device_list: - eval("self." + code) - - if self.data["devices"][aDevice]["type"] in hive_type: - self.config.battery.append(d["id"]) - - if "action" in HIVE_TYPES["Switch"]: - for action in self.data["actions"]: - a = self.data["actions"][action] # noqa: F841 - eval("self." + ACTIONS) - - return self.deviceList - - @staticmethod - def epochTime(date_time: any, pattern: str, action: str): - """date/time conversion to epoch. - - Args: - date_time (any): epoch time or date and time to use. - pattern (str): Pattern for converting to epoch. - action (str): Convert from/to. - - Returns: - any: Converted time. - """ - if action == "to_epoch": - pattern = "%d.%m.%Y %H:%M:%S" - epochtime = int(time.mktime(time.strptime(str(date_time), pattern))) - return epochtime - elif action == "from_epoch": - date = datetime.fromtimestamp(int(date_time)).strftime(pattern) - return date diff --git a/requirements.txt b/requirements_all.txt similarity index 100% rename from requirements.txt rename to requirements_all.txt diff --git a/requirements_test.txt b/requirements_test.txt deleted file mode 100644 index b654c22..0000000 --- a/requirements_test.txt +++ /dev/null @@ -1 +0,0 @@ -tox \ No newline at end of file diff --git a/requirements_test_all.txt b/requirements_test_all.txt new file mode 100644 index 0000000..5f45c9e --- /dev/null +++ b/requirements_test_all.txt @@ -0,0 +1,4 @@ +pyhiveapi +pytest +pytest-cov +tox \ No newline at end of file diff --git a/setup.cfg b/setup.cfg index c81ed27..11dbf31 100644 --- a/setup.cfg +++ b/setup.cfg @@ -3,8 +3,8 @@ name = pyhiveapi description = A Python library to interface with the Hive API keywords = Hive API Library license = MIT -author = Rendili -author_email = rendili@outlook.com +author = KJonline +author_email = kjtech6@outlook.com url = https://github.com/Pyhive/pyhiveapi project_urls = Source = https://github.com/Pyhive/Pyhiveapi @@ -20,13 +20,13 @@ classifiers = [options] -package_dir = - = pyhiveapi +package_dir = + = src packages = find: python_requires = >=3.8.* [options.packages.find] -where = pyhiveapi +where = src [build-system] requires = ["setuptools>=40.6.2", "wheel", "unasync"] @@ -67,3 +67,9 @@ warn_incomplete_stub = true warn_redundant_casts = true warn_unused_configs = true +[tool:pytest] +minversion = 6.0 +addopts = -ra -q +testpaths = + tests/pyhiveapi + tests/apyhiveapi \ No newline at end of file diff --git a/setup.py b/setup.py index 941e974..b7cd88a 100644 --- a/setup.py +++ b/setup.py @@ -7,7 +7,7 @@ from setuptools import setup -def requirements_from_file(filename="requirements.txt"): +def requirements_from_file(filename="requirements_all.txt"): """Get requirements from file.""" with open(os.path.join(os.path.dirname(__file__), filename)) as r: reqs = r.read().strip().split("\n") @@ -33,5 +33,5 @@ def requirements_from_file(filename="requirements.txt"): ) }, install_requires=requirements_from_file(), - extras_require={"dev": requirements_from_file("requirements_test.txt")}, + extras_require={"dev": requirements_from_file("requirements_test_all.txt")}, ) diff --git a/pyhiveapi/apyhiveapi/__init__.py b/src/apyhiveapi/__init__.py similarity index 67% rename from pyhiveapi/apyhiveapi/__init__.py rename to src/apyhiveapi/__init__.py index 7436221..926049b 100644 --- a/pyhiveapi/apyhiveapi/__init__.py +++ b/src/apyhiveapi/__init__.py @@ -1,11 +1,18 @@ """__init__.py.""" # pylint: skip-file +import os + if __name__ == "pyhiveapi": from .api.hive_api import HiveApi as API # noqa: F401 from .api.hive_auth import HiveAuth as Auth # noqa: F401 + + PATH = os.path.dirname(os.path.realpath(__file__)) + "/test_data/" + PATH = PATH.replace("/pyhiveapi/", "/apyhiveapi/") else: from .api.hive_async_api import HiveApiAsync as API # noqa: F401 from .api.hive_auth_async import HiveAuthAsync as Auth # noqa: F401 + PATH = os.path.dirname(os.path.realpath(__file__)) + "/test_data/" + from .helper.const import SMS_REQUIRED # noqa: F401 from .hive import Hive # noqa: F401 diff --git a/pyhiveapi/apyhiveapi/action.py b/src/apyhiveapi/action.py similarity index 93% rename from pyhiveapi/apyhiveapi/action.py rename to src/apyhiveapi/action.py index c4af98f..4343e60 100644 --- a/pyhiveapi/apyhiveapi/action.py +++ b/src/apyhiveapi/action.py @@ -84,14 +84,14 @@ async def setStatusOn(self, device: dict): final = False if device["hiveID"] in self.session.data.actions: - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.actions[device["hiveID"]] data.update({"enabled": True}) send = json.dumps(data) resp = await self.session.api.setAction(device["hiveID"], send) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -109,13 +109,13 @@ async def setStatusOff(self, device: dict): final = False if device["hiveID"] in self.session.data.actions: - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.actions[device["hiveID"]] data.update({"enabled": False}) send = json.dumps(data) resp = await self.session.api.setAction(device["hiveID"], send) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final diff --git a/pyhiveapi/apyhiveapi/alarm.py b/src/apyhiveapi/alarm.py similarity index 98% rename from pyhiveapi/apyhiveapi/alarm.py rename to src/apyhiveapi/alarm.py index d85fdfb..4250542 100644 --- a/pyhiveapi/apyhiveapi/alarm.py +++ b/src/apyhiveapi/alarm.py @@ -58,7 +58,7 @@ async def setMode(self, device: dict, mode: str): device["hiveID"] in self.session.data.devices and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() resp = await self.session.api.setAlarm(mode=mode) if resp["original"] == 200: final = True diff --git a/pyhiveapi/apyhiveapi/api/__init__.py b/src/apyhiveapi/api/__init__.py similarity index 100% rename from pyhiveapi/apyhiveapi/api/__init__.py rename to src/apyhiveapi/api/__init__.py diff --git a/pyhiveapi/apyhiveapi/api/hive_api.py b/src/apyhiveapi/api/hive_api.py similarity index 98% rename from pyhiveapi/apyhiveapi/api/hive_api.py rename to src/apyhiveapi/api/hive_api.py index c81e282..9151f26 100644 --- a/pyhiveapi/apyhiveapi/api/hive_api.py +++ b/src/apyhiveapi/api/hive_api.py @@ -96,7 +96,7 @@ def refreshTokens(self, tokens={}): info = self.request("POST", url, jsc) data = json.loads(info.text) if "token" in data and self.session: - self.session.updateTokens(data) + self.session.update_tokens(data) self.urls.update({"base": data["platform"]["endpoint"]}) self.urls.update({"camera": data["platform"]["cameraPlatform"]}) self.json_return.update({"original": info.status_code}) @@ -142,11 +142,11 @@ def getAll(self): return json_return - def getAlarm(self, homeID=None): + def getAlarm(self, home_id=None): """Build and query alarm endpoint.""" if self.session is not None: - homeID = self.session.config.homeID - url = self.urls["base"] + self.urls["alarm"] + homeID + home_id = self.session.config.home_id + url = self.urls["base"] + self.urls["alarm"] + home_id try: info = self.request("GET", url) self.json_return.update({"original": info.status_code}) diff --git a/pyhiveapi/apyhiveapi/api/hive_async_api.py b/src/apyhiveapi/api/hive_async_api.py similarity index 98% rename from pyhiveapi/apyhiveapi/api/hive_async_api.py rename to src/apyhiveapi/api/hive_async_api.py index e38e72e..f894e84 100644 --- a/pyhiveapi/apyhiveapi/api/hive_async_api.py +++ b/src/apyhiveapi/api/hive_async_api.py @@ -130,7 +130,7 @@ async def refreshTokens(self): if self.json_return["original"] == 200: info = self.json_return["parsed"] if "token" in info: - await self.session.updateTokens(info) + await self.session.update_tokens(info) self.baseUrl = info["platform"]["endpoint"] self.cameraBaseUrl = info["platform"]["cameraPlatform"] return True @@ -155,7 +155,7 @@ async def getAll(self): async def getAlarm(self): """Build and query alarm endpoint.""" json_return = {} - url = self.urls["alarm"] + self.session.config.homeID + url = self.urls["alarm"] + self.session.config.home_id try: resp = await self.request("get", url) json_return.update({"original": resp.status}) @@ -307,7 +307,7 @@ async def setAlarm(self, **kwargs): + "}" ) - url = f"{self.urls['alarm']}{self.session.config.homeID}" + url = f"{self.urls['alarm']}{self.session.config.home_id}" try: await self.isFileBeingUsed() resp = await self.request("post", url, data=jsc) diff --git a/pyhiveapi/apyhiveapi/api/hive_auth.py b/src/apyhiveapi/api/hive_auth.py similarity index 100% rename from pyhiveapi/apyhiveapi/api/hive_auth.py rename to src/apyhiveapi/api/hive_auth.py diff --git a/pyhiveapi/apyhiveapi/api/hive_auth_async.py b/src/apyhiveapi/api/hive_auth_async.py similarity index 100% rename from pyhiveapi/apyhiveapi/api/hive_auth_async.py rename to src/apyhiveapi/api/hive_auth_async.py diff --git a/pyhiveapi/apyhiveapi/camera.py b/src/apyhiveapi/camera.py similarity index 96% rename from pyhiveapi/apyhiveapi/camera.py rename to src/apyhiveapi/camera.py index 7ccd006..1c9a8be 100644 --- a/pyhiveapi/apyhiveapi/camera.py +++ b/src/apyhiveapi/camera.py @@ -90,11 +90,11 @@ async def setCameraOn(self, device: dict, mode: str): device["hiveID"] in self.session.data.devices and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() resp = await self.session.api.setState(mode=mode) if resp["original"] == 200: final = True - await self.session.getCamera() + await self.session.get_camera() return final @@ -113,11 +113,11 @@ async def setCameraOff(self, device: dict, mode: str): device["hiveID"] in self.session.data.devices and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() resp = await self.session.api.setState(mode=mode) if resp["original"] == 200: final = True - await self.session.getCamera() + await self.session.get_camera() return final diff --git a/pyhiveapi/apyhiveapi/device_attributes.py b/src/apyhiveapi/device_attributes.py similarity index 100% rename from pyhiveapi/apyhiveapi/device_attributes.py rename to src/apyhiveapi/device_attributes.py diff --git a/src/apyhiveapi/heating.py b/src/apyhiveapi/heating.py new file mode 100644 index 0000000..e69de29 diff --git a/pyhiveapi/apyhiveapi/helper/__init__.py b/src/apyhiveapi/helper/__init__.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/__init__.py rename to src/apyhiveapi/helper/__init__.py diff --git a/src/apyhiveapi/helper/const.py b/src/apyhiveapi/helper/const.py new file mode 100644 index 0000000..994d6b0 --- /dev/null +++ b/src/apyhiveapi/helper/const.py @@ -0,0 +1,51 @@ +"""Constants for Pyhiveapi.""" +SMS_REQUIRED = "SMS_MFA" + +# HTTP return codes. +HTTP_OK = 200 +HTTP_CREATED = 201 +HTTP_ACCEPTED = 202 +HTTP_MOVED_PERMANENTLY = 301 +HTTP_BAD_REQUEST = 400 +HTTP_UNAUTHORIZED = 401 +HTTP_FORBIDDEN = 403 +HTTP_NOT_FOUND = 404 +HTTP_METHOD_NOT_ALLOWED = 405 +HTTP_UNPROCESSABLE_ENTITY = 422 +HTTP_TOO_MANY_REQUESTS = 429 +HTTP_INTERNAL_SERVER_ERROR = 500 +HTTP_BAD_GATEWAY = 502 +HTTP_SERVICE_UNAVAILABLE = 503 + + +HIVETOHA = { + "Alarm": {"home": "armed_home", "away": "armed_away", "asleep": "armed_night"}, + "Attribute": {True: "Online", False: "Offline"}, + "Boost": {None: "OFF", False: "OFF"}, + "Heating": {False: "OFF", "ENABLED": True, "DISABLED": False}, + "Hotwater": {"MANUAL": "ON", None: "OFF", False: "OFF"}, + "Hub": { + "Status": {True: 1, False: 0}, + "Smoke": {True: 1, False: 0}, + "Dog": {True: 1, False: 0}, + "Glass": {True: 1, False: 0}, + }, + "Light": {"ON": True, "OFF": False}, + "Sensor": { + "OPEN": True, + "CLOSED": False, + True: "Online", + False: "Offline", + }, + "Switch": {"ON": True, "OFF": False}, +} + +HIVE_TYPES = { + "Hub": ["hub", "sense"], + "Thermo": ["thermostatui", "trv"], + "Heating": ["heating", "trvcontrol"], + "Hotwater": ["hotwater"], + "Light": ["warmwhitelight", "tuneablelight", "colourtuneablelight"], + "Sensor": ["motionsensor", "contactsensor"], + "Switch": ["activeplug"], +} diff --git a/pyhiveapi/apyhiveapi/helper/debugger.py b/src/apyhiveapi/helper/debugger.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/debugger.py rename to src/apyhiveapi/helper/debugger.py diff --git a/pyhiveapi/apyhiveapi/helper/hive_exceptions.py b/src/apyhiveapi/helper/hive_exceptions.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/hive_exceptions.py rename to src/apyhiveapi/helper/hive_exceptions.py diff --git a/src/apyhiveapi/helper/hive_helper.py b/src/apyhiveapi/helper/hive_helper.py new file mode 100644 index 0000000..c007615 --- /dev/null +++ b/src/apyhiveapi/helper/hive_helper.py @@ -0,0 +1,564 @@ +"""Helper class for pyhiveapi.""" +# pylint: skip-file +import datetime +import operator + +from .const import HIVE_TYPES + + +class HiveHelper: + """Hive helper class.""" + + def __init__(self, session: object = None): + """Hive Helper. + + Args: + session (object, optional): Interact with hive account. Defaults to None. + """ + self.session = session + + def getDeviceName(self, n_id: str): + """Resolve a id into a name. + + Args: + n_id (str): ID of a device. + + Returns: + str: Name of device. + """ + try: + product_name = self.session.data.products[n_id]["state"]["name"] + except KeyError: + product_name = False + + try: + device_name = self.session.data.devices[n_id]["state"]["name"] + except KeyError: + device_name = False + + if product_name: + return product_name + elif device_name: + return device_name + elif n_id == "No_ID": + return "Hive" + else: + return n_id + + def deviceRecovered(self, n_id: str): + """Register that a device has recovered from being offline. + + Args: + n_id (str): ID of the device. + """ + # name = HiveHelper.getDeviceName(n_id) + if n_id in self.session.config.errorList: + self.session.config.errorList.pop(n_id) + + def getDeviceFromID(self, n_id: str): + """Get product/device data from ID. + + Args: + n_id (str): ID of the device. + + Returns: + dict: Device data. + """ + data = False + try: + data = self.session.devices[n_id] + except KeyError: + pass + + return data + + def getDeviceData(self, product: dict): + """Get device from product data. + + Args: + product (dict): Product data. + + Returns: + [type]: Device data. + """ + device = product + type = product["type"] + if type in ("heating", "hotwater"): + for aDevice in self.session.data.devices: + if self.session.data.devices[aDevice]["type"] in HIVE_TYPES["Thermo"]: + try: + if ( + product["props"]["zone"] + == self.session.data.devices[aDevice]["props"]["zone"] + ): + device = self.session.data.devices[aDevice] + except KeyError: + pass + elif type == "trvcontrol": + device = self.session.data.devices[product["props"]["trvs"][0]] + elif type == "warmwhitelight" and product["props"]["model"] == "SIREN001": + device = self.session.data.devices[product["parent"]] + elif type == "sense": + device = self.session.data.devices[product["parent"]] + else: + device = self.session.data.devices[product["id"]] + + return device + + def convertMinutesToTime(self, minutes_to_convert: str): + """Convert minutes string to datetime. + + Args: + minutes_to_convert (str): minutes in string value. + + Returns: + timedelta: time object of the minutes. + """ + hours_converted, minutes_converted = divmod(minutes_to_convert, 60) + converted_time = datetime.datetime.strptime( + str(hours_converted) + ":" + str(minutes_converted), "%H:%M" + ) + converted_time_string = converted_time.strftime("%H:%M") + return converted_time_string + + def getScheduleNNL(self, hive_api_schedule: list): + """Get the schedule now, next and later of a given nodes schedule. + + Args: + hive_api_schedule (list): Schedule to parse. + + Returns: + dict: Now, Next and later values. + """ + schedule_now_and_next = {} + date_time_now = datetime.datetime.now() + date_time_now_day_int = date_time_now.today().weekday() + + days_t = ( + "monday", + "tuesday", + "wednesday", + "thursday", + "friday", + "saturday", + "sunday", + ) + + days_rolling_list = list(days_t[date_time_now_day_int:] + days_t)[:7] + + full_schedule_list = [] + + for day_index in range(0, len(days_rolling_list)): + current_day_schedule = hive_api_schedule[days_rolling_list[day_index]] + current_day_schedule_sorted = sorted( + current_day_schedule, + key=operator.itemgetter("start"), + reverse=False, + ) + + for current_slot in range(0, len(current_day_schedule_sorted)): + current_slot_custom = current_day_schedule_sorted[current_slot] + + slot_date = datetime.datetime.now() + datetime.timedelta(days=day_index) + slot_time = self.convertMinutesToTime(current_slot_custom["start"]) + slot_time_date_s = slot_date.strftime("%d-%m-%Y") + " " + slot_time + slot_time_date_dt = datetime.datetime.strptime( + slot_time_date_s, "%d-%m-%Y %H:%M" + ) + if slot_time_date_dt <= date_time_now: + slot_time_date_dt = slot_time_date_dt + datetime.timedelta(days=7) + + current_slot_custom["Start_DateTime"] = slot_time_date_dt + full_schedule_list.append(current_slot_custom) + + fsl_sorted = sorted( + full_schedule_list, + key=operator.itemgetter("Start_DateTime"), + reverse=False, + ) + + schedule_now = fsl_sorted[-1] + schedule_next = fsl_sorted[0] + schedule_later = fsl_sorted[1] + + schedule_now["Start_DateTime"] = schedule_now[ + "Start_DateTime" + ] - datetime.timedelta(days=7) + + schedule_now["End_DateTime"] = schedule_next["Start_DateTime"] + schedule_next["End_DateTime"] = schedule_later["Start_DateTime"] + schedule_later["End_DateTime"] = fsl_sorted[2]["Start_DateTime"] + + schedule_now_and_next["now"] = schedule_now + schedule_now_and_next["next"] = schedule_next + schedule_now_and_next["later"] = schedule_later + + return schedule_now_and_next + + def getHeatOnDemandDevice(self, device: dict): + """Use TRV device to get the linked thermostat device. + + Args: + device ([dictionary]): [The TRV device to lookup.] + + Returns: + [dictionary]: [Gets the thermostat device linked to TRV.] + """ + trv = self.session.data.products.get(device["HiveID"]) + thermostat = self.session.data.products.get(trv["state"]["zone"]) + return thermostat + + async def call_sensor_function(self, device): + """Helper to decide which function to call.""" + if device["hiveType"] == "SMOKE_CO": + return await self.session.hub.getSmokeStatus(device) + if device["hiveType"] == "DOG_BARK": + return await self.session.hub.getDogBarkStatus(device) + if device["hiveType"] == "GLASS_BREAK": + return await self.session.hub.getGlassBreakStatus(device) + if device["hiveType"] == "Camera_Temp": + return await self.session.camera.getCameraTemperature(device) + if device["hiveType"] == "Heating_Current_Temperature": + return await self.session.heating.getCurrentTemperature(device) + if device["hiveType"] == "Heating_Target_Temperature": + return await self.session.heating.getTargetTemperature(device) + if device["hiveType"] == "Heating_State": + return await self.session.heating.getState(device) + if device["hiveType"] in ("Heating_Mode", "Hotwater_Mode", "Mode"): + return await self.session.heating.getMode(device) + if device["hiveType"] == "Heating_Boost": + return await self.session.heating.getBoostStatus(device) + if device["hiveType"] == "Hotwater_State": + return await self.session.hotwater.getState(device) + if device["hiveType"] == "Hotwater_Boost": + return await self.session.hotwater.getBoost(device) + if device["hiveType"] == "Battery": + return await self.session.attr.getBattery(device["device_id"]) + if device["hiveType"] in ("Availability", "Connectivity"): + return await self.online(device) + if device["hiveType"] == "Power": + return await self.session.switch.getPowerUsage(device) + return None + + async def call_products_to_add(self, entity_type, product): + """Helper to add a product to the list.""" + if entity_type == "sense": + self.session.add_list( + "binary_sensor", + product, + haName="Glass Detection", + hiveType="GLASS_BREAK", + ) + self.session.add_list( + "binary_sensor", product, haName="Smoke Detection", hiveType="SMOKE_CO" + ) + self.session.add_list( + "binary_sensor", + product, + haName="Dog Bark Detection", + hiveType="DOG_BARK", + ) + if entity_type == "heating": + self.session.add_list( + "climate", + product, + temperatureunit=self.session.data["user"]["temperatureUnit"], + ) + self.session.add_list( + "switch", + product, + haName=" Heat on Demand", + hiveType="Heating_Heat_On_Demand", + category="config", + ) + self.session.add_list( + "sensor", + product, + haName=" Current Temperature", + hiveType="Heating_Current_Temperature", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Target Temperature", + hiveType="Heating_Target_Temperature", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" State", + hiveType="Heating_State", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Heating_Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Boost", + hiveType="Heating_Boost", + category="diagnostic", + ) + if entity_type == "trvcontrol": + self.session.add_list( + "climate", + product, + temperatureunit=self.session.data["user"]["temperatureUnit"], + ) + self.session.add_list( + "sensor", + product, + haName=" Current Temperature", + hiveType="Heating_Current_Temperature", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Target Temperature", + hiveType="Heating_Target_Temperature", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" State", + hiveType="Heating_State", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Heating_Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Boost", + hiveType="Heating_Boost", + category="diagnostic", + ) + if entity_type == "hotwater": + self.session.add_list( + "water_heater", + product, + ) + self.session.add_list( + "sensor", + product, + haName="Hotwater State", + hiveType="Hotwater_State", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName="Hotwater Mode", + hiveType="Hotwater_Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName="Hotwater Boost", + hiveType="Hotwater_Boost", + category="diagnostic", + ) + if entity_type == "activeplug": + self.session.add_list("switch", product) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Power", + hiveType="Power", + category="diagnostic", + ) + if entity_type == "warmwhitelight": + self.session.add_list("light", product) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "tuneablelight": + self.session.add_list("light", product) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "colourtuneablelight": + self.session.add_list("light", product) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "hivecamera": + self.session.add_list("camera", product) + self.session.add_list( + "sensor", + product, + haName=" Mode", + hiveType="Mode", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + self.session.add_list( + "sensor", + product, + haName=" Temperature", + hiveType="Camera_Temp", + category="diagnostic", + ) + if entity_type == "motionsensor": + self.session.add_list("binary_sensor", product) + if entity_type == "contactsensor": + self.session.add_list("binary_sensor", product) + return None + + async def call_devices_to_add(self, entity_type, device): + """Helper to add a device to the list.""" + if entity_type == "contactsensor": + self.session.add_list( + "sensor", + device, + haName=" Battery Level", + hiveType="Battery", + category="diagnostic", + ) + self.session.add_list( + "sensor", + device, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "hub": + self.session.add_list( + "binary_sensor", + device, + haName="Hive Hub Status", + hiveType="Connectivity", + category="diagnostic", + ) + if entity_type == "motionsensor": + self.session.add_list( + "sensor", + device, + haName=" Battery Level", + hiveType="Battery", + category="diagnostic", + ) + self.session.add_list( + "sensor", + device, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "sense": + self.session.add_list( + "binary_sensor", + device, + haName="Hive Hub Status", + hiveType="Connectivity", + ) + if entity_type == "siren": + self.session.add_list("alarm_control_panel", device) + if entity_type == "thermostatui": + self.session.add_list( + "sensor", + device, + haName=" Battery Level", + hiveType="Battery", + category="diagnostic", + ) + self.session.add_list( + "sensor", + device, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + if entity_type == "trv": + self.session.add_list( + "sensor", + device, + haName=" Battery Level", + hiveType="Battery", + category="diagnostic", + ) + self.session.add_list( + "sensor", + device, + haName=" Availability", + hiveType="Availability", + category="diagnostic", + ) + + async def call_action_to_add(self, action): + """Helper to add an action to the list.""" + await self.session.add_list( + "switch", + action, + hiveName=action["name"], + haName=action["name"], + hiveType="action", + ) diff --git a/pyhiveapi/apyhiveapi/helper/hivedataclasses.py b/src/apyhiveapi/helper/hivedataclasses.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/hivedataclasses.py rename to src/apyhiveapi/helper/hivedataclasses.py diff --git a/pyhiveapi/apyhiveapi/helper/logger.py b/src/apyhiveapi/helper/logger.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/logger.py rename to src/apyhiveapi/helper/logger.py diff --git a/pyhiveapi/apyhiveapi/helper/map.py b/src/apyhiveapi/helper/map.py similarity index 100% rename from pyhiveapi/apyhiveapi/helper/map.py rename to src/apyhiveapi/helper/map.py diff --git a/pyhiveapi/apyhiveapi/hive.py b/src/apyhiveapi/hive.py similarity index 100% rename from pyhiveapi/apyhiveapi/hive.py rename to src/apyhiveapi/hive.py diff --git a/pyhiveapi/apyhiveapi/hotwater.py b/src/apyhiveapi/hotwater.py similarity index 92% rename from pyhiveapi/apyhiveapi/hotwater.py rename to src/apyhiveapi/hotwater.py index ae83d82..48ce873 100644 --- a/pyhiveapi/apyhiveapi/hotwater.py +++ b/src/apyhiveapi/hotwater.py @@ -127,14 +127,14 @@ async def setMode(self, device: dict, new_mode: str): final = False if device["hiveID"] in self.session.data.products: - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], device["hiveID"], mode=new_mode ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -155,14 +155,14 @@ async def setBoostOn(self, device: dict, mins: int): and device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], device["hiveID"], mode="BOOST", boost=mins ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -182,14 +182,14 @@ async def setBoostOff(self, device: dict): and await self.getBoost(device) == "ON" and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] prev_mode = data["props"]["previous"]["mode"] resp = await self.session.api.setState( data["type"], device["hiveID"], mode=prev_mode ) if resp["original"] == 200: - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() final = True return final diff --git a/pyhiveapi/apyhiveapi/hub.py b/src/apyhiveapi/hub.py similarity index 100% rename from pyhiveapi/apyhiveapi/hub.py rename to src/apyhiveapi/hub.py diff --git a/pyhiveapi/apyhiveapi/light.py b/src/apyhiveapi/light.py similarity index 95% rename from pyhiveapi/apyhiveapi/light.py rename to src/apyhiveapi/light.py index 703abe7..ce9c416 100644 --- a/pyhiveapi/apyhiveapi/light.py +++ b/src/apyhiveapi/light.py @@ -180,7 +180,7 @@ async def setStatusOff(self, device: dict): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], device["hiveID"], status="OFF" @@ -188,7 +188,7 @@ async def setStatusOff(self, device: dict): if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -207,7 +207,7 @@ async def setStatusOn(self, device: dict): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( @@ -215,7 +215,7 @@ async def setStatusOn(self, device: dict): ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -235,7 +235,7 @@ async def setBrightness(self, device: dict, n_brightness: int): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], @@ -245,7 +245,7 @@ async def setBrightness(self, device: dict, n_brightness: int): ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -265,7 +265,7 @@ async def setColorTemp(self, device: dict, color_temp: int): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] if data["type"] == "tuneablelight": @@ -284,7 +284,7 @@ async def setColorTemp(self, device: dict, color_temp: int): if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -304,7 +304,7 @@ async def setColor(self, device: dict, new_color: list): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( @@ -317,7 +317,7 @@ async def setColor(self, device: dict, new_color: list): ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final diff --git a/pyhiveapi/apyhiveapi/plug.py b/src/apyhiveapi/plug.py similarity index 96% rename from pyhiveapi/apyhiveapi/plug.py rename to src/apyhiveapi/plug.py index c86e535..60d10e5 100644 --- a/pyhiveapi/apyhiveapi/plug.py +++ b/src/apyhiveapi/plug.py @@ -66,14 +66,14 @@ async def setStatusOn(self, device: dict): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], data["id"], status="ON" ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final @@ -92,14 +92,14 @@ async def setStatusOff(self, device: dict): device["hiveID"] in self.session.data.products and device["deviceData"]["online"] ): - await self.session.hiveRefreshTokens() + await self.session.hive_refresh_tokens() data = self.session.data.products[device["hiveID"]] resp = await self.session.api.setState( data["type"], data["id"], status="OFF" ) if resp["original"] == 200: final = True - await self.session.getDevices(device["hiveID"]) + await self.session.get_devices() return final diff --git a/pyhiveapi/apyhiveapi/sensor.py b/src/apyhiveapi/sensor.py similarity index 87% rename from pyhiveapi/apyhiveapi/sensor.py rename to src/apyhiveapi/sensor.py index ceb0b99..2c1c323 100644 --- a/pyhiveapi/apyhiveapi/sensor.py +++ b/src/apyhiveapi/sensor.py @@ -1,6 +1,6 @@ """Hive Sensor Module.""" # pylint: skip-file -from .helper.const import HIVE_TYPES, HIVETOHA, sensor_commands +from .helper.const import HIVE_TYPES, HIVETOHA class HiveSensor: @@ -108,17 +108,11 @@ async def getSensor(self, device: dict): elif device["hiveID"] in self.session.data.products: data = self.session.data.products.get(device["hiveID"], {}) - if ( - dev_data["hiveType"] in sensor_commands - or dev_data.get("custom", None) in sensor_commands - ): - code = sensor_commands.get( - dev_data["hiveType"], - sensor_commands.get(dev_data["custom"]), - ) + sensor_state = self.session.helper.call_sensor_function(dev_data) + if sensor_state is not None: dev_data.update( { - "status": {"state": await eval(code)}, + "status": {"state": sensor_state}, "deviceData": data.get("props", None), "parentDevice": data.get("parent", None), } diff --git a/src/apyhiveapi/session.py b/src/apyhiveapi/session.py new file mode 100644 index 0000000..e69de29 diff --git a/pyhiveapi/apyhiveapi/data/alarm.json b/src/apyhiveapi/test_data/alarm.json similarity index 100% rename from pyhiveapi/apyhiveapi/data/alarm.json rename to src/apyhiveapi/test_data/alarm.json diff --git a/src/apyhiveapi/test_data/cameraImage.json b/src/apyhiveapi/test_data/cameraImage.json new file mode 100644 index 0000000..8671e10 --- /dev/null +++ b/src/apyhiveapi/test_data/cameraImage.json @@ -0,0 +1,12 @@ +{ + "parsed": { + "events": [ + { + "thumbnailUrls": [ + "https://test.com/image" + ], + "hasRecording": true + } + ] + } +} \ No newline at end of file diff --git a/src/apyhiveapi/test_data/cameraRecording.json b/src/apyhiveapi/test_data/cameraRecording.json new file mode 100644 index 0000000..569e0e7 --- /dev/null +++ b/src/apyhiveapi/test_data/cameraRecording.json @@ -0,0 +1,3 @@ +{ + "parsed": "https://test.com/video" +} \ No newline at end of file diff --git a/pyhiveapi/apyhiveapi/data/data.json b/src/apyhiveapi/test_data/data.json similarity index 100% rename from pyhiveapi/apyhiveapi/data/data.json rename to src/apyhiveapi/test_data/data.json diff --git a/tests/API/async_auth.py b/tests/API/async_auth.py deleted file mode 100644 index b439b42..0000000 --- a/tests/API/async_auth.py +++ /dev/null @@ -1 +0,0 @@ -"""Test file.""" diff --git a/tests/__init__.py b/tests/__init__.py index d203639..cf2ff68 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1 +1 @@ -"""Tests for pyhiveapi.""" +"""Apyhiveapi tests.""" diff --git a/tests/apyhiveapi/__init__.py b/tests/apyhiveapi/__init__.py new file mode 100644 index 0000000..cf2ff68 --- /dev/null +++ b/tests/apyhiveapi/__init__.py @@ -0,0 +1 @@ +"""Apyhiveapi tests.""" diff --git a/tests/apyhiveapi/test_hub.py b/tests/apyhiveapi/test_hub.py new file mode 100644 index 0000000..e5fd03e --- /dev/null +++ b/tests/apyhiveapi/test_hub.py @@ -0,0 +1,126 @@ +"""Tests for the hub object.""" + +import pytest + +from tests.common import MockSession + + +@pytest.mark.asyncio +async def test_hub_get_smoke_status_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products[smoke_sensor["hiveID"]]["props"]["sensors"][ + "SMOKE_CO" + ]["active"] = True + state = await hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state == 1 + + +@pytest.mark.asyncio +async def test_hub_get_smoke_status_not_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + + state = await hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state == 0 + + +@pytest.mark.asyncio +async def test_hub_get_smoke_status_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products.pop(smoke_sensor["hiveID"]) + state = await hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state is None + + +@pytest.mark.asyncio +async def test_hub_get_glass_break_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][0] + state = await hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state == 1 + + +@pytest.mark.asyncio +async def test_hub_get_glass_break_not_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products[glass_sensor["hiveID"]]["props"]["sensors"][ + "GLASS_BREAK" + ]["active"] = False + state = await hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state == 0 + + +@pytest.mark.asyncio +async def test_hub_get_glass_break_detection_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][0] + hive_session.session.data.products.pop(glass_sensor["hiveID"]) + state = await hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state is None + + +@pytest.mark.asyncio +async def test_hub_get_dog_bark_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + hive_session.session.data.products[dog_sensor["hiveID"]]["props"]["sensors"][ + "DOG_BARK" + ]["active"] = True + state = await hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state == 1 + + +@pytest.mark.asyncio +async def test_hub_get_dog_bark_not_detected(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + state = await hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state == 0 + + +@pytest.mark.asyncio +async def test_hub_get_dog_bark_detection_status_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + hive_session.session.data.products.pop(dog_sensor["hiveID"]) + state = await hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state is None diff --git a/tests/apyhiveapi/test_plug.py b/tests/apyhiveapi/test_plug.py new file mode 100644 index 0000000..8238124 --- /dev/null +++ b/tests/apyhiveapi/test_plug.py @@ -0,0 +1,202 @@ +"""Tests for the switch and plug object.""" + +from unittest.mock import patch + +import pytest + +from tests.common import MockSession + + +@pytest.mark.asyncio +async def test_switch_update_switch_online(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + device_data = await hive_session.switch.getSwitch(switch) + + assert device_data != {} + + +@pytest.mark.asyncio +async def test_switch_update_switch_offline(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + switch["deviceData"]["online"] = False + device_data = await hive_session.switch.getSwitch(switch) + + assert device_data["hiveID"] in hive_session.session.config.errorList + + +@pytest.mark.asyncio +async def test_switch_get_plug_state(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + state = await hive_session.switch.getSwitchState(switch) + + assert state in (True, False) + + +@pytest.mark.asyncio +async def test_switch_get_plug_state_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + hive_session.session.data.products.pop(switch["hiveID"]) + state = await hive_session.switch.getState(switch) + + assert state is None + + +@pytest.mark.asyncio +async def test_switch_get_heat_on_demand_state(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][0] + state = await hive_session.switch.getSwitchState(switch) + + assert state in (True, False) + + +@pytest.mark.asyncio +async def test_switch_turn_on_successfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOn(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_switch_turn_on_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOn(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_switch_turn_off_successfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOff(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_switch_turn_off_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOff(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_switch_heat_on_demand_turn_on_successfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][0] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOn(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_switch_heat_on_demand_turn_on_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][0] + + with patch( + "apyhiveapi.api.hive_async_api.HiveApiAsync.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = await hive_session.switch.turnOff(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +@pytest.mark.asyncio +async def test_plug_get_power_usage(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + power_usage = await hive_session.switch.getPowerUsage(switch) + + assert power_usage is not None + + +@pytest.mark.asyncio +async def test_plug_get_power_usage_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + await hive.async_start_session() + hive_session = hive.async_hive + switch = hive_session.session.device_list["switch"][1] + hive_session.session.data.products.pop(switch["hiveID"]) + state = await hive_session.switch.getPowerUsage(switch) + + assert state is None diff --git a/tests/apyhiveapi/test_session.py b/tests/apyhiveapi/test_session.py new file mode 100644 index 0000000..5a2be39 --- /dev/null +++ b/tests/apyhiveapi/test_session.py @@ -0,0 +1,14 @@ +"""Tests for the session object.""" + +import pytest + +from tests.common import MockSession + + +@pytest.mark.asyncio +async def test_start_session(): + """Test a session can be started.""" + hive = MockSession() + device_list = await hive.async_start_session() + + assert len(device_list) > 0 diff --git a/tests/common.py b/tests/common.py index 95f9d4a..10a5e35 100644 --- a/tests/common.py +++ b/tests/common.py @@ -1,10 +1,57 @@ -"""Mock services for tests.""" +"""Test the helper method for writing tests.""" # pylint: skip-file +from dataclasses import dataclass +from apyhiveapi import Hive as HiveAsync + +from pyhiveapi import Hive as HiveSync + +USERNAME = "use@file.com" +PASSWORD = "Test12345" +TEMP_CONFIG = { + "username": USERNAME, + "password": PASSWORD, + "options": {"scan_interval": 120}, +} + + +@dataclass class MockConfig: """Mock config for tests.""" + username: str + password: str + device_data: dict + tokens: dict + options: dict + +@dataclass class MockDevice: - """Mock Device for tests.""" + """Mock config for tests.""" + + username: str + password: str + device_data: dict + tokens: dict + options: dict + + +class MockSession: + """Mock Session for tests.""" + + def __init__(self): + """Initialize the Mock Session.""" + self.async_hive = None + self.sync_hive = None + + def sync_start_session(self): + """Start a sync session.""" + self.sync_hive = HiveSync(username=USERNAME, password=PASSWORD) + return self.sync_hive.start_session(TEMP_CONFIG) + + async def async_start_session(self): + """Start a async session.""" + self.async_hive = HiveAsync(username=USERNAME, password=PASSWORD) + return await self.async_hive.start_session(TEMP_CONFIG) diff --git a/tests/pyhiveapi/__init__.py b/tests/pyhiveapi/__init__.py new file mode 100644 index 0000000..e84ef15 --- /dev/null +++ b/tests/pyhiveapi/__init__.py @@ -0,0 +1 @@ +"""Tests for library.""" diff --git a/tests/pyhiveapi/test_hub.py b/tests/pyhiveapi/test_hub.py new file mode 100644 index 0000000..46d3c69 --- /dev/null +++ b/tests/pyhiveapi/test_hub.py @@ -0,0 +1,115 @@ +"""Tests for the hub object.""" + +from tests.common import MockSession + + +def test_hub_get_smoke_status_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products[smoke_sensor["hiveID"]]["props"]["sensors"][ + "SMOKE_CO" + ]["active"] = True + state = hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state == 1 + + +def test_hub_get_smoke_status_not_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + + state = hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state == 0 + + +def test_hub_get_smoke_status_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + smoke_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products.pop(smoke_sensor["hiveID"]) + state = hive_session.hub.getSmokeStatus(smoke_sensor) + + assert state is None + + +def test_hub_get_glass_break_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][0] + state = hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state == 1 + + +def test_hub_get_glass_break_not_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][1] + hive_session.session.data.products[glass_sensor["hiveID"]]["props"]["sensors"][ + "GLASS_BREAK" + ]["active"] = False + state = hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state == 0 + + +def test_hub_get_glass_break_detection_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + glass_sensor = hive_session.session.device_list["binary_sensor"][0] + hive_session.session.data.products.pop(glass_sensor["hiveID"]) + state = hive_session.hub.getGlassBreakStatus(glass_sensor) + + assert state is None + + +def test_hub_get_dog_bark_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + hive_session.session.data.products[dog_sensor["hiveID"]]["props"]["sensors"][ + "DOG_BARK" + ]["active"] = True + state = hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state == 1 + + +def test_hub_get_dog_bark_not_detected(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + state = hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state == 0 + + +def test_hub_get_dog_bark_detection_status_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + dog_sensor = hive_session.session.device_list["binary_sensor"][2] + hive_session.session.data.products.pop(dog_sensor["hiveID"]) + state = hive_session.hub.getDogBarkStatus(dog_sensor) + + assert state is None diff --git a/tests/pyhiveapi/test_plug.py b/tests/pyhiveapi/test_plug.py new file mode 100644 index 0000000..50b81a8 --- /dev/null +++ b/tests/pyhiveapi/test_plug.py @@ -0,0 +1,187 @@ +"""Tests for the switch and plug object.""" + +from unittest.mock import patch + +from tests.common import MockSession + + +def test_switch_update_switch_online(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + device_data = hive_session.switch.getSwitch(switch) + + assert device_data != {} + + +def test_switch_update_switch_offline(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + switch["deviceData"]["online"] = False + device_data = hive_session.switch.getSwitch(switch) + + assert device_data["hiveID"] in hive_session.session.config.errorList + + +def test_switch_get_plug_state(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + state = hive_session.switch.getSwitchState(switch) + + assert state in (True, False) + + +def test_switch_get_plug_state_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + hive_session.session.data.products.pop(switch["hiveID"]) + state = hive_session.switch.getState(switch) + + assert state is None + + +def test_switch_get_heat_on_demand_state(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][0] + state = hive_session.switch.getSwitchState(switch) + + assert state in (True, False) + + +def test_switch_turn_on_successfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOn(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +def test_switch_turn_on_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOn(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +def test_switch_turn_off_successfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOff(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +def test_switch_turn_off_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOff(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +def test_switch_heat_on_demand_turn_on_successfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][0] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 200, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOn(switch) + + assert result is True + assert len(api_call.mock_calls) == 1 + + +def test_switch_heat_on_demand_turn_on_unsuccessfully(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][0] + + with patch( + "pyhiveapi.api.hive_api.HiveApi.setState", + return_value={"original": 401, "parsed": {}}, + ) as api_call: + result = hive_session.switch.turnOff(switch) + + assert result is False + assert len(api_call.mock_calls) == 1 + + +def test_plug_get_power_usage(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + power_usage = hive_session.switch.getPowerUsage(switch) + + assert power_usage is not None + + +def test_plug_get_power_usage_with_key_error(): + """Test a session can be started.""" + hive = MockSession() + hive.sync_start_session() + hive_session = hive.sync_hive + switch = hive_session.session.device_list["switch"][1] + hive_session.session.data.products.pop(switch["hiveID"]) + state = hive_session.switch.getPowerUsage(switch) + + assert state is None diff --git a/tests/pyhiveapi/test_session.py b/tests/pyhiveapi/test_session.py new file mode 100644 index 0000000..d6e1aad --- /dev/null +++ b/tests/pyhiveapi/test_session.py @@ -0,0 +1,11 @@ +"""Tests for the session object.""" + +from tests.common import MockSession + + +def test_start_session(): + """Test a session can be started.""" + hive = MockSession() + device_list = hive.sync_start_session() + + assert len(device_list) > 0 diff --git a/tests/test_hub.py b/tests/test_hub.py deleted file mode 100644 index 6c83435..0000000 --- a/tests/test_hub.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Test hub framework.""" - - -def test_hub_smoke(): - """Test for hub smoke.""" - result = None - - assert result