parameters: (
      filepath: str,
      progress_callback: Callable[[float], None],
      acceptable_files: Sequence[str],
      mandatory_files: Sequence[str],
      chunk_size: int = 1024,
  )
return_statement: return file_paths, file_sizes
docstring: Unzip an update file
  
      The update file must contain
  
      - a file called rootfs.ext4
      - a file called rootfs.ext4.hash
  
      It may contain
  
      - a file called rootfs.ext4.hash.sig
  
      These will all be unzipped (discarding their leading directories) to
      the same file as the zipfile.
  
      This function is blocking and takes a while. It calls ``progress_callback``
      to indicate update progress with a number between 0 and 1 indicating
      overall archive unzip progress.
  
      :param filepath: The path zipfile to unzip. The contents will be in its
                       directory
      :param progress_callback: A callable taking a number between 0 and 1 that
                                will be called periodically to check progress.
                                This is for user display; it may not reach 1.0
                                exactly.
      :param acceptable_files: A list of files to unzip if found. Others will be
                               ignored.
      :param mandatory_files: A list of files to raise an error about if they're
                              not in the zip. Should probably be a subset of
                              ``acceptable_files``.
      :param chunk_size: If specified, the size of the chunk to read and write.
                         If not specified, will default to 1024
      :return: Two dictionaries, the first mapping file names to paths and the
               second mapping file names to sizes
  
      :raises FileMissing: If a mandatory file is missing
docstring_summary: Unzip an update file
docstring_tokens: Unzip an update file
function: def unzip_update(
      filepath: str,
      progress_callback: Callable[[float], None],
      acceptable_files: Sequence[str],
      mandatory_files: Sequence[str],
      chunk_size: int = 1024,
  ) -> Tuple[Mapping[str, Optional[str]], Mapping[str, int]]:
      """Unzip an update file
  
      The update file must contain
  
      - a file called rootfs.ext4
      - a file called rootfs.ext4.hash
  
      It may contain
  
      - a file called rootfs.ext4.hash.sig
  
      These will all be unzipped (discarding their leading directories) to
      the same file as the zipfile.
  
      This function is blocking and takes a while. It calls ``progress_callback``
      to indicate update progress with a number between 0 and 1 indicating
      overall archive unzip progress.
  
      :param filepath: The path zipfile to unzip. The contents will be in its
                       directory
      :param progress_callback: A callable taking a number between 0 and 1 that
                                will be called periodically to check progress.
                                This is for user display; it may not reach 1.0
                                exactly.
      :param acceptable_files: A list of files to unzip if found. Others will be
                               ignored.
      :param mandatory_files: A list of files to raise an error about if they're
                              not in the zip. Should probably be a subset of
                              ``acceptable_files``.
      :param chunk_size: If specified, the size of the chunk to read and write.
                         If not specified, will default to 1024
      :return: Two dictionaries, the first mapping file names to paths and the
               second mapping file names to sizes
  
      :raises FileMissing: If a mandatory file is missing
      """
      assert chunk_size
      total_size = 0
      written_size = 0
      to_unzip: List[zipfile.ZipInfo] = []
      file_paths: Dict[str, Optional[str]] = {fn: None for fn in acceptable_files}
      file_sizes: Dict[str, int] = {fn: 0 for fn in acceptable_files}
      LOG.info(f"Unzipping {filepath}")
      with zipfile.ZipFile(filepath, "r") as zf:
          files = zf.infolist()
          remaining_filenames = [fn for fn in acceptable_files]
          for fi in files:
              if fi.filename in acceptable_files:
                  to_unzip.append(fi)
                  total_size += fi.file_size
                  remaining_filenames.remove(fi.filename)
                  LOG.debug(f"Found {fi.filename} ({fi.file_size}B)")
              else:
                  LOG.debug(f"Ignoring {fi.filename}")
  
          for name in remaining_filenames:
              if name in mandatory_files:
                  raise FileMissing(f"File {name} missing from zip")
  
          for fi in to_unzip:
              uncomp_path = os.path.join(os.path.dirname(filepath), fi.filename)
              with zf.open(fi) as zipped, open(uncomp_path, "wb") as unzipped:
                  LOG.debug(f"Beginning unzip of {fi.filename} to {uncomp_path}")
                  while True:
                      chunk = zipped.read(chunk_size)
                      unzipped.write(chunk)
                      written_size += len(chunk)
                      progress_callback(written_size / total_size)
                      if len(chunk) != chunk_size:
                          break
                  file_paths[fi.filename] = uncomp_path
                  file_sizes[fi.filename] = fi.file_size
                  LOG.debug(f"Unzipped {fi.filename} to {uncomp_path}")
      LOG.info(
          f"Unzipped {filepath}, results: \n\t"
          + "\n\t".join(
              [f"{k}: {file_paths[k]} ({file_sizes[k]}B)" for k in file_paths.keys()]
          )
      )
      return file_paths, file_sizes
function_tokens: def unzip_update ( filepath : str , progress_callback : Callable [ [ float ] , None ] , acceptable_files : Sequence [ str ] , mandatory_files : Sequence [ str ] , chunk_size : int = 1024 , ) -> Tuple [ Mapping [ str , Optional [ str ] ] , Mapping [ str , int ] ] : assert chunk_size total_size = 0 written_size = 0 to_unzip : List [ zipfile . ZipInfo ] = [ ] file_paths : Dict [ str , Optional [ str ] ] = { fn : None for fn in acceptable_files } file_sizes : Dict [ str , int ] = { fn : 0 for fn in acceptable_files } LOG . info ( f"Unzipping {filepath}" ) with zipfile . ZipFile ( filepath , "r" ) as zf : files = zf . infolist ( ) remaining_filenames = [ fn for fn in acceptable_files ] for fi in files : if fi . filename in acceptable_files : to_unzip . append ( fi ) total_size += fi . file_size remaining_filenames . remove ( fi . filename ) LOG . debug ( f"Found {fi.filename} ({fi.file_size}B)" ) else : LOG . debug ( f"Ignoring {fi.filename}" ) for name in remaining_filenames : if name in mandatory_files : raise FileMissing ( f"File {name} missing from zip" ) for fi in to_unzip : uncomp_path = os . path . join ( os . path . dirname ( filepath ) , fi . filename ) with zf . open ( fi ) as zipped , open ( uncomp_path , "wb" ) as unzipped : LOG . debug ( f"Beginning unzip of {fi.filename} to {uncomp_path}" ) while True : chunk = zipped . read ( chunk_size ) unzipped . write ( chunk ) written_size += len ( chunk ) progress_callback ( written_size / total_size ) if len ( chunk ) != chunk_size : break file_paths [ fi . filename ] = uncomp_path file_sizes [ fi . filename ] = fi . file_size LOG . debug ( f"Unzipped {fi.filename} to {uncomp_path}" ) LOG . info ( f"Unzipped {filepath}, results: \n\t" + "\n\t" . join ( [ f"{k}: {file_paths[k]} ({file_sizes[k]}B)" for k in file_paths . keys ( ) ] ) ) return file_paths , file_sizes