[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 os
 import re
 import tempfile
 
 from telemetry import benchmark
-from telemetry.core import bitmap
+from telemetry.image_processing import image_util
+from telemetry.image_processing import rgba_color
 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:
+    if (x < 0 or y < 0 or x > image_util.Width(screenshot) or
+        y > image_util.Height(screenshot)):
       raise page_test.Failure(
           'Expected pixel location [%d, %d] is out of range on [%d, %d] image' %
-          (x, y, screenshot.width, screenshot.height))
+          (x, y, image_util.Width(screenshot), image_util.Height(screenshot)))
 
-    actual_color = screenshot.GetPixelColor(x, y)
-    expected_color = bitmap.RgbaColor(
+    actual_color = image_util.GetPixelColor(screenshot, x, y)
+    expected_color = rgba_color.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) + "]")
@@ skipping 17 matching lines @@
   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)
+    image_util.WritePngFile(png_image, image_path)
 
   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:
+    if ref_png is not None:
       # Save the reference image.
       # This ensures that we get the right revision number.
       self._WriteImage(
           os.path.join(img_dir, full_image_name), ref_png)
 
       # Save the difference image.
-      diff_png = screenshot.Diff(ref_png)
+      diff_png = image_util.Diff(screenshot, ref_png)
       self._WriteImage(
           os.path.join(img_dir, 'DIFF_' + full_image_name), diff_png)
 
   ###
   ### Cloud storage code path -- the bots use this.
   ###
 
   def _ComputeGpuInfo(self, tab):
     if ((self.vendor_id and self.device_id) or
         (self.vendor_string and self.device_string)):
@@ skipping 32 matching lines @@
       img_name,
       page.revision,
       self._FormatGpuInfo(tab))
 
   def _UploadBitmapToCloudStorage(self, bucket, name, bitmap, public=False):
     # This sequence of steps works on all platforms to write a temporary
     # PNG to disk, following the pattern in bitmap_unittest.py. The key to
     # avoiding PermissionErrors seems to be to not actually try to write to
     # the temporary file object, but to re-open its name for all operations.
     temp_file = tempfile.NamedTemporaryFile().name
-    bitmap.WritePngFile(temp_file)
+    image_util.WritePngFile(bitmap, temp_file)
     cloud_storage.Insert(bucket, name, temp_file, publicly_readable=public)
 
   def _ConditionallyUploadToCloudStorage(self, img_name, page, tab, screenshot):
     """Uploads the screenshot to cloud storage as the reference image
     for this test, unless it already exists. Returns True if the
     upload was actually performed."""
     if not self.options.refimg_cloud_storage_bucket:
       raise Exception('--refimg-cloud-storage-bucket argument is required')
     cloud_name = self._FormatReferenceImageName(img_name, page, tab)
     if not cloud_storage.Exists(self.options.refimg_cloud_storage_bucket,
                                 cloud_name):
       self._UploadBitmapToCloudStorage(self.options.refimg_cloud_storage_bucket,
                                        cloud_name,
                                        screenshot)
       return True
     return False
 
   def _DownloadFromCloudStorage(self, img_name, page, tab):
     """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 image_util.FromPngFile(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' % (
       image_name, self.options.build_revision)
     self._UploadBitmapToCloudStorage(
       base_bucket + '/gen', image_name_with_revision, screenshot,
       public=True)
-    if ref_img:
+    if ref_img is not None:
       self._UploadBitmapToCloudStorage(
         base_bucket + '/ref', image_name_with_revision, ref_img, public=True)
-      diff_img = screenshot.Diff(ref_img)
+      diff_img = image_util.Diff(screenshot, ref_img)
       self._UploadBitmapToCloudStorage(
         base_bucket + '/diff', image_name_with_revision, diff_img,
         public=True)
     print ('See http://%s.commondatastorage.googleapis.com/'
            'view_test_results.html?%s for this run\'s test results') % (
       error_image_cloud_storage_bucket, upload_dir)
 
   def _ValidateScreenshotSamples(self, url,
                                  screenshot, expectations, device_pixel_ratio):
     """Samples the given screenshot and verifies pixel color values.
@@ skipping 41 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)