Давно мучал вопрос передачи показаний давления системы отопления со штатного манометра газового котла. Для этого несколько лет назад была приобретена камера ESP32-CAM и интегрирована в Home Assistant посредством ESPHome.

Камеру я направил прямо на манометр, что позволило мне периодически вручную удаленно контролировать давление в системе отопления, и, при необходимости, открывать кран подачи воды в систему (также удалённо).

Для того, чтобы хоть как-то облагородить этот "аналоговый" процесс в Node-RED была создана автоматизация, завязанная на контроле значений энергопотребления котла и насосов:

Работало это так:

2. В Телеграм приходит уведомление об ошибке котла;

3. Я лезу в интерфейс HA и в камере смотрю какое давление;

4. Если давление низкое - открываю кран подпитки.

Уже неплохо, но руки-то чешутся, поэтому я решил, что изображение надо распознать и вывести давление в HA в нормальном виде.

Выбор способа оцифровки

Помимо Raspberry Pi 4B, на котором у меня крутится HA у меня есть самосборный NAS c Xpenology на борту. Поэтому я решил задействовать в этом деле его.

Установка контейнера Python в Xpenology

Первым делом в Container Manager загружаем образ Python 3-10-slim

Создание контейнера:

1. Открываем вкладку Образы, выбираем python и нажимаем Запустить;

2. Как-нибудь называем контейнер;

3. На этапе Настройка томов:

   • Нажимаем Добавить папку

   • Выбираем папку /volume1/docker/gauge

   • В контейнере указываем путь: /app

4. Нажимаем Далее, потом Применить

Далее временно включаем доступ к NAS по SSH в панели управления:

И подключаемся к с серверу с помощью Putty

Вводим логин, пароль, переключаемся на root, получаем ID контейнера:

docker ps

Заходим в контейнер:

docker exec -it ID_контейнера bash

Устанавливаем необходимые библиотеки:

apt update && apt install -y python3-pip libgl1 libglib2.0-0
apt-get install -y python3-pip python3-dev
apt-get install -y libglib2.0-0 libsm6 libxext6 libxrender-dev
apt-get install -y qt5-qmake qtbase5-dev-tools qtchooser
apt-get install -y libx11-dev libgl1-mesa-glx libfontconfig1 libxkbcommon0
apt-get install -y libxcb-xinerama0
pip install opencv-python numpy
pip3 install opencv-python matplotlib

Создаем скрипт gauge_reader.py и кладем его в папку docker\gauge на NAS. К нему мы вернемся чуть позже.

Настройка Home Assistant

Теперь мне нужно положить в эту же папку, где лежит скрипт само изображение, которое нужно распознать. Для этого делаем следующее:

1. В HA переходим в Настройки, Хранилище, Добавить сетевое хранилище, указываем путь к папке docker на NAS:

После создаем автоматизацию, которая будет каждый час включать светодиод на ESP32-CAM, делать снимок, выкладывать его в папку gauge, которую мы подключили шагом выше, и выключать светодиод:

alias: ESP Cam snapshot каждый час
description: ""
triggers:
  - minutes: 0
    trigger: time_pattern
actions:
  - target:
      entity_id: switch.pressure_boiler_flash_led
    action: switch.turn_on
    data: {}
  - delay:
      hours: 0
      minutes: 0
      seconds: 10
      milliseconds: 0
  - data:
      entity_id: camera.pressure_boiler_my_camera
      filename: /media/gauge/sample.jpg
    action: camera.snapshot
  - delay:
      hours: 0
      minutes: 0
      seconds: 10
      milliseconds: 0
  - target:
      entity_id: switch.pressure_boiler_flash_led
    action: switch.turn_off
    data: {}
mode: single

И сразу создадим сенсор, в который будет присылать NAS результат распознавания давления:

sensor:
    - name: "Boiler Pressure"
      state_topic: "sensors/boiler/pressure"
      unit_of_measurement: "bar"
      device_class: pressure
      unique_id: pressure_boiler

Подготовка фото

Теперь надо немного поколдовать с самим объектом - манометром. Спустя пару десятков итераций со скриптом я пришел к тому, что несмотря на все ухищрения, качество изображения не позволяет со 100% результатом распознать стрелку манометра. Поэтому я наклеил вокруг манометра квадрат для возможности правильной обрезки и коррекции перспективы. Получилось так:

