Неделя 16 - Промышленное компьютерное зрение и MLOps
1. Industrial CV: Особенности и требования
1.1. Ключевые отличия промышленных решений
Надежность: 24/7 работа в harsh-условиях
Производительность: Реальное время (<100 мс/кадр)
Точность: >99% для критических задач
Интеграция: PLC, SCADA, MES системы
1.2. Типовые промышленные задачи
mermaid
Copy
graph LR
A[Контроль качества] --> B[Дефектоскопия]
A --> C[Измерения]
D[Логистика] --> E[Чтение штрих-кодов]
D --> F[Сортировка]
G[Безопасность] --> H[Детекция PPE]
G --> I[Мониторинг зон]
2. MLOps для компьютерного зрения
2.1. CI/CD пайплайн для CV моделей
python
Copy
# Пример GitHub Actions для автоматического тестирования
name: CV Model CI
on: [push]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: '3.8'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest opencv-python
- name: Run unit tests
run: |
pytest tests/
- name: Model validation
run: |
python validate_model.py --min-accuracy 0.95
2.2. Мониторинг моделей в production
python
Copy
from prometheus_client import start_http_server, Gauge
import time
# Метрики Prometheus
MODEL_ACCURACY = Gauge('model_accuracy', 'Точность модели в продакшн')
INFERENCE_TIME = Gauge('inference_time_ms', 'Время инференса в мс')
DRIFT_SCORE = Gauge('feature_drift_score', 'Косор данных')
def monitor_model_performance():
start_http_server(8000)
while True:
# Расчет метрик
accuracy = calculate_rolling_accuracy()
inference_time = measure_inference_time()
drift = calculate_drift_score()
# Установка значений метрик
MODEL_ACCURACY.set(accuracy)
INFERENCE_TIME.set(inference_time)
DRIFT_SCORE.set(drift)
time.sleep(60)
3. Оптимизация для промышленных решений
3.1. Ускорение инференса
python
Copy
# TensorRT оптимизация
import tensorrt as trt
# Конвертация ONNX -> TensorRT
logger = trt.Logger(trt.Logger.INFO)
builder = trt.Builder(logger)
network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
parser = trt.OnnxParser(network, logger)
with open("model.onnx", "rb") as f:
parser.parse(f.read())
config = builder.create_builder_config()
config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)
serialized_engine = builder.build_serialized_network(network, config)
with open("model.engine", "wb") as f:
f.write(serialized_engine)
3.2. Развертывание на edge-устройствах
dockerfile
Copy
# Dockerfile для NVIDIA Jetson
FROM nvcr.io/nvidia/l4t-base:r32.6.1
# Установка зависимостей
RUN apt-get update && apt-get install -y \
python3-pip \
libopencv-dev \
&& rm -rf /var/lib/apt/lists/*
# Установка TensorRT-оптимизированных библиотек
RUN pip3 install --extra-index-url https://developer.download.nvidia.com/compute/redist/jp/v46 \
tensorrt==8.2.1.8 \
nvidia-pyindex \
onnx-graphsurgeon
# Копирование модели и кода
COPY trt_model.engine /app/
COPY inference.py /app/
WORKDIR /app
CMD ["python3", "inference.py"]
4. Реальные кейсы промышленного CV
4.1. Контроль качества на конвейере
python
Copy
class QualityInspector:
def __init__(self):
self.defect_model = load_defect_detector()
self.measurement_model = load_measurement_model()
self.ok_count = 0
self.defect_count = 0
def process_frame(self, frame):
# Детекция дефектов
defects = self.defect_model.detect(frame)
# Проверка размеров
measurements = self.measurement_model.measure(frame)
# Принятие решения
is_ok = not defects and self._check_tolerances(measurements)
# Логирование статистики
if is_ok:
self.ok_count += 1
return "OK"
else:
self.defect_count += 1
return {"status": "REJECT", "defects": defects, "measurements": measurements}
def _check_tolerances(self, measurements):
return all(m['value'] >= m['min'] and m['value'] <= m['max']
for m in measurements)
4.2. Интеграция с промышленными системами
python
Copy
import pyads
class PLCInterface:
def __init__(self, plc_ip='192.168.1.10'):
self.plc = pyads.Connection(plc_ip, 851)
self.plc.open()
def send_defect_signal(self, defect_type):
# Запись в PLC переменные
self.plc.write_by_name('MAIN.defect_type', defect_type, pyads.PLCTYPE_INT)
self.plc.write_by_name('MAIN.defect_detected', True, pyads.PLCTYPE_BOOL)
def reset_signal(self):
self.plc.write_by_name('MAIN.defect_detected', False, pyads.PLCTYPE_BOOL)
5. Практическое задание
1. Реализуйте CI/CD пайплайн для вашей CV модели
2. Оптимизируйте модель с помощью TensorRT
3. Разработайте систему мониторинга дрейфа данных
4. Создайте Docker-образ для развертывания на NVIDIA Jetson
5. Сымитируйте интеграцию с PLC (используйте pyads или OPC UA)
6. Итоги
Промышленные решения требуют особого подхода к надежности и интеграции
MLOps обеспечивает стабильность CV систем в production
Оптимизация под edge-устройства критически важна
Реальные кейсы требуют комплексного решения от детекции до интеграции
Заключение: Вы освоили полный цикл разработки промышленных систем компьютерного зрения - от алгоритмов до внедрения!