fuchsia_web/av_testing/av_sync_tests.py - chromium/src - Git at Google

 #!/usr/bin/env vpython3
 # 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.
 """ Executes Audio / Video performance tests against a smart display device.
     This script needs to be executed from the build output folder, e.g.
     out/fuchsia/."""

 import logging
 import multiprocessing
 import os
 import shutil
 import subprocess
 import sys
 import time

 from contextlib import AbstractContextManager
 from pathlib import Path

 import camera
 import server
 import video_analyzer

 TEST_SCRIPTS_ROOT = os.path.join(os.path.dirname(__file__), '..', '..',
                                  'build', 'fuchsia', 'test')
 sys.path.append(TEST_SCRIPTS_ROOT)

 import monitors
 import version
 from chrome_driver_wrapper import ChromeDriverWrapper
 from common import get_build_info, get_ffx_isolate_dir, get_free_local_port
 from isolate_daemon import IsolateDaemon
 from run_webpage_test import capture_devtools_addr


 HTTP_SERVER_PORT = get_free_local_port()
 LOG_DIR = os.environ.get('ISOLATED_OUTDIR', '/tmp')
 TEMP_DIR = os.environ.get('TMPDIR', '/tmp')

 VIDEOS =[
     '720p24fpsH264_gangnam_sync.mp4',
     '720p24fpsVP9_gangnam_sync.webm',
 ]


 class StartProcess(AbstractContextManager):
     """Starts a multiprocessing.Process."""

     def __init__(self, target, args, terminate: bool):
         self._proc = multiprocessing.Process(target=target, args=args)
         self._terminate = terminate

     def __enter__(self):
         self._proc.start()

     def __exit__(self, exc_type, exc_value, traceback):
         if self._terminate:
             self._proc.terminate()
         self._proc.join()
         if not self._terminate:
             assert self._proc.exitcode == 0


 def parameters_of(file: str) -> camera.Parameters:
     result = camera.Parameters()
     result.file = file
     # Recorded videos are huge, instead of placing them into the LOG_DIR
     # which will be uploaded to CAS output, use TEMP_DIR provided by
     # luci-swarming to be cleaned up automatically after the test run.
     result.output_path = TEMP_DIR
     # max_frames controls the maximum number of umcompressed frames in the
     # memory. And if the number of uncompressed frames reaches the max_frames,
     # the basler camera recorder will fail. The camera being used may use up to
     # 388,800 bytes per frame, or ~467MB for 1200 frames, setting the max_frames
     # to 1200 would avoid OOM. Also limit the fps to 120 to ensure the processor
     # of the host machine is capable to compress the camera stream on time
     # without exhausting the in-memory frame queue.
     # TODO(crbug.com/40935291): These two values need to be adjusted to reach
     # 300fps for a more accurate analysis result when the test is running on
     # more performant host machines.
     result.max_frames = 1200
     result.fps = 120
     # All the videos now being used are 30s long.
     result.duration_sec = 30
     return result


 def run_video_perf_test(file: str, driver: ChromeDriverWrapper,
                         host: str) -> None:
     driver.get(f'http://{host}:{HTTP_SERVER_PORT}/video.html?file={file}')
     camera_params = parameters_of(file)
     original_video = os.path.join(server.VIDEO_DIR, file)
     # Ensure the original video won't be overwritten.
     assert camera_params.video_file != original_video
     with StartProcess(camera.start, [camera_params], False):
         video = driver.find_element_by_id('video')
         video.click()
     # Video playback should finish almost within the same time as the camera
     # recording, and this check is only necessary to indicate a very heavy
     # network laggy and buffering.
     # TODO(crbug.com/40935291): May need to adjust the strategy here, the
     # final frame / barcode is considered laggy and drops the score.
     with monitors.time_consumption(file, 'video_perf', 'playback', 'laggy'):
         while not driver.execute_script('return arguments[0].ended;', video):
             time.sleep(1)
     logging.warning('Video %s finished', file)

     results = video_analyzer.from_original_video(camera_params.video_file,
                                                  original_video)

     def record(key: str) -> None:
         # If the video_analyzer does not generate any result, treat it as an
         # error and use the default value to filter them out instead of failing
         # the tests.
         # TODO(crbug.com/40935291): Revise the default value for errors.
         monitors.average(file, key).record(results.get(key, -128))

     record('smoothness')
     record('freezing')
     record('dropped_frame_count')
     record('total_frame_count')
     record('dropped_frame_percentage')
     logging.warning('Video analysis result of %s: %s', file, results)

     # Move the info csv to the cas-output for debugging purpose. Video files
     # are huge and will be ignored.
     shutil.move(camera_params.info_file, LOG_DIR)


 def run_test(proc: subprocess.Popen) -> None:
     device, port = capture_devtools_addr(proc, LOG_DIR)
     logging.warning('DevTools is now running on %s:%s', device, port)
     # webpage test may update the fuchsia version, so get build_info after its
     # finish.
     logging.warning('Chrome version %s %s', version.chrome_version_str(),
                     version.git_revision())
     build_info = get_build_info()
     logging.warning('Fuchsia build info %s', build_info)
     monitors.tag(version.chrome_version_str(), build_info.version,
                  version.chrome_version_str() + '/' + build_info.version)
     # Replace the last byte to 1, by default it's the ip address of the host
     # machine being accessible on the device.
     host = '.'.join(device.split('.')[:-1] + ['1'])
     proxy_host = os.environ.get('GCS_PROXY_HOST')
     if proxy_host:
         # This is a hacky way to get the ip address of the host machine
         # being accessible on the device by the fuchsia managed docker image.
         host = proxy_host + '0'
     with ChromeDriverWrapper((device, port)) as driver:
         for file in VIDEOS:
             run_video_perf_test(file, driver, host)


 def main() -> int:
     proc = subprocess.Popen([
         os.path.join(TEST_SCRIPTS_ROOT,
                      'run_test.py'), 'webpage', '--out-dir=.',
         '--browser=web-engine-shell', '--device', f'--logs-dir={LOG_DIR}'
     ],
                             env={
                                 **os.environ, 'CHROME_HEADLESS': '1'
                             })
     try:
         run_test(proc)
         return 0
     except:
         # Do not dump the results unless the tests were passed successfully to
         # avoid polluting the metrics.
         monitors.clear()
         raise
     finally:
         proc.terminate()
         proc.wait()
         # May need to merge with the existing file created by run_test.py
         # webpage process.
         monitors.dump(os.path.join(LOG_DIR, 'invocations'))


 if __name__ == '__main__':
     logging.warning('Running %s with env %s', sys.argv, os.environ)
     # Setting a temporary isolate daemon dir and share it with the webpage
     # runner.
     with StartProcess(server.start, [HTTP_SERVER_PORT], True), \
          IsolateDaemon.IsolateDir():
         logging.warning('ffx daemon is running in %s', get_ffx_isolate_dir())
         sys.exit(main())
	#!/usr/bin/env vpython3
	# 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.
	""" Executes Audio / Video performance tests against a smart display device.
	This script needs to be executed from the build output folder, e.g.
	out/fuchsia/."""

	import logging
	import multiprocessing
	import os
	import shutil
	import subprocess
	import sys
	import time

	from contextlib import AbstractContextManager
	from pathlib import Path

	import camera
	import server
	import video_analyzer

	TEST_SCRIPTS_ROOT = os.path.join(os.path.dirname(__file__), '..', '..',
	'build', 'fuchsia', 'test')
	sys.path.append(TEST_SCRIPTS_ROOT)

	import monitors
	import version
	from chrome_driver_wrapper import ChromeDriverWrapper
	from common import get_build_info, get_ffx_isolate_dir, get_free_local_port
	from isolate_daemon import IsolateDaemon
	from run_webpage_test import capture_devtools_addr


	HTTP_SERVER_PORT = get_free_local_port()
	LOG_DIR = os.environ.get('ISOLATED_OUTDIR', '/tmp')
	TEMP_DIR = os.environ.get('TMPDIR', '/tmp')

	VIDEOS =[
	'720p24fpsH264_gangnam_sync.mp4',
	'720p24fpsVP9_gangnam_sync.webm',
	]


	class StartProcess(AbstractContextManager):
	"""Starts a multiprocessing.Process."""

	def __init__(self, target, args, terminate: bool):
	self._proc = multiprocessing.Process(target=target, args=args)
	self._terminate = terminate

	def __enter__(self):
	self._proc.start()

	def __exit__(self, exc_type, exc_value, traceback):
	if self._terminate:
	self._proc.terminate()
	self._proc.join()
	if not self._terminate:
	assert self._proc.exitcode == 0


	def parameters_of(file: str) -> camera.Parameters:
	result = camera.Parameters()
	result.file = file
	# Recorded videos are huge, instead of placing them into the LOG_DIR
	# which will be uploaded to CAS output, use TEMP_DIR provided by
	# luci-swarming to be cleaned up automatically after the test run.
	result.output_path = TEMP_DIR
	# max_frames controls the maximum number of umcompressed frames in the
	# memory. And if the number of uncompressed frames reaches the max_frames,
	# the basler camera recorder will fail. The camera being used may use up to
	# 388,800 bytes per frame, or ~467MB for 1200 frames, setting the max_frames
	# to 1200 would avoid OOM. Also limit the fps to 120 to ensure the processor
	# of the host machine is capable to compress the camera stream on time
	# without exhausting the in-memory frame queue.
	# TODO(crbug.com/40935291): These two values need to be adjusted to reach
	# 300fps for a more accurate analysis result when the test is running on
	# more performant host machines.
	result.max_frames = 1200
	result.fps = 120
	# All the videos now being used are 30s long.
	result.duration_sec = 30
	return result


	def run_video_perf_test(file: str, driver: ChromeDriverWrapper,
	host: str) -> None:
	driver.get(f'http://{host}:{HTTP_SERVER_PORT}/video.html?file={file}')
	camera_params = parameters_of(file)
	original_video = os.path.join(server.VIDEO_DIR, file)
	# Ensure the original video won't be overwritten.
	assert camera_params.video_file != original_video
	with StartProcess(camera.start, [camera_params], False):
	video = driver.find_element_by_id('video')
	video.click()
	# Video playback should finish almost within the same time as the camera
	# recording, and this check is only necessary to indicate a very heavy
	# network laggy and buffering.
	# TODO(crbug.com/40935291): May need to adjust the strategy here, the
	# final frame / barcode is considered laggy and drops the score.
	with monitors.time_consumption(file, 'video_perf', 'playback', 'laggy'):
	while not driver.execute_script('return arguments[0].ended;', video):
	time.sleep(1)
	logging.warning('Video %s finished', file)

	results = video_analyzer.from_original_video(camera_params.video_file,
	original_video)

	def record(key: str) -> None:
	# If the video_analyzer does not generate any result, treat it as an
	# error and use the default value to filter them out instead of failing
	# the tests.
	# TODO(crbug.com/40935291): Revise the default value for errors.
	monitors.average(file, key).record(results.get(key, -128))

	record('smoothness')
	record('freezing')
	record('dropped_frame_count')
	record('total_frame_count')
	record('dropped_frame_percentage')
	logging.warning('Video analysis result of %s: %s', file, results)

	# Move the info csv to the cas-output for debugging purpose. Video files
	# are huge and will be ignored.
	shutil.move(camera_params.info_file, LOG_DIR)


	def run_test(proc: subprocess.Popen) -> None:
	device, port = capture_devtools_addr(proc, LOG_DIR)
	logging.warning('DevTools is now running on %s:%s', device, port)
	# webpage test may update the fuchsia version, so get build_info after its
	# finish.
	logging.warning('Chrome version %s %s', version.chrome_version_str(),
	version.git_revision())
	build_info = get_build_info()
	logging.warning('Fuchsia build info %s', build_info)
	monitors.tag(version.chrome_version_str(), build_info.version,
	version.chrome_version_str() + '/' + build_info.version)
	# Replace the last byte to 1, by default it's the ip address of the host
	# machine being accessible on the device.
	host = '.'.join(device.split('.')[:-1] + ['1'])
	proxy_host = os.environ.get('GCS_PROXY_HOST')
	if proxy_host:
	# This is a hacky way to get the ip address of the host machine
	# being accessible on the device by the fuchsia managed docker image.
	host = proxy_host + '0'
	with ChromeDriverWrapper((device, port)) as driver:
	for file in VIDEOS:
	run_video_perf_test(file, driver, host)


	def main() -> int:
	proc = subprocess.Popen([
	os.path.join(TEST_SCRIPTS_ROOT,
	'run_test.py'), 'webpage', '--out-dir=.',
	'--browser=web-engine-shell', '--device', f'--logs-dir={LOG_DIR}'
	],
	env={
	**os.environ, 'CHROME_HEADLESS': '1'
	})
	try:
	run_test(proc)
	return 0
	except:
	# Do not dump the results unless the tests were passed successfully to
	# avoid polluting the metrics.
	monitors.clear()
	raise
	finally:
	proc.terminate()
	proc.wait()
	# May need to merge with the existing file created by run_test.py
	# webpage process.
	monitors.dump(os.path.join(LOG_DIR, 'invocations'))


	if __name__ == '__main__':
	logging.warning('Running %s with env %s', sys.argv, os.environ)
	# Setting a temporary isolate daemon dir and share it with the webpage
	# runner.
	with StartProcess(server.start, [HTTP_SERVER_PORT], True), \
	IsolateDaemon.IsolateDir():
	logging.warning('ffx daemon is running in %s', get_ffx_isolate_dir())
	sys.exit(main())