🎞️ Резюме видео как услуга 🐍

👋 Здравствуйте!

Уважаемые разработчики,

Вам нравится пособие “одна картинка стоит тысячи слов” ? Я делаю!
Давайте проверим, работает ли он для «Картина стоит тысячи кадров» Отказ
В этом руководстве вы увидите следующее:
- Как понять содержание видео в мигании,
- менее чем за 300 строк кода Python (3.9).

Вот визуальный сводный пример, сгенерированный из 2’42 “видео из 35 последовательностей (также известных как видео снимки):

Примечание. Сводка – это сетка, где каждая ячейка является рамкой, представляющей видео выстрел.

🔭 Целью

В этом руководстве есть 2 цели, 1 практические и 1 технические:

Автоматически генерировать визуальные сводки видео
Создайте обработку конвейера с этими свойствами:
- Управлялся (всегда готов и легко настроить)
- Масштабируемость (возможность глотать несколько видео параллельно)
- не стоит ничего, когда нет использовал

🛠️ инструменты

Несколько инструментов достаточно:

Пространство для хранения для видео и результатов
Несоведочное решение для запуска кода
Модель машинного обучения для анализа видео
Библиотека для извлечения кадров из видео
Библиотека для создания визуальных сводок

🧱 архитектура

Вот возможная архитектура, используя 3 Услуги Google Cloud ( Облачное хранилище , Облачные функции и Видео интеллект API ):

Трубопровод обработки следует следующим образом:

Вы загружаете видео в 1-е ведро (ведро – это место для хранения в облаке)
Событие загрузки автоматически запускает 1-й функцию
Функция отправляет запрос на API видео интеллекта для обнаружения выстрелов
API видео интеллекта анализирует видео и загружает результаты (аннотации) во 2-й ведре
Событие загрузки триггеры 2-го функции
Функция загружает как аннотацию, так и видеофайлов
Функция отображает и загружает резюме к 3-м ведрю
Резюме видео готов!

🐍 Библиотеки Python

Клиентские библиотеки с открытым исходным кодом позволяют интерфейсу с сервисами Google Cloud в идиоматическом Python. Вы будете использовать следующее:

Облачное хранилище
- Управление загрузками и загрузками
- https://pypi.org/project/google-cloud-storage
Видео интеллект API.
- Проанализировать видео
- https://pypi.org/project/google-cloud-videointelligence

Вот выбор из 2 дополнительных библиотек Python для графических нужд:

Openc.
- Чтобы извлечь видеокадры
- Есть даже без головы без головы (без функций графического интерфейса), которая идеально подходит для обслуживания
- https://pypi.org/project/opencv-python-headless
Подушка
- Генерировать визуальные резюме
- Подушка это очень популярная библиотека изображений, как обширная, так и простая в использовании
- https://pypi.org/project/Pillow

⚙️. Настройка проекта

Предполагая, что у вас есть учетная запись Google Cloud, вы можете настроить архитектуру из облачной оболочки с gcloud и ГЮТИЛ команды. Это позволяет вам сценарировать все с нуля воспроизводимым способом.

Переменные среды

# Project
PROJECT_NAME="Visual Summary"
PROJECT_ID="visual-summary-REPLACE_WITH_UNIQUE_SUFFIX"
# Cloud Storage region (https://cloud.google.com/storage/docs/locations)
GCS_REGION="europe-west1"
# Cloud Functions region (https://cloud.google.com/functions/docs/locations)
GCF_REGION="europe-west1"
# Source
GIT_REPO="cherry-on-py"
PROJECT_SRC=~/$PROJECT_ID/$GIT_REPO/gcf_video_summary

# Cloud Storage buckets (environment variables)
export VIDEO_BUCKET="b1-videos_${PROJECT_ID}"
export ANNOTATION_BUCKET="b2-annotations_${PROJECT_ID}"
export SUMMARY_BUCKET="b3-summaries_${PROJECT_ID}"

Примечание. Вы можете использовать имя пользователя GitHUB в качестве уникального суффикса.

Новый проект

gcloud projects create $PROJECT_ID \
  --name="$PROJECT_NAME" \
  --set-as-default

