Из любопытства была у меня идея попробовать снять показания со счётчика удалённо, но так как был уже установлен древний дубовый счётчик Энергомера СЕ102, то вкладываться в замену не хотелось. Однако межповерочный интервал начал подходить к концу и я начал задумываться насчёт замены счётчика на более продвинутый (с интерфейсами связи). Начал выбирать варианты в виде LoRaWan, Zigbee, RS-485, но оказалось, что по современному законодательству счётчики электроэнергии меняет ресурсоснабжающая организация (так называемый «гарантирующий поставщик»). С надеждой ждал, что электрик придёт и поставит современный навороченный счётчик с кучей интерфейсов, но всё оказалось не совсем так...
Представитель ресурсоснабжающей организации установил новый счётчик МИР С-05.10–230-5(80)‑G2Z1B‑KNQ‑S-D.
Оказалось, что у него из интерфейсов связи с внешним миром LTE, Zigbee и оптопорт... Путь в LTE мне конечно сразу оказался закрыт из‑за того, что туда ресурсоснабжающая организация установила свою сим‑карту. Про Zigbee я сначала обрадовался, но затем узнал, что протокол Zigbee, к которому мы привыкли в «умных домах», имеет мало общего с тем, что установлено в счётчике электроэнергии. Поэтому остался всего один вариант — оптопорт...
Для реализации связи со счётчиком понадобится USB-оптоголовка (которая стоит неприятно в районе 3000 рублей)
А также программа DLMS/COSEM.
Однако доступ ко счётчику по оптопорту защищён парольной защитой. Поэтому я направил запрос в ресурсоснабжающую организацию на получение пароля для считывания показаний счётчика электроэнергии.
В ресурсоснабжающей организации удивились моему запросу и со мной связался главный инженер с вопросом: «А зачем вам это надо???» Услышав мой эпикриз, мне сказали, что вместо парольного доступа к оптопорту мне дадут пин‑код для Bluetooth соединения со счётчиком с помощью мобильного приложения, но только после того, как я подтвержу свою личность и своё право собственности на имущество. (почему‑то ранее в описании я нигде не обратил внимание на то, что у счётчика есть bluetooth интерфейс.
Бюрократическую процедуру подтверждения личности и права собственности опущу и скажу, что пин‑код я в конце‑концов всё‑таки получил, после чего занялся беспроводной связью с помощью мобильного приложения...
Оказалось, что для связи со счётчиком используется фирменное приложение «МИР ДП».
Которое отсутствует в Play Market и его нужно скачивать в виде .apk файла по ссылке https://mir-omsk.ru/products/mobilnoe-prilozhenie-mir-dp/
Приложение я без проблем установил, по Bluetooth соединился со счётчиком и в довольно симпатичном интерфейсе увидел все нужные данные со счётчика.
На этом, казалось бы, историю можно заканчивать, но в моём случае самое интересное только начиналось, так как я решил попробовать интегрировать показания счётчика в систему "умного дома" HomeAssistant. А для этого мобильное приложение явно не годится...
Поэтому я установил неподалёку от счётчика Raspberry Pi Zero 2W (который имеет встроенный Bluetooth-интерфейс) и начал с помощью SSH-соединения устанавливать соединение между счётчиком и Raspberry.
В первую очередь установил необходимые пакеты на Raspberry:
sudo apt update sudo apt install -y bluez bluez-tools bluetooth python3-pip python3 -m pip install --break-system-packages bleak sudo systemctl enable bluetooth sudo systemctl restart bluetooth
Затем запустил bluetoothctl (встроенный командный интерфейс (интерактивная консольная утилита) для управления Bluetooth на устройствах под управлением Linux) включил его питание и подготовил к сопряжению
Затем необходимо "просканировать эфир", чтобы понять, есть ли счётчик в сети. Для этого в интерфейсе bluetoothctl необходимо ввести команду scan on
В результате среди кучи соседского bluetooth-оборудования и остальных счётчиков электроэнергии обнаружился и мой счётчик. В bluetooth-сканере все счётчики МИР С-05 отображаются как C05-..., где ... - это серийный номер счётчика. В приложении он тоже засветился, поэтому сопоставить несложно.
Затем ввожу команду devices чтобы отобразить список найденных устройств:
Из списка устройств я выяснил, что мой счётчик имеет MAC-адрес E4:06:BF:87:CD:69. Поэтому я запускаю команду pairing (сопряжение) в bluetoothctl, для установления доверенных отношений между Raspberry Pi и другим счётчиком по bluetooth.
В ходе сопряжения нужно быстро напечатать yes чтобы "успеть" :)
Далее по очереди применяем команды trust "mac-адрес", connect "mac-адрес" и info "mac-aдрес.
После установления соединения со счётчиком у него необходимо получить общий атрибутный профиль GATT (Generic Attribute Profile) командой menu gatt
В результате я понял уже следующее: подключение держится без pairing и список сервисов уже есть. Теперь надо понять, какие из них служебные, а где реальный обмен с приложением. Без pairing счётчик подключается и держит GATT-сессию. Сервис d14d6ee-fd63-4fa1-bfa4-8f47b42119f0 - это типовой Silicon Labs OTA service, его можно пока игнорировать. Сервис 4880c12c-fdcb-4077-8920-a450d7f9b907 с характеристикой fec26ec4-6d71-4442-9f81-55bc21d658d6 совпадает с типовым Silicon Labs SPP-over-BLE каналом: одна характеристика для notify и write without response.
Далее для проверки работоспособности сервисов написал скрипт на питоне:
import asyncio from bleak import BleakClient ADDR = "E4:06:BF:87:CD:69" CHARS = [ "00002a29-0000-1000-8000-00805f9b34fb", "00002a24-0000-1000-8000-00805f9b34fb", "00002a25-0000-1000-8000-00805f9b34fb", "00002a27-0000-1000-8000-00805f9b34fb", "00002a26-0000-1000-8000-00805f9b34fb", "2bb1d64f-451a-4541-ac89-24df683309dc", "d24a5138-1448-48ea-a983-f7df274c6d89", "4335a5be-fbbd-464e-8659-1a4212239e3e", "b3f7e595-2951-42fa-879e-0d9dfa5e846e", "fec26ec4-6d71-4442-9f81-55bc21d658d6", "f7bf3564-fb6d-4e53-88a4-5e37e0326063", ] def fmt(data: bytes) -> str: try: txt = data.decode("utf-8", errors="strict") return f'hex={data.hex(" ")} | utf8={txt!r}' except Exception: return f'hex={data.hex(" ")}' async def main(): def cb(sender, data): print(f"NOTIFY {sender}: {fmt(bytes(data))}") async with BleakClient(ADDR, timeout=20.0) as client: print("connected:", client.is_connected) for uuid in CHARS: try: data = await client.read_gatt_char(uuid) print(f"READ {uuid}: {fmt(bytes(data))}") except Exception as e: print(f"READ {uuid}: ERROR: {e}") try: await client.start_notify("fec26ec4-6d71-4442-9f81-55bc21d658d6", cb) print("notify enabled on fec26...") except Exception as e: print("notify fec26 error:", e) await asyncio.sleep(10) asyncio.run(main())
В результате выполнения которого получил вот такой вывод:
connected: True
READ 00002a29-0000-1000-8000-00805f9b34fb: hex=4e 50 4f 22 4d 49 52 22 22 | utf8='NPO"MIR""'
READ 00002a24-0000-1000-8000-00805f9b34fb: hex=43 30 35 | utf8='C05'
READ 00002a25-0000-1000-8000-00805f9b34fb: hex=35 30 39 31 32 35 32 35 31 36 31 38 31 38 | utf8='50912525161818'
READ 00002a27-0000-1000-8000-00805f9b34fb: hex=31 | utf8='1'
READ 00002a26-0000-1000-8000-00805f9b34fb: hex=76 32 2e 38 31 2e 39 2e 30 | utf8='v2.81.9.0'
READ 2bb1d64f-451a-4541-ac89-24df683309dc: hex=1a 07 | utf8='\x1a\x07'
READ d24a5138-1448-48ea-a983-f7df274c6d89: ERROR: (<BleakGATTProtocolErrorCode.READ_NOT_PERMITTED: 2>, 'GATT Protocol Error: Read Not Permitted')
READ 4335a5be-fbbd-464e-8659-1a4212239e3e: hex=00 | utf8='\x00'
READ b3f7e595-2951-42fa-879e-0d9dfa5e846e: ERROR: (<BleakGATTProtocolErrorCode.READ_NOT_PERMITTED: 2>, 'GATT Protocol Error: Read Not Permitted')
READ fec26ec4-6d71-4442-9f81-55bc21d658d6: ERROR: (<BleakGATTProtocolErrorCode.READ_NOT_PERMITTED: 2>, 'GATT Protocol Error: Read Not Permitted')
READ f7bf3564-fb6d-4e53-88a4-5e37e0326063: ERROR: (<BleakGATTProtocolErrorCode.READ_NOT_PERMITTED: 2>, 'GATT Protocol Error: Read Not Permitted')
notify enabled on fec26...
Таким образом получилось, что я с помощью BLE-команд получил информацию о производителе: NPO"MIR"" модели: C05 серийнике: 50912525161818 прошивке: v2.81.9.0
А вот все остальные параметры - это тёмный лес. Без знания работы протокола обмена получить данные со счётчика почти нереально... И вот тут начинается следующий объёмный этап в виде реверс-инжиниринга.
Для этого нужен только мобильный телефон на базе операционной системы Android с root-доступом и компьютер.
В первую очередь переходим в настройки телефона и несколько раз нажимаем пункт "версия ядра", после чего активируется внутреннее меню Android в настройках "<> Для разработчиков". После чего необходимо перейти в это меню и включить "snoop-логи HCI Bluetooth" и выключить "фильтрацию snoop-логов".
Затем необходимо включить "Отладка по USB" и включить "Отладка суперпользователем".
Далее, с включёнными snoop-логами необходимо выключить и включить Bluetooth в телефоне, запустить приложение "МИР ДП" и соединиться со счётчиком, после чего считать с него показания в приложении. Далее закрыть приложение "МИР ДП" и выключить Bluetooth. Затем подключить телефон кабелем к компьютеру и перейти в папку с утилитами adb tools где сначала проверить соединение командой adb devices, а затем скачать bagreport с логами командой adb bugreport bugreport.zip
Внутри скачанного архива будет нужный snoop_log по пути /FS/data/misc/bluetooth/logs/
Кстати логи можно получить ещё вот таким набором команд в adb
adb shell su -c "cp /data/misc/bluetooth/logs/btsnoop_hci.log /sdcard/Download/btsnoop_hci.log && chmod 666 /sdcard/Download/btsnoop_hci.log" adb pull /sdcard/Download/btsnoop_hci.log
И в результате в папке с adb-утилитами появится btsnoop_hci.log. Однако он сохраняется в бинарном формате и выглядит не особо удобочитаемо для человеческого глаза:
Конечно можно было скормить его wireshark и расшифровать, но мне уже настолько надоела эта процедура, что я скормил этот лог нейронке и вот что она мне выдала:
Таким образом получается, что у нас получилось вытащить из логов hci правильную последовательность команд bluetooth для счётчика. Осталось только на основании этой последовательности попробовать вытащить эти данные из счётчика на Raspberry по BLE. Для этого делаю скрипт на питоне последовательных запросов параметров (если за один раз запросить - то будет обрыв. такой уж протокол у счётчика):
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Считывание показаний со счётчика МИР С-05 по BLE. Что делает скрипт: 1. Подключается к счётчику по BLE. 2. Выполняет служебную инициализацию, которую делает штатное приложение. 3. Включает приём notify-уведомлений. 4. Последовательно отправляет команды чтения. 5. Печатает ответы счётчика: - общий расход - Т1 - Т2 - дату/время - ток - напряжение - и другие параметры, если счётчик их отдаёт Важно: - обычный BLE pairing для этого скрипта не требуется; - связь идёт через GATT characteristics; - логика инициализации и команды были восстановлены экспериментально по фактическому обмену штатного Android-приложения. Зависимость: pip install bleak """ import asyncio from bleak import BleakClient # ============================================================ # НАСТРОЙКИ # ============================================================ # BLE MAC-адрес счётчика. # При необходимости замени на свой. ADDR = "E4:06:BF:87:CD:69" # UUID характеристик, найденных экспериментально. # # UUID_D24A: # Служебная характеристика для инициализации. # # UUID_B3F7: # Ещё одна служебная характеристика для инициализации. # # UUID_FEC2: # Главный рабочий канал: # - сюда пишутся команды # - отсюда приходят notify-ответы UUID_D24A = "d24a5138-1448-48ea-a983-f7df274c6d89" UUID_B3F7 = "b3f7e595-2951-42fa-879e-0d9dfa5e846e" UUID_FEC2 = "fec26ec4-6d71-4442-9f81-55bc21d658d6" # ============================================================ # НАБОР КОМАНД # ============================================================ # Это последовательность команд, которая показала рабочий результат. # Команды отправляются в UUID_FEC2. # # Комментарии по смыслу: # - "time" : запрос времени # - "energy" : старт чтения блока энергий # - "next" : переход к следующему пункту/экрану # - "params" : переход к текущим параметрам # - "month" : по факту в эксперименте снова выводил блок текущих параметров, # поэтому его назначение пока не до конца подтверждено CMDS = [ ("time", bytes.fromhex("0001fdc11f")), ("energy", bytes.fromhex("0001eee34d")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("params", bytes.fromhex("000102dfef")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("month", bytes.fromhex("000101ef8c")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ] # ============================================================ # ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ # ============================================================ def pretty(data: bytes) -> str: """ Красивое представление бинарного ответа. Возвращает: - hex-строку - текстовую интерпретацию в cp1251 Почему cp1251: Счётчик отдаёт русские строки именно в этой кодировке. UTF-8 тут не подходит. """ text = data.decode("cp1251", errors="ignore") # В ответах много нулевых байтов; заменяем их на пробелы, # чтобы текст читался понятнее. text = text.replace("\x00", " ") return f"hex={data.hex(' ')}\ntext={text}" def notify_callback(sender: int, data: bytearray) -> None: """ Обработчик notify-уведомлений от счётчика. sender: внутренний идентификатор характеристики в bleak/BlueZ data: полезная нагрузка уведомления """ raw = bytes(data) print("\n--- NOTIFY ---") print(pretty(raw)) # ============================================================ # ОСНОВНАЯ ЛОГИКА # ============================================================ async def main() -> None: """ Основная процедура: - подключение - поиск характеристик - инициализация - включение notify - отправка команд - ожидание ответов """ # Создаём BLE-клиент и подключаемся к счётчику. async with BleakClient(ADDR, timeout=20.0) as client: print("connected:", client.is_connected) # Получаем объекты характеристик по UUID. # # Объект характеристики нужен bleak для чтения/записи/notify. ch_d24a = client.services.get_characteristic(UUID_D24A) ch_b3f7 = client.services.get_characteristic(UUID_B3F7) ch_fec2 = client.services.get_characteristic(UUID_FEC2) # Базовая проверка: если какая-то характеристика не нашлась, # значит либо устройство не то, либо сервисы не прочитались. if ch_d24a is None: raise RuntimeError(f"Characteristic {UUID_D24A} not found") if ch_b3f7 is None: raise RuntimeError(f"Characteristic {UUID_B3F7} not found") if ch_fec2 is None: raise RuntimeError(f"Characteristic {UUID_FEC2} not found") # -------------------------------------------------------- # ШАГ 1. СЛУЖЕБНАЯ ИНИЦИАЛИЗАЦИЯ # -------------------------------------------------------- # # Эта часть была подсмотрена в работе штатного приложения. # Без неё счётчик либо не отвечает, либо отвечает не так, как нужно. # Отправляем 0x01 в характеристику B3F7. # По смыслу это выглядит как включение/разрешение дальнейшего обмена. await client.write_gatt_char(ch_b3f7, b"\x01", response=True) print("init: b3f7 <- 01") # Отправляем 4 байта в характеристику D24A. # Это тоже обязательная часть стартовой инициализации. await client.write_gatt_char(ch_d24a, bytes.fromhex("9de40000"), response=True) print("init: d24a <- 9d e4 00 00") # -------------------------------------------------------- # ШАГ 2. ВКЛЮЧАЕМ NOTIFY # -------------------------------------------------------- # # После этого счётчик сможет присылать ответы асинхронно. # Все ответы будут обрабатываться функцией notify_callback. await client.start_notify(ch_fec2, notify_callback) print("notify enabled on fec2") # Небольшая пауза, чтобы устройство успело войти в нужное состояние. await asyncio.sleep(0.5) # -------------------------------------------------------- # ШАГ 3. ПОСЛЕДОВАТЕЛЬНО ОТПРАВЛЯЕМ КОМАНДЫ # -------------------------------------------------------- # # Команды идут в рабочую характеристику FEC2. # Ответы приходят не как return write-функции, # а отдельными notify-уведомлениями. for name, cmd in CMDS: print(f"\nSEND {name}: {cmd.hex(' ')}") # response=False: # запись без ожидания подтверждения на уровне GATT write response. # Это соответствует тому, как устройство работает в данном случае. await client.write_gatt_char(ch_fec2, cmd, response=False) # Даём счётчику время сформировать и прислать очередной ответ. await asyncio.sleep(1.2) # -------------------------------------------------------- # ШАГ 4. ЖДЁМ ПОСЛЕДНИЕ УВЕДОМЛЕНИЯ # -------------------------------------------------------- # # После последней команды полезно немного подождать, # потому что ответы могут прийти не мгновенно. await asyncio.sleep(5) # ============================================================ # ТОЧКА ВХОДА # ============================================================ if __name__ == "__main__": asyncio.run(main())
И в результате выполнения получаем на Raspberry ответ от счётчика:
connected: True init: b3f7 <- 01 init: d24a <- 9d e4 00 00 notify enabled on fec2
SEND time: 00 01 fd c1 1f
— NOTIFY — hex=00 14 06 11 09 11 02 11 01 11 01 11 02 09 06 31 35 3a 31 34 00 20 10 d9 text= 15:14 Щ
SEND energy: 00 01 ee e3 4d
— NOTIFY — hex=00 3f 03 11 2f 09 1f c0 ea f2 e8 e2 2e fd ed 2e 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 ef f0 ff ec 2e 00 07 ff 00 08 01 00 01 00 00 09 07 31 33 37 2e 34 34 00 09 06 ea c2 f2 2a f7 00 00 6d eb text= ?/ Актив.эн. прям. я 137.44 кВт*ч mл
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 41 03 11 2f 09 21 c0 ea f2 e8 e2 2e fd ed 2e 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 ef f0 ff ec 2e f2 2e 31 00 07 ff 01 08 01 00 01 00 00 09 07 31 30 34 2e 32 31 00 09 06 ea c2 f2 2a f7 00 02 49 28 text= A/ !Актив.эн. прям.т.1 104.21 кВт*ч I(
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 40 03 11 2f 09 21 c0 ea f2 e8 e2 2e fd ed 2e 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 ef f0 ff ec 2e f2 2e 32 00 07 ff 02 08 01 00 01 00 00 09 06 33 33 2e 32 33 00 09 06 ea c2 f2 2a f7 00 00 45 6a text= @/ !Актив.эн. прям.т.2 33.23 кВт*ч Ej
SEND next: 00 01 08 7e a5
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 24 03 11 34 09 07 20 c2 d0 c5 cc df 00 07 ff 01 09 00 00 01 00 00 09 09 31 35 3a 31 34 3a 34 33 00 09 01 00 2e 40 ce text= $4 ВРЕМЯ я 15:14:43 .@О
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 3f 03 11 2f 09 1f c0 ea f2 e8 e2 2e fd ed 2e 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 ef f0 ff ec 2e 00 07 ff 00 08 01 00 01 00 00 09 07 31 33 37 2e 34 34 00 09 06 ea c2 f2 2a f7 00 00 6d eb text= ?/ Актив.эн. прям. я 137.44 кВт*ч mл
SEND params: 00 01 02 df ef
— NOTIFY — hex=00 24 01 11 01 11 05 11 17 09 1a 20 20 20 d2 c5 ca d3 d9 c8 c5 20 20 20 20 20 cf c0 d0 c0 cc c5 d2 d0 db 20 00 00 8f cf text= $ ␦ ТЕКУЩИЕ ПАРАМЕТРЫ ЏП
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 31 02 11 01 11 10 11 01 11 05 07 ff 01 09 00 00 01 00 00 09 0e 20 20 20 20 20 20 20 20 c2 d0 c5 cc df 00 09 09 31 35 3a 31 34 3a 34 37 00 09 01 00 33 56 60 text= 1я ВРЕМЯ 15:14:47 3V`
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 30 02 11 02 11 10 11 01 11 05 07 ff 02 09 00 00 01 00 00 09 0d 20 20 20 20 20 20 20 20 c4 c0 d2 c0 00 09 09 32 39 2e 30 33 2e 32 36 00 09 01 00 09 0e e3 ДАТА 29.03.26 г
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 2f 02 11 03 11 10 11 01 11 05 07 ff 00 07 1f 00 01 00 00 09 0f 20 20 20 20 20 20 d2 ce ca 20 d4 c0 c7 db 00 09 05 33 2e 33 36 00 09 02 c0 00 00 5b 5b text= /я ТОК ФАЗЫ 3.36 А [[
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 44 02 11 04 11 10 11 01 11 05 07 ff 00 07 20 00 01 00 00 09 22 20 20 20 20 20 20 cd c0 cf d0 df c6 c5 cd c8 c5 20 20 20 20 20 20 20 20 20 20 20 20 20 d4 c0 c7 db 00 09 07 32 33 32 2e 37 39 00 09 02 c2 00 09 27 ec text= Dя " НАПРЯЖЕНИЕ ФАЗЫ 232.79 В 'м
SEND month: 00 01 01 ef 8c
— NOTIFY — hex=00 24 01 11 01 11 05 11 17 09 1a 20 20 20 d2 c5 ca d3 d9 c8 c5 20 20 20 20 20 cf c0 d0 c0 cc c5 d2 d0 db 20 00 2e 4a 63 text= $ ␦ ТЕКУЩИЕ ПАРАМЕТРЫ .Jc
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 31 02 11 01 11 10 11 01 11 05 07 ff 01 09 00 00 01 00 00 09 0e 20 20 20 20 20 20 20 20 c2 d0 c5 cc df 00 09 09 31 35 3a 31 34 3a 35 33 00 09 01 00 06 8e 56 text= 1я ВРЕМЯ 15:14:53 ЋV
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 30 02 11 02 11 10 11 01 11 05 07 ff 02 09 00 00 01 00 00 09 0d 20 20 20 20 20 20 20 20 c4 c0 d2 c0 00 09 09 32 39 2e 30 33 2e 32 36 00 09 01 00 09 0e e3 ДАТА 29.03.26 г
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 2f 02 11 03 11 10 11 01 11 05 07 ff 00 07 1f 00 01 00 00 09 0f 20 20 20 20 20 20 d2 ce ca 20 d4 c0 c7 db 00 09 05 33 2e 34 32 00 09 02 c0 00 00 5d 85 text= /я ТОК ФАЗЫ 3.42 А ]…
SEND next: 00 01 08 7e a5
— NOTIFY — hex=00 44 02 11 04 11 10 11 01 11 05 07 ff 00 07 20 00 01 00 00 09 22 20 20 20 20 20 20 cd c0 cf d0 df c6 c5 cd c8 c5 20 20 20 20 20 20 20 20 20 20 20 20 20 d4 c0 c7 db 00 09 07 32 33 32 2e 37 34 00 09 02 c2 00 09 cd e6 text= Dя " НАПРЯЖЕНИЕ ФАЗЫ 232.74 В Нж
А следовательно у нас победа! Мы смогли напрямую получить данные со счётчика МИР С05 по bluetooth на Raspberry. Осталось только "причесать" код, чтобы выбранные данные сохранялись в JSON для последующей работы с "умным домом" Home Assistant.
Но хотелось бы всё-таки немного конкретнее определиться со строкой инициализации, подсмотренной snoop_log. Ведь где-то в ней зашит пин-код, который выдала организация. Чтобы ускорить процесс я снова скормил нейронке лог и попросил вычислить пин-код производителя в строке инициализации. И вот что она вычислила:
Исходя из полученных знаний добавил в код Raspberry пин-код в виде константы. И в результате работающий код получается такой:
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Считывание показаний со счётчика МИР С-05 по BLE. Что делает скрипт: 1. Подключается к счётчику по BLE. 2. Выполняет служебную инициализацию, которую делает штатное приложение. 3. Формирует авторизационный пакет из PIN-кода. 4. Включает приём notify-уведомлений. 5. Последовательно отправляет команды чтения. 6. Печатает ответы счётчика: - общий расход - Т1 - Т2 - дату/время - ток - напряжение - и другие параметры, если счётчик их отдаёт. Важно: - обычный BLE pairing для этого скрипта не требуется; - связь идёт через GATT characteristics; - логика инициализации и команды были восстановлены экспериментально по фактическому обмену штатного Android-приложения; - предполагается, что PIN передаётся в характеристику D24A как 16-битное little-endian число + 2 нулевых байта. Для PIN 58525 это даёт пакет: 9d e4 00 00 Зависимость: pip install bleak """ import asyncio from bleak import BleakClient # ============================================================ # НАСТРОЙКИ # ============================================================ # BLE MAC-адрес счётчика. # При необходимости замени на свой. ADDR = "E4:06:BF:87:CD:69" # PIN-код, выданный поставщиком для данного счётчика. # ВАЖНО: # Это не обязательно "Bluetooth PIN" в системном смысле. # Это прикладной код авторизации, # который приложение передаёт в бинарном виде. PIN = 58525 # UUID характеристик, найденных экспериментально. # # UUID_D24A: # Служебная характеристика для передачи авторизационного пакета. # # UUID_B3F7: # Служебная характеристика для начальной инициализации сеанса. # # UUID_FEC2: # Главный рабочий канал: # - сюда пишутся команды # - отсюда приходят notify-ответы UUID_D24A = "d24a5138-1448-48ea-a983-f7df274c6d89" UUID_B3F7 = "b3f7e595-2951-42fa-879e-0d9dfa5e846e" UUID_FEC2 = "fec26ec4-6d71-4442-9f81-55bc21d658d6" # ============================================================ # НАБОР КОМАНД # ============================================================ # Это последовательность команд, которая показала рабочий результат. # Команды отправляются в UUID_FEC2. # # Комментарии по смыслу: # - "time" : запрос времени # - "energy" : старт чтения блока энергий # - "next" : переход к следующему пункту/экрану # - "params" : переход к текущим параметрам # - "month" : в эксперименте эта команда снова возвращала # блок текущих параметров, поэтому её назначение # пока не подтверждено окончательно CMDS = [ ("time", bytes.fromhex("0001fdc11f")), ("energy", bytes.fromhex("0001eee34d")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("params", bytes.fromhex("000102dfef")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("month", bytes.fromhex("000101ef8c")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ] # ============================================================ # ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ # ============================================================ def build_auth_payload(pin: int) -> bytes: """ Формирует бинарный авторизационный пакет из PIN-кода. Логика преобразования: 1. Берём PIN как целое число. 2. Упаковываем его в 2 байта little-endian. Пример: 58525 (dec) = 0xE49D little-endian -> 9d e4 3. Добавляем ещё 2 нулевых байта. Итого: 9d e4 00 00 Почему именно так: В BLE-логе штатного приложения перед чтением данных в характеристику D24A записывался пакет: 9d e4 00 00 который численно совпадает с PIN 58525. Ограничение: Такой формат вмещает значения только 0..65535, то есть максимум 16-битный PIN. """ if not (0 <= pin <= 0xFFFF): raise ValueError("PIN must be in range 0..65535") return pin.to_bytes(2, "little") + b"\x00\x00" def pretty(data: bytes) -> str: """ Красивое представление бинарного ответа. Возвращает: - hex-строку - текстовую интерпретацию в cp1251 Почему cp1251: Счётчик отдаёт русские строки именно в этой кодировке. UTF-8 тут не подходит. """ text = data.decode("cp1251", errors="ignore") # В ответах много нулевых байтов; заменяем их на пробелы, # чтобы текст читался понятнее. text = text.replace("\x00", " ") return f"hex={data.hex(' ')}\ntext={text}" def notify_callback(sender: int, data: bytearray) -> None: """ Обработчик notify-уведомлений от счётчика. sender: внутренний идентификатор характеристики в bleak/BlueZ data: полезная нагрузка уведомления """ raw = bytes(data) print("\n--- NOTIFY ---") print(pretty(raw)) # ============================================================ # ОСНОВНАЯ ЛОГИКА # ============================================================ async def main() -> None: """ Основная процедура: - подключение - поиск характеристик - формирование авторизации из PIN - инициализация - включение notify - отправка команд - ожидание ответов """ # Заранее формируем авторизационный пакет из PIN. auth_payload = build_auth_payload(PIN) print(f"PIN: {PIN}") print(f"auth payload: {auth_payload.hex(' ')}") # Создаём BLE-клиент и подключаемся к счётчику. async with BleakClient(ADDR, timeout=20.0) as client: print("connected:", client.is_connected) # Получаем объекты характеристик по UUID. # # Объект характеристики нужен bleak для чтения/записи/notify. ch_d24a = client.services.get_characteristic(UUID_D24A) ch_b3f7 = client.services.get_characteristic(UUID_B3F7) ch_fec2 = client.services.get_characteristic(UUID_FEC2) # Базовая проверка: если какая-то характеристика не нашлась, # значит либо устройство не то, либо сервисы не прочитались. if ch_d24a is None: raise RuntimeError(f"Characteristic {UUID_D24A} not found") if ch_b3f7 is None: raise RuntimeError(f"Characteristic {UUID_B3F7} not found") if ch_fec2 is None: raise RuntimeError(f"Characteristic {UUID_FEC2} not found") # -------------------------------------------------------- # ШАГ 1. СЛУЖЕБНАЯ ИНИЦИАЛИЗАЦИЯ # -------------------------------------------------------- # # Эта часть была подсмотрена в работе штатного приложения. # Без неё счётчик либо не отвечает, либо отвечает не так, как нужно. # Отправляем 0x01 в характеристику B3F7. # По смыслу это похоже на включение/разрешение сеанса обмена. await client.write_gatt_char(ch_b3f7, b"\x01", response=True) print("init: b3f7 <- 01") # Отправляем авторизационный пакет в характеристику D24A. # Пакет формируется автоматически из PIN, # чтобы было понятно, откуда берутся байты, # и чтобы можно было легко заменить PIN при необходимости. await client.write_gatt_char(ch_d24a, auth_payload, response=True) print(f"init: d24a <- {auth_payload.hex(' ')}") # -------------------------------------------------------- # ШАГ 2. ВКЛЮЧАЕМ NOTIFY # -------------------------------------------------------- # # После этого счётчик сможет присылать ответы асинхронно. # Все ответы будут обрабатываться функцией notify_callback. await client.start_notify(ch_fec2, notify_callback) print("notify enabled on fec2") # Небольшая пауза, чтобы устройство успело войти в нужное состояние. await asyncio.sleep(0.5) # -------------------------------------------------------- # ШАГ 3. ПОСЛЕДОВАТЕЛЬНО ОТПРАВЛЯЕМ КОМАНДЫ # -------------------------------------------------------- # # Команды идут в рабочую характеристику FEC2. # Ответы приходят не как return write-функции, # а отдельными notify-уведомлениями. for name, cmd in CMDS: print(f"\nSEND {name}: {cmd.hex(' ')}") # response=False: # запись без ожидания подтверждения на уровне GATT write response. # Это соответствует тому, как устройство работает в данном случае. await client.write_gatt_char(ch_fec2, cmd, response=False) # Даём счётчику время сформировать и прислать очередной ответ. await asyncio.sleep(1.2) # -------------------------------------------------------- # ШАГ 4. ЖДЁМ ПОСЛЕДНИЕ УВЕДОМЛЕНИЯ # -------------------------------------------------------- # # После последней команды полезно немного подождать, # потому что ответы могут прийти не мгновенно. await asyncio.sleep(5) # ============================================================ # ТОЧКА ВХОДА # ============================================================ if __name__ == "__main__": asyncio.run(main())
Далее осталось переделать выводимые данные в JSON для того, чтобы Home Assistant мог нормально работать с данными со счётчика. Для этого модифицируем код до такого состояния
#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Считывание данных со счётчика МИР С-05 по BLE с выводом результата в виде JSON. Что делает скрипт: 1. Подключается к счётчику по BLE. 2. Выполняет служебную инициализацию. 3. Формирует авторизационный пакет из PIN. 4. Включает notify на рабочей характеристике. 5. Последовательно отправляет команды. 6. Разбирает ответы и складывает значения в словарь. 7. В конце печатает один JSON-объект. Скрипт рассчитан на использование в Home Assistant, MQTT, cron/systemd и других автоматизациях, где нужен чистый JSON без лишнего текстового мусора. Зависимость: pip install bleak """ import asyncio import json import re from typing import Optional from bleak import BleakClient # ============================================================ # НАСТРОЙКИ # ============================================================ # MAC-адрес счётчика ADDR = "E4:06:BF:87:CD:69" # PIN-код счётчика PIN = 58525 # UUID характеристик UUID_D24A = "d24a5138-1448-48ea-a983-f7df274c6d89" UUID_B3F7 = "b3f7e595-2951-42fa-879e-0d9dfa5e846e" UUID_FEC2 = "fec26ec4-6d71-4442-9f81-55bc21d658d6" # ============================================================ # КОМАНДЫ ОБМЕНА # ============================================================ CMDS = [ ("time", bytes.fromhex("0001fdc11f")), ("energy", bytes.fromhex("0001eee34d")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("params", bytes.fromhex("000102dfef")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ("next", bytes.fromhex("0001087ea5")), ] # ============================================================ # ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ # ============================================================ def build_auth_payload(pin: int) -> bytes: # Формирует бинарный пакет авторизации из PIN. if not (0 <= pin <= 0xFFFF): raise ValueError("PIN must be in range 0..65535") return pin.to_bytes(2, "little") + b"\x00\x00" def decode_cp1251(data: bytes) -> str: # Декодирует ответ счётчика из cp1251. return data.decode("cp1251", errors="ignore").replace("\x00", " ") def normalize_spaces(text: str) -> str: # Убирает лишние пробелы и переводит текст в более удобный вид. return re.sub(r"\s+", " ", text).strip() def extract_float(text: str) -> Optional[float]: # Ищет в строке первое число вида 123.45 и возвращает float. # Если число не найдено, возвращает None. match = re.search(r"(\d+\.\d+)", text) if match: try: return float(match.group(1)) except ValueError: return None return None # ============================================================ # ХРАНИЛИЩЕ РЕЗУЛЬТАТОВ # ============================================================ # Сюда будут складываться значения, которые мы вытащим из notify-ответов. result = { "meter_mac": ADDR, "pin_used": PIN, "total_kwh": None, "t1_kwh": None, "t2_kwh": None, "date": None, "time": None, "current_a": None, "voltage_v": None, } # ============================================================ # ПАРСИНГ ОТВЕТОВ # ============================================================ def parse_notify_payload(data: bytes) -> None: """ Разбирает один notify-ответ счётчика и, если находит полезные данные, записывает их в глобальный словарь result. Ориентируемся на текстовые фрагменты в cp1251: - Актив.эн. ... прям. - Актив.эн. ... прям.т.1 - Актив.эн. ... прям.т.2 - ДАТА - ВРЕМЯ - ТОК ФАЗЫ - НАПРЯЖЕНИЕ ФАЗЫ """ global result raw_text = decode_cp1251(data) text = normalize_spaces(raw_text) # Общая активная энергия # # Важно: # У T1/T2 тоже есть "Актив.эн.", поэтому сначала проверяем более # специфичные варианты T1/T2, а потом уже общий total. if "Актив.эн." in text and "прям.т.1" in text: value = extract_float(text) if value is not None: result["t1_kwh"] = value return if "Актив.эн." in text and "прям.т.2" in text: value = extract_float(text) if value is not None: result["t2_kwh"] = value return # Общая энергия без тарифного хвоста if "Актив.эн." in text and "прям." in text and "прям.т.1" not in text and "прям.т.2" not in text: value = extract_float(text) if value is not None: result["total_kwh"] = value return # Дата if "ДАТА" in text: match = re.search(r"(\d{2}\.\d{2}\.\d{2})", text) if match: result["date"] = match.group(1) return # Время if "ВРЕМЯ" in text: match = re.search(r"(\d{2}:\d{2}:\d{2}|\d{2}:\d{2})", text) if match: result["time"] = match.group(1) return # Ток фазы if "ТОК ФАЗЫ" in text: value = extract_float(text) if value is not None: result["current_a"] = value return # Напряжение фазы if "НАПРЯЖЕНИЕ" in text and "ФАЗЫ" in text: value = extract_float(text) if value is not None: result["voltage_v"] = value return def notify_callback(sender: int, data: bytearray) -> None: # Обработчик notify-ответов от счётчика. # Вместо печати сырых пакетов мы сразу пытаемся распарсить данные # и обновить словарь result. parse_notify_payload(bytes(data)) # ============================================================ # ОСНОВНАЯ ЛОГИКА # ============================================================ async def main() -> None: """ Основная процедура: - создаём авторизационный пакет - подключаемся - инициализируем сеанс - включаем notify - отправляем команды - ждём ответы - печатаем один JSON """ auth_payload = build_auth_payload(PIN) async with BleakClient(ADDR, timeout=20.0) as client: # Получаем характеристики ch_d24a = client.services.get_characteristic(UUID_D24A) ch_b3f7 = client.services.get_characteristic(UUID_B3F7) ch_fec2 = client.services.get_characteristic(UUID_FEC2) if ch_d24a is None: raise RuntimeError(f"Characteristic {UUID_D24A} not found") if ch_b3f7 is None: raise RuntimeError(f"Characteristic {UUID_B3F7} not found") if ch_fec2 is None: raise RuntimeError(f"Characteristic {UUID_FEC2} not found") # Служебная инициализация: # 1. включаем сеанс # 2. отправляем авторизационный пакет из PIN await client.write_gatt_char(ch_b3f7, b"\x01", response=True) await client.write_gatt_char(ch_d24a, auth_payload, response=True) # Включаем notify на основном канале ответов await client.start_notify(ch_fec2, notify_callback) # Небольшая пауза после инициализации await asyncio.sleep(0.5) # Отправляем все команды по очереди for _name, cmd in CMDS: await client.write_gatt_char(ch_fec2, cmd, response=False) await asyncio.sleep(1.2) # Ждём последние ответы await asyncio.sleep(5) # После завершения сеанса печатаем только JSON. # ensure_ascii=False нужен, чтобы русский текст не превращался в \uXXXX print(json.dumps(result, ensure_ascii=False, indent=2)) # ============================================================ # ТОЧКА ВХОДА # ============================================================ if __name__ == "__main__": asyncio.run(main())
В результате выполнения скрипта мы получаем на выходе вот такие данные:
{ “meter_mac”: “E4:06:BF:87:CD:69”, “pin_used”: 58525, “total_kwh”: 138.22, “t1_kwh”: 104.99, “t2_kwh”: null, “date”: “29.03.26”, “time”: “16:20:36”, “current_a”: 3.36, “voltage_v”: 232.42 }
Таким образом главную задачу считаю выполненной. И дальше в планах только перенос этой функции на ESP32 с Raspberry.
P.S.: Я понимаю, что в целом задача возможно высосана из пальца в рамках одного счётчика и одной квартиры, но возможно кому-то пригодятся мои куски кода или реализации для каких-то более глобальных и интересных задач.