text: """
  Simple utils to save and load from disk.
  """
  import joblib
  import gzip
  import pickle
  import os
  import tempfile
  import tarfile
  import zipfile
  import logging
  from urllib.request import urlretrieve
  from typing import Any, Iterator, List, Optional, Tuple, Union, cast, IO
  
  import pandas as pd
  import numpy as np
  
  import deepchem as dc
  
  logger = logging.getLogger(__name__)
  
  
  def pad_array(x: np.ndarray,
                shape: Union[Tuple, int],
                fill: float = 0.0,
                both: bool = False) -> np.ndarray:
    """
    Pad an array with a fill value.
  
    Parameters
    ----------
    x: np.ndarray
      A numpy array.
    shape: Tuple or int
      Desired shape. If int, all dimensions are padded to that size.
    fill: float, optional (default 0.0)
      The padded value.
    both: bool, optional (default False)
      If True, split the padding on both sides of each axis. If False,
      padding is applied to the end of each axis.
  
    Returns
    -------
    np.ndarray
      A padded numpy array
    """
    x = np.asarray(x)
    if not isinstance(shape, tuple):
      shape = tuple(shape for _ in range(x.ndim))
    pad = []
    for i in range(x.ndim):
      diff = shape[i] - x.shape[i]
      assert diff >= 0
      if both:
        a, b = divmod(diff, 2)
        b += a
        pad.append((a, b))
      else:
        pad.append((0, diff))
    pad = tuple(pad)  # type: ignore
    x = np.pad(x, pad, mode='constant', constant_values=fill)
    return x
  
  
  def get_data_dir() -> str:
    """Get the DeepChem data directory.
  
    Returns
    -------
    str
      The default path to store DeepChem data. If you want to
      change this path, please set your own path to `DEEPCHEM_DATA_DIR`
      as an environment variable.
    """
    if 'DEEPCHEM_DATA_DIR' in os.environ:
      return os.environ['DEEPCHEM_DATA_DIR']
    return tempfile.gettempdir()
  
  
  def download_url(url: str,
                   dest_dir: str = get_data_dir(),
                   name: Optional[str] = None):
    """Download a file to disk.
  
    Parameters
    ----------
    url: str
      The URL to download from
    dest_dir: str
      The directory to save the file in
    name: str
      The file name to save it as.  If omitted, it will try to extract a file name from the URL
    """
    if name is None:
      name = url
      if '?' in name:
        name = name[:name.find('?')]
      if '/' in name:
        name = name[name.rfind('/') + 1:]
    urlretrieve(url, os.path.join(dest_dir, name))
  
  
  def untargz_file(file: str,
                   dest_dir: str = get_data_dir(),
                   name: Optional[str] = None):
    """Untar and unzip a .tar.gz file to disk.
  
    Parameters
    ----------
    file: str
      The filepath to decompress
    dest_dir: str
      The directory to save the file in
    name: str
      The file name to save it as.  If omitted, it will use the file name
    """
    if name is None:
      name = file
    tar = tarfile.open(name)
    tar.extractall(path=dest_dir)
    tar.close()
  
  
  def unzip_file(file: str,
                 dest_dir: str = get_data_dir(),
                 name: Optional[str] = None):
    """Unzip a .zip file to disk.
  
    Parameters
    ----------
    file: str
      The filepath to decompress
    dest_dir: str
      The directory to save the file in
    name: str
      The directory name to unzip it to.  If omitted, it will use the file name
    """
    if name is None:
      name = file
    if dest_dir is None:
      dest_dir = os.path.join(get_data_dir, name)
    with zipfile.ZipFile(file, "r") as zip_ref:
      zip_ref.extractall(dest_dir)
  
  
  def load_image_files(input_files: List[str]) -> np.ndarray:
    """Loads a set of images from disk.
  
    Parameters
    ----------
    input_files: List[str]
      List of image filenames.
  
    Returns
    -------
    np.ndarray
      A numpy array that contains loaded images. The shape is, `(N,...)`.
  
    Notes
    -----
    This method requires Pillow to be installed.
    The supported file types are PNG and TIF.
    """
    try:
      from PIL import Image
    except ModuleNotFoundError:
      raise ImportError("This function requires Pillow to be installed.")
  
    images = []
    for input_file in input_files:
      _, extension = os.path.splitext(input_file)
      extension = extension.lower()
      if extension == ".png":
        image = np.array(Image.open(input_file))
        images.append(image)
      elif extension == ".tif":
        im = Image.open(input_file)
        imarray = np.array(im)
        images.append(imarray)
      else:
        raise ValueError("Unsupported image filetype for %s" % input_file)
    return np.array(images)
  
  
  def load_sdf_files(input_files: List[str],
                     clean_mols: bool = True,
                     tasks: List[str] = [],
                     shard_size: Optional[int] = None) -> Iterator[pd.DataFrame]:
    """Load SDF file into dataframe.
  
    Parameters
    ----------
    input_files: List[str]
      List of filenames
    clean_mols: bool, default True
      Whether to sanitize molecules.
    tasks: List[str], default []
      Each entry in `tasks` is treated as a property in the SDF file and is
      retrieved with `mol.GetProp(str(task))` where `mol` is the RDKit mol
      loaded from a given SDF entry.
    shard_size: int, default None
      The shard size to yield at one time.
  
    Returns
    -------
    Iterator[pd.DataFrame]
      Generator which yields the dataframe which is the same shard size.
  
    Notes
    -----
    This function requires RDKit to be installed.
    """
    try:
      from rdkit import Chem
    except ModuleNotFoundError:
      raise ImportError("This function requires RDKit to be installed.")
  
    df_rows = []
    for input_file in input_files:
      # Tasks are either in .sdf.csv file or in the .sdf file itself
      has_csv = os.path.isfile(input_file + ".csv")
      # Structures are stored in .sdf file
      logger.info("Reading structures from %s." % input_file)
      suppl = Chem.SDMolSupplier(str(input_file), clean_mols, False, False)
      for ind, mol in enumerate(suppl):
        if mol is None:
          continue
        smiles = Chem.MolToSmiles(mol)
        df_row = [ind, smiles, mol]
        if not has_csv:  # Get task targets from .sdf file
          for task in tasks:
            df_row.append(mol.GetProp(str(task)))
        df_rows.append(df_row)
        if shard_size is not None and len(df_rows) == shard_size:
          if has_csv:
            mol_df = pd.DataFrame(df_rows, columns=('mol_id', 'smiles', 'mol'))
            raw_df = next(load_csv_files([input_file + ".csv"], shard_size=None))
            yield pd.concat([mol_df, raw_df], axis=1, join='inner')
          else:
            mol_df = pd.DataFrame(
                df_rows, columns=('mol_id', 'smiles', 'mol') + tuple(tasks))
            yield mol_df
          # Reset aggregator
          df_rows = []
  
      # Handle final leftovers for this file
      if len(df_rows) > 0:
        if has_csv:
          mol_df = pd.DataFrame(df_rows, columns=('mol_id', 'smiles', 'mol'))
          raw_df = next(load_csv_files([input_file + ".csv"], shard_size=None))
          yield pd.concat([mol_df, raw_df], axis=1, join='inner')
        else:
          mol_df = pd.DataFrame(
              df_rows, columns=('mol_id', 'smiles', 'mol') + tuple(tasks))
          yield mol_df
        df_rows = []
  
  
  def load_csv_files(input_files: List[str],
                     shard_size: Optional[int] = None) -> Iterator[pd.DataFrame]:
    """Load data as pandas dataframe from CSV files.
  
    Parameters
    ----------
    input_files: List[str]
      List of filenames
    shard_size: int, default None
      The shard size to yield at one time.
  
    Returns
    -------
    Iterator[pd.DataFrame]
      Generator which yields the dataframe which is the same shard size.
    """
    # First line of user-specified CSV *must* be header.
    shard_num = 1
    for input_file in input_files:
      if shard_size is None:
        yield pd.read_csv(input_file)
      else:
        logger.info("About to start loading CSV from %s" % input_file)
        for df in pd.read_csv(input_file, chunksize=shard_size):
          logger.info(
              "Loading shard %d of size %s." % (shard_num, str(shard_size)))
          df = df.replace(np.nan, str(""), regex=True)
          shard_num += 1
          yield df
  
  
  def load_json_files(input_files: List[str],
                      shard_size: Optional[int] = None) -> Iterator[pd.DataFrame]:
    """Load data as pandas dataframe.
  
    Parameters
    ----------
    input_files: List[str]
      List of json filenames.
    shard_size: int, default None
      Chunksize for reading json files.
  
    Returns
    -------
    Iterator[pd.DataFrame]
      Generator which yields the dataframe which is the same shard size.
  
    Notes
    -----
    To load shards from a json file into a Pandas dataframe, the file
    must be originally saved with ``df.to_json('filename.json', orient='records', lines=True)``
    """
    shard_num = 1
    for input_file in input_files:
      if shard_size is None:
        yield pd.read_json(input_file, orient='records', lines=True)
      else:
        logger.info("About to start loading json from %s." % input_file)
        for df in pd.read_json(
            input_file, orient='records', chunksize=shard_size, lines=True):
          logger.info(
              "Loading shard %d of size %s." % (shard_num, str(shard_size)))
          df = df.replace(np.nan, str(""), regex=True)
          shard_num += 1
          yield df
  
  
  def load_pickle_file(input_file: str) -> Any:
    """Load from single, possibly gzipped, pickle file.
  
    Parameters
    ----------
    input_file: str
      The filename of pickle file. This function can load from
      gzipped pickle file like `XXXX.pkl.gz`.
  
    Returns
    -------
    Any
      The object which is loaded from the pickle file.
    """
    if ".gz" in input_file:
      with gzip.open(input_file, "rb") as unzipped_file:
        return pickle.load(cast(IO[bytes], unzipped_file))
    else:
      with open(input_file, "rb") as opened_file:
        return pickle.load(opened_file)
  
  
  def load_pickle_files(input_files: List[str]) -> Iterator[Any]:
    """Load dataset from pickle files.
  
    Parameters
    ----------
    input_files: List[str]
      The list of filenames of pickle file. This function can load from
      gzipped pickle file like `XXXX.pkl.gz`.
  
    Returns
    -------
    Iterator[Any]
      Generator which yields the objects which is loaded from each pickle file.
    """
    for input_file in input_files:
      yield load_pickle_file(input_file)
  
  
  def load_data(input_files: List[str],
                shard_size: Optional[int] = None) -> Iterator[Any]:
    """Loads data from files.
  
    Parameters
    ----------
    input_files: List[str]
      List of filenames.
    shard_size: int, default None
      Size of shard to yield
  
    Returns
    -------
    Iterator[Any]
      Iterator which iterates over provided files.
  
    Notes
    -----
    The supported file types are SDF, CSV and Pickle.
    """
    if len(input_files) == 0:
      raise ValueError("The length of `filenames` must be more than 1.")
  
    file_type = _get_file_type(input_files[0])
    if file_type == "sdf":
      if shard_size is not None:
        logger.info("Ignoring shard_size for sdf input.")
      for value in load_sdf_files(input_files):
        yield value
    elif file_type == "csv":
      for value in load_csv_files(input_files, shard_size):
        yield value
    elif file_type == "pickle":
      if shard_size is not None:
        logger.info("Ignoring shard_size for pickle input.")
      for value in load_pickle_files(input_files):
        yield value
  
  
  def _get_file_type(input_file: str) -> str:
    """Get type of input file. Must be csv/pkl/sdf/joblib file."""
    filename, file_extension = os.path.splitext(input_file)
    # If gzipped, need to compute extension again
    if file_extension == ".gz":
      filename, file_extension = os.path.splitext(filename)
    if file_extension == ".csv":
      return "csv"
    elif file_extension == ".pkl":
      return "pickle"
    elif file_extension == ".joblib":
      return "joblib"
    elif file_extension == ".sdf":
      return "sdf"
    else:
      raise ValueError("Unrecognized extension %s" % file_extension)
  
  
  def save_to_disk(dataset: Any, filename: str, compress: int = 3):
    """Save a dataset to file.
  
    Parameters
    ----------
    dataset: str
      A data saved
    filename: str
      Path to save data.
    compress: int, default 3
      The compress option when dumping joblib file.
    """
    if filename.endswith('.joblib'):
      joblib.dump(dataset, filename, compress=compress)
    elif filename.endswith('.npy'):
      np.save(filename, dataset)
    else:
      raise ValueError("Filename with unsupported extension: %s" % filename)
  
  
  def load_from_disk(filename: str) -> Any:
    """Load a dataset from file.
  
    Parameters
    ----------
    filename: str
      A filename you want to load data.
  
    Returns
    -------
    Any
      A loaded object from file.
    """
    name = filename
    if os.path.splitext(name)[1] == ".gz":
      name = os.path.splitext(name)[0]
    extension = os.path.splitext(name)[1]
    if extension == ".pkl":
      return load_pickle_file(filename)
    elif extension == ".joblib":
      return joblib.load(filename)
    elif extension == ".csv":
      # First line of user-specified CSV *must* be header.
      df = pd.read_csv(filename, header=0)
      df = df.replace(np.nan, str(""), regex=True)
      return df
    elif extension == ".npy":
      return np.load(filename, allow_pickle=True)
    else:
      raise ValueError("Unrecognized filetype for %s" % filename)
  
  
  def load_dataset_from_disk(save_dir: str) -> Tuple[bool, Optional[Tuple[
      "dc.data.DiskDataset", "dc.data.DiskDataset", "dc.data.DiskDataset"]], List[
          "dc.trans.Transformer"]]:
    """Loads MoleculeNet train/valid/test/transformers from disk.
  
    Expects that data was saved using `save_dataset_to_disk` below. Expects the
    following directory structure for `save_dir`:
    save_dir/
      |
      ---> train_dir/
      |
      ---> valid_dir/
      |
      ---> test_dir/
      |
      ---> transformers.pkl
  
    Parameters
    ----------
    save_dir: str
      Directory name to load datasets.
  
    Returns
    -------
    loaded: bool
      Whether the load succeeded
    all_dataset: Tuple[DiskDataset, DiskDataset, DiskDataset]
      The train, valid, test datasets
    transformers: Transformer
      The transformers used for this dataset
  
    See Also
    --------
    save_dataset_to_disk
    """
  
    train_dir = os.path.join(save_dir, "train_dir")
    valid_dir = os.path.join(save_dir, "valid_dir")
    test_dir = os.path.join(save_dir, "test_dir")
    if not os.path.exists(train_dir) or not os.path.exists(
        valid_dir) or not os.path.exists(test_dir):
      return False, None, list()
    loaded = True
    train = dc.data.DiskDataset(train_dir)
    valid = dc.data.DiskDataset(valid_dir)
    test = dc.data.DiskDataset(test_dir)
    train.memory_cache_size = 40 * (1 << 20)  # 40 MB
    all_dataset = (train, valid, test)
    transformers = load_transformers(save_dir)
    return loaded, all_dataset, transformers
  
  
  def save_dataset_to_disk(
      save_dir: str, train: "dc.data.DiskDataset", valid: "dc.data.DiskDataset",
      test: "dc.data.DiskDataset", transformers: List["dc.trans.Transformer"]):
    """Utility used by MoleculeNet to save train/valid/test datasets.
  
    This utility function saves a train/valid/test split of a dataset along
    with transformers in the same directory. The saved datasets will take the
    following structure:
    save_dir/
      |
      ---> train_dir/
      |
      ---> valid_dir/
      |
      ---> test_dir/
      |
      ---> transformers.pkl
  
    Parameters
    ----------
    save_dir: str
      Directory name to save datasets to.
    train: DiskDataset
      Training dataset to save.
    valid: DiskDataset
      Validation dataset to save.
    test: DiskDataset
      Test dataset to save.
    transformers: List[Transformer]
      List of transformers to save to disk.
  
    See Also
    --------
    load_dataset_from_disk
    """
    train_dir = os.path.join(save_dir, "train_dir")
    valid_dir = os.path.join(save_dir, "valid_dir")
    test_dir = os.path.join(save_dir, "test_dir")
    train.move(train_dir)
    valid.move(valid_dir)
    test.move(test_dir)
    save_transformers(save_dir, transformers)
  
  
  def load_transformers(save_dir: str) -> List["dc.trans.Transformer"]:
    """Load the transformers for a MoleculeNet dataset from disk."""
    with open(os.path.join(save_dir, "transformers.pkl"), 'rb') as f:
      return pickle.load(f)
  
  
  def save_transformers(save_dir: str,
                        transformers: List["dc.trans.Transformer"]):
    """Save the transformers for a MoleculeNet dataset to disk."""
    with open(os.path.join(save_dir, "transformers.pkl"), 'wb') as f:
      pickle.dump(transformers, f)
  