Метод WS /api/v2/fridge/stream-capture

Документация API: Потоковая верификация продуктов через реальное время (Camera Stream) с использованием буферной очереди Apache Kafka

Published

July 1, 2026

1 Функциональное назначение

Метод предназначен для обработки потока изображений (кадров) в реальном времени, поступающих непосредственно с камеры мобильного приложения во время наведения на продукт. В отличие от пакетной загрузки готовых файлов, данный контур минимизирует задержку ввода (Latency) и защищает вычислительные ресурсы ИИ-сервера (PyTorch/GPU) от перегрузки при высокой частоте кадров (FPS).

Бэкенд-шлюз FastAPI в этой схеме выступает в роли ультрабыстрого легковесного ретранслятора, который сгружает бинарные кадры в брокер сообщений Apache Kafka, выполняющий роль амортизатора нагрузки (Load Smoother). ИИ-сервис асинхронно вычитывает кадры из очереди, производит инференс и возвращает результат распознавания обратно в WebSocket-сессию пользователя.

2 Протокол взаимодействия (Потоковый Контракт)

  • Протокол: WebSockets (WSS)
  • Маршрут (Эндпоинт): /api/v2/fridge/stream-capture/{session_id}
  • Тип обмена: Дуплексный (Двунаправленный асинхронный поток)
  • Базовый формат кадров (Frame Payload): Бинарный поток сжатых изображений (image/jpeg или image/webp) + JSON-управляющие сообщения.

2.1 Спецификация заголовков при установлении соединения (Handshake Headers)

Перед переключением протокола (Upgrade) с HTTP на WebSockets клиент Dio / WebSocketChannel обязан передать валидные метаданные.

Заголовок Обязательный Описание Пример значения
Upgrade Да Системный заголовок для переключения протокола websocket
Connection Да Системный заголовок для переключения протокола Upgrade
Authorization Да Короткоживущий Access Token для валидации сессии пользователя Bearer eyJhbGciOiJIUzI1Ni...
X-Request-ID Да Сквозной ID WebSocket-сессии для агрегации логов в Jaeger/Loki stream-7a2b-9c1d-00ef

2.2 Спецификация структуры потоковых сообщений (Stream Payload)

В рамках одной открытой сессии клиент отправляет бинарные данные, а сервер возвращает структурированный JSON с результатами ИИ-классификации.

2.2.1 Входящий поток от клиента (Client-to-Server, Binary/Text)

Мобильное приложение отправляет кадры в бинарном формате для экономии трафика. Кадры сжимаются на стороне Flutter до разрешения 640x480 (дефолт для MobileNetV2).

Тип фрейма Обязательный Описание Формат данных
Binary Frame Да Сырые байты текущего кадра с камеры смартфона ByteBuf (image/jpeg)
Text Frame (Ping) Нет Системный фрейм поддержания соединения (каждые 5 сек) {"type": "PING"}

2.2.2 Исходящий поток от сервера (Server-to-Client, JSON)

Бэкенд возвращает результаты распознавания только тогда, когда ИИ-сервис преодолевает порог уверенности (Confidence Threshold > 0.85).

Поле Тип Обязательный Описание
event_type String Да Тип события потока
confidence Float Да Степень уверенности ИИ в распознавании объекта
predicted_sku String Да Распознанный технический код продукта питания
frame_status String Да Инструкция для UI (HOLD — удерживать, DETECTED — зафиксирован)

3 Схема обработки запроса пользователя (Mermaid)

На диаграмме представлена сквозная архитектура стрим-контура (Версия 2). Бэкенд-шлюз FastAPI изолирован от ИИ-нагрузки: он работает как транслятор фреймов в Apache Kafka. Kafka выступает амортизатором (буфером), из которого ИИ-сервис асинхронно забирает кадры на инференс, предотвращая падение системы при высоком FPS.

