[Telemetry] Migrate bitmap.py from bitmaptools.cc to numpy

content/test/gpu/gpu_tests/cloud_storage_test_base.py

 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Base classes for a test and validator which upload results
 (reference images, error images) to cloud storage."""
 
+import cv2
 import os
 import re
 import tempfile
 
 from telemetry import benchmark
 from telemetry.core import bitmap
 from telemetry.page import page_test
 from telemetry.util import cloud_storage
 
 
 test_data_dir = os.path.abspath(os.path.join(
     os.path.dirname(__file__), '..', '..', 'data', 'gpu'))
 
 default_generated_data_dir = os.path.join(test_data_dir, 'generated')
 
 error_image_cloud_storage_bucket = 'chromium-browser-gpu-tests'
 
 def _CompareScreenshotSamples(screenshot, expectations, device_pixel_ratio):
   for expectation in expectations:
     location = expectation["location"]
     x = int(location[0] * device_pixel_ratio)
     y = int(location[1] * device_pixel_ratio)
 
     if x < 0 or y < 0 or x > screenshot.width or y > screenshot.height:
       raise page_test.Failure(
           'Expected pixel location [%d, %d] is out of range on [%d, %d] image' %
           (x, y, screenshot.width, screenshot.height))
 
-    actual_color = screenshot.GetPixelColor(x, y)
-    expected_color = bitmap.RgbaColor(
-        expectation["color"][0],
-        expectation["color"][1],
-        expectation["color"][2])
-    if not actual_color.IsEqual(expected_color, expectation["tolerance"]):
-      raise page_test.Failure('Expected pixel at ' + str(location) +
-          ' to be ' +
-          str(expectation["color"]) + " but got [" +
-          str(actual_color.r) + ", " +
-          str(actual_color.g) + ", " +
-          str(actual_color.b) + "]")
+    actual_color = screenshot.image[y][x]
+    expected_color = (expectation["color"][2], expectation["color"][1],
+                      expectation["color"][0])
+    if not bitmap.ColorsAreEqual(actual_color, expected_color,
+                                 expectation["tolerance"]):
+      raise page_test.Failure(
+          'Expected pixel at %s to be %s but got [%s, %s, %s]' %
+          (location, expectation["color"], actual_color[2], actual_color[1],
+           actual_color[0]))
 
 class ValidatorBase(page_test.PageTest):
   def __init__(self):
     super(ValidatorBase, self).__init__()
     # Parameters for cloud storage reference images.
     self.vendor_id = None
     self.device_id = None
     self.vendor_string = None
     self.device_string = None
     self.msaa = False
 
   ###
   ### Routines working with the local disk (only used for local
   ### testing without a cloud storage account -- the bots do not use
   ### this code path).
   ###
 
   def _UrlToImageName(self, url):
     image_name = re.sub(r'^(http|https|file)://(/*)', '', url)
     image_name = re.sub(r'\.\./', '', image_name)
     image_name = re.sub(r'(\.|/|-)', '_', image_name)
     return image_name
 
   def _WriteImage(self, image_path, png_image):
     output_dir = os.path.dirname(image_path)
     if not os.path.exists(output_dir):
       os.makedirs(output_dir)
-    png_image.WritePngFile(image_path)
+    cv2.imwrite(image_path, png_image.image)
 
   def _WriteErrorImages(self, img_dir, img_name, screenshot, ref_png):
     full_image_name = img_name + '_' + str(self.options.build_revision)
     full_image_name = full_image_name + '.png'
 
     # Always write the failing image.
     self._WriteImage(
         os.path.join(img_dir, 'FAIL_' + full_image_name), screenshot)
 
     if ref_png:
@@ skipping 78 matching lines @@
     """Downloads the reference image for the given test from cloud
     storage, returning it as a Telemetry Bitmap object."""
     # TODO(kbr): there's a race condition between the deletion of the
     # temporary file and gsutil's overwriting it.
     if not self.options.refimg_cloud_storage_bucket:
       raise Exception('--refimg-cloud-storage-bucket argument is required')
     temp_file = tempfile.NamedTemporaryFile().name
     cloud_storage.Get(self.options.refimg_cloud_storage_bucket,
                       self._FormatReferenceImageName(img_name, page, tab),
                       temp_file)
-    return bitmap.Bitmap.FromPngFile(temp_file)
+    return bitmap.Bitmap.FromImageFile(temp_file)
 
   def _UploadErrorImagesToCloudStorage(self, image_name, screenshot, ref_img):
     """For a failing run, uploads the failing image, reference image (if
     supplied), and diff image (if reference image was supplied) to cloud
     storage. This subsumes the functionality of the
     archive_gpu_pixel_test_results.py script."""
     machine_name = re.sub('\W+', '_', self.options.test_machine_name)
     upload_dir = '%s_%s_telemetry' % (self.options.build_revision, machine_name)
     base_bucket = '%s/runs/%s' % (error_image_cloud_storage_bucket, upload_dir)
     image_name_with_revision = '%s_%s.png' % (
@@ skipping 59 matching lines @@
         default='')
     group.add_option('--test-machine-name',
         help='Name of the test machine. Specifying this argument causes this '
         'script to upload failure images and diffs to cloud storage directly, '
         'instead of relying on the archive_gpu_pixel_test_results.py script.',
         default='')
     group.add_option('--generated-dir',
         help='Overrides the default on-disk location for generated test images '
         '(only used for local testing without a cloud storage account)',
         default=default_generated_data_dir)