diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..751fbbb --- /dev/null +++ b/README.md @@ -0,0 +1,6 @@ +You should pair Tion S3 device with your bluetooth host by two steps: + 1. pressing for 5 second poer button at device + 2. running ./pair.py device_mac_address while button blinks +before use this module! + +If you want to use MagicAir API please follow https://github.com/airens/tion \ No newline at end of file diff --git a/__init__.py b/__init__.py deleted file mode 100644 index a6606e7..0000000 --- a/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -from .tion import tion -from .s3 import s3 diff --git a/pair.py b/pair.py new file mode 100755 index 0000000..3360b2c --- /dev/null +++ b/pair.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 +import sys +from tion import s3 as s3device + +s3 = s3device(sys.argv[1]) +s3.pair() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..dc1a339 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +bleak==0.9.1 \ No newline at end of file diff --git a/s3.py b/s3.py deleted file mode 100644 index dbc70aa..0000000 --- a/s3.py +++ /dev/null @@ -1,5 +0,0 @@ -from . import tion - -class s3(tion): - def get() -> dict: - return {} diff --git a/setup.py b/setup.py index 6e44b50..6458d1c 100644 --- a/setup.py +++ b/setup.py @@ -1,8 +1,17 @@ +from __future__ import print_function from setuptools import setup, find_packages +import os +import tion_btle.const + + +here = os.path.abspath(os.path.dirname(__file__)) setup( - name='tion', - version='1.0', - packages=find_packages(), + name='tion_btle', + version=tion_btle.const.VERSION, long_description="Module for working with Tion breezers", + url='https://github.com/TionAPI/tion_python/tree/dev', + install_requires=['bleak==0.9.1'], + description='Python module for interacting with Tion breezers', + packages=find_packages(), ) diff --git a/tests/decode_temperature.py b/tests/decode_temperature.py new file mode 100755 index 0000000..67ffc85 --- /dev/null +++ b/tests/decode_temperature.py @@ -0,0 +1,22 @@ +#!/usr/bin/python +import sys +import os +import unittest + +PACKAGE_PARENT = '..' +SCRIPT_DIR = os.path.dirname(os.path.realpath(os.path.join(os.getcwd(), os.path.expanduser(__file__)))) +sys.path.append(os.path.normpath(os.path.join(SCRIPT_DIR, PACKAGE_PARENT))) + +import tion + + +class TestDecodeTemperature(unittest.TestCase, tion.tion): + def test_positive(self): + self.assertEqual(self.decode_temperature(0x09), 9, "Should be 9") + + def test_negative(self): + self.assertEqual(self.decode_temperature(0xFF), -1, "Should be -1") + + +if __name__ == '__main__': + unittest.main() diff --git a/tion.py b/tion.py deleted file mode 100644 index 422def8..0000000 --- a/tion.py +++ /dev/null @@ -1,43 +0,0 @@ -import abc - -class tion: - @abc.abstractmethod - def _send_request(self, request: bytearray) -> bytearray: - """ Send request to device - - Args: - request : array of bytes to send to device - Returns: - array of bytes with device response - """ - pass - - @abc.abstractmethod - def _decode_response(self, response: bytearray) -> dict: - """ Decode response from device - - Args: - response: array of bytes with data from device, taken from _send_request - Returns: - dictionary with device response - """ - pass - - @abc.abstractmethod - def _encode_request(self, request: dict) -> bytearray: - """ Encode dictionary of request to byte array - - Args: - request: dictionry with request - Returns: - Byte array for sending to device - """ - pass - - @abc.abstractmethod - def get() -> dict: - """ Get device information - Returns: - dictionay with device paramters - """ - pass diff --git a/tion_btle/__init__.py b/tion_btle/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tion_btle/const.py b/tion_btle/const.py new file mode 100644 index 0000000..81343d7 --- /dev/null +++ b/tion_btle/const.py @@ -0,0 +1 @@ +VERSION = '0.2.0' diff --git a/tion_btle/s3.py b/tion_btle/s3.py new file mode 100644 index 0000000..e1f6810 --- /dev/null +++ b/tion_btle/s3.py @@ -0,0 +1,174 @@ +import logging +import asyncio +from bleak import exc +from typing import Callable + +if __package__ == "": + from tion_btle.tion import tion, TionException, TionExceptionGet +else: + from .tion import tion, TionException, TionExceptionGet + +import time + +logging.basicConfig(level=logging.DEBUG) +_LOGGER = logging.getLogger(__name__) + + +class s3(tion): + uuid = "6e400001-b5a3-f393-e0a9-e50e24dcca9e" + uuid_write = "6e400002-b5a3-f393-e0a9-e50e24dcca9e" + uuid_notify = "6e400003-b5a3-f393-e0a9-e50e24dcca9e" + + modes = ['recirculation', 'mixed'] + _btle = None + + command_prefix = 61 + command_suffix = 90 + + command_PAIR = 5 + command_REQUEST_PARAMS = 1 + command_SET_PARAMS = 2 + + def __init__(self, mac: str): + super().__init__(mac) + + async def pair(self): + async def get_pair_command() -> bytearray: + return await self.create_command(self.command_PAIR) + + _LOGGER.setLevel("DEBUG") + _LOGGER.debug("Going to pair with %s" % self.mac) + _LOGGER.debug("Connecting") + try: + await self._connect() + _LOGGER.debug("BT Pairing") + await self._btle.pair() + pair_command = await get_pair_command() + _LOGGER.debug("Sending pair command %s to %s", bytes(pair_command).hex(), self.uuid_write) + await self._try_write(request=pair_command) + _LOGGER.debug("Done!") + finally: + _LOGGER.debug("Disconnecting") + await self._disconnect() + + async def create_command(self, command: int) -> bytearray: + command_special = 1 if command == self.command_PAIR else 0 + return bytearray([self.command_prefix, command, command_special, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + self.command_suffix]) + + async def get(self) -> dict: + async def get_status_command() -> bytearray: + return await self.create_command(self.command_REQUEST_PARAMS) + + async def decode_response(response: bytearray) -> dict: + async def process_mode(mode_code: int) -> str: + try: + mode = self.modes[mode_code] + except IndexError: + mode = 'outside' + return mode + + result = {} + try: + self.fan_speed = int(list("{:02x}".format(response[2]))[1]) + result = {"code": 200, + "mode": await process_mode(int(list("{:02x}".format(response[2]))[0])), + "fan_speed": self.fan_speed, + "heater_temp": response[3], + "heater": await self._process_status(response[4] & 1), + "status": await self._process_status(response[4] >> 1 & 1), + "timer": await self._process_status(response[4] >> 2 & 1), + "sound": await self._process_status(response[4] >> 3 & 1), + "out_temp": await self.decode_temperature(response[7]), + "in_temp": await self.decode_temperature(response[8]), + "filter_remain": response[10] * 256 + response[9], + "time": "{}:{}".format(response[11], response[12]), + "request_error_code": response[13], + "productivity": response[14], + "fw_version": "{:02x}{:02x}".format(response[18], response[17])} + + if result["heater"] == "off": + result["is_heating"] = "off" + else: + if result["in_temp"] < result["heater_temp"] and result["out_temp"] - result["heater_temp"] < 3: + result["is_heating"] = "on" + else: + result["is_heating"] = "off" + + except IndexError as e: + result = {"code": 400, + "error": "Got bad response from Tion '%s': %s while parsing" % (response, str(e))} + finally: + return result + + try: + await self._connect() + await self._enable_notifications() + await self._try_write(request=await get_status_command()) + + i = 0 + try: + while i < 10 and not self._delegation.have_new_data: + await asyncio.sleep(1) + i += 1 + if i < 10: + byte_response = self._delegation.data + else: + _LOGGER.warning("Waiting data too long. Will use data from uuid_notify (%s)." % self.uuid_notify) + byte_response = await self._btle.read_gatt_char(self.uuid_notify) + result = await decode_response(byte_response) + except exc.BleakError as e: + _LOGGER.error("Got %s while waiting for notification", str(e)) + raise TionExceptionGet('get', str(e)) + except (exc.BleakError, TionException, TionExceptionGet) as e: + _LOGGER.error(str(e)) + raise TionException('get', str(e)) + finally: + await self._disconnect() + + return result + + async def set(self, request: dict, keep_connection=False): + async def encode_request(request: dict) -> bytearray: + async def encode_mode(mode: str) -> int: + return self.modes.index(mode) if mode in self.modes else 2 + + async def encode_status(status: str) -> int: + return self.statuses.index(status) if status in self.statuses else 0 + + try: + if request["fan_speed"] == 0: + del request["fan_speed"] + request["status"] = "off" + except KeyError: + pass + + try: + current_settings = await self.get() + except TionExceptionGet: + raise TionException('set', 'Could not get current settings!') + + _LOGGER.debug("set: got '%s' settings" % current_settings) + settings = {**current_settings, **request} + new_settings = await self.create_command(self.command_SET_PARAMS) + new_settings[2] = int(settings["fan_speed"]) + new_settings[3] = int(settings["heater_temp"]) + new_settings[4] = await encode_mode(settings["mode"]) + new_settings[5] = await encode_status(settings["heater"]) | (await encode_status(settings["status"]) << 1) | ( + await encode_status(settings["sound"]) << 3) + return new_settings + try: + await self._connect() + try: + encoded_request = await encode_request(request) + except (KeyError, TionException) as e: + _LOGGER.warning("Could not create encoded settings command: %s" % str(e)) + return + + await self._try_write(request=encoded_request) + + except (exc.BleakError, TionException, TionExceptionGet) as e: + _LOGGER.error(str(e)) + raise TionException('set', str(e)) + finally: + await self._disconnect() diff --git a/tion_btle/tion.py b/tion_btle/tion.py new file mode 100644 index 0000000..f9b1626 --- /dev/null +++ b/tion_btle/tion.py @@ -0,0 +1,188 @@ +import abc +import logging +import time +import bleak +import asyncio +from typing import Callable +from bleak import BleakClient +from bleak import exc + +_LOGGER = logging.getLogger(__name__) + + +class TionDelegation: + def __init__(self): + self._have_new_data: bool = False + self._data: bytearray = bytearray() + + def handleNotification(self, handle: int, data: bytearray): + self._data = data + self._have_new_data = True + _LOGGER.debug("Got data in %d response %s", handle, bytes(data).hex()) + + @property + def data(self) -> bytearray: + self._have_new_data = False + return self._data + + @property + def have_new_data(self) -> bool: + return self._have_new_data + + +class TionException(Exception): + def __init__(self, expression, message): + self.expression = expression + self.message = message + + +class TionExceptionGet(TionException): + def __init__(self, expression, message): + super().__init__(expression, message) + + +class tion: + statuses = ['off', 'on'] + uuid_notify: str = "" + uuid_write: str = "" + + def __init__(self, mac: str): + self._mac: str = mac + self._btle: BleakClient = BleakClient(self.mac) + self._delegation = TionDelegation() + self._fan_speed = 0 + self._connections: int = 0 + + @abc.abstractmethod + async def _send_request(self, request: bytearray) -> bytearray: + """ Send request to device + + Args: + request : array of bytes to send to device + Returns: + array of bytes with device response + """ + pass + + @abc.abstractmethod + def _decode_response(self, response: bytearray) -> dict: + """ Decode response from device + + Args: + response: array of bytes with data from device, taken from _send_request + Returns: + dictionary with device response + """ + pass + + @abc.abstractmethod + async def _encode_request(self, request: dict) -> bytearray: + """ Encode dictionary of request to byte array + + Args: + request: dictionry with request + Returns: + Byte array for sending to device + """ + pass + + @abc.abstractmethod + async def get(self) -> dict: + """ Get device information + Returns: + dictionay with device paramters + """ + pass + + @property + def mac(self): + return self._mac + + async def decode_temperature(self, raw: bytes) -> int: + """ Converts temperature from bytes with addition code to int + Args: + raw: raw temperature value from Tion + Returns: + Integer value for temperature + """ + barrier = 0b10000000 + if (raw < barrier): + result = raw + else: + result = -(~(result - barrier) + barrier + 1) + + return result + + async def _process_status(self, code: int) -> str: + try: + status = self.statuses[code] + except IndexError: + status = 'unknown' + return status + + @property + async def connection_status(self): + status = "connected" if await self._btle.is_connected() else "disc" + _LOGGER.debug("connection_status is %s" % status) + return status + + async def _connect(self): + self._connections += 1 + if self.mac == "dummy": + _LOGGER.info("Dummy connect") + return + + if await self.connection_status == "disc": + try: + await self._btle.connect() + except exc.BleakError as e: + _LOGGER.warning("Got %s exception", str(e)) + await asyncio.sleep(2) + raise e + else: + _LOGGER.debug("Already connected!") + + async def _disconnect(self): + self._connections -= 1 + if await self.connection_status != "disc": + if self.mac != "dummy": + if self._connections == 0: + await self._btle.disconnect() + # drop _services_resolved flag, that not dropped while disconnect + self._btle._services_resolved = False + + async def _try_write(self, request: bytearray): + if self.mac != "dummy": + _LOGGER.debug("Writing %s to %s", bytes(request).hex(), self.uuid_write) + return await self._btle.write_gatt_char( + self.uuid_write, + request, + False + ) + else: + _LOGGER.info("Dummy write") + return "dummy write" + + async def _enable_notifications(self): + _LOGGER.debug("Enabling notification") + try: + result = await self._btle.start_notify(self.uuid_notify, self._delegation.handleNotification) + except bleak.exc.BleakError as e: + _LOGGER.warning("Got exception %s while enabling notifications!" % str(e)) + raise e + return result + + @property + def fan_speed(self): + return self._fan_speed + + @fan_speed.setter + def fan_speed(self, new_speed: int): + if 0 <= new_speed <= 6: + self._fan_speed = new_speed + + else: + _LOGGER.warning("Incorrect new fan speed. Will use 1 instead") + self._fan_speed = 1 + + # self.set({"fan_speed": new_speed})