Commit b9d616c0 authored by Bharath's avatar Bharath
Browse files

First draft of stripped down featurization

parent d619c4a8

deepchem/datasets/__init__.py

+38 −25
Original line number Diff line number Diff line
@@ -60,16 +60,24 @@ class Dataset(object): 
        raise ValueError("No metadata found.")



  @staticmethod

  def write_dataframe(val, data_dir, featurizers=None, tasks=None,

                      raw_data=None, basename=None):

  def write_dataframe(val, data_dir, featurizer=None, tasks=None,

                      raw_data=None, basename=None, mol_id_field="mol_id"):

    """Writes data from dataframe to disk."""

    if featurizers is not None and tasks is not None:

      feature_types = [featurizer.__class__.__name__ for featurizer in featurizers]

    if featurizer is not None and tasks is not None:

      feature_type = featurizer.__class__.__name__

      (basename, df) = val

      # TODO(rbharath): This is a hack. clean up.

      if not len(df):

        return None

      ids, X, y, w = _df_to_numpy(df, feature_types, tasks)

      ############################################################## DEBUG

      print("About to call convert_df_to_numpy")

      print("mol_id_field")

      print(mol_id_field)

      ############################################################## DEBUG

      ids, X, y, w = convert_df_to_numpy(df, feature_type, tasks, mol_id_field)

      ############################################################## DEBUG

      print("convert_df_to_numpy returned successfully")

      ############################################################## DEBUG

    else:

      ids, X, y, w = raw_data

      basename = ""
@@ -588,11 +596,11 @@ def compute_sums_and_nb_sample(tensor, W=None): 


# The following are all associated with Dataset, but are separate functions to

# make it easy to use multiprocessing.

def _df_to_numpy(df, feature_types, tasks):

def convert_df_to_numpy(df, feature_type, tasks, mol_id_field):

  """Transforms a featurized dataset df into standard set of numpy arrays"""

  if not set(feature_types).issubset(df.keys()):

  if feature_type not in df.keys():

    raise ValueError(

        "Featurized data does not support requested feature_types.")

        "Featurized data does not support requested feature_type.")

  # perform common train/test split across all tasks

  n_samples = df.shape[0]

  n_tasks = len(tasks)
@@ -600,38 +608,43 @@ def _df_to_numpy(df, feature_types, tasks): 
      np.reshape(np.array(df[task].values), (n_samples, 1)) for task in tasks])

  w = np.ones((n_samples, n_tasks))

  missing = np.zeros_like(y).astype(int)

  tensors = []

  all_features = []

  feature_shape = None

  ############################################################## DEBUG

  print("convert_df_to_numpy --- about to loop through data.")

  ############################################################## DEBUG

  for ind in range(n_samples):

    ############################################################### DEBUG

    #print("sample %d" % ind)

    ############################################################### DEBUG

    datapoint = df.iloc[ind]

    feature_list = []

    for feature_type in feature_types:

      feature_list.append(datapoint[feature_type])

    try:

      features = np.squeeze(np.concatenate(feature_list))

    features = np.squeeze(datapoint[feature_type])

    ############################################################### DEBUG

    #print("features.size")

    ############################################################### DEBUG

    if features.size == 0:

      features = np.zeros(feature_shape)

        tensors.append(features)

      all_features.append(features)

      missing[ind, :] = 1

      continue

      for feature_ind, val in enumerate(features):

        if features[feature_ind] == "":

          features[feature_ind] = 0.

      features = features.astype(float)

    if feature_shape is None:

      feature_shape = features.shape

    except ValueError:

      missing[ind, :] = 1

      continue

    for task in range(n_tasks):

      if y[ind, task] == "":

        missing[ind, task] = 1

    if features.shape != feature_shape:

      missing[ind, :] = 1

      continue

    tensors.append(features)

  x = np.stack(tensors)

  sorted_ids = df["mol_id"]

    ############################################################### DEBUG

    #print("Done processing sample")

    ############################################################### DEBUG

    all_features.append(features)

  x = np.stack(all_features)

  ############################################################## DEBUG

  print("mol_id_field")

  print(mol_id_field)

  ############################################################## DEBUG

  sorted_ids = df[mol_id_field]



  # Set missing data to have weight zero

  # TODO(rbharath): There's a better way to do this with numpy indexing