sequenceDiagram
    autonumber
    actor User as Пользователь (Камера)
    participant APP as Мобильное приложение (Flutter)
    participant API as Бэкенд-шлюз (FastAPI)
    participant KAFKA as Брокер событий (Kafka)
    participant GRPC as ИИ-сервис (PyTorch)
    participant DB as Реляционная СУБД (PostgreSQL)

    %% УСТАНОВЛЕНИЕ СОЕДИНЕНИЯ
    User->>APP: Наводит камеру на продукт (Томат)
    APP->>API: HTTP GET /stream-capture (Handshake Upgrade)
    activate API
    API-->>APP: 101 Switching Protocols (WebSocket открыт)
    
    %% ПОТОКОВАЯ ПЕРЕДАЧА КАДРОВ
    loop Кадр за кадром (Streaming FPS)
        APP->>API: Binary Frame (Сырые байты JPEG кадра)
        API->>KAFKA: Отправляет байты в топик "media.camera.stream.v2"
    end
    deactivate API

    %% АСИНХРОННЫЙ ИИ-КОНТУР
    activate GRPC
    KAFKA->>GRPC: Вычитка кадров из топика (Consumer Group)
    note over GRPC: Шаг 10: PyTorch (MobileNetV2) расчет весов
    GRPC->>GRPC: Объект идентифицирован (TOMATO, confidence: 0.94)
    
    %% ОБРАТНАЯ СВЯЗЬ ЧЕРЕЗ RPC
    GRPC->>API: gRPC StreamResponse(predicted_sku='TOMATO')
    deactivate GRPC
    activate API
    API->>APP: JSON: STREAM_PREDICTION_UPDATED (TOMATO)
    note over APP: UI подсвечивает рамку и пишет "Томат (94%)"
    deactivate API

    %% КЛИЕНТСКАЯ ФИКСАЦИЯ И МАРШРУТИЗАЦИЯ
    User->>APP: Нажимает на кнопку затвора "Зафиксировать"
    APP->>API: Текстовый фрейм: {"action": "FREEZE_FRAME", "sku": "TOMATO"}
    activate API
    note over API: Шаг 16: Определение типа и ветвление таблиц
    API->>DB: INSERT INTO pending_products_buffer (status='PROCESSING')
    API->>APP: JSON: SESSION_FROZEN (Переход на экран ввода веса)
    deactivate API
    note over APP: Стрим камеры закрывается, юзер вводит вес вручную

4 Расшифровка шагов

  • Шаг 1 (User -> APP): Пользователь открывает экран «Умная камера» во Flutter-приложении и наводит объектив смартфона на продукт (например, лежащий на столе томат).
  • Шаг 2 (APP -> API): Приложение инициирует HTTP GET запрос на эндпоинт /stream-capture, передавая авторизационный токен и уникальный X-Request-ID. Запрашивается обновление протокола до WebSockets.
  • Шаг 3 (API -> APP): FastAPI производит валидацию токена. Если сессия подтверждена, сервер возвращает статус HTTP 101 Switching Protocols. Устанавливается постоянное дуплексное TCP-соединение. Бэкенд переходит в режим ожидания фреймов (activate API).
  • Шаг 4 (APP -> API): Внутри бесконечного цикла мобильное приложение захватывает кадры с видеобуфера камеры, сжимает их до разрешения 640x480 в формат JPEG и непрерывно отправляет в WebSocket-канал в виде бинарных фреймов (Binary Frame).
  • Шаг 5 (API -> KAFKA): Бэкенд-шлюз FastAPI работает в асинхронном неблокирующем режиме. Он не сохраняет файлы на диск и не вызывает ИИ-модели. Получив бинарный фрейм, он моментально упаковывает его в массив байт и выполняет метод producer.send(), отправляя кадр в топик media.camera.stream.v2. На этом контекст шлюза освобождается (deactivate API).
  • Шаг 6 (KAFKA -> GRPC): ИИ-сервис на базе PyTorch работает как изолированный потребитель (Consumer). Он вычитывает бинарные кадры из топика media.camera.stream.v2. Если приложение шлет 30 кадров в секунду, а видеокарта (GPU) успевает обработать только 15, Kafka удерживает оставшиеся 15 кадров в очереди, не перегружая память ИИ-сервиса.
  • Шаг 7 (GRPC -> GRPC): Внутренний шаг ИИ-сервиса: каждый извлеченный из очереди кадр пропускается через сверточную нейросеть MobileNetV2. Модель рассчитывает вектор вероятностей.
  • Шаг 8 (GRPC -> GRPC): Модель фиксирует совпадение: на картинке обнаружен объект с кодом класса, соответствующим техническому SKU "TOMATO", с высоким коэффициентом уверенности (confidence = 0.94).
  • Шаг 9 (GRPC -> API): ИИ-сервис открывает обратный gRPC-канал к шлюзу и отправляет сообщение StreamResponse(predicted_sku='TOMATO', confidence=0.94). Воркер ИИ-сервиса уходит на чтение следующего кадра из Kafka.
  • Шаг 10 (API -> APP): FastAPI принимает gRPC-ответ, находит нужную WebSocket-сессию пользователя по session_id и отправляет текстовый JSON-пакет с событием STREAM_PREDICTION_UPDATED.
  • Шаг 11 (APP -> APP): Интерфейс мобильного приложения мгновенно реагирует на сигнал: поверх превью камеры отрисовывается зеленая рамка-таргет, а пользователю выводится подсказка: «Томат (94%). Удерживайте для фиксации».
  • Шаг 12 (User -> APP): Пользователь видит, что ИИ правильно распознал продукт, и нажимает на экране физическую кнопку затвора «Зафиксировать продукт».
  • Шаг 13 (APP -> API): Вместо отправки тяжелой фотографии, приложение шлет в открытый WebSocket легковесный текстовый JSON-фрейм команды: {"action": "FREEZE_FRAME", "sku": "TOMATO"}.
  • Шаг 14 (API -> DB): Шлюз FastAPI перехватывает команду фиксации, останавливает трансляцию кадров из Kafka и обращается к PostgreSQL. Так как ИИ определил тип как одиночный продукт, срабатывает логика маршрутизации: бэкенд выполняет INSERT INTO pending_products_buffer с дефолтным статусом PROCESSING.
  • Шаг 15 (API -> APP): Сервер отправляет в вебсокет финальный статус SESSION_FROZEN. Соединение закрывается. На стороне Flutter экран камеры блокируется, и приложение автоматически перенаправляет пользователя на форму ручного ввода, где ему остается только указать примерный вес томата (например, 0.3 кг), как это описано в стандарте Версии 1.

5 Спецификация сообщений брокера и вилок исключений

5.1 Спецификация топика Apache Kafka: media.camera.stream.v2

В этот топик бэкенд-шлюз FastAPI непрерывно сбрасывает кадры, полученные из бинарных фреймов WebSocket. Для оптимизации пропускной способности данные передаются в бинарном виде (сериализация ByteArray), а метаданные упаковываются в заголовки сообщения (Kafka Headers).

  • Стратегия партиционирования (Partition Key): session_id (гарантирует, что все кадры в рамках одной сессии камеры попадают на одну партицию и обрабатываются ИИ строго в хронологическом порядке).
  • Формат Key: String (значение session_id).
  • Формат Value: Byte[] (сырой бинарный поток сжатого JPEG-изображения).
  • Конфигурация заголовков сообщения (Kafka Headers):
Ключ заголовка Тип данных Описание Пример значения
x-request-id String Сквозной ID сессии для распределенного трейсинга stream-7a2b-9c1d-00ef
frame_timestamp String (ISO) Время захвата кадра на мобильном устройстве 2026-07-01T21:10:05.123Z
user_id String Идентификатор владельца сессии usr-8822-fa41

5.2 Спецификация управляющих текстовых фреймов (JSON)

5.2.1 Команда фиксации кадра от клиента (Шаг 13)

Отправляется в WebSocket-канал в текстовом формате при нажатии пользователем кнопки затвора.

{
  "action": "FREEZE_FRAME",
  "timestamp": "2026-07-01T21:10:12.450Z",
  "payload": {
    "verified_sku": "TOMATO",
    "confidence_at_freeze": 0.942
  }
}

5.2.2 Ответ сервера о заморозке сессии (Шаг 15)

Финальное системное сообщение от FastAPI, подтверждающее успешную запись в буфер и закрытие стрима.

{
  "event_type": "SESSION_FROZEN",
  "timestamp": "2026-07-01T21:10:13.110Z",
  "data": {
    "draft_id": "8a71d11e-9500-4b11-9a2c-d2b0d7b3dcba",
    "target_buffer_table": "pending_products_buffer",
    "next_ui_step": "NAVIGATE_TO_MANUAL_WEIGHT_INPUT"
  }
}

5.3 Вилки исключений и стратегии обработки ошибок (Failures & Fail-Safe)

Стриминговый контур чувствителен к сетевым задержкам и стабильности соединений. Система обрабатывает три основных аварийных сценария:

5.3.1 Сценарий А: Отказ ИИ-сервиса или падение gRPC (Зависание инференса)

Если ИИ-сервис уходит в offline или перестает отвечать на gRPC-запросы шлюза FastAPI, срабатывает стратегия Client-Side Timeout / Fallback.

  • Поведение системы: Если в течение 3 секунд от сервера не приходит ни одного обновления STREAM_PREDICTION_UPDATED, Flutter-приложение мягко переключает интерфейс в режим ручной съемки. На экран выводится тост-уведомление, а кнопка затвора меняет логику: при нажатии отправляется стандартный пакетный HTTP POST /product-photo по сценарию Версии 1. Система не блокирует пользователя.

5.3.2 Сценарий Б: Ошибка переполнения буфера или отказ Kafka (HTTP 503)

Выбрасывается на Шаге 5, если брокер Kafka недоступен или внутренний буфер бэкенда переполнен из-за аномально высокого FPS от клиента.

{
  "error_code": "ERR-STREAM-BUFFER-OVERFLOW",
  "message": "Поток камеры временно перегружен. Пожалуйста, удерживайте камеру неподвижно.",
  "details": {
    "reason": "Kafka broker partition leader is unavailable or local producer memory buffer is full."
  }
}

5.3.3 Сценарий В: Разрыв сетевого соединения в процессе стриминга (Close Code)

Если у пользователя пропадает 4G/Wi-Fi соединение на Шаге 4, WebSocket-сессия аварийно закрывается операционной системой.

  • Поведение системы: FastAPI ловит системное событие WebSocketDisconnect. Бэкенд немедленно отправляет команду в Kafka на удаление промежуточных необработанных кадров этой сессии, очищая партицию. В базе данных PostgreSQL транзакция черновика аннулируется. При восстановлении сети Flutter-приложение генерирует новый session_id и начинает процесс с Шага 2.