1
    2
    3
    4
    5
    6
    7
    8
    9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59

fuchsia_web / av_testing / camera.py [blame]

# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
""" Starts an av_sync_record process to record the video from the camera. """

import logging
import os.path
import subprocess

from dataclasses import dataclass
from typing import Optional


@dataclass
class Parameters:
    fps: Optional[int] = None
    output_path: str = '/tmp'
    file: str = 'video'
    duration_sec: Optional[int] = None
    max_frames: Optional[int] = None
    serial_number: Optional[str] = None

    @property
    def video_file(self):
        assert self.output_path
        assert self.file
        return os.path.join(self.output_path, f'{self.file}.mkv')

    @property
    def info_file(self):
        assert self.output_path
        assert self.file
        return os.path.join(self.output_path, f'{self.file}.info.csv')


def start(parameters: Parameters) -> None:
    """Starts an av_sync_record process to record the video from the camera.
    Executing of this function shouldn't be terminated as it would create a
    bad constructed video file."""
    assert parameters.output_path
    assert parameters.file
    BINARY = '/usr/local/cipd/av_sync_record/av_sync_record'
    assert os.path.isfile(BINARY)
    args = [
        BINARY, '--gid=', '--uid=',
        f'--camera_info_path={parameters.info_file}',
        f'--video_output={parameters.video_file}'
    ]
    if parameters.fps:
        args.append(f'--camera_fps={parameters.fps}')
    if parameters.duration_sec:
        args.append(f'--duration={parameters.duration_sec}s')
    if parameters.max_frames:
        args.append(f'--max_frames={parameters.max_frames}')
    if parameters.serial_number:
        args.append(f'--serial_number={parameters.serial_number}')
    logging.warning('Camera starts recording to %s', parameters.output_path)
    subprocess.run(args, check=True)
    logging.warning('Camera finishes recording.')