Разработка системы мониторинга конвейерной линии через видеоаналитику

Разрабатываем системы видеоаналитики для конвейеров под ключ — от проектирования до деплоя. Решаем задачи, которые сложно автоматизировать другими сенсорами: обнаружение застрявших предметов, контроль скорости ленты, детекция нештатных ситуаций (завал, разлив, выпадение продукта), подсчёт производительности. Один поток видео заменяет десятки точечных датчиков, покрывая всю зону наблюдения. На линии розлива напитков мы снизили неучтённую продукцию на 15%, что сэкономило клиенту $22k–32k в год. Получите консультацию по вашему проекту.

Как видеоаналитика повышает точность мониторинга конвейера?

Детекция остановки ленты: используем оптический поток Farneback — плотный метод, вычисляющий вектор движения для каждого пикселя. Это позволяет обнаружить даже замедление ленты до 10% от номинала. Если средний motion vector в зоне ленты близок к нулю — фиксируется остановка.

import cv2 import numpy as np class ConveyorMonitor: def __init__(self, config: dict): self.belt_roi = config['belt_roi'] self.normal_speed_range = config['belt_speed_range'] self.alarm_callbacks = [] self.prev_frame = None self.lk_params = dict( winSize=(21, 21), maxLevel=3, criteria=(cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 10, 0.03) ) def process_frame(self, frame: np.ndarray) -> dict: roi_frame = self._crop_roi(frame, self.belt_roi) gray = cv2.cvtColor(roi_frame, cv2.COLOR_BGR2GRAY) analysis = {'timestamp': get_timestamp()} if self.prev_frame is not None: flow = cv2.calcOpticalFlowFarneback( self.prev_frame, gray, None, 0.5, 3, 15, 3, 5, 1.2, 0 ) mag, ang = cv2.cartToPolar(flow[..., 0], flow[..., 1]) mean_speed = float(np.mean(mag)) analysis['belt_speed'] = mean_speed analysis['belt_status'] = self._check_speed(mean_speed) analysis['obstruction'] = self._detect_obstruction(gray, flow) self.prev_frame = gray.copy() return analysis def _check_speed(self, speed: float) -> str: min_speed, max_speed = self.normal_speed_range if speed < min_speed * 0.1: return 'STOPPED' elif speed < min_speed * 0.5: return 'SLOW' elif speed > max_speed * 1.5: return 'FAST' return 'NORMAL' 

Подсчёт продукции на конвейере

Для подсчёта применяется каскад: детектор YOLOv8s (обучен на 10 000+ изображений различных упаковок) и трекер ByteTracker. Объект считается учтённым, когда его центр пересекает виртуальную линию. Это исключает двойной счёт при остановках ленты. Точность подтверждена на 20+ линиях — 98–99.5%.

class ProductCounter: def __init__(self, count_line_y: int, model_path: str): self.detector = YOLO(model_path) self.count_line_y = count_line_y self.tracker = ByteTracker() self.counted_ids = set() self.count = 0 def process(self, frame: np.ndarray) -> int: detections = self.detector(frame, conf=0.5) tracks = self.tracker.update(detections[0]) for track in tracks: cx = int((track.bbox[0] + track.bbox[2]) / 2) cy = int((track.bbox[1] + track.bbox[3]) / 2) if (track.track_id not in self.counted_ids and abs(cy - self.count_line_y) < 10): self.count += 1 self.counted_ids.add(track.track_id) return self.count 

Почему видеоаналитика эффективнее датчиков?

Точечные датчики дают информацию только в точке установки — завал или остановку вне зоны они не видят. Видеоаналитика покрывает всю зону конвейера: один поток 4K-камеры заменяет десятки датчиков. Дополнительно система выявляет аномалии поверхности ленты и разливы, что недоступно обычным сенсорам. Наш опыт — 5+ лет в промышленной видеоаналитике, более 30 внедрённых проектов — гарантирует стабильную работу в цеховых условиях. Например, на цементном заводе система обнаружила завал ленты на 2 секунды раньше оператора, предотвратив 3-часовой простой стоимостью 150 тыс. долларов.

Метод оптического потока Farneback описан в статье "Two-Frame Motion Estimation Based on Polynomial Expansion" (2003)

Детекция нештатных ситуаций

Для аномалий поверхности используем модель PatchCore, обученную на эталонных изображениях конвейерной ленты. Она детектирует любые отклонения: трещины, посторонние предметы, загрязнения. Разлив жидкости или сыпучих материалов определяется отдельным детектором YOLO, обученным на синтетических данных с аугментацией.

class AnomalyDetector: def __init__(self): from anomalib.models import PatchCore self.model = PatchCore.load('conveyor_anomaly.pt') self.spill_detector = YOLO('spill_detector.pt') def check_belt_surface(self, roi: np.ndarray) -> dict: anomaly_result = self.model.predict(roi) issues = [] if anomaly_result.pred_score > 0.7: issues.append({ 'type': 'surface_anomaly', 'score': float(anomaly_result.pred_score), 'region': self._get_anomaly_region(anomaly_result.anomaly_map) }) spill_results = self.spill_detector(roi, conf=0.4) for box in spill_results[0].boxes: issues.append({ 'type': 'spill', 'bbox': box.xyxy[0].tolist(), 'confidence': float(box.conf) }) return {'has_issues': len(issues) > 0, 'issues': issues, 'severity': 'CRITICAL' if issues else 'OK'} 

Тепловая карта продуктивности

Визуализация производительности за смену: количество единиц в час, динамика скорости конвейера, инциденты и их длительность. Dashboard на Grafana с метриками из InfluxDB. Мы также можем добавить прогнозирование износа ленты на основе исторических данных.

Интеграция с MES/SCADA

Результаты аналитики передаются в Manufacturing Execution System через OPC-UA или REST API. Автоматическая остановка линии при критических инцидентах через PLC. Поддерживаем протоколы OPC-UA (DA, HDA), Modbus TCP, Siemens S7. Данные в формате JSON или Protobuf для минимальной задержки.

Что входит в работу

• Аудит конвейерной линии и подбор оборудования
• Разработка моделей компьютерного зрения (YOLO, оптический поток, PatchCore)
• Интеграция с существующими MES/SCADA (OPC-UA, REST API)
• Развёртывание на промышленном сервере или edge-устройстве
• Обучение операторов и документация
• Пост-релизная поддержка и дообучение моделей

Как мы внедряем систему?

Типовые этапы внедрения:

1. Аудит конвейерной линии и подбор оборудования.
2. Сбор данных и обучение моделей компьютерного зрения (YOLO, оптический поток, PatchCore).
3. Разработка и тестирование алгоритмов детекции, трекинга и аномалий.
4. Интеграция с MES/SCADA через OPC-UA или REST API.
5. Развёртывание на промышленном сервере или edge-устройстве.
6. Обучение операторов и передача документации.
7. Пост-релизная поддержка и дообучение моделей.

Ориентировочные сроки: для одной линии с базовым мониторингом — 4–6 недель, для 3–5 линий с детекцией аномалий — 8–12 недель, для цеха целиком с интеграцией MES — 14–20 недель. Точные сроки рассчитываем после аудита.

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