def classify_directory_using_build_fingerprint()

in ioXt/uraniborg/scripts/python/automate_observation.py [0:0]

• 61 lines of code
• 9 McCabe index (conditional complexity)

def classify_directory_using_build_fingerprint(adb_wrapper,
                                               source,
                                               results_dir,
                                               logger):
  """Decides which directory in results/ to dump new result to.

  This is a renewed method that makes use of build fingerprint to do
  result classification. This method allows for better usability when a user
  is obtaining many results for the same device or build.

  Args:
    adb_wrapper: An AdbWrapper instance.
    source: the source directory (on target device) containing new results.
    results_dir: the umbrella results/ directory.
    logger: A logger object to log debug or error messages.

  Returns:
    A string representing the final directory (on host) where results are pulled
    to. <code>None</code> is returned if any failure is encountered along the
    way.
  """
  # need to grab the build.txt to a tmp location
  tmp_file = "/tmp/device_build.txt"
  adb_pull_failed = False
  if not adb_wrapper.pull("{}/build.txt".format(source), tmp_file):
    logger.error("Failed to pull build.txt from device results dir.")
    adb_pull_failed = True

  if adb_pull_failed:
    logger.debug("Attempting to grab files using adb backup instead...")
    compressed_backup_filepath = "/tmp/hubble_results.ab"
    decompressed_backup_filepath = "/tmp/hubble_results.tar"
    logger.warning("Manual intervention required: Please select "
                   "`Back up my data` to proceed")
    if not adb_wrapper.backup(compressed_backup_filepath, HUBBLE_PACKAGE_NAME):
      logger.error("Failed to use `adb backup` to pull result files.")
      return None

    # now we have to decompress the backup file
    logger.debug("Reading from %s into buffer...", compressed_backup_filepath)
    compressed_backup_filecontent = ""
    with open(compressed_backup_filepath, "rb") as f_in:
      compressed_backup_filecontent = f_in.read()

    # we'll need to skip 24 bytes to skip the Android backup header
    logger.debug("Decompressing backup file...")
    decompressed_content = zlib.decompress(compressed_backup_filecontent[24:])
    with open(decompressed_backup_filepath, "wb") as f_out:
      f_out.write(decompressed_content)

    tarfile_obj = io.BytesIO(decompressed_content)
    tar_obj = tarfile.open(fileobj=tarfile_obj)
    base_result_path = "/tmp/untarred_hubble_results"
    logger.debug("Extracting result files into %s", base_result_path)
    tar_obj.extractall(base_result_path)

    # we overwrite tmp_file to reuse existing logic
    tmp_file = os.path.join(base_result_path, "apps", HUBBLE_PACKAGE_NAME,
                            "ef", "results", "build.txt")
    logger.debug("tmp_file: %s", tmp_file)

  build_info = ""
  with open(tmp_file, "r") as f_in:
    build_info = f_in.read()
  build_json = json.loads(build_info)
  build_fingerprint = build_json["buildInfo"][0]["fingerprint"]
  target_dir_parent = os.path.join(results_dir, build_fingerprint)

  if not os.path.exists(target_dir_parent):
    logger.debug("{} does not exist yet. Creating...".format(target_dir_parent))
    os.makedirs(target_dir_parent)

  target_dir = ""
  for i in range(1000):
    target_dir = os.path.join(target_dir_parent, "{0:03d}".format(i))
    logger.debug("Testing {} as target directory.".format(target_dir))
    if not os.path.exists(target_dir):
      logger.debug("{} does not exist yet! Using it!".format(target_dir))
      break

  if adb_pull_failed:
    source_dir = os.path.join("/tmp/untarred_hubble_results/apps",
                              HUBBLE_PACKAGE_NAME,
                              "ef",
                              "results")
    shutil.move(source_dir, target_dir)
    return target_dir
  else:
    if adb_wrapper.pull(source, target_dir):
      return target_dir
  return None