В первой статье цикла я разобрал real-time-конвейер видеоаналитики на C++20. Затем мы настроили систему через браузерный Viewer и безопасно опубликовали её через VPS. Теперь посмотрим на Python-реализацию того же сервера — глазами разработчика, привыкшего к RAII, std::thread, явному владению памятью и статической типизации.

Главная сложность такого перехода — не синтаксис. Написать цикл, класс или HTTP-обработчик на Python можно довольно быстро. Гораздо важнее понять другую модель выполнения: что именно хранит переменная, когда создаётся копия кадра, зачем одновременно нужны asyncio, потоки и процессы, где проявляется GIL и почему сборщик мусора не заменяет управление ресурсами.

Разбирать это будем не на игрушечных примерах, а на архитектуре реального сервера: RTSP-вход, OpenCV, детектор движения, две модели YOLO через ONNX Runtime, браузерный просмотр, HLS, запись фрагментов и PostgreSQL.

Одна задача — две инженерные модели

Python- и C++-серверы реализуют один REST-контракт и решают одну задачу, но устроены по-разному.

Эта таблица показывает важную вещь: многие свойства системы определяет не язык. Кодек HLS всё равно исполняет FFmpeg, модель YOLO — ONNX Runtime, а декодирование RTSP — OpenCV и системные библиотеки. Язык в первую очередь меняет способ оркестрации, контроля состояния и жизненного цикла объектов.

Имя — не объект

После C++ легко бессознательно читать присваивание как создание нового значения. В Python имя обычно лишь начинает указывать на уже существующий объект:

second = frame

Теперь second и frame ссылаются на один массив. Если изменить пиксели через одно имя, изменение будет видно через другое.

Для независимого изображения нужна явная копия:

copy = frame.copy()

Срез NumPy обычно тоже не копирует данные, а создаёт представление:

crop = frame[y:y + height, x:x + width]
crop[:, :, 1] = 0  # может изменить исходный frame

Это похоже на поведение cv::Mat: копирование заголовка не означает копирование пикселей, а ROI может разделять буфер с исходным изображением. Поэтому C++-разработчику сама идея знакома — важно только не забывать, что в Python она распространяется почти на все изменяемые объекты.

Цена ошибки особенно заметна на видео. Один BGR-кадр 2304×1296 занимает:

2304 × 1296 × 3 = 8 957 952 байта ≈ 8,54 МиБ

Лишняя копия на каждом этапе при 30 кадрах в секунду быстро превращается в сотни мегабайт трафика памяти. Поэтому в конвейере кадр остаётся внутри видеопроцесса, а между процессами передаются только JPEG-превью и небольшие события.

Для конфигурации ситуация обратная. Если рабочий поток должен получить неизменяемый снимок параметров, поверхностного dict.copy() может быть недостаточно: вложенные списки и словари останутся общими. В таком месте осознанно применяется copy.deepcopy() или создаётся новая типизированная структура.

Что GIL запрещает — и чего он не запрещает

Короткая формулировка «в Python потоки не работают параллельно» слишком груба. В обычном CPython GIL не позволяет двум потокам одного процесса одновременно выполнять Python-байткод. Но видеосервер большую часть тяжёлой работы выполняет не в Python-байткоде.

• VideoCapture.read() ждёт сеть и декодер;

• NumPy считает в нативном коде;

• OpenCV выполняет обработку в C++;

• ONNX Runtime запускает нативный inference;

• FFmpeg работает отдельным процессом;

• очереди и condition variables часто находятся в ожидании.

Нативные библиотеки обычно освобождают GIL на время длительной операции. Поэтому потоки остаются полезными для координации ввода-вывода и C/C++-ядер.

А вот такой код масштабироваться по ядрам не будет:

def threshold_in_python(frame):
    for y in range(frame.shape[0]):
        for x in range(frame.shape[1]):
            if frame[y, x, 0] > 120:
                frame[y, x] = 255

Здесь внутренний цикл исполняет Python-байткод и удерживает GIL. Правильное решение — выразить операцию через NumPy или OpenCV:

mask = frame[:, :, 0] > 120
frame[mask] = 255

Практическое правило простое: потоки подходят, если задача ждёт I/O либо надолго уходит в библиотеку, освобождающую GIL. Для тяжёлого чистого Python-кода нужны процессы, векторизация или нативное расширение.

asyncio — не ещё один вид потока

FastAPI поддерживает async def, и отсюда возникает опасное предположение: если обработчик асинхронный, любая операция внутри него перестаёт блокировать сервер. Это не так.