Create in progress for [https://cloudresourcemanager.googleapis.com/v1/projects/PROJECT_ID].
Waiting for [operations/cp...] to finish...done.
Enabling service [cloudapis.googleapis.com] on project [PROJECT_ID]...
Operation "operations/acf..." finished successfully.
Updated property [core/project] to [PROJECT_ID].

Биллинг Счет

# Link project with billing account (single account)
BILLING_ACCOUNT=$(gcloud beta billing accounts list \
    --format 'value(name)')
# Link project with billing account (specific one among multiple accounts)
BILLING_ACCOUNT=$(gcloud beta billing accounts list \
    --format 'value(name)' \
    --filter "displayName='My Billing Account'")

gcloud beta billing projects link $PROJECT_ID --billing-account $BILLING_ACCOUNT

billingAccountName: billingAccounts/XXXXXX-YYYYYY-ZZZZZZ
billingEnabled: true
name: projects/PROJECT_ID/billingInfo
projectId: PROJECT_ID

Ведра

# Create buckets with uniform bucket-level access
gsutil mb -b on -c regional -l $GCS_REGION gs://$VIDEO_BUCKET
gsutil mb -b on -c regional -l $GCS_REGION gs://$ANNOTATION_BUCKET
gsutil mb -b on -c regional -l $GCS_REGION gs://$SUMMARY_BUCKET

Creating gs://VIDEO_BUCKET/...
Creating gs://ANNOTATION_BUCKET/...
Creating gs://SUMMARY_BUCKET/...

Вы можете проверить, как это выглядит в Облачная консоль :

Сервисная учетная запись

Создайте учетную запись услуг. Это только для целей развития (не требуется для производства). Это предоставляет вам учетные данные для выполнения вашего кода локально.

mkdir ~/$PROJECT_ID
cd ~/$PROJECT_ID

SERVICE_ACCOUNT_NAME="dev-service-account"
SERVICE_ACCOUNT="${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
gcloud iam service-accounts create $SERVICE_ACCOUNT_NAME
gcloud iam service-accounts keys create ~/$PROJECT_ID/key.json --iam-account $SERVICE_ACCOUNT

Created service account [SERVICE_ACCOUNT_NAME].
created key [...] of type [json] as [~/PROJECT_ID/key.json] for [SERVICE_ACCOUNT]

Установите Google_aplication_credentials Переменная среды и убедитесь, что она указывает на ключ учетной записи службы. Когда вы запускаете код приложения в текущем сеансе оболочки, клиентские библиотеки будут использовать эти учетные данные для аутентификации. Если вы открываете новую сеанс оболочки, снова установите переменную.

export GOOGLE_APPLICATION_CREDENTIALS=~/$PROJECT_ID/key.json
cat $GOOGLE_APPLICATION_CREDENTIALS

{
  "type": "service_account",
  "project_id": "PROJECT_ID",
  "private_key_id": "...",
  "private_key": "-----BEGIN PRIVATE KEY-----\n...",
  "client_email": "SERVICE_ACCOUNT",
  ...
}

Авторизуйте учетную запись услуг для доступа к ведрам:

IAM_BINDING="serviceAccount:${SERVICE_ACCOUNT}:roles/storage.objectAdmin"
gsutil iam ch $IAM_BINDING gs://$VIDEO_BUCKET
gsutil iam ch $IAM_BINDING gs://$ANNOTATION_BUCKET
gsutil iam ch $IAM_BINDING gs://$SUMMARY_BUCKET

Апис

Несколько API включены по умолчанию:

gcloud services list

NAME                              TITLE
bigquery.googleapis.com           BigQuery API
bigquerystorage.googleapis.com    BigQuery Storage API
cloudapis.googleapis.com          Google Cloud APIs
clouddebugger.googleapis.com      Cloud Debugger API
cloudtrace.googleapis.com         Cloud Trace API
datastore.googleapis.com          Cloud Datastore API
logging.googleapis.com            Cloud Logging API
monitoring.googleapis.com         Cloud Monitoring API
servicemanagement.googleapis.com  Service Management API
serviceusage.googleapis.com       Service Usage API
sql-component.googleapis.com      Cloud SQL
storage-api.googleapis.com        Google Cloud Storage JSON API
storage-component.googleapis.com  Cloud Storage

Включите видеовыносные функции, функции облачных и облачных API:

gcloud services enable \
  videointelligence.googleapis.com \
  cloudfunctions.googleapis.com \
  cloudbuild.googleapis.com

Operation "operations/acf..." finished successfully.

Примечание. Cloud Build генерирует контейнерные изображения для облачных функций при развертывании.

Исходный код

Получить исходный код:

cd ~/$PROJECT_ID
git clone https://github.com/PicardParis/$GIT_REPO.git

Cloning into 'GIT_REPO'...
...

🧠 Видео анализ

Обнаружение видео выстрел

API Video Intelligence – это предварительно обученная модель обучения машины, которая может анализировать видео. Одним из множественных функций является обнаружение видео выстрела. Для функции 1-го облаков вот возможная основная функция, вызывающая Annotate_Video () с Shot_change_detection характерная черта:

from google.cloud import storage, videointelligence

def launch_shot_detection(video_uri: str, annot_bucket: str):
    """Detect video shots (asynchronous operation)

    Results will be stored in  with this naming convention:
    - video_uri: gs://video_bucket/path/to/video.ext
    - annot_uri: gs://annot_bucket/video_bucket/path/to/video.ext.json
    """
    print(f"Launching shot detection for <{video_uri}>...")
    features = [videointelligence.Feature.SHOT_CHANGE_DETECTION]
    video_blob = storage.Blob.from_string(video_uri)
    video_bucket = video_blob.bucket.name
    path_to_video = video_blob.name
    annot_uri = f"gs://{annot_bucket}/{video_bucket}/{path_to_video}.json"
    request = dict(features=features, input_uri=video_uri, output_uri=annot_uri)

    video_client = videointelligence.VideoIntelligenceServiceClient()
    video_client.annotate_video(request)

Местное развитие и тесты

Перед развертыванием функции вам нужно разработать и тестировать ее. Создайте виртуальную среду Python 3 и активируйте ее:

cd ~/$PROJECT_ID
python3 -m venv venv
source venv/bin/activate

Установите зависимости:

pip install -r $PROJECT_SRC/gcf1_detect_shots/requirements.txt

Проверьте зависимости:

pip list

Package                        Version
----------------------------------- ----------
...
google-cloud-storage           1.42.3
google-cloud-videointelligence 2.3.3
...

Вы можете использовать основной объем, чтобы проверить функцию в режиме сценария:

import os

ANNOTATION_BUCKET = os.getenv("ANNOTATION_BUCKET", "")
assert ANNOTATION_BUCKET, "Undefined ANNOTATION_BUCKET environment variable"

if __name__ == "__main__":
    # Local tests only (service account needed)
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "video_uri", type=str, help="gs://video_bucket/path/to/video.ext"
    )
    args = parser.parse_args()
    launch_shot_detection(args.video_uri, ANNOTATION_BUCKET)

