Привет, Хабр! Меня зовут Андрей Бирюков. Я независимый эксперт в области ИТ и ИБ, преподаю в учебных центрах и пишу статьи и книги. И сегодня мы поговорим о том, как можно обнаружить проблемы в работе оборудования задолго до того, как оно выйдет из строя.

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

Из подобных проблем нужно сделать один важный вывод: большинство поломок не случаются внезапно. Они вызревают неделями, просто никто не смотрит туда, где зарождается проблема. Потому что штатные системы мониторинга показывают только «всё в норме», пока не становится слишком поздно.

Сегодня мы соберем работающий цифровой двойник. Не за миллион, не на основе дорогостоящих датчиков и ПО, и не за полгода. Его можно будет собрать за выходные, на Python, ESP32 и открытых библиотеках. И он начнет предсказывать проблемы раньше, чем ваш мастер по ремонту достанет телефон.

Почему заводской мониторинг врет

Для начала давайте разберемся, почему заводские системы мониторинга дают неверные результаты. Стандартная система сбора данных с ЧПУ или насоса обычно показывает три вещи: температура в норме, давление в норме, ток в норме. Проблема в том, что к моменту, когда эти параметры выходят за красную черту, оборудование уже убито.

Проблема в том, что реальная физика разрушения выглядит иначе. Микровибрации нарастают за 200-300 часов до отказа. Спектр вибрации меняется, когда в подшипнике появляется микротрещина диаметром с волос. Но штатные ПЛК (программируемые логические контроллеры) не умеют анализировать спектр — они просто смотрят, не превысило ли значение порог.

Цифровой двойник решает эту проблему принципиально иначе. Он не ждет, пока датчик заорет «SOS». Он учится на истории работы станка и кричит сам, когда видит неладное.

Как это работает: вы строите модель «нормального поведения» станка. Если реальные данные отклоняются от того, что модель считает нормой — вы получаете сигнал за дни или недели до поломки. Без ложных срабатываний каждые пять минут.

Что нам понадобится

Как я и обещал в начале, мы не будем применять какое-либо дорогостоящее оборудование, а вместо этого воспользуемся компонентами DIY (сделай сам), которые можно прибрести на маркетплейсах по вменяемой цене. Да, серьезные инженеры могут заявить, что для промышленных задач эти компоненты не подходят из-за недостаточной точности и надежности, но я предлагаю сначала развернуть и потестировать подобную систему, а затем, когда вы упретесь в ее потолок, уже думать, чем можно заменить «слабые» компоненты.

Итак, нам потребуется: ESP32 Dev Board. Выбирайте с USB-C, удобнее прошивать.

Датчик MPU6050 (акселерометр + гироскоп + термометр). Нам нужны его оси X, Y, Z.

Макетная плата и провода dupont. Для сборки прототипа этого будет достаточно. Но, когда дойдет до промышленной эксплуатации, конечно, придется паять.

Также, потребуется блок питания 5V/1A для ESP32 (можно запитывать от USB, но для цеха берите стабильный). Завод есть завод.

И важная деталь, о которой забывают многие разработчики DIY – корпус. В пыльном цеху плата болтающаяся на станке сама быстро выйдет из строя.

Также, нам потребуется следующий софт:

• Python 3.9+.

• MicroPython прошивка для ESP32.

• Mosquitto MQTT брокер (поставим на любой компьютер в локальной сети или на Raspberry Pi).

• Библиотеки: umqtt.simple, mpu6050 (для ESP32), на сервере — paho-mqtt, numpy, scipy, scikit-learn, fastapi, jinja2, aiohttp.

Что важно: не нужно покупать промышленные датчики за 50 тысяч рублей. Для старта MPU6050 более чем достаточно — он выдает 16-битные данные по трем осям с частотой до 400 кГц. Этого хватит, чтобы поймать вибрацию подшипника качения.

Железная часть: крепим и подключаем