asyncio даёт конкурентность, пока корутина регулярно возвращает управление циклу событий через await. Обычный блокирующий вызов внутри неё останавливает тот же поток, который обслуживает остальные HTTP-запросы.

Плохой вариант:

@app.get("/history")
async def history():
    return repository.load_all()  # синхронный запрос к БД

Если load_all() работает долго, event loop ждёт вместе с ним. Блокирующую операцию можно вынести в рабочий поток:

@app.get("/history")
async def history():
    return await asyncio.to_thread(repository.load_all)

Тот же приём подходит для редких фоновых процедур:

async def maintain_history():
    while True:
        await asyncio.to_thread(history.remove_expired)
        await asyncio.sleep(3600)

Но отправлять через to_thread() каждый кадр в YOLO не стоит. Для постоянного потока данных лучше создать долгоживущий worker с ограниченным входным слотом: меньше накладных расходов, понятнее владение моделью и проще контролировать отставание.

Три уровня конкурентности в одном сервере

В Python-версии одновременно используются корутины, потоки и процессы. Это не дублирование, а разделение задач по их природе.

Основной процесс
├── FastAPI / Uvicorn
├── asyncio: REST, периодическая очистка истории
├── RuntimeState и PostgreSQL
└── Queue команд и событий
          │
          ▼
Видеопроцесс
├── поток захвата RTSP
├── worker Motion
├── worker YOLO
├── worker Tracking YOLO
├── сборка кадра для просмотра
└── управление FFmpeg
          │
          ▼
Отдельные процессы FFmpeg: HLS и запись

asyncio удобно обслуживает большое число коротких HTTP-операций и таймеров. Потоки внутри видеопроцесса связывают OpenCV, ONNX Runtime и FFmpeg без копирования каждого кадра между адресными пространствами. Отдельный процесс изолирует тяжёлый видеоконвейер от REST API: ошибка декодера или длительный inference не должны останавливать приём управляющих запросов.

По этой же причине Uvicorn запускается с одним worker-процессом. Несколько независимых workers создали бы несколько экземпляров состояния, подключений к камере и владельцев HLS-каталога. Увеличение числа HTTP-процессов здесь не является бесплатным способом масштабирования.

Граница процесса должна быть узкой

Для создания видеопроцесса используется контекст spawn. Он запускает новый интерпретатор и заставляет явно определить данные, которые могут пересечь границу процесса.

В сервере есть две ограниченные очереди:

• очередь команд: открыть поток, изменить конфигурацию, начать запись, остановиться;

• очередь событий: состояние подключения, JPEG, FPS, ошибка, профиль и обновление истории.

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

commands.put({
    "type": "configure",
    "payload": configuration,
})

events.put({
    "type": "preview",
    "jpeg": encoded_frame,
    "fps": current_fps,
})

Сырые numpy.ndarray через эту границу не ходят. Иначе каждый кадр пришлось бы сериализовать, копировать и передавать через межпроцессный канал. При размере 8,54 МиБ это уничтожило бы преимущество изоляции.

Ограниченный размер очереди тоже принципиален. Неограниченная очередь маскирует перегрузку: сервер продолжает принимать кадры, но пользователь смотрит всё более далёкое прошлое. В real-time-системе свежесть часто важнее полноты.

Не очередь кадров, а слот последнего значения

Если камера выдаёт 25 кадров в секунду, а YOLO обрабатывает 10, FIFO-очередь неизбежно начинает расти. Увеличение памяти лишь откладывает проблему. Через минуту результат может быть точным, но относиться к уже неактуальной сцене.

Вместо этого worker хранит один ожидающий пакет:

class LatestSlot:
    def __init__(self):
        self._condition = threading.Condition()
        self._pending = None
        self._closed = False

    def offer(self, packet):
        with self._condition:
            self._pending = packet
            self._condition.notify()

    def take(self):
        with self._condition:
            self._condition.wait_for(
                lambda: self._pending is not None or self._closed
            )
            packet, self._pending = self._pending, None
            return packet

Для полного детектора можно принимать новый кадр только тогда, когда слот свободен. Для tracking-ветки полезнее заменять ещё не начатый пакет более свежим. В обоих случаях задержка остаётся ограниченной, а система явно учитывает skipped и replaced, а не скрывает перегрузку в глубине очереди.

Эта архитектурная идея одинакова в C++ и Python. Меняются примитивы синхронизации, но не физика потока данных.

Состояние: короткая блокировка, длинная работа снаружи

Основной процесс хранит текущее состояние сервера: активную сессию, режим записи, последние метрики и параметры просмотра. Доступ защищён threading.RLock.