Примечание: вы уже экспортировали Annotation_bucket Переменная среды ранее в сеансе оболочки; Вы также будете использовать его позже на этапе развертывания. Это делает код универсальным и позволяет вам повторно использовать его независимо от выходного ведра.

Проверьте функцию:

VIDEO_PATH="cloud-samples-data/video/gbikes_dinosaur.mp4"
VIDEO_URI="gs://$VIDEO_PATH"
python $PROJECT_SRC/gcf1_detect_shots/main.py $VIDEO_URI

Launching shot detection for ...

Примечание: тестовое видео расположен во внешнем ведре. Это работает, потому что видео является общедоступным.

Подождите минутку и убедитесь, что аннотации были сгенерированы:

gsutil ls -r gs://$ANNOTATION_BUCKET

964  YYYY-MM-DDThh:mm:ssZ  gs://ANNOTATION_BUCKET/VIDEO_PATH.json
TOTAL: 1 objects, 964 bytes (964 B)

Проверьте последние 200 байтов файла аннотации:

gsutil cat -r -200 gs://$ANNOTATION_BUCKET/$VIDEO_PATH.json

}
    }, {
      "start_time_offset": {
        "seconds": 28,
        "nanos": 166666000
      },
      "end_time_offset": {
        "seconds": 42,
        "nanos": 766666000
      }
    } ]
  } ]
}

ПРИМЕЧАНИЕ. Это начинающие и конечные позиции последнего видео выстрела. Все кажется хорошо.

Очистите, когда закончите:

gsutil rm gs://$ANNOTATION_BUCKET/$VIDEO_PATH.json

deactivate

rm -rf venv

Точка входа в функцию

def gcf_detect_shots(data, context):
    """Cloud Function triggered by a new Cloud Storage object"""
    video_bucket = data["bucket"]
    path_to_video = data["name"]
    video_uri = f"gs://{video_bucket}/{path_to_video}"
    launch_shot_detection(video_uri, ANNOTATION_BUCKET)

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

Развертывание функций

Разверните 1-й функцию:

GCF_NAME="gcf1_detect_shots"
GCF_SOURCE="$PROJECT_SRC/gcf1_detect_shots"
GCF_ENTRY_POINT="gcf_detect_shots"
GCF_TRIGGER_BUCKET="$VIDEO_BUCKET"
GCF_ENV_VARS="ANNOTATION_BUCKET=$ANNOTATION_BUCKET"
GCF_MEMORY="128MB"

gcloud functions deploy $GCF_NAME \
  --runtime python39 \
  --source $GCF_SOURCE \
  --entry-point $GCF_ENTRY_POINT \
  --update-env-vars $GCF_ENV_VARS \
  --trigger-bucket $GCF_TRIGGER_BUCKET \
  --region $GCF_REGION \
  --memory $GCF_MEMORY \
  --quiet

Примечание. Память по умолчанию, выделенная для облачной функции, составляет 256 МБ (возможные значения – 128 МБ, 256 МБ, 512 МБ, 1024 МБ и 2048 МБ). Поскольку функция не имеет никаких потребностей памяти или процессора (он отправляет простой запрос API), достаточно минимальной настройки памяти.

Deploying function (may take a while - up to 2 minutes)...done.
availableMemoryMb: 128
entryPoint: gcf_detect_shots
environmentVariables:
  ANNOTATION_BUCKET: b2-annotations...
eventTrigger:
  eventType: google.storage.object.finalize
...
status: ACTIVE
timeout: 60s
updateTime: 'YYYY-MM-DDThh:mm:ss.mmmZ'
versionId: '1'

Примечание: Annotation_bucket Переменная среды определяется с помощью --update-env-vars флаг. Использование переменной среды позволяет развертывать тот же код с разными ведрами для триггера и вывода.

Вот как это выглядит в Облачная консоль :

Производственные испытания

Обязательно проверьте функцию в производстве. Скопируйте видео в ведро видео:

VIDEO_NAME="gbikes_dinosaur.mp4"
SRC_URI="gs://cloud-samples-data/video/$VIDEO_NAME"
DST_URI="gs://$VIDEO_BUCKET/$VIDEO_NAME"

gsutil cp $SRC_URI $DST_URI

Copying gs://cloud-samples-data/video/gbikes_dinosaur.mp4 [Content-Type=video/mp4]...
- [1 files][ 62.0 MiB/ 62.0 MiB]
Operation completed over 1 objects/62.0 MiB.

Запрос журналов, чтобы проверить, что функция была вызвана:

gcloud functions logs read --region $GCF_REGION

LEVEL  NAME               EXECUTION_ID  TIME_UTC  LOG
D      gcf1_detect_shots  ...           ...       Function execution started
I      gcf1_detect_shots  ...           ...       Launching shot detection for ...
D      gcf1_detect_shots  ...           ...       Function execution took 874 ms, finished with status: 'ok'

Подождите минутку и проверьте корпус аннотации:

gsutil ls -r gs://$ANNOTATION_BUCKET

Вы должны увидеть файл аннотации:

gs://ANNOTATION_BUCKET/VIDEO_BUCKET/:
gs://ANNOTATION_BUCKET/VIDEO_BUCKET/VIDEO_NAME.json

1-я функция работает!

🎞️ Визуальное резюме

Структура кода

Интересно разделить код на 2 основных класса:

StorageHelper для локального файла и Управление объектом облачного хранения
Видео процессор Для графической обработки

Вот возможная основная функция:

class VideoProcessor:
    @staticmethod
    def generate_summary(annot_uri: str, output_bucket: str):
        """ Generate a video summary from video shot annotations """
        try:
            with StorageHelper(annot_uri, output_bucket) as storage:
                with VideoProcessor(storage) as video_proc:
                    print("Generating summary...")
                    image = video_proc.render_summary()
                    video_proc.upload_summary_as_jpeg(image)
        except Exception:
            logging.exception("Could not generate summary from <%s>", annot_uri)

Примечание. Если исключения подняты, это удобно входить в систему с регистрация .Exception () Чтобы получить трассировку стека в журналах производства.

Классный хранилище

Класс управляет следующим:

Извлечение и разбор аннотаций видео съемки
Загрузка исходных видео
Загрузка сгенерированных визуальных резюме
Имена файлов

class StorageHelper:
    """Local+Cloud storage helper

    - Uses a temp dir for local processing (e.g. video frame extraction)
    - Paths are relative to this temp dir (named after the output bucket)

    Naming convention:
    - video_uri:                 gs://video_bucket/path/to/video.ext
    - annot_uri:    gs://annot_bucket/video_bucket/path/to/video.ext.json
    - video_path:                     video_bucket/path/to/video.ext
    - summary_path:                   video_bucket/path/to/video.ext.SUFFIX
    - summary_uri: gs://output_bucket/video_bucket/path/to/video.ext.SUFFIX
    """

    client = storage.Client()
    video_shots: list[VideoShot]
    video_path: Path
    video_local_path: Path
    upload_bucket: storage.Bucket

    def __init__(self, annot_uri: str, output_bucket: str):
        if not annot_uri.endswith(ANNOT_EXT):
            raise RuntimeError(f"annot_uri must end with <{ANNOT_EXT}>")
        self.video_shots = self.get_video_shots(annot_uri)
        self.video_path = self.video_path_from_uri(annot_uri)
        temp_root = Path(tempfile.gettempdir(), output_bucket)
        temp_root.mkdir(parents=True, exist_ok=True)
        self.video_local_path = temp_root.joinpath(self.video_path)
        self.upload_bucket = self.client.bucket(output_bucket)

Исходное видео обрабатывается в с Оперативный менеджер контекста:

    def __enter__(self):
        self.download_video()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.video_local_path.unlink()

Примечание. После загрузки видео использует пространство памяти в /TMP RAM Disk (единственное пиректируемое пространство для функции без сервеса). Лучше всего удалять временные файлы, когда они больше не нужны, чтобы избежать потенциальных ошибок в памяти на будущих вызовах функции.

Аннотации видео могут быть получены методами место хранения. Blob.download_as_text () и json.loads () :

    def get_video_shots(self, annot_uri: str) -> list[VideoShot]:
        json_blob = storage.Blob.from_string(annot_uri, self.client)
        api_response: dict = json.loads(json_blob.download_as_text())
        single_video_results: dict = api_response["annotation_results"][0]
        annotations: list = single_video_results["shot_annotations"]
        return [VideoShot.from_dict(annotation) for annotation in annotations]

Пашин разбирается с этим Видеошот Класс помощника:

class VideoShot(NamedTuple):
    """Video shot start/end positions in nanoseconds"""

    pos1_ns: int
    pos2_ns: int
    NANOS_PER_SECOND = 10 ** 9

    @classmethod
    def from_dict(cls, annotation: dict) -> "VideoShot":
        def time_offset_in_ns(time_offset) -> int:
            seconds: int = time_offset.get("seconds", 0)
            nanos: int = time_offset.get("nanos", 0)
            return seconds * cls.NANOS_PER_SECOND + nanos

        pos1_ns = time_offset_in_ns(annotation["start_time_offset"])
        pos2_ns = time_offset_in_ns(annotation["end_time_offset"])
        return cls(pos1_ns, pos2_ns)

Конвенция об именах была выбрана для сохранения последовательных объектных путей между различными ведрами. Это также позволяет выводить видео путь от URI аннотации:

    def video_path_from_uri(self, annot_uri: str) -> Path:
        annot_blob = storage.Blob.from_string(annot_uri)
        return Path(annot_blob.name[: -len(ANNOT_EXT)])

Видео прямо загружено с место хранения. Blob.download_to_filename () :

    def download_video(self):
        video_uri = f"gs://{self.video_path.as_posix()}"
        blob = storage.Blob.from_string(video_uri, self.client)
        print(f"Downloading -> {self.video_local_path}")
        self.video_local_path.parent.mkdir(parents=True, exist_ok=True)
        blob.download_to_filename(self.video_local_path)

Наоборот, результаты могут быть загружены место хранения. Blob.upload_from_string () :

    def upload_summary(self, image_bytes: bytes, image_type: str):
        path = self.summary_path(image_type)
        blob = self.upload_bucket.blob(path.as_posix())
        content_type = f"image/{image_type}"
        print(f"Uploading -> {blob.name}")
        blob.upload_from_string(image_bytes, content_type)

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

И, наконец, вот возможная конвенция именования для сводных файлов:

    def summary_path(self, image_type: str) -> Path:
        video_name = self.video_path.name
        shot_count = len(self.video_shots)
        suffix = f"summary{shot_count:03d}.{image_type}"
        summary_name = f"{video_name}.{suffix}"
        return Path(self.video_path.parent, summary_name)

Видеопроцессор класса

Класс управляет следующим:

Добыча видеокадра
Визуальное сводное поколение

import cv2 as cv
from PIL import Image

from storage_helper import StorageHelper

PilImage = Image.Image
ImageSize = NamedTuple("ImageSize", [("w", int), ("h", int)])