Начнем с правильного монтажа — от этого зависит 80% успеха. Бессмысленно собирать сложную систему анализа, если датчик просто лежит рядом со станком. Как НЕ надо монтировать: не вешайте датчик на длинных проводах (будет больше помех), не крепите на двусторонний скотч (да, все мы знаем, что скотч и синяя изолента — лучшие друзья инженера, но он гасит высокочастотные вибрации, а нам как раз надо их поймать, так что лучше не надо). Наконец, не ставьте датчик на пластиковый корпус редуктора.

Правильным будет максимально жесткое крепление датчика к корпусу подшипникового узла, использование для монтажа эпоксидной смолы или винтовое соединение через маленькую переходную пластину. И вообще, в промышленных условиях стоит избегать «крепление на соплях», которое часто используется в домашних проектах (липучки, скотч, изолента и т. д.). В условиях завода все это очень быстро выйдет из строя.

И расположить датчик нужно так, чтобы одна из его осей (лучше X или Y) совпадала с направлением вращения вала.

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

Шаг 1. Подключение

Мы перейдем сразу к подключению ESP32 к MPU6050 (стандартная распиновка I2C):

Шаг 2. Прошивка

Код на MicroPython будет иметь следующий вид:

from machine import Pin, SoftI2C

import mpu6050
import time

# Инициализируем I2C
i2c = SoftI2C(scl=Pin(22), sda=Pin(21))

# Создаем объект датчика
mpu = mpu6050.accel(i2c)

# Проверяем, что датчик отвечает
print("Датчик MPU6050 найден:", i2c.scan())  # Должен вернуть [104] или [0x68]

# Пробное чтение
while True:
    data = mpu.get_values()
    print(f"Acc X: {data['AcX']:6.0f} | Acc Y: {data['AcY']:6.0f} | Acc Z: {data['AcZ']:6.0f}")
    time.sleep(0.5)

Если видите меняющиеся числа — датчик жив. Если нет — то по классике, проверьте подключение GND и VCC (ничего не сгорело?) и контакты SDA/SCL, не перепутаны ли они местами.

Шаг 3. MQTT: заставляем станок говорить

MQTT — это протокол для передачи телеметрии. Легкий, надежный, с ним работает полмира IoT. У нас будет один брокер (сервер), который принимает сообщения от ESP32, и один клиент (наш Python-скрипт на сервере), который эти сообщения забирает.

Установку брокера Mosquitto можно выполнить на обычном компьютере или Raspberry-подобном устройстве:

Настраивать Mosquitto глобально не обязательно — для теста он работает с доступом для всех в локальной сети (порт 1883). Когда дойдет до продуктива, не забудьте поставить пароль и отключите анонимный доступ.

Код для ESP32 (добавляем Wi-Fi и MQTT отправку):

import network
from umqtt.simple import MQTTClient
import ujson
import time
from machine import Pin, SoftI2C
import mpu6050

# --- Wi-Fi настройки ---

SSID = "Ваша_сеть"
PASSWORD = "Ваш_пароль"

# --- MQTT настройки ---
MQTT_BROKER = "192.168.1.100"  # IP компьютера, где запущен Mosquitto
MQTT_TOPIC = b"machine/vibration"

# --- Подключаемся к Wi-Fi ---
sta_if = network.WLAN(network.STA_IF)
sta_if.active(True)
sta_if.connect(SSID, PASSWORD)
print("Connecting to WiFi", end="")
while not sta_if.isconnected():
    print(".", end="")
    time.sleep(0.5)
print(" Connected!")

# --- Подключаем датчик ---
i2c = SoftI2C(scl=Pin(22), sda=Pin(21))
mpu = mpu6050.accel(i2c)

# --- Подключаемся к MQTT ---
client = MQTTClient(b"esp32_001", MQTT_BROKER)
client.connect()
print("Connected to MQTT broker")

# --- Основной цикл ---

while True:
    data = mpu.get_values()

    # Формируем JSON
    payload = ujson.dumps({
        "acc_x": data["AcX"],
        "acc_y": data["AcY"],
        "acc_z": data["AcZ"],
        "gyro_x": data["GyX"],
        "gyro_y": data["GyY"],
        "gyro_z": data["GyZ"],
        "temp": data["Tmp"]  # температура с MPU6050
    })

    client.publish(MQTT_TOPIC, payload)
    print("Sent:", payload)
    time.sleep(0.1)  # 10 измерений в секунду — достаточно

Здесь стоит обратить внимание на один важный момент: time.sleep(0.1) дает частоту 10 Гц. Для анализа вибрации подшипников качения этого достаточно. Если хотите ловить дефекты зубчатых передач — ставьте sleep(0.02) (50 Гц), но будьте готовы к большему объему данных.

Далее проверяем, что данные доходят (на сервере, в отдельном терминале):

mosquitto_sub -h localhost -t "machine" -v

Если видите поток JSON — все работает. Нет — проверьте IP брокера в коде ESP32 и фаервол на сервере (порт 1883 должен быть открыт).

Шаг 4. Пишем цифрового двойника

Теперь переходим к самому интересному. Цифровой двойник — это не просто «собиратель логов». Это модель, которая учится на истории работы станка и знает, какая вибрация для него «нормальна», а какая — предвестник смерти.

Мы построим детектор аномалий на изолирующем лесе (Isolation Forest) — алгоритме, который специально создан для поиска выбросов в многомерных данных. Он не требует размеченных данных о поломках, только историю нормальной работы.

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

Процесс построения такого дерева состоит из следующих шагов:

1. Выбирается случайный признак из набора данных.

2. Выбирается случайное значение разбиения в диапазоне значений этого признака.

3. Данные разделяются на две группы в зависимости от того, выше или ниже выбранного порога находится значение признака.

4. Процесс повторяется рекурсивно для каждой полученной группы до тех пор, пока не будет достигнута максимальная глубина дерева или не будет выполнено другое условие остановки.

Здесь ключевой концепцией является длина пути (количество разбиений), необходимого для изоляции точки данных в изолирующем дереве. Аномалии, будучи редкими и отличающимися от большинства данных, обычно требуют меньше разбиений для изоляции, так как они с большей вероятностью выходят за пределы типичного диапазона значений для выбранных признаков. Обычные точки данных, которые более схожи между собой, могут потребовать большего количества разбиений.

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

Структура нашего проекта на сервере:

digital_twin/

Далее мы рассмотрим содержимое каждого из скриптов, входящих в проект.

Шаг 5. Сбор данных для обучения (collector.py)

Для начала нам нужно собрать эталонные данные. Для этого мы запустим сбор данных на 2-3 дня нормальной работы станка.

import paho.mqtt.client as mqtt
import json
import csv
from datetime import datetime

CSV_FILE = "data/normal_operation.csv"

def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode())
    timestamp = datetime.now().isoformat()

    # Добавляем временную метку
    payload["timestamp"] = timestamp

    # Сохраняем в CSV

    with open(CSV_FILE, "a", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=payload.keys())
        if f.tell() == 0:
            writer.writeheader()
        writer.writerow(payload)
    print(f"Saved: {payload['acc_x']}, {payload['acc_y']}, {payload['acc_z']}")

client = mqtt.Client()
client.connect("localhost", 1883)
client.subscribe("machine/vibration")
client.on_message = on_message
print("Collecting normal operation data...")
client.loop_forever()

Шаг 6. Обучаем модель аномалий (model_train.py)

Когда у нас накопилось достаточно данных (хотя бы 10 000 строк), запускаем обучение. Isolation Forest анализирует распределение точек в 6-мерном пространстве (acc_x, acc_y, acc_z, gyro_x, gyro_y, gyro_z) и помечает редкие комбинации как аномальные.

import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
import joblib

# Загружаем собранные данные
df = pd.read_csv("data/normal_operation.csv")

# Берем только числовые колонки (без timestamp)
features = ["acc_x", "acc_y", "acc_z", "gyro_x", "gyro_y", "gyro_z"]
X = df[features].values
print(f"Training on {len(X)} samples")

# Обучаем изолирующий лес
# contamination — ожидаемая доля аномалий (для здорового станка ~1%)
model = IsolationForest(
    contamination=0.01,
    random_state=42,
    n_estimators=100
)
model.fit(X)

# Сохраняем модель
joblib.dump(model, "models/anomaly_model.pkl")
print("Model saved to models/anomaly_model.pkl")

# Проверяем, как модель видит обучающие данные
predictions = model.predict(X)
anomalies = sum(predictions == -1)
print(f"Anomalies found in training data: {anomalies} ({anomalies/len(X)*100:.2f}%)")

Если модель нашла аномалии в обучающих данных — не страшно. Значит, даже в «нормальной» работе были кратковременные выбросы (удары, переходные режимы) и модель их запомнит.

Шаг 7. Реалтайм детектор (detector.py)

Запускаем и радуемся. Каждые 10 секунд агрегируем данные за последнюю минуту и прогоняем через модель.

import paho.mqtt.client as mqtt
import json
import numpy as np
import joblib
from collections import deque
import time

# Загружаем модель
model = joblib.load("models/anomaly_model.pkl")
# Буфер для накопления данных (последние 600 точек = 1 минута при 10 Гц)
BUFFER_SIZE = 600
buffer = deque(maxlen=BUFFER_SIZE)

def extract_features(window_data):
   # Извлекаем статистические признаки из окна данных
    window = np.array(window_data)
    features = []
    for i in range(6):  # 6 осей
        features.append(np.mean(window[:, i]))
        features.append(np.std(window[:, i]))
        features.append(np.max(window[:, i]) - np.min(window[:, i]))
    return np.array(features).reshape(1, -1)

def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode())
    vec = [payload["acc_x"], payload["acc_y"], payload["acc_z"],
           payload["gyro_x"], payload["gyro_y"], payload["gyro_z"]]
    buffer.append(vec)

    # Каждые 10 секунд анализируем
    if len(buffer) == BUFFER_SIZE:
        features = extract_features(buffer)
        prediction = model.predict(features)[0]
        if prediction == -1:  # Аномалия
            anomaly_score = model.decision_function(features)[0]
            print(f" ANOMALY DETECTED! Score: {anomaly_score:.3f}")
            # Здесь вызываем отправку в Telegram
            send_alert(features, anomaly_score)
        else:
            print(f" Normal: {len(buffer)} samples processed")
        buffer.clear()

def send_alert(features, score):
    # Заглушка — реализуем в следующем разделе
    print(f"ALERT: anomaly score = {score:.3f}")
client = mqtt.Client()
client.connect("localhost", 1883)
client.subscribe("machine/vibration")
client.on_message = on_message
print("Real-time anomaly detector started")
client.loop_forever()

Почему статистика работает лучше порогов

Здесь многие могут задать вопрос: «А почему нельзя просто поставить порог по вибрации, как в любом виброметре?»

В качестве ответа давайте рассмотрим небольшой пример. У вас станок точит детали, соответственно, при обработке толстой заготовки вибрация выше, при тонкой — ниже. Если вы поставите жесткий порог, то на толстых деталях будете получать ложные срабатывания каждые пять минут. Если поднимете порог — пропустите реальную поломку на тонкой детали.

Цифровой двойник видит всю картину. Он знает: «Ага, сейчас у нас режим толстой заготовки, и для этого режима вибрация чуть выше — это норма. Но вот этот всплеск на третьей гармонике — совсем не норма».

Изолирующий лес работает именно так. Он строит многомерные границы «нормальности» и бьет тревогу, когда точка оказывается далеко от всех известных кластеров.

И еще один вопрос: почему Isolation Forest, а не LSTM. Здесь все тоже довольно просто. Во-первых, Isolation Forest не требует GPU для обучения и обучается за секунды на 100 000 точек. Также “лес” интерпретируем, то есть можно понять, какая ось дала аномалию. Наконец, он не требует размеченных данных о поломках.

При этом LSTM и другие нейросетевые методы дают более высокую точность, но для старта и первых результатов Isolation Forest — идеальный выбор.

Шаг 8. Заставьте робота орать в Telegram

Ладно, вы сидите и смотрите на вывод скриптов или построенный график. Но если станок ломается ночью, а вы спите — какой смысл в этом? Ответ очевиден – нам нужно добавить Telegram-бота.