Замок нужен не потому, что GIL якобы отсутствует. Составная операция над несколькими полями всё равно должна быть атомарной с точки зрения приложения, а нативный код между отдельными инструкциями может освобождать GIL.

Правильный шаблон выглядит так:

with state.lock:
    snapshot = state.make_snapshot()

# SQL, OpenCV, сеть и сериализация выполняются без lock
result = build_response(snapshot)

Под блокировкой создаётся короткий согласованный снимок. Декодирование, inference, SQL-запросы и отправка ответа выполняются уже снаружи. Иначе один медленный клиент способен задержать весь сервер.

Сборщик мусора не заменяет RAII

Python освобождает память объектов автоматически, но внешние ресурсы требуют детерминированного завершения не меньше, чем в C++:

• RTSP-соединение нужно закрыть;

• поток — попросить остановиться и join();

• дочерний процесс — завершить и дождаться;

• stdin FFmpeg — закрыть, чтобы кодировщик корректно дописал файл;

• зависший FFmpeg — сначала terminate(), затем при необходимости kill();

• пул соединений с БД — закрыть после остановки фоновых задач.

Ближайший аналог локального RAII — контекстный менеджер:

with open(path, "rb") as source:
    data = source.read()

Для ресурса с более сложным жизненным циклом применяется try/finally:

worker.start()
try:
    run_server()
finally:
    worker.stop()
    worker.join(timeout=5)

При завершении приложения ресурсы закрываются в обратном порядке: сначала прекращается приём новой работы, затем останавливаются фоновые задачи, видеопроцесс, FFmpeg и база данных. Надежда на то, что интерпретатор «сам всё соберёт», приводит к повреждённым записям, зависшим процессам и занятым портам.

Python защищает от многих вариантов use-after-free, но не гарантирует отсутствие гонок и своевременное освобождение внешних ресурсов.

Обработчик FastAPI должен оставаться тонким

REST-метод в такой системе не место для видеоконвейера. Его работа обычно состоит из пяти шагов:

1. Проверить роль пользователя.

2. Разобрать и провалидировать входные данные.

3. Кратко обновить RuntimeState.

4. Отправить команду видеопроцессу.

5. Вернуть небольшой JSON-ответ.

Например:

@app.post("/recording/start")
async def start_recording(request: RecordingRequest):
    require_operator()

    with runtime.lock:
        runtime.recording_requested = True

    video_commands.put({
        "type": "start_recording",
        "payload": request.model_dump(),
    })
    return {"accepted": True}

HTTP-обработчик подтверждает приём команды, но не ждёт, пока FFmpeg откроет файл и запишет первый GOP. Фактическое состояние позже приходит через очередь событий. Это обычная модель команд и событий, а не особенность FastAPI.

Что действительно сравнивать в Python и C++

Некорректно приписывать языку всё поведение сервера. Если обе реализации используют OpenCV, ONNX Runtime и FFmpeg, значительная часть нагрузки выполняется одним и тем же нативным кодом.

Python даёт быструю разработку, удобную интеграцию с ML-экосистемой и компактный код оркестрации. C++ даёт детерминированное время жизни, статическую типизацию, предсказуемый контроль памяти и свободу от GIL в собственных CPU-циклах.

Но выбор языка не исправит неверную архитектуру. Неограниченная очередь будет накапливать задержку и в Python, и в C++. Лишнее копирование восьмимегабайтного кадра останется дорогим в обоих случаях. А перенос Python-цикла по пикселям в отдельный поток не превратит его в NumPy.

Итоговая модель

Python-видеосервер удобно держать в голове как девять правил:

1. FastAPI-процесс владеет REST API, авторизацией, состоянием и PostgreSQL.

2. Видеопроцесс владеет OpenCV и ONNX Runtime.

3. Захват хранит последний кадр, а не бесконечную историю.

4. Motion и две ветки YOLO работают независимо.

5. Координатор собирает кадр из последних доступных результатов.

6. FFmpeg остаётся отдельным системным процессом.

7. Через межпроцессную границу проходят команды, события и JPEG, но не сырые кадры.

8. Очереди и слоты ограничены, поэтому перегрузка не превращается в растущую задержку.

9. Каждый внешний ресурс имеет явного владельца и процедуру остановки.

Если принять эту модель, Python перестаёт выглядеть «упрощённым C++». Это другой уровень оркестрации вокруг тех же нативных библиотек и тех же ограничений real-time-системы.

В следующей статье посмотрим на Rust-версию того же сервера: как Arc<Frame>, move-семантика и Send/Sync меняют владение кадрами и межпоточную передачу — и где строгие гарантии Rust всё равно не заменяют архитектуру и измерения.