deepchem/featurizers/featurize.py

+95 −94
Original line number Diff line number Diff line
@@ -28,11 +28,11 @@ from deepchem.utils.save import get_input_type 
import time

############################################################## DEBUG



def _process_helper(row, loader, fields, input_type):

  return loader._process_raw_sample(input_type, row, fields)

#def _process_helper(row, loader, fields, input_type):

#  return loader._process_raw_sample(input_type, row, fields)



def featurize_map_function(args):

  try:

  #try:

  ############################################################## DEBUG

  time1 = time.time()

  ############################################################## DEBUG
@@ -42,15 +42,18 @@ def featurize_map_function(args): 
  log("About to featurize shard.", loader.verbosity)

  write_fn = partial(

      Dataset.write_dataframe, data_dir=data_dir,

        featurizers=loader.featurizers, tasks=loader.tasks)

    process_fn = partial(_process_helper, loader=loader,

                         fields=raw_df_shard.keys(),

                         input_type=input_type)

      featurizer=loader.featurizer, tasks=loader.tasks,

      mol_id_field=loader.id_field)

  #process_fn = partial(_process_helper, loader=loader,

  #                     fields=raw_df_shard.keys(),

  #                     input_type=input_type)

  ############################################################## DEBUG

  shard_time1 = time.time()

  ############################################################## DEBUG

  metadata_row = loader._featurize_shard(

        raw_df_shard, process_fn, write_fn, shard_num, input_type)

  #    raw_df_shard, process_fn, write_fn, shard_num, input_type)

  ############################################################## DEBUG

      raw_df_shard, write_fn, shard_num, input_type)

  ############################################################## DEBUG

  shard_time2 = time.time()

  print("SHARD FEATURIZATION TOOK %0.3f s" % (shard_time2-shard_time1))
@@ -61,26 +64,26 @@ def featurize_map_function(args): 
  print("FEATURIZATION MAP FUNCTION TOOK %0.3f s" % (time2-time1))

  ############################################################## DEBUG

  return metadata_row

  except:

    print("Shard %d featurization crashed!" % shard_num)

    return None





def _process_field(val):

  """Parse data in a field."""

  if (isinstance(val, numbers.Number) or isinstance(val, np.ndarray)):

    return val

  elif isinstance(val, list):

    return [_process_field(elt) for elt in val]

  elif isinstance(val, str):

    try:

      return float(val)

    except ValueError:

      return val

  elif isinstance(val, Chem.Mol):

    return val

  else:

    raise ValueError("Field of unrecognized type: %s" % str(val))

  #except:

  #  print("Shard %d featurization crashed!" % shard_num)

  #  return None





#def _process_field(val):

#  """Parse data in a field."""

#  if (isinstance(val, numbers.Number) or isinstance(val, np.ndarray)):

#    return val

#  elif isinstance(val, list):

#    return [_process_field(elt) for elt in val]

#  elif isinstance(val, str):

#    try:

#      return float(val)

#    except ValueError:

#      return val

#  elif isinstance(val, Chem.Mol):

#    return val

#  else:

#    raise ValueError("Field of unrecognized type: %s" % str(val))



class DataFeaturizer(object):

  """
@@ -92,9 +95,7 @@ class DataFeaturizer(object): 


  def __init__(self, tasks, smiles_field=None,

               id_field=None, threshold=None,

               protein_pdb_field=None, ligand_pdb_field=None,

               ligand_mol2_field=None, mol_field=None,

               featurizers=[],

               mol_field=None, featurizer=None,

               verbosity=None, log_every_n=1000):

    """Extracts data from input as Pandas data frame"""

    if not isinstance(tasks, list):
@@ -108,15 +109,11 @@ class DataFeaturizer(object): 
    else:

      self.id_field = id_field

    self.threshold = threshold

    self.protein_pdb_field = protein_pdb_field

    self.ligand_pdb_field = ligand_pdb_field

    self.ligand_mol2_field = ligand_mol2_field

    self.mol_field = mol_field

    self.user_specified_features = None

    for featurizer in featurizers:

    if isinstance(featurizer, UserDefinedFeaturizer):

      self.user_specified_features = featurizer.feature_fields 

    self.featurizers = featurizers

    self.featurizer = featurizer

    self.log_every_n = log_every_n



  def featurize(self, input_files, data_dir, shard_size=8192,
@@ -148,7 +145,7 @@ class DataFeaturizer(object): 
    data_iterator = it.izip(

        it.repeat((self, shard_size, input_type, data_dir)),

        enumerate(load_data(input_files, shard_size, self.verbosity)))

    ###### TODO(rbharath): Turns out python map is terrible and exhausts the

    ###### Turns out python map is terrible and exhausts the

    ###### generator as given. Solution seems to be to to manually pull out N elements

    ###### from iterator, then to map on only those N elements. BLECH. Python

    ###### should do a better job here.
@@ -191,15 +188,19 @@ class DataFeaturizer(object): 
    ############################################################## DEBUG

    return dataset 



  def _featurize_shard(self, raw_df_shard, process_fn, write_fn, shard_num, input_type):

  #def _featurize_shard(self, raw_df_shard, process_fn, write_fn, shard_num,

  #                     input_type):

  ############################################################## DEBUG

  def _featurize_shard(self, raw_df_shard, write_fn, shard_num, input_type):

  ############################################################## DEBUG

    """Featurizes a shard of an input dataframe."""

    ############################################################## DEBUG

    time1 = time.time()

    ############################################################## DEBUG

    log("Applying processing transformation to shard.",

        self.verbosity)

    raw_df_shard = raw_df_shard.apply(

        process_fn, axis=1, reduce=False)

    #raw_df_shard = raw_df_shard.apply(

    #    process_fn, axis=1, reduce=False)

    ############################################################## DEBUG

    time2 = time.time()

    print("PROCESSING TRANSFORMATION TOOK %0.3f s" % (time2-time1))
@@ -208,26 +209,29 @@ class DataFeaturizer(object): 
    time1 = time.time()

    ############################################################## DEBUG

    log("About to standardize dataframe.")

    df_shard = self._standardize_df(raw_df_shard) 

    #df_shard = self._standardize_df(raw_df_shard) 

    df_shard = raw_df_shard

    ############################################################## DEBUG

    time2 = time.time()

    print("STANDARDIZATION TOOK %0.3f s" % (time2-time1))

    ############################################################## DEBUG

  

    field = "mol" if input_type == "sdf" else "smiles"

    for featurizer in self.featurizers:

    field = self.mol_field if input_type == "sdf" else self.smiles_field 

    log("Currently featurizing feature_type: %s"

          % featurizer.__class__.__name__, self.verbosity)

      if isinstance(featurizer, UserDefinedFeaturizer):

        self._add_user_specified_features(df_shard, featurizer)

      elif isinstance(featurizer, Featurizer):

        self._featurize_mol(df_shard, featurizer, field=field)

      elif isinstance(featurizer, ComplexFeaturizer):

        self._featurize_complexes(df_shard, featurizer)

        % self.featurizer.__class__.__name__, self.verbosity)

    if isinstance(self.featurizer, UserDefinedFeaturizer):

      self._add_user_specified_features(df_shard, self.featurizer)

    elif isinstance(self.featurizer, Featurizer):

      self._featurize_mol(df_shard, self.featurizer, field=field)

    elif isinstance(self.featurizer, ComplexFeaturizer):

      self._featurize_complexes(df_shard, self.featurizer)

    basename = "shard-%d" % shard_num 

    ############################################################## DEBUG

    time1 = time.time()

    ############################################################## DEBUG

    ############################################################## DEBUG

    print("About to invoke write_fn")

    ############################################################## DEBUG

    metadata_row = write_fn((basename, df_shard))

    ############################################################## DEBUG

    time2 = time.time()
@@ -242,50 +246,44 @@ class DataFeaturizer(object): 
        return True

    return False



  def _process_raw_sample(self, input_type, row, fields):

    """Extract information from row data."""

    data = {}

    if input_type == "csv":

      for ind, field in enumerate(fields):

        data[field] = _process_field(row[ind])

    elif input_type in ["pandas-pickle", "pandas-joblib", "sdf"]:

      for field in fields:

        data[field] = _process_field(row[field])

    else:

      raise ValueError("Unrecognized input_type")

    if self.threshold is not None:

      for task in self.tasks:

        raw = _process_field(data[task])

        if not isinstance(raw, float):

          raise ValueError("Cannot threshold non-float fields.")

        data[field] = 1 if raw > self.threshold else 0

    return data



  def _standardize_df(self, ori_df):

    """Copy specified columns to new df with standard column names.



    TODO(rbharath): I think think function is now unnecessary (since the

                    dataframes are only temporary and not on disk). Should

                    be able to remove this function.

    """

    df = pd.DataFrame(ori_df[[self.id_field]])

    df.columns = ["mol_id"]

    if self.smiles_field is not None:

      df["smiles"] = ori_df[[self.smiles_field]]

    for task in self.tasks:

      df[task] = ori_df[[task]]

    if self.user_specified_features is not None:

      for feature in self.user_specified_features:

        df[feature] = ori_df[[feature]]

    if self.mol_field is not None:

      df["mol"] = ori_df[[self.mol_field]]

    if self.protein_pdb_field is not None:

      df["protein_pdb"] = ori_df[[self.protein_pdb_field]]

    if self.ligand_pdb_field is not None:

      df["ligand_pdb"] = ori_df[[self.ligand_pdb_field]]

    if self.ligand_mol2_field is not None:

      df["ligand_mol2"] = ori_df[[self.ligand_mol2_field]]

    return df

  #def _process_raw_sample(self, input_type, row, fields):

  #  """Extract information from row data."""

  #  data = {}

  #  if input_type == "csv":

  #    for ind, field in enumerate(fields):

  #      data[field] = _process_field(row[ind])

  #  elif input_type in ["pandas-pickle", "pandas-joblib", "sdf"]:

  #    for field in fields:

  #      data[field] = _process_field(row[field])

  #  else:

  #    raise ValueError("Unrecognized input_type")

  #  if self.threshold is not None:

  #    for task in self.tasks:

  #      raw = _process_field(data[task])

  #      if not isinstance(raw, float):

  #        raise ValueError("Cannot threshold non-float fields.")

  #      data[field] = 1 if raw > self.threshold else 0

  #  return data



  #def _standardize_df(self, ori_df):

  #  """Copy specified columns to new df with standard column names.



  #  TODO(rbharath): I think think function is now unnecessary (since the

  #                  dataframes are only temporary and not on disk). Should

  #                  be able to remove this function.

  #  """

  #  df = pd.DataFrame(ori_df[[self.id_field]])

  #  df.columns = ["mol_id"]

  #  if self.smiles_field is not None:

  #    df["smiles"] = ori_df[[self.smiles_field]]

  #  for task in self.tasks:

  #    df[task] = ori_df[[task]]

  #  if self.user_specified_features is not None:

  #    for feature in self.user_specified_features:

  #      df[feature] = ori_df[[feature]]

  #  if self.mol_field is not None:

  #    df["mol"] = ori_df[[self.mol_field]]

  #  return df



  def _featurize_complexes(self, df, featurizer, parallel=True,

                           worker_pool=None):
@@ -377,6 +375,9 @@ class DataFeaturizer(object): 
    ############################################################## DEBUG

    log("Aggregating User-Specified Features", self.verbosity)

    features_data = []

    ############################################################## DEBUG

    df[featurizer.feature_fields] = df[featurizer.feature_fields].apply(pd.to_numeric)

    ############################################################## DEBUG

    for ind, row in df.iterrows():

      # pandas rows are tuples (row_num, row_data)

      feature_list = []

CRA Git | Maintained and supported by SUSTech CRA and CCSE