Для этого выполним следующие действия:

1. Создаем бота у @BotFather (в Telegram). Получаем токен вида 123456:ABC-DEF1234.

2. Узнаем ваш chat_id (напишите боту @userinfobot что-нибудь).

3. Создаем скрипт alert.py:

import requests
import json

TELEGRAM_TOKEN = "ваш_токен_от_botfather"
CHAT_ID = "ваш_id"

def send_telegram_alert(message):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    payload = {
        "chat_id": CHAT_ID,
        "text": message,
        "parse_mode": "HTML"
    }
    try:
        response = requests.post(url, json=payload)
        response.raise_for_status()
        print("Alert sent to Telegram")
    except Exception as e:
        print(f"Failed to send: {e}")

# Интегрируем в detector.py
def send_alert(features, score):
    message = f"""
<b>ПРЕДУПРЕЖДЕНИЕ О ПОЛОМКЕ СТАНКА</b>
<b>Уровень аномалии:</b> {score:.3f}
<b>Время:</b> {time.ctime()}
<b>Рекомендация:</b> Провести внеплановый осмотр подшипникового узла
<i>Ожидаемое время до отказа:</i> ~{estimate_remaining_life(score)} часов
    """
    send_telegram_alert(message)

def estimate_remaining_life(anomaly_score):
    #Грубая оценка: чем выше score, тем ближе смерть
    # Нормализуем score из диапазона (-0.5..0) в часы
    # Чем ближе к нулю, тем хуже
    if anomaly_score > -0.1:
        return 2
    elif anomaly_score > -0.3:
        return 12
    else:
        return 48

Теперь, когда модель находит аномалию, вы получаете сообщение в Telegram. Даже если вы в отпуске.

Шаг 9 и последующие... Что может пойти не так (и как это чинить)

Разберем основные проблемы, которые могут произойти в процессе настройки и эксплуатации нашего двойника.

Проблема: ложные срабатывания каждые 5 минут

Причина: модель обучилась на слишком коротком интервале и не видит нормальных переходных режимов (разгон, останов, смена инструмента). Решение: соберите данные минимум за 3 дня нормальной работы, включив все режимы. Или увеличьте contamination в модели до 0.05 — тогда она будет считать аномалией только 5% точек.

Проблема: ESP32 отваливается от Wi-Fi

Причина: в цеху могут быть помехи от моторов и сварки. Решение: добавьте реконнект в код ESP32:

while True:
    if not sta_if.isconnected():
        sta_if.connect(SSID, PASSWORD)
        time.sleep(5)
        continue
    # остальной код

 Проблема: модель не ловит поломку, которая реально случилась

Причина: ваши «нормальные» данные уже содержали деградацию. Вы начали обучение на больном станке. Решение: переобучите модель на данных сразу после капремонта. Нет таких данных? Используйте публичные датасеты для калибровки, например WADI от IEEE  или NASA Bearing Dataset.

 Проблема: слишком много данных, буфер переполняется

 Решение: уменьшите BUFFER_SIZE до 300 (30 секунд) или агрегируйте данные на самом ESP32 — отправляйте не каждые 0.1 секунды, а среднее за секунду.

Подведем итоги

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

Но это только начало. Дальше вы можете попробовать выполнить различные эксперименты. Например, подключите датчик тока на ESP32 и добавьте потребление двигателя в признаки. Замените Isolation Forest на LSTM-autoencoder, когда соберете достаточно данных о поломках. Добавьте спектральный анализ — преобразуйте временной ряд в частотную область и ищите аномалии в спектре. Главное — вы перестали гадать, когда сломается станок. Удачи в цеху и берегите подшипники.

Хотите глубже разобраться, как создавать цифровые модели реальных объектов, связывать их с данными и использовать для мониторинга, анализа и прогнозирования? Обратите внимание на курс «Цифровые двойники» — на нём рассматривают полный цикл разработки таких систем: от сбора телеметрии и построения модели до интеграции с промышленной инфраструктурой.

А чтобы не пропускать новые статьи, анонсы и подборки, подписывайтесь на канал OTUS в MAX.