class VideoProcessor:
    storage: StorageHelper
    video: cv.VideoCapture
    cell_size: ImageSize
    grid_size: ImageSize

    def __init__(self, storage: StorageHelper):
        self.storage = storage

Открытие и закрытие видео обработано в с Оперативный менеджер контекста:

    def __enter__(self):
        video_path = self.storage.video_local_path
        self.video = cv.VideoCapture(str(video_path))
        if not self.video.isOpened():
            raise RuntimeError(f"Could not open video <{video_path}>")
        self.compute_grid_dimensions()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.video.release()

Резюме видео – это сетка клеток, которые могут быть отображены в одной петле с двумя генераторами:

    def render_summary(self, shot_ratio: float = 0.5) -> PilImage:
        grid_img = Image.new("RGB", self.grid_size, RGB_BACKGROUND)

        img_and_pos_iter = zip(self.gen_cell_img(shot_ratio), self.gen_cell_pos())
        for cell_img, cell_pos in img_and_pos_iter:
            cell_img.thumbnail(self.cell_size)  # Makes it smaller if needed
            grid_img.paste(cell_img, cell_pos)

        return grid_img

Примечание: shot_ratio установлен на 0,5 По умолчанию для извлечения видео выстрелил средние рамки.

Первый генератор дает мобильные изображения:

    def gen_cell_img(self, shot_ratio: float) -> Iterator[PilImage]:
        assert 0.0 <= shot_ratio <= 1.0
        MS_IN_NS = 10 ** 6
        for video_shot in self.storage.video_shots:
            pos1_ns, pos2_ns = video_shot
            pos_ms = (pos1_ns + shot_ratio * (pos2_ns - pos1_ns)) / MS_IN_NS
            yield self.frame_at_position(pos_ms)

Второй генератор дает элементы клеток:

    def gen_cell_pos(self) -> Iterator[tuple[int, int]]:
        cell_x, cell_y = 0, 0
        while True:
            yield cell_x, cell_y
            cell_x += self.cell_size.w
            if self.grid_size.w <= cell_x:  # Move to next row?
                cell_x, cell_y = 0, cell_y + self.cell_size.h

Opencv Легко позволяет извлекать видеокадры в заданном положении:

    def frame_at_position(self, pos_ms: float) -> PilImage:
        self.video.set(cv.CAP_PROP_POS_MSEC, pos_ms)
        _, cv_frame = self.video.read()
        return Image.fromarray(cv.cvtColor(cv_frame, cv.COLOR_BGR2RGB))

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

    def compute_grid_dimensions(self):
        shot_count = len(self.storage.video_shots)
        if shot_count < 1:
            raise RuntimeError(f"Expected 1+ video shots (got {shot_count})")
        # Try to preserve the video aspect ratio
        # Consider cells as pixels and try to fit them in a square
        cols = rows = int(shot_count ** 0.5 + 0.5)
        if cols * rows < shot_count:
            cols += 1
        cell_w = int(self.video.get(cv.CAP_PROP_FRAME_WIDTH))
        cell_h = int(self.video.get(cv.CAP_PROP_FRAME_HEIGHT))
        if SUMMARY_MAX_SIZE.w < cell_w * cols:
            scale = SUMMARY_MAX_SIZE.w / (cell_w * cols)
            cell_w = int(scale * cell_w)
            cell_h = int(scale * cell_h)
        self.cell_size = ImageSize(cell_w, cell_h)
        self.grid_size = ImageSize(cell_w * cols, cell_h * rows)

Наконец, Подушка Дает полный контроль на изображениях сериализации:

    def upload_summary_as_jpeg(self, image: PilImage):
        mem_file = BytesIO()
        image_type = "jpeg"
        jpeg_save_parameters = dict(optimize=True, progressive=True)
        image.save(mem_file, format=image_type, **jpeg_save_parameters)

        image_bytes = mem_file.getvalue()
        self.storage.upload_summary(image_bytes, image_type)

Примечание. Работа с изображениями в памяти позволяет избежать управления локальными файлами и использует меньше памяти.

Местное развитие и тесты

Вы можете использовать основной объем, чтобы проверить функцию в режиме сценария:

import os

from video_processor import VideoProcessor

SUMMARY_BUCKET = os.getenv("SUMMARY_BUCKET", "")
assert SUMMARY_BUCKET, "Undefined SUMMARY_BUCKET environment variable"

if __name__ == "__main__":
    # Local tests only (service account needed)
    import argparse

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "annot_uri", type=str, help="gs://annotation_bucket/path/to/video.ext.json"
    )
    args = parser.parse_args()
    VideoProcessor.generate_summary(args.annot_uri, SUMMARY_BUCKET)

Проверьте функцию:

cd ~/$PROJECT_ID
python3 -m venv venv
source venv/bin/activate

pip install -r $PROJECT_SRC/gcf2_generate_summary/requirements.txt

VIDEO_NAME="gbikes_dinosaur.mp4"
ANNOTATION_URI="gs://$ANNOTATION_BUCKET/$VIDEO_BUCKET/$VIDEO_NAME.json"

python $PROJECT_SRC/gcf2_generate_summary/main.py $ANNOTATION_URI

Downloading -> /tmp/SUMMARY_BUCKET/VIDEO_BUCKET/VIDEO_NAME
Generating summary...
Uploading -> VIDEO_BUCKET/VIDEO_NAME.summary004.jpeg

Примечание. Загруженное видео сводная информация показывает 4 снимка.

Очистить:

deactivate
rm -rf venv

Точка входа в функцию

