Управление Tion S3 и его подключение к умному дому

У TIONofficial есть замечательный продукт: бризер — система активной приточной вентиляции с фильтрами (и теперь с подогревом уличного воздуха). Такие относительно большие ящики, которые через дырку в стене засасывают уличный воздух, прогоняют через фильтры и тадам: в комнате чистый и свежий воздух.

Смотреть на них я начал несколько лет назад, тогда эти устройства управлялись обычным ИК пультом и интегрировать их в любой умный дом было относительно не сложно: умный ИК пульт + сграбить нужные команды управления. Новые же девайсы управляются модно, со смартфона, а пульт перестал быть оптическим. И вот тут с интеграцией возникла проблема: через телефон управлять можно, а вот к контроллеру умного дома уже не подключишь.

Есть еще базовая станция Magic air с выносным датчиком CO2, которая управляет бризером на основе показаний датчика, но возможность управления бризером контроллером умного дома через такую базовую станцию тоже под вопросом.

Что ж, настало время посмотреть как же эту задачу решает телефон и сделать свой сервис для управления бризером.

В спецификации на сайте сказано, что нужен телефон с Bluetooth 4.0 и выше. Приложений два: Magic air и Tion remote. Разбиралось второе, поскольку оно проще и в нем нет специфики работы с базовой станцией.

После декомпиляции быстро нашлись классы, отвечающие за взаимодействие (кстати, разные для версии S3 и S3 lite): приложение и бризер обмениваются пакетам по 20 байт, в начале и конце которого фиксированные символы (= и Z соответственно), второй байт в пакете — передаваемая команда (из enum).

Доступных действий три: pair, decode и encode. Ими и займемся.

pair

Бризер не транслирует ничего во внешний мир. В инструкции сказано что чтобы подружить его с телефоном или пультом нужно на 5 секунд зажать кнопку управления. В этот момент его можно обнаружить и передавать любые команды, на которые он будет отвечать.

Но даже если подать BT команду "pair", после выхода из этого режима, при попытке подключения, бризер будет тут же сбрасывать соединение.

Чтобы это не происходило нужно за время "дружелюбного" режима передать ему pair пакет, который отличается от обычного пакета еще и тем, что 3 байт выставляется в единицу. В результате получаем вот такой базовый сборщик команд:

command_PAIR = 5
command_REQUEST_PARAMS = 1
command_SET_PARAMS = 2
command_prefix = 61
command_suffix = 90

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

и после передачи пакета с command_PAIR, к устройству можно подключаться в любой момент.

Большой проблемой для меня стало разобраться куда же этот пакет нужно передать: устройство танслирует несколько characteristic и из декомпилированного приложения я так и не понял что куда нужно писать. Но нашелся класс с описанием uuid и человеческими названиями этих uuid.

Как вскоре выяснилось, pair-пакет нужно писать в DeviceUUID. В итоге pair делается так:

uuid = "6e400001-b5a3-f393-e0a9-e50e24dcca9e"
def _get_pair_command(self) -> bytearray:
    return self.create_command(self.command_PAIR)
def pair(self, mac: str):
    self._btle.connect(mac, btle.ADDR_TYPE_RANDOM)
    characteristic = self._btle.getServiceByUUID(self.uuid).getCharacteristics()[0]
    characteristic.write(bytes(self._get_pair_command()))
    self._btle.disconnect()

Получение и декодирование данных

После того как стало возможно нормально общаться с бризером захотелось получать от него что-нибудь разумное, что отображается на экране приложения.
Исходные данные можно получить если прочитать из той же характеристики, что используется для pair.

response =  self._btle.getServiceByUUID(self.uuid).getCharacteristics()[0].read()

А на параметры 20 байт ответа разбираются вот так:

statuses = [ 'off', 'on' ]
modes = [ 'recirculation', 'mixed' ]

def _process_mode(self, mode_code: int) -> str:
    try:
      result = self.modes[mode_code]
    except IndexError:
      result = 'outside'
    return result

def _process_status(self, code: int) -> str:
    try:
      result = self.statuses[code]
    except IndexError:
      result = 'unknown'
    return result

def _decode_response(self, response: bytearray) -> dict:
    return {
      "heater": self._process_status(response[4] & 1),
      "status": self._process_status(response[4] >> 1 & 1),
      "sound": self._process_status(response[4] >> 3 & 1),
      "mode": self._process_mode(int(list("{:02x}".format(response[2]))[0])),
      "fan_speed": int(list("{:02x}".format(response[2]))[1]),
      "heater_temp": response[3],
      "in_temp": response[8],
      "filter_remain": response[10]*256 + response[9],
      "time": "{}:{}".format(response[11],response[12]),
      "request_error_code": response[13],
      "fw_version": "{:02x}{:02x}".format(response[16],response[17])
    }

Но тут выяснилось, что при последовательных чтениях данные не обновляются. Чтобы их обновить нужно прочитать данные из notify характеристики и послать команду command_REQUEST_PARAMS в write характеристику. После этого функция для получения актуальных данных с бризера начинает выглядеть примерно так:

uuid_write = "6e400002-b5a3-f393-e0a9-e50e24dcca9e"
uuid_notify = "6e400003-b5a3-f393-e0a9-e50e24dcca9e"

def _connect(self, mac: str,  new_connection = True):
    if new_connection:
      self._btle.connect(mac, btle.ADDR_TYPE_RANDOM)
      for tc in self._btle.getCharacteristics():
        if tc.uuid == self.uuid_notify:
          self.notify = tc
        if tc.uuid == self.uuid_write:
          self.write = tc
...
def get(self) -> dict:
    response = ""

    self.notify.read()
    self.write.write(self._get_status_command())
    response =  self._btle.getServiceByUUID(self.uuid).getCharacteristics()[0].read()

    return self._decode_response(response)

Теперь каждый вызов get выдает актуальные данные с бризера.

Читать хорошо, но хочется и управлять

Получение данных с бризера — хорошо, но цель в том, чтобы ночью уменьшать мощность вентилятора (сильно шумит, зараза), управлять оборотами на основании данных mhz19 итд.

За запись данных отвечает пакет с командой command_SET_PARAMS. В пакете всегда содержится полный набор устанавливаемых данных, поэтому если хочется поменять что-то одно, нужно прочитать текущее состояние, изменить то, что нужно, и после этого передать полный набор параметров обратно на бризер.

def _encode_mode(self, mode: str) -> int:
  return self.modes.index(mode) if mode in self.modes else 2

def _encode_status(self, status: str) -> int:
  return self.statuses.index(status) if status in self.statuses else 0

def _encode_request(self, request: dict) -> bytearray:
  settings = {**self.get(), **request}
  new_settings = self.create_command(self.command_SET_PARAMS)
  new_settings[2] = settings["fan_speed"]
  new_settings[3] = settings["heater_temp"]
  new_settings[4] = self._encode_mode(settings["mode"])
  new_settings[5] = self._encode_status(settings["heater"]) | (self._encode_status(settings["status"])<<1) | (self._encode_status(settings["sound"])<<3)
  return new_settings

Пишем это все в характеристику uuid == uuid_write.
Неясной осталась судьба еще пары байт пакета, игнорирование которых видимого эффекта на работу не оказало.

Как этим всем пользоваться

Результаты изысканий выше собраны в модуль для python3.
На коленке написан простой API сервер, который транслирует json-запросы, через модуль управления, в бризер и выдает json ответы.

В Home assistant сказано, что есть: rest сенсор, rest команды управления и компонент fan, управление которым делается через rest команды.

И все это добро запущено на малине.

Дальше в планах — отдельный компонент для Home assistant, чтобы обойтись без промежуточного API сервера, который очень нужен поскольку обещаниями интеграции бризеров в умный дом Тион кормит публику с 2016 года (если я правильно помню историю общения с пользователями в официальной группе ВК(https://vk.com/tion_ru)), платформ управления великое множество, а с rest API могут взаимодействовать все.

Ну и хотелось бы, при помощи энтузиастов добавить поддержку lite версии, которую я могу изучать только теоретически.

И небольшая ложка дегтя: если использовать с HomeAssist, то сервер спустя некоторое время перестает успевать обрабатывать запросы. Видимо, HA делает одновременные запросы, которые не очень хорошо обрабатываются сервером.

P.S.: Issue, PR и вопросы в коментах приветствуются: если возникнут вопросы — постараюсь помочь с интеграцией, в меру своих знаний и возможностей.