Написание скрипта (хвала нейросетям!)

Теперь переходим к скрипту. Задачу по определению давления разбиваем на следующие шаги:

1. Обнаруживаем на фото четырехугольник;

2. Корректируем перспективу - превращаем четырехугольник в квадрат и обрезаем изображение;

3. Определяем центр и стрелку манометра;

4. Замеряем угол отклонения стрелки и по таблице калибровки переводим угол-градусы в давление-бары;

5. Отрисовываем некоторые этапы для контроля выполнения кода скрипта;

6. Отправляем полученные данные в mqtt HA.

import cv2
import numpy as np
import math
import logging
import paho.mqtt.client as mqtt
from typing import Optional, Tuple, List

# Настройка логгирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Конфигурация
class Config:
    MQTT_BROKER = "192.168.1.****"
    MQTT_PORT = ****
    MQTT_TOPIC = "sensors/boiler/pressure"
    MQTT_USER = "****"
    MQTT_PASS = "****"
    MQTT_TIMEOUT = 5
    
    CALIBRATION = {
        215: 0.0,
        155: 1.0,
        90: 2.0,
        29: 3.0,
        327: 4.0
    }
    
    OUTPUT_DIR = '/app/'
    OUTPUT_SIZE = 600
    BLUR_SIZE = (5, 5)
    ADAPTIVE_THRESH_BLOCK = 11
    ADAPTIVE_THRESH_C = 2
    HOUGH_CIRCLES_PARAMS = {
        'dp': 1,
        'minDist': 100,
        'param1': 50,
        'param2': 30,
        'minRadius': 100,
        'maxRadius': 200
    }
    HOUGH_LINES_PARAMS = {
        'rho': 1,
        'theta': np.pi/180,
        'threshold': 50,
        'minLineLength': 150,
        'maxLineGap': 10
    }

def detect_quadrilateral(image: np.ndarray) -> Tuple[Optional[List[Tuple[int, int]]], np.ndarray]:
    """Определение углов четырехугольника вокруг манометра"""
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    _, thresh = cv2.threshold(gray, 50, 255, cv2.THRESH_BINARY_INV)
    
    kernel = np.ones((5,5), np.uint8)
    thresh = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)
    thresh = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)
    
    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    
    if not contours:
        return None, image.copy()
    
    largest_contour = max(contours, key=cv2.contourArea)
    epsilon = 0.03 * cv2.arcLength(largest_contour, True)
    approx = cv2.approxPolyDP(largest_contour, epsilon, True)
    
    if len(approx) != 4:
        for eps in [0.02, 0.04, 0.05]:
            epsilon = eps * cv2.arcLength(largest_contour, True)
            approx = cv2.approxPolyDP(largest_contour, epsilon, True)
            if len(approx) == 4:
                break
        else:
            return None, image.copy()
    
    points = np.array([point[0] for point in approx], dtype=np.float32)
    center = np.mean(points, axis=0)
    
    def sort_key(point):
        return math.atan2(point[1] - center[1], point[0] - center[0])
    
    sorted_points = sorted(points, key=sort_key, reverse=True)
    bottom_idx = np.argmax([p[1] for p in sorted_points])
    sorted_points = np.roll(sorted_points, -bottom_idx, axis=0)
    
    if bottom_idx == 1:
        sorted_points = np.roll(sorted_points, -1, axis=0)
    
    corners = [tuple(map(int, p)) for p in sorted_points]
    
    vis = image.copy()
    for i, corner in enumerate(corners):
        cv2.circle(vis, corner, 10, (0, 255, 0), -1)
        cv2.putText(vis, f"{i+1}", (corner[0]+15, corner[1]+15), 
                   cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
    
    for i in range(4):
        cv2.line(vis, corners[i], corners[(i+1)%4], (255, 0, 0), 2)
    
    return corners, vis

def correct_perspective(image: np.ndarray, corners: List[Tuple[int, int]]) -> np.ndarray:
    """Коррекция перспективы на основе найденных углов"""
    size = Config.OUTPUT_SIZE
    dst_points = np.array([
        [0, size-1],
        [size-1, size-1],
        [size-1, 0],
        [0, 0]
    ], dtype=np.float32)
    
    src_points = np.array(corners, dtype=np.float32)
    matrix = cv2.getPerspectiveTransform(src_points, dst_points)
    corrected = cv2.warpPerspective(image, matrix, (size, size))
    
    return corrected

def enhance_image(image: np.ndarray, roi: Tuple[int, int, int, int]) -> Tuple[np.ndarray, np.ndarray]:
    """Улучшение изображения и выделение стрелки"""
    x, y, w, h = roi
    cropped = image[y:y+h, x:x+w]
    gray = cv2.cvtColor(cropped, cv2.COLOR_BGR2GRAY)
    blurred = cv2.GaussianBlur(gray, Config.BLUR_SIZE, 0)
    thresh = cv2.adaptiveThreshold(
        blurred, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
        cv2.THRESH_BINARY_INV, Config.ADAPTIVE_THRESH_BLOCK, Config.ADAPTIVE_THRESH_C
    )
    return thresh, cropped

def find_dial_center(image: np.ndarray, roi: Tuple[int, int, int, int]) -> Tuple[Tuple[int, int, int], np.ndarray]:
    """Определение центра циферблата"""
    x, y, w, h = roi
    cropped = image[y:y+h, x:x+w]
    gray = cv2.cvtColor(cropped, cv2.COLOR_BGR2GRAY)
    vis = cropped.copy()
    
    circles = cv2.HoughCircles(gray, cv2.HOUGH_GRADIENT, **Config.HOUGH_CIRCLES_PARAMS)
    
    if circles is not None:
        circles = np.uint16(np.around(circles))
        best = circles[0][0]
        cv2.circle(vis, (best[0], best[1]), best[2], (0, 255, 0), 2)
        cv2.circle(vis, (best[0], best[1]), 2, (0, 0, 255), 3)
        center_vis_path = f"{Config.OUTPUT_DIR}center_detection.jpg"
        cv2.imwrite(center_vis_path, vis)
        return (x + best[0], y + best[1], best[2]), vis
    
    center = (x + w//2, y + h//2, min(w, h)//2)
    cv2.circle(vis, (w//2, h//2), min(w, h)//2, (0, 255, 0), 2)
    center_vis_path = f"{Config.OUTPUT_DIR}center_detection.jpg"
    cv2.imwrite(center_vis_path, vis)
    return center, vis

def find_arrow_angle(thresh: np.ndarray, original: np.ndarray, 
                    roi: Tuple[int, int, int, int], 
                    center_info: Tuple[int, int, int]) -> Tuple[Optional[float], Optional[Tuple[int, int]], np.ndarray]:
    """Поиск угла стрелки"""
    x, y, w, h = roi
    cX, cY, radius = center_info
    cropped = original[y:y+h, x:x+w]
    cX_roi, cY_roi = cX - x, cY - y
    
    params = Config.HOUGH_LINES_PARAMS.copy()
    params['minLineLength'] = radius * 0.5
    
    edges = cv2.Canny(thresh, 50, 150)
    lines = cv2.HoughLinesP(edges, **params)
    
    if lines is not None:
        filtered = []
        for line in lines:
            x1, y1, x2, y2 = line[0]
            dist1 = math.hypot(x1 - cX_roi, y1 - cY_roi)
            dist2 = math.hypot(x2 - cX_roi, y2 - cY_roi)
            if min(dist1, dist2) < radius * 0.3:
                filtered.append(line)
        
        if filtered:
            longest = max(filtered, key=lambda l: math.hypot(l[0][0]-l[0][2], l[0][1]-l[0][3]))
            x1, y1, x2, y2 = longest[0]
            
            tip = (x2, y2) if math.hypot(x1-cX_roi, y1-cY_roi) < math.hypot(x2-cX_roi, y2-cY_roi) else (x1, y1)
            angle = math.degrees(math.atan2(cY_roi - tip[1], tip[0] - cX_roi)) % 360
            
            cv2.line(cropped, (cX_roi, cY_roi), tip, (0, 0, 255), 3)
            return angle, (cX, cY), cropped
    
    return None, None, cropped

def setup_mqtt() -> mqtt.Client:
    """Настройка MQTT клиента"""
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    client.username_pw_set(Config.MQTT_USER, Config.MQTT_PASS)
    client.connect(Config.MQTT_BROKER, Config.MQTT_PORT, Config.MQTT_TIMEOUT)
    return client

def calibrate_pressure(angle: float) -> float:
    """Калибровка давления"""
    angles = np.array(sorted(Config.CALIBRATION.keys()))
    pressures = np.array([Config.CALIBRATION[a] for a in angles])
    return float(np.interp(angle, angles, pressures))

def send_mqtt(pressure: float) -> bool:
    """Отправка данных через MQTT"""
    try:
        client = setup_mqtt()
        client.publish(Config.MQTT_TOPIC, f"{pressure:.2f}")
        client.disconnect()
        return True
    except Exception as e:
        logger.error(f"Ошибка MQTT: {str(e)}")
        return False

def process_image(image_path: str) -> Optional[float]:
    """Основная функция обработки изображения"""
    try:
        image = cv2.imread(image_path)
        if image is None:
            raise FileNotFoundError(f"Не удалось загрузить изображение: {image_path}")
        
        quad_corners, quad_vis = detect_quadrilateral(image)
        if quad_corners is None:
            logger.error("Не удалось обнаружить четырехугольник")
            return None
            
        cv2.imwrite(f"{Config.OUTPUT_DIR}quadrilateral_detection.jpg", quad_vis)
        corrected_image = correct_perspective(image, quad_corners)
        cv2.imwrite(f"{Config.OUTPUT_DIR}corrected_perspective.jpg", corrected_image)
        
        corrected_roi = (0, 0, Config.OUTPUT_SIZE, Config.OUTPUT_SIZE)
        thresh, cropped = enhance_image(corrected_image, corrected_roi)
        center, _ = find_dial_center(corrected_image, corrected_roi)
        angle, _, processed = find_arrow_angle(thresh, corrected_image, corrected_roi, center)
        
        if angle is None:
            logger.error("Не удалось определить угол стрелки")
            return None
        
        pressure = calibrate_pressure(angle)
        cv2.putText(processed, f"Angle: {angle:.1f}°", (20, 40), 
                   cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        cv2.putText(processed, f"Pressure: {pressure:.2f} bar", (20, 80),
                   cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        
        cv2.imwrite(f"{Config.OUTPUT_DIR}processed.jpg", processed)
        return pressure
        
    except Exception as e:
        logger.error(f"Ошибка обработки: {str(e)}")
        return None

def main():
    image_path = '/app/sample.jpg'
    pressure = process_image(image_path)
    if pressure is not None:
        send_mqtt(pressure)

if __name__ == "__main__":
    main()

Код скрипта писал трудяга DeepSeek под моим чутким руководством. Сначала пытался это сделать в ChatGPT, но после достижения бесплатного лимита запросов он резко "тупел" и дело шло уже не так хорошо:)

Настройка автозапуска скрипта в Xpenology

Осталось дело за малым - настраиваем на NAS автозапуск скрипта. Переходим в Панель управления, Планировщик задач, Создать, Запланированная задача, Скрипт заданный пользователем.

Во вкладке Общие указываем, что задача должна выполняться от пользователя root. Расписание - ежедневно каждый час. В Настройки задач в поле Выполнить команду прописываем docker exec ID_контейнера python3 /app/gauge_reader.py.

Результат

Теперь раз в час HA делает фотографию манометра и кладет ее в папку NAS. А NAS спустя пять минут распознает это дело шлёт в mqtt фактическое давление в системе отопления.

Да, я в курсе, что сейчас уже есть DIY zigbee манометры. Может быть я соберусь с мыслями, возьму паяльник и врежусь в систему отопления и в систему водоснабжения, чтобы можно было автоматизировать включение крана подпитки системы отопления только в случае наличия холодной воды. Но пока пусть поработает так.

Почему не реализовал распознавание изображения на Raspberry Pi? Просто так захотелось))

Сама идея взаимодействия двух устройств в домашней сети показалась интересной.