def gcf_generate_summary(data, context):
    """Cloud Function triggered by a new Cloud Storage object"""
    annotation_bucket = data["bucket"]
    path_to_annotation = data["name"]
    annot_uri = f"gs://{annotation_bucket}/{path_to_annotation}"
    VideoProcessor.generate_summary(annot_uri, SUMMARY_BUCKET)

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

Развертывание функций

GCF_NAME="gcf2_generate_summary"
GCF_SOURCE="$PROJECT_SRC/gcf2_generate_summary"
GCF_ENTRY_POINT="gcf_generate_summary"
GCF_TRIGGER_BUCKET="$ANNOTATION_BUCKET"
GCF_ENV_VARS="SUMMARY_BUCKET=$SUMMARY_BUCKET"
GCF_TIMEOUT="540s"
GCF_MEMORY="512MB"

gcloud functions deploy $GCF_NAME \
  --runtime python39 \
  --source $GCF_SOURCE \
  --entry-point $GCF_ENTRY_POINT \
  --update-env-vars $GCF_ENV_VARS \
  --trigger-bucket $GCF_TRIGGER_BUCKET \
  --region $GCF_REGION \
  --timeout $GCF_TIMEOUT \
  --memory $GCF_MEMORY \
  --quiet

Примечания:

Тайм-аут по умолчанию для облачной функции составляет 60 секунд. Поскольку вы развертываете фоновую функцию с потенциально длинными процессами, установите его на максимальное значение (540 минут).
Вам также необходимо немного натянуть память для видео и изображений. В зависимости от размера ваших видео и максимального разрешения ваших выходных данных, или если вам нужно генерировать суммы быстрее (размер памяти и скорость VCPU соотносится), вы можете использовать более высокое значение (1024 МБ или 2048 МБ).

Deploying function (may take a while - up to 2 minutes)...done.
availableMemoryMb: 512
entryPoint: gcf_generate_summary
environmentVariables:
  SUMMARY_BUCKET: b3-summaries...
...
status: ACTIVE
timeout: 540s
updateTime: 'YYYY-MM-DDThh:mm:ss.mmmZ'
versionId: '1'

Вот как это выглядит в Облачная консоль :

Производственные испытания

Обязательно проверьте функцию в производстве. Вы можете загрузить файл аннотации в 2-м ведре:

VIDEO_NAME="gbikes_dinosaur.mp4"
ANNOTATION_FILE="$VIDEO_NAME.json"
ANNOTATION_URI="gs://$ANNOTATION_BUCKET/$VIDEO_BUCKET/$ANNOTATION_FILE"
gsutil cp $ANNOTATION_URI .
gsutil cp $ANNOTATION_FILE $ANNOTATION_URI
rm $ANNOTATION_FILE

Примечание. Это повторно использует предыдущий локальный файл аннотации теста и перезаписывает его. Перезапись файла в ведре также триггеры прилагаемые функции.

Подождите несколько секунд и запрашивайте журналы, чтобы проверить, что функция была запущена:

gcloud functions logs read --region $GCF_REGION

LEVEL  NAME                   EXECUTION_ID  TIME_UTC  LOG
...
D      gcf2_generate_summary  ...           ...       Function execution started
I      gcf2_generate_summary  ...           ...       Downloading -> /tmp/SUMMARY_BUCKET/VIDEO_BUCKET/VIDEO_NAME
I      gcf2_generate_summary  ...           ...       Generating summary...
I      gcf2_generate_summary  ...           ...       Uploading -> VIDEO_BUCKET/VIDEO_NAME.summary004.jpeg
D      gcf2_generate_summary  ...           ...       Function execution took 11591 ms, finished with status: 'ok'

2-я функция оперативна, а трубопровод на месте! Теперь вы можете выполнять сквозные тесты, копируя новые видео в 1-е ведро.

Полученные результаты

Загрузите сгенерированное резюме на вашем компьютере:

cd ~/$PROJECT_ID
gsutil cp -r gs://$SUMMARY_BUCKET/**.jpeg .
cloudshell download *.jpeg

Вот визуальное резюме для gbikes_dinosaur.mp4 (4 обнаружены выстрелы):

Вы также можете напрямую предварительно просмотреть файл из Облачная консоль :

🍒 вишня на PY 🐍

Теперь глазурь на пироге (или «вишне на пироге», как мы говорим по-французски) …

На основании той же архитектуры и кода вы можете добавить несколько функций:
- Запустить обработку для видео из других ведер
- Создать резюме в нескольких форматах (таких как JPEG, PNG, WebP)
- Создание анимированных резюме (также в нескольких форматах, таких как GIF, PNG, WebP)
Обегайте архитектуру, чтобы дублировать 2 предмета:
- Функция обнаружения видео выстрела, чтобы запускать его в качестве конечной точки HTTP
- Функция сводки генерации для обработки анимированных изображений
Адаптируйте код для поддержки новых функций:
- Анимированные Параметр для генерации еще или анимированных резюме
- Сохранить и загрузить результаты в нескольких форматах

Архитектура (V2)

A. Обнаружение видео выстрел также может быть вызвано вручную с HTTP Get Proke
B. До сих пор и анимированные резюме генерируются в 2 функциях параллельно
C. Резюме загружаются в нескольких форматах изображения

HTTP входная точка

def gcf_detect_shots_http(request):
    """Cloud Function triggered by an HTTP GET request"""
    if request.method != "GET":
        return ("Please use a GET request", 403)
    if not request.args or "video_uri" not in request.args:
        return ('Please specify a "video_uri" parameter', 400)
    video_uri = request.args["video_uri"]
    launch_shot_detection(video_uri, ANNOTATION_BUCKET)
    return f"Launched shot detection for <{video_uri}>"

Примечание: Это тот же код, что и gcf_detect_shots С помощью параметра Video URI предоставляется из запроса на получение.

Развертывание функций

GCF_NAME="gcf1_detect_shots_http"
GCF_SOURCE="$PROJECT_SRC/gcf1_detect_shots"
GCF_ENTRY_POINT="gcf_detect_shots_http"
GCF_TRIGGER_BUCKET="$VIDEO_BUCKET"
GCF_ENV_VARS="ANNOTATION_BUCKET=$ANNOTATION_BUCKET"
GCF_MEMORY="128MB"

gcloud functions deploy $GCF_NAME \
  --runtime python39 \
  --source $GCF_SOURCE \
  --entry-point $GCF_ENTRY_POINT \
  --update-env-vars $GCF_ENV_VARS \
  --trigger-http \
  --region $GCF_REGION \
  --memory $GCF_MEMORY \
  --quiet

Вот как это выглядит в Облачная консоль :

Поддержка анимации

Добавить Анимированные Опция в основной функции:

class VideoProcessor:
    @staticmethod
-   def generate_summary(annot_uri: str, output_bucket: str):
+   def generate_summary(annot_uri: str, output_bucket: str, animated=False):
        """ Generate a video summary from video shot annotations """
        try:
            with StorageHelper(annot_uri, output_bucket) as storage:
                with VideoProcessor(storage) as video_proc:
                    print("Generating summary...")
-                   image = video_proc.render_summary()
-                   video_proc.upload_summary_as_jpeg(image)
+                   if animated:
+                       video_proc.generate_summary_animations()
+                   else:
+                       video_proc.generate_summary_stills()
        except Exception:
            logging.exception("Could not generate summary from <%s>", annot_uri)

Определите форматы, которые вы заинтересованы в создании:

ImageFormat = NamedTuple("ImageFormat", [("type", str), ("save_parameters", dict)])

IMAGE_JPEG = ImageFormat("jpeg", dict(optimize=True, progressive=True))
IMAGE_GIF = ImageFormat("gif", dict(optimize=True))
IMAGE_PNG = ImageFormat("png", dict(optimize=True))
IMAGE_WEBP = ImageFormat("webp", dict(lossless=False, quality=80, method=1))
SUMMARY_STILL_FORMATS = (IMAGE_JPEG, IMAGE_PNG, IMAGE_WEBP)
SUMMARY_ANIMATED_FORMATS = (IMAGE_GIF, IMAGE_PNG, IMAGE_WEBP)

Добавьте поддержку для генерации неподвижных и анимированных резюме в разных форматах:

    def generate_summary_stills(self):
        image = self.render_summary()
        for image_format in SUMMARY_STILL_FORMATS:
            self.upload_summary([image], image_format)

    def generate_summary_animations(self):
        frame_count = ANIMATION_FRAMES
        images = []
        for frame_index in range(frame_count):
            shot_ratio = (frame_index + 1) / (frame_count + 1)
            print(f"shot_ratio: {shot_ratio:.0%}")
            image = self.render_summary(shot_ratio)
            images.append(image)
        for image_format in SUMMARY_ANIMATED_FORMATS:
            self.upload_summary(images, image_format)

Сериализация все еще может проходить в одной функции:

    def upload_summary(self, images: list[PilImage], image_format: ImageFormat):
        if not images:
            raise RuntimeError("Empty image list")
        mem_file = BytesIO()
        image_type = image_format.type
        save_parameters = image_format.save_parameters.copy()
        if animated := 1 < len(images):
            save_parameters |= dict(
                save_all=True,
                append_images=images[1:],
                duration=ANIMATION_FRAME_DURATION_MS,
                loop=0,  # Infinite loop
            )
        images[0].save(mem_file, format=image_type, **save_parameters)

        image_bytes = mem_file.getvalue()
        self.storage.upload_summary(image_bytes, image_type, animated)

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

Добавить Анимированные Дополнительный параметр для StorageHelper класс:

class StorageHelper:
-    def upload_summary(self, image_bytes: bytes, image_type: str):
-       path = self.summary_path(image_type)
+    def upload_summary(self, image_bytes: bytes, image_type: str, animated=False):
+       path = self.summary_path(image_type, animated)
        blob = self.upload_bucket.blob(path.as_posix())
        content_type = f"image/{image_type}"
        print(f"Uploading -> {blob.name}")
        blob.upload_from_string(image_bytes, content_type)

-   def summary_path(self, image_type: str) -> Path:
+   def summary_path(self, image_type: str, animated=False) -> Path:
        video_name = self.video_path.name
        shot_count = self.shot_count()
-       suffix = f"summary{shot_count:03d}.{image_type}"
+       still_or_anim = "anim" if animated else "still"
+       suffix = f"summary{shot_count:03d}_{still_or_anim}.{image_type}"
        summary_name = f'{video_name}.{suffix}'
        return Path(self.video_path.parent, summary_name)

И, наконец, добавьте Анимированные Дополнительная переменная среды в точке входа:

...
+ANIMATED = os.getenv("ANIMATED", "0") == "1"

def gcf_generate_summary(data, context):
    ...
-   VideoProcessor.generate_summary(annot_uri, SUMMARY_BUCKET)
+   VideoProcessor.generate_summary(annot_uri, SUMMARY_BUCKET, ANIMATED)

if __name__ == '__main__':
    ...
-   VideoProcessor.generate_summary(args.annot_uri, SUMMARY_BUCKET)
+   VideoProcessor.generate_summary(args.annot_uri, SUMMARY_BUCKET, ANIMATED)

Развертывание функций

Дублируйте 2-й функцию с дополнительными Анимированные Переменная среды:

GCF_NAME="gcf2_generate_summary_animated"
GCF_SOURCE="$PROJECT_SRC/gcf2_generate_summary"
GCF_ENTRY_POINT="gcf_generate_summary"
GCF_TRIGGER_BUCKET="$ANNOTATION_BUCKET"
GCF_ENV_VARS1="SUMMARY_BUCKET=$SUMMARY_BUCKET"
GCF_ENV_VARS2="ANIMATED=1"
GCF_TIMEOUT="540s"
GCF_MEMORY="2048MB"

gcloud functions deploy $GCF_NAME \
  --runtime python39 \
  --source $GCF_SOURCE \
  --entry-point $GCF_ENTRY_POINT \
  --update-env-vars $GCF_ENV_VARS1 \
  --update-env-vars $GCF_ENV_VARS2 \
  --trigger-bucket $GCF_TRIGGER_BUCKET \
  --region $GCF_REGION \
  --timeout $GCF_TIMEOUT \
  --memory $GCF_MEMORY \
  --quiet

Вот как это выглядит в Облачная консоль :

🎉 Окончательные тесты

Конечная точка HTTP позволяет вам вызвать трубопровод с помощью запроса GET:

GCF_NAME="gcf1_detect_shots_http"
VIDEO_URI="gs://cloud-samples-data/video/visionapi.mp4"
GCF_URL="https://$GCF_REGION-$PROJECT_ID.cloudfunctions.net/$GCF_NAME?video_uri=$VIDEO_URI"

curl $GCF_URL -H "Authorization: bearer $(gcloud auth print-identity-token)"

Launched shot detection for

Примечание: тестовое видео расположен во внешнем ведре, но является общедоступным.

Кроме того, скопируйте один или несколько видео в видеокод. Вы можете перетаскивать видео:

Затем видео обработаны параллельно. Вот несколько журналов:

LEVEL NAME                           EXECUTION_ID ... LOG
...
D     gcf2_generate_summary_animated f6n6tslsfwdu ... Function execution took 49293 ms, finished with status: 'ok'
I     gcf2_generate_summary          yd1vqabafn17 ... Uploading -> b1-videos.../JaneGoodall.mp4.summary035_still.png
I     gcf2_generate_summary_animated qv9b03814jjk ... shot_ratio: 43%
I     gcf2_generate_summary          yd1vqabafn17 ... Uploading -> b1-videos.../JaneGoodall.mp4.summary035_still.webp
D     gcf2_generate_summary          yd1vqabafn17 ... Function execution took 54616 ms, finished with status: 'ok'
I     gcf2_generate_summary_animated g4d2wrzxz2st ... shot_ratio: 71%
...
D     gcf2_generate_summary          amwmov1wk0gn ... Function execution took 65256 ms, finished with status: 'ok'
I     gcf2_generate_summary_animated 7pp882fz0x84 ... shot_ratio: 57%
I     gcf2_generate_summary_animated i3u830hsjz4r ... Uploading -> b1-videos.../JaneGoodall.mp4.summary035_anim.png
I     gcf2_generate_summary_animated i3u830hsjz4r ... Uploading -> b1-videos.../JaneGoodall.mp4.summary035_anim.webp
D     gcf2_generate_summary_animated i3u830hsjz4r ... Function execution took 70862 ms, finished with status: 'ok'
...

В 3-м ведре вы найдете все еще и анимированные резюме:

Вы уже видели еще резюме для Как введение в это руководство. В анимированной версии и только в 6 кадрах вы получаете еще лучшее представление о том, что Все видео около:

Если вы не хотите держать свой проект, вы можете удалить его:

gcloud projects delete $PROJECT_ID

➕ Еще одна вещь

Насколько большой код кодовой базы?

first_line_after_licence=16
find $PROJECT_SRC -name '*.py' -exec tail -n +$first_line_after_licence {} \; | grep -v "^$" | wc -l

Количество линий Python:

Видео Анализ и обработка, с различными вариантами, работают менее чем за 300 строк читаемого Python.
Меньше линии, меньше жуков!
🔥🐍 Миссия выполнена! 🐍🔥.

🖖 Увидимся

Я надеюсь, что вы оценили это руководство и хотели бы читать ваш отзыв . Вы также можете Следуй за мной в Twitter Отказ

⏳ обновления

2021-10-08: Обновлено последними версиями библиотеки + Python 3.7 → 3.9

Оригинал: “https://dev.to/googlecloud/auto-generate-video-summaries-with-a-machine-learning-model-and-a-serverless-pipeline-324i”

👋 Здравствуйте!

🔭 Целью

🛠️ инструменты

🧱 архитектура

🐍 Библиотеки Python

⚙️. Настройка проекта

Переменные среды

Новый проект

Биллинг Счет

Ведра

Сервисная учетная запись

Апис

Исходный код

🧠 Видео анализ

Обнаружение видео выстрел

Местное развитие и тесты

Точка входа в функцию

Развертывание функций

Производственные испытания

🎞️ Визуальное резюме

Структура кода

Классный хранилище

Видеопроцессор класса

Местное развитие и тесты

Точка входа в функцию

Развертывание функций

Производственные испытания

Полученные результаты

🍒 вишня на PY 🐍

Архитектура (V2)

HTTP входная точка

Развертывание функций

Поддержка анимации

Развертывание функций

🎉 Окончательные тесты

➕ Еще одна вещь

🖖 Увидимся

⏳ обновления

Читайте ещё по теме: