Loading deepchem/data/__init__.py +6 −1 Original line number Diff line number Diff line Loading @@ -6,7 +6,12 @@ from __future__ import division from __future__ import unicode_literals # TODO(rbharath): Get rid of * import from deepchem.data.datasets import * from deepchem.data.datasets import pad_features from deepchem.data.datasets import pad_batch from deepchem.data.datasets import Dataset from deepchem.data.datasets import NumpyDataset from deepchem.data.datasets import DiskDataset from deepchem.data.supports import * from deepchem.data.data_loader import DataLoader from deepchem.data.data_loader import CSVLoader import deepchem.data.tests deepchem/data/data_loader.py +177 −106 Original line number Diff line number Diff line Loading @@ -16,13 +16,126 @@ from rdkit import Chem import time import sys from deepchem.utils.save import log from deepchem.utils.save import save_to_disk from deepchem.utils.save import load_pickle_from_disk from deepchem.feat import Featurizer, ComplexFeaturizer from deepchem.utils.save import load_csv_files from deepchem.utils.save import load_sdf_files from deepchem.feat import UserDefinedFeaturizer from deepchem.data import DiskDataset from deepchem.utils.save import load_data from deepchem.utils.save import get_input_type def convert_df_to_numpy(df, tasks, id_field, verbose=False): """Transforms a dataframe containing deepchem input into numpy arrays""" n_samples = df.shape[0] n_tasks = len(tasks) time1 = time.time() y = np.hstack([ np.reshape(np.array(df[task].values), (n_samples, 1)) for task in tasks]) time2 = time.time() w = np.ones((n_samples, n_tasks)) missing = np.zeros_like(y).astype(int) feature_shape = None #for ind in range(n_samples): # for task in range(n_tasks): # if y[ind, task] == "": # missing[ind, task] = 1 #x_list = list(df[feature_type].values) #valid_inds = np.array([1 if elt.size > 0 else 0 for elt in x_list], dtype=bool) #x_list = [elt for (is_valid, elt) in zip(valid_inds, x_list) if is_valid] #x = np.squeeze(np.array(x_list)) sorted_ids = df[id_field].values # Set missing data to have weight zero for ind in range(n_samples): for task in range(n_tasks): if missing[ind, task]: y[ind, task] = 0. w[ind, task] = 0. #sorted_ids = sorted_ids[valid_inds] #y = y[valid_inds] #w = w[valid_inds] # Adding this assertion in to avoid ill-formed outputs. #assert len(sorted_ids) == len(x) == len(y) == len(w) assert len(sorted_ids) == len(y) == len(w) #return sorted_ids, x, y.astype(float), w.astype(float) return sorted_ids, y.astype(float), w.astype(float) #def write_dataframe(basename, df, data_dir, feature_type, tasks=None, # raw_data=None, mol_id_field="mol_id", # verbose=False): # """Writes data from dataframe to disk.""" # if featurizer is not None and tasks is not None: # feature_type = featurizer.__class__.__name__ # time1 = time.time() # ids, X, y, w = convert_df_to_numpy( # df, feature_type, tasks, mol_id_field) # time2 = time.time() # log("TIMING: convert_df_to_numpy took %0.3f s" % (time2-time1), verbose) # else: # ids, X, y, w = raw_data # assert X.shape[0] == y.shape[0] # assert y.shape == w.shape # assert len(ids) == X.shape[0] # return DiskDataset.write_data_to_disk( # data_dir, basename, tasks, X, y, w, ids) def featurize_smiles_df(df, featurizer, field, log_every_N=1000, verbose=True): """Featurize individual compounds in dataframe. Given a featurizer that operates on individual chemical compounds or macromolecules, compute & add features for that compound to the features dataframe """ sample_elems = df[field].tolist() features = [] for ind, elem in enumerate(sample_elems): mol = Chem.MolFromSmiles(elem) if ind % log_every_N == 0: log("Featurizing sample %d" % ind, verbose) features.append(featurizer.featurize([mol])) return np.array(features) def get_user_specified_features(df, featurizer, verbose=True): """Extract and merge user specified features. Merge features included in dataset provided by user into final features dataframe Three types of featurization here: 1) Molecule featurization -) Smiles string featurization -) Rdkit MOL featurization 2) Complex featurization -) PDB files for interacting molecules. 3) User specified featurizations. """ time1 = time.time() df[featurizer.feature_fields] = df[featurizer.feature_fields].apply(pd.to_numeric) X_shard = df.as_matrix(columns=featurizer.feature_fields) time2 = time.time() log("TIMING: user specified processing took %0.3f s" % (time2-time1), verbose) return X_shard def featurize_mol_df(df, featurizer, field, verbose=True, log_every_N=1000): """Featurize individual compounds in dataframe. Featurizes .sdf files, so the 3-D structure should be preserved so we use the rdkit "mol" object created from .sdf instead of smiles string. Some featurizers such as CoulombMatrix also require a 3-D structure. Featurizing from .sdf is currently the only way to perform CM feautization. """ sample_elems = df[field].tolist() features = [] for ind, mol in enumerate(sample_elems): if ind % log_every_N == 0: log("Featurizing sample %d" % ind, verbose) features.append(featurizer.featurize([mol])) return np.array(features) class DataLoader(object): """ Loading Loading @@ -57,110 +170,68 @@ class DataLoader(object): log("Loading raw samples now.", self.verbose) log("shard_size: %d" % shard_size, self.verbose) # Allow users to specify a single file for featurization if not isinstance(input_files, list): input_files = [input_files] if data_dir is not None: if not os.path.exists(data_dir): os.makedirs(data_dir) else: data_dir = tempfile.mkdtemp() if not len(input_files): return None input_type = get_input_type(input_files[0]) metadata_rows = [] for shard_num, elt in enumerate(load_data(input_files, shard_size)): time1 = time.time() metadata_row = self._featurize_shard(elt, data_dir, shard_num, input_type) if metadata_row is not None: metadata_rows.append(metadata_row) time2 = time.time() log("TIMING: shard %d took %0.3f s" % (shard_num, time2-time1), self.verbose) def shard_generator(): for shard_num, shard in enumerate(self.get_shards(input_files, shard_size)): time1 = time.time() dataset = DiskDataset(data_dir=data_dir, metadata_rows=metadata_rows, reload=True) X = self.featurize_shard(shard) ids, y, w = convert_df_to_numpy(shard, self.tasks, self.id_field) time2 = time.time() print("TIMING: dataset construction took %0.3f s" % (time2-time1), log("TIMING: featurizing shard %d took %0.3f s" % (shard_num, time2-time1), self.verbose) return dataset yield ids, X, y, w return DiskDataset(shard_generator(), data_dir, self.tasks) def _featurize_shard(self, df_shard, data_dir, shard_num, input_type): """Featurizes a shard of an input dataframe.""" field = self.mol_field if input_type == "sdf" else self.smiles_field field_type = "mol" if input_type == "sdf" else "smiles" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) 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, field_type=field_type) basename = "shard-%d" % shard_num time1 = time.time() metadata_row = DiskDataset.write_dataframe( basename, df_shard, data_dir=data_dir, featurizer=self.featurizer, tasks=self.tasks, mol_id_field=self.id_field) time2 = time.time() log("TIMING: writing metadata row took %0.3f s" % (time2-time1), self.verbose) return metadata_row def _featurize_mol(self, df, featurizer, parallel=True, field_type="mol", field=None): """Featurize individual compounds. def get_shards(self, input_files, shard_size): """Stub for children classes.""" raise NotImplementedError Given a featurizer that operates on individual chemical compounds or macromolecules, compute & add features for that compound to the features dataframe def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" raise NotImplementedError When featurizing a .sdf file, the 3-D structure should be preserved so we use the rdkit "mol" object created from .sdf instead of smiles string. Some featurizers such as CoulombMatrix also require a 3-D structure. Featurizing from .sdf is currently the only way to perform CM feautization. TODO(rbharath): Needs to be merged with _featurize_compounds class CSVLoader(DataLoader): """ assert field_type in ["mol", "smiles"] assert field is not None sample_elems = df[field].tolist() features = [] for ind, elem in enumerate(sample_elems): if field_type == "smiles": mol = Chem.MolFromSmiles(elem) else: mol = elem if ind % self.log_every_n == 0: log("Featurizing sample %d" % ind, self.verbose) features.append(featurizer.featurize([mol])) df[featurizer.__class__.__name__] = features Handles loading of CSV files. """ def get_shards(self, input_files, shard_size, verbose=True): """Defines a generator which returns data for each shard""" return load_csv_files(input_files, shard_size, verbose=verbose) def _add_user_specified_features(self, df, featurizer): """Merge user specified features. def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) return featurize_smiles_df(shard, self.featurizer, field=self.smiles_field) Merge features included in dataset provided by user into final features dataframe class UserCSVLoader(DataLoader): """ Handles loading of CSV files with user-defined featurizers. """ def get_shards(self, input_files, shard_size): """Defines a generator which returns data for each shard""" return load_csv_files(input_files, shard_size, verbose=verbose) Three types of featurization here: def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) assert isinstance(self.featurizer, UserDefinedFeaturizer) return get_user_specified_features(shard, self.featurizer) 1) Molecule featurization -) Smiles string featurization -) Rdkit MOL featurization 2) Complex featurization -) PDB files for interacting molecules. 3) User specified featurizations. class SDFLoader(DataLoader): """ time1 = time.time() df[featurizer.feature_fields] = df[featurizer.feature_fields].apply(pd.to_numeric) X_shard = df.as_matrix(columns=featurizer.feature_fields) df[featurizer.__class__.__name__] = [np.array(elt) for elt in X_shard.tolist()] time2 = time.time() log("TIMING: user specified processing took %0.3f s" % (time2-time1), self.verbose) Handles loading of SDF files. """ def get_shard(self, input_files, shard_size): """Defines a generator which returns data for each shard""" return load_sdf_files(input_files) def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) return featurize_mol_df(shard, self.featurizer, field=self.mol_field) deepchem/data/datasets.py +85 −230 Original line number Diff line number Diff line Loading @@ -343,70 +343,43 @@ class NumpyDataset(Dataset): newx, newy, neww = fn(self._X, self._y, self._w) return NumpyDataset(newx, newy, neww, self._ids[:]) class DiskDataset(Dataset): """ A Dataset that is stored as a set of files on disk. """ def __init__(self, data_dir=None, tasks=[], metadata_rows=None, raw_data=None, reload=False, verbose=False): def __init__(self, shard_generator=None, data_dir=None, tasks=[], reload=False, verbose=True): """ Turns featurized dataframes into numpy files, writes them & metadata to disk. """ if data_dir is not None: if not os.path.exists(data_dir): os.makedirs(data_dir) else: data_dir = tempfile.mkdtemp() self.data_dir = data_dir self.verbose = verbose if not reload or not os.path.exists(self._get_metadata_filename()): if metadata_rows is not None: self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() elif raw_data is not None: metadata_rows = [] ids, X, y, w = raw_data metadata_rows.append( DiskDataset.write_data_to_disk( self.data_dir, "data", tasks, X, y, w, ids)) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() else: # Create an empty metadata dataframe to be filled at a later time basename = "metadata" metadata_rows = [DiskDataset.write_data_to_disk( self.data_dir, basename, tasks)] self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() else: log("Loading pre-existing metadata file.", self.verbose) if reload: log("Loading pre-existing dataset.", self.verbose) if os.path.exists(self._get_metadata_filename()): self.metadata_df = load_from_disk(self._get_metadata_filename()) else: raise ValueError("No metadata found.") return @staticmethod def write_dataframe(basename, df, data_dir, featurizer=None, tasks=None, raw_data=None, mol_id_field="mol_id", verbose=False): """Writes data from dataframe to disk.""" if featurizer is not None and tasks is not None: feature_type = featurizer.__class__.__name__ # TODO(rbharath): This is a hack. clean up. if not len(df): return None metadata_rows = [] time1 = time.time() ids, X, y, w = convert_df_to_numpy( df, feature_type, tasks, mol_id_field) for shard_num, (ids, X, y, w) in enumerate(shard_generator): basename = "shard-%d" % shard_num metadata_rows.append( DiskDataset.write_data_to_disk( self.data_dir, basename, tasks, X, y, w, ids)) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() time2 = time.time() log("TIMING: convert_df_to_numpy took %0.3f s" % (time2-time1), verbose) else: ids, X, y, w = raw_data assert X.shape[0] == y.shape[0] assert y.shape == w.shape assert len(ids) == X.shape[0] return DiskDataset.write_data_to_disk( data_dir, basename, tasks, X, y, w, ids) print("TIMING: dataset construction took %0.3f s" % (time2-time1), self.verbose) @staticmethod def construct_metadata(metadata_entries): Loading Loading @@ -457,7 +430,7 @@ class DiskDataset(Dataset): reshard_dir = tempfile.mkdtemp() new_metadata = [] # Write data in new shards ind = 0 def generator(): tasks = self.get_task_names() X_next = np.zeros((0,) + self.get_data_shape()) y_next = np.zeros((0,) + (len(tasks),)) Loading @@ -473,19 +446,10 @@ class DiskDataset(Dataset): y_batch, y_next = y_next[:shard_size], y_next[shard_size:] w_batch, w_next = w_next[:shard_size], w_next[shard_size:] ids_batch, ids_next = ids_next[:shard_size], ids_next[shard_size:] new_basename = "reshard-%d" % ind new_metadata.append(DiskDataset.write_data_to_disk( reshard_dir, new_basename, tasks, X_batch, y_batch, w_batch, ids_batch)) ind += 1 yield (ids_batch, X_batch, y_batch, w_batch) # Handle spillover from last shard new_basename = "reshard-%d" % ind new_metadata.append(DiskDataset.write_data_to_disk( reshard_dir, new_basename, tasks, X_next, y_next, w_next, ids_next)) ind += 1 # Get new metadata rows resharded_dataset = DiskDataset( data_dir=reshard_dir, tasks=tasks, metadata_rows=new_metadata, verbosity=self.verbosity) yield (ids_next, X_next, y_next, w_next) resharded_dataset = DiskDataset(generator(), data_dir=reshard_dir) shutil.rmtree(self.data_dir) shutil.move(reshard_dir, self.data_dir) self.metadata_df = resharded_dataset.metadata_df Loading Loading @@ -632,13 +596,12 @@ class DiskDataset(Dataset): out_dir = tempfile.mkdtemp() tasks = self.get_task_names() metadata_rows = [] def generator(): for shard_num, row in self.metadata_df.iterrows(): X, y, w, ids = self.get_shard(shard_num) newx, newy, neww = fn(X, y, w) basename = "dataset-%d" % shard_num metadata_rows.append(DiskDataset.write_data_to_disk( out_dir, basename, tasks, newx, newy, neww, ids)) return DiskDataset(data_dir=out_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=out_dir) @staticmethod def from_numpy(X, y, w=None, ids=None, tasks=None, data_dir=None): Loading Loading @@ -669,15 +632,11 @@ class DiskDataset(Dataset): os.makedirs(merge_dir) else: merge_dir = tempfile.mkdtemp() Xs, ys, ws, all_ids = [], [], [], [] metadata_rows = [] def generator(): for ind, dataset in enumerate(datasets): X, y, w, ids = (dataset.X, dataset.y, dataset.w, dataset.ids) basename = "dataset-%d" % ind tasks = dataset.get_task_names() metadata_rows.append( DiskDataset.write_data_to_disk(merge_dir, basename, tasks, X, y, w, ids)) return DiskDataset(data_dir=merge_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=merge_dir) def subset(self, shard_nums, subset_dir=None): """Creates a subset of the original dataset on disk.""" Loading @@ -688,14 +647,14 @@ class DiskDataset(Dataset): subset_dir = tempfile.mkdtemp() tasks = self.get_task_names() metadata_rows = [] def generator(): for shard_num, row in self.metadata_df.iterrows(): if shard_num not in shard_nums: continue X, y, w, ids = self.get_shard(shard_num) basename = "dataset-%d" % shard_num metadata_rows.append(DiskDataset.write_data_to_disk( subset_dir, basename, tasks, X, y, w, ids)) return DiskDataset(data_dir=subset_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=subset_dir) def sparse_shuffle(self): """Shuffling that exploits data sparsity to shuffle large datasets. Loading Loading @@ -739,51 +698,6 @@ class DiskDataset(Dataset): self.verbose) ######################################################### TIMING def shuffle(self, iterations=1): """Shuffles this dataset on disk to have random order.""" #np.random.seed(9452) for _ in range(iterations): metadata_rows = [] tasks = self.get_task_names() # Shuffle the arrays corresponding to each row in metadata_df n_rows = len(self.metadata_df.index) len_data = len(self) print("ABOUT TO SHUFFLE DATA ONCE") for i in range(n_rows): # Select random row to swap with j = np.random.randint(n_rows) row_i, row_j = self.metadata_df.iloc[i], self.metadata_df.iloc[j] metadata_rows.append(row_i) # Useful to avoid edge cases, but perhaps there's a better solution if i == j: continue basename_i, basename_j = row_i["basename"], row_j["basename"] X_i, y_i, w_i, ids_i = self.get_shard(i) X_j, y_j, w_j, ids_j = self.get_shard(j) n_i, n_j = X_i.shape[0], X_j.shape[0] # Join two shards and shuffle them at random. X = np.vstack([X_i, X_j]) y = np.vstack([y_i, y_j]) w = np.vstack([w_i, w_j]) ids = np.concatenate([ids_i, ids_j]) permutation = np.random.permutation(n_i + n_j) X, y, w, ids = (X[permutation], y[permutation], w[permutation], ids[permutation]) X_i, y_i, w_i, ids_i = X[:n_i], y[:n_i], w[:n_i], ids[:n_i] X_j, y_j, w_j, ids_j = X[n_i:], y[n_i:], w[n_i:], ids[n_i:] DiskDataset.write_data_to_disk( self.data_dir, basename_i, tasks, X_i, y_i, w_i, ids_i) DiskDataset.write_data_to_disk( self.data_dir, basename_j, tasks, X_j, y_j, w_j, ids_j) assert len(self) == len_data # Now shuffle order of rows in metadata_df random.shuffle(metadata_rows) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() def shuffle_each_shard(self): """Shuffles elements within each shard of the datset.""" tasks = self.get_task_names() Loading Loading @@ -827,8 +741,6 @@ class DiskDataset(Dataset): tasks = self.get_task_names() DiskDataset.write_data_to_disk(self.data_dir, basename, tasks, X, y, w, ids) # TODO(rbharath): This change for general object types seems a little # kludgey. Is there a more principled approach to support general objects? def select(self, indices, select_dir=None): """Creates a new dataset from a selection of indices from self. Loading @@ -846,12 +758,12 @@ class DiskDataset(Dataset): select_dir = tempfile.mkdtemp() # Handle edge case with empty indices if not len(indices): return DiskDataset( data_dir=select_dir, metadata_rows=[]) return DiskDataset([], data_dir=select_dir) indices = np.array(sorted(indices)).astype(int) count, indices_count = 0, 0 metadata_rows = [] tasks = self.get_task_names() def generator(): for shard_num, (X, y, w, ids) in enumerate(self.itershards()): shard_len = len(X) # Find indices which rest in this shard Loading @@ -866,17 +778,14 @@ class DiskDataset(Dataset): y_sel = y[shard_inds] w_sel = w[shard_inds] ids_sel = ids[shard_inds] basename = "dataset-%d" % shard_num metadata_rows.append( DiskDataset.write_data_to_disk( select_dir, basename, tasks, X_sel, y_sel, w_sel, ids_sel)) yield (ids_sel, X_sel, y_sel, w_sel) # Updating counts indices_count += num_shard_elts count += shard_len # Break when all indices have been used up already if indices_count >= len(indices): break return DiskDataset(data_dir=select_dir, metadata_rows=metadata_rows) return return DiskDataset(generator(), data_dir=select_dir) @property def ids(self): Loading Loading @@ -955,57 +864,3 @@ class DiskDataset(Dataset): def get_label_stds(self): """Return pandas series of label stds.""" return self.metadata_df["y_stds"] # The following are all associated with Dataset, but are separate functions to # make it easy to use multiprocessing. def convert_df_to_numpy(df, feature_type, tasks, mol_id_field, verbose=False): """Transforms a dataframe containing deepchem input into numpy arrays""" if feature_type not in df.keys(): raise ValueError( "Featurized data does not support requested feature_type %s." % feature_type) # perform common train/test split across all tasks n_samples = df.shape[0] n_tasks = len(tasks) time1 = time.time() y = np.hstack([ np.reshape(np.array(df[task].values), (n_samples, 1)) for task in tasks]) time2 = time.time() log("TIMING: convert_df_to_numpy y computation took %0.3f s" % (time2-time1), verbose) w = np.ones((n_samples, n_tasks)) missing = np.zeros_like(y).astype(int) feature_shape = None time1 = time.time() for ind in range(n_samples): for task in range(n_tasks): if y[ind, task] == "": missing[ind, task] = 1 x_list = list(df[feature_type].values) valid_inds = np.array([1 if elt.size > 0 else 0 for elt in x_list], dtype=bool) x_list = [elt for (is_valid, elt) in zip(valid_inds, x_list) if is_valid] x = np.squeeze(np.array(x_list)) time2 = time.time() log("TIMING: convert_df_to_numpy x computation took %0.3f s" % (time2-time1), verbose) sorted_ids = df[mol_id_field].values # Set missing data to have weight zero time1 = time.time() for ind in range(n_samples): for task in range(n_tasks): if missing[ind, task]: y[ind, task] = 0. w[ind, task] = 0. time2 = time.time() log("TIMING: convert_df_to_numpy missing elts computation took %0.3f s" % (time2-time1), verbose) sorted_ids = sorted_ids[valid_inds] y = y[valid_inds] w = w[valid_inds] # Adding this assertion in to avoid ill-formed outputs. assert len(sorted_ids) == len(x) == len(y) == len(w) return sorted_ids, x, y.astype(float), w.astype(float) deepchem/data/tests/test_data_loader.py +8 −8 File changed.Preview size limit exceeded, changes collapsed. Show changes deepchem/feat/base_classes.py +1 −3 Original line number Diff line number Diff line Loading @@ -68,9 +68,7 @@ class Featurizer(object): mols : iterable RDKit Mol objects. """ if self.conformers and isinstance(mols, types.GeneratorType): mols = list(mols) features = [] for i, mol in enumerate(mols): if verbose and i % log_every_n == 0: Loading Loading
deepchem/data/__init__.py +6 −1 Original line number Diff line number Diff line Loading @@ -6,7 +6,12 @@ from __future__ import division from __future__ import unicode_literals # TODO(rbharath): Get rid of * import from deepchem.data.datasets import * from deepchem.data.datasets import pad_features from deepchem.data.datasets import pad_batch from deepchem.data.datasets import Dataset from deepchem.data.datasets import NumpyDataset from deepchem.data.datasets import DiskDataset from deepchem.data.supports import * from deepchem.data.data_loader import DataLoader from deepchem.data.data_loader import CSVLoader import deepchem.data.tests
deepchem/data/data_loader.py +177 −106 Original line number Diff line number Diff line Loading @@ -16,13 +16,126 @@ from rdkit import Chem import time import sys from deepchem.utils.save import log from deepchem.utils.save import save_to_disk from deepchem.utils.save import load_pickle_from_disk from deepchem.feat import Featurizer, ComplexFeaturizer from deepchem.utils.save import load_csv_files from deepchem.utils.save import load_sdf_files from deepchem.feat import UserDefinedFeaturizer from deepchem.data import DiskDataset from deepchem.utils.save import load_data from deepchem.utils.save import get_input_type def convert_df_to_numpy(df, tasks, id_field, verbose=False): """Transforms a dataframe containing deepchem input into numpy arrays""" n_samples = df.shape[0] n_tasks = len(tasks) time1 = time.time() y = np.hstack([ np.reshape(np.array(df[task].values), (n_samples, 1)) for task in tasks]) time2 = time.time() w = np.ones((n_samples, n_tasks)) missing = np.zeros_like(y).astype(int) feature_shape = None #for ind in range(n_samples): # for task in range(n_tasks): # if y[ind, task] == "": # missing[ind, task] = 1 #x_list = list(df[feature_type].values) #valid_inds = np.array([1 if elt.size > 0 else 0 for elt in x_list], dtype=bool) #x_list = [elt for (is_valid, elt) in zip(valid_inds, x_list) if is_valid] #x = np.squeeze(np.array(x_list)) sorted_ids = df[id_field].values # Set missing data to have weight zero for ind in range(n_samples): for task in range(n_tasks): if missing[ind, task]: y[ind, task] = 0. w[ind, task] = 0. #sorted_ids = sorted_ids[valid_inds] #y = y[valid_inds] #w = w[valid_inds] # Adding this assertion in to avoid ill-formed outputs. #assert len(sorted_ids) == len(x) == len(y) == len(w) assert len(sorted_ids) == len(y) == len(w) #return sorted_ids, x, y.astype(float), w.astype(float) return sorted_ids, y.astype(float), w.astype(float) #def write_dataframe(basename, df, data_dir, feature_type, tasks=None, # raw_data=None, mol_id_field="mol_id", # verbose=False): # """Writes data from dataframe to disk.""" # if featurizer is not None and tasks is not None: # feature_type = featurizer.__class__.__name__ # time1 = time.time() # ids, X, y, w = convert_df_to_numpy( # df, feature_type, tasks, mol_id_field) # time2 = time.time() # log("TIMING: convert_df_to_numpy took %0.3f s" % (time2-time1), verbose) # else: # ids, X, y, w = raw_data # assert X.shape[0] == y.shape[0] # assert y.shape == w.shape # assert len(ids) == X.shape[0] # return DiskDataset.write_data_to_disk( # data_dir, basename, tasks, X, y, w, ids) def featurize_smiles_df(df, featurizer, field, log_every_N=1000, verbose=True): """Featurize individual compounds in dataframe. Given a featurizer that operates on individual chemical compounds or macromolecules, compute & add features for that compound to the features dataframe """ sample_elems = df[field].tolist() features = [] for ind, elem in enumerate(sample_elems): mol = Chem.MolFromSmiles(elem) if ind % log_every_N == 0: log("Featurizing sample %d" % ind, verbose) features.append(featurizer.featurize([mol])) return np.array(features) def get_user_specified_features(df, featurizer, verbose=True): """Extract and merge user specified features. Merge features included in dataset provided by user into final features dataframe Three types of featurization here: 1) Molecule featurization -) Smiles string featurization -) Rdkit MOL featurization 2) Complex featurization -) PDB files for interacting molecules. 3) User specified featurizations. """ time1 = time.time() df[featurizer.feature_fields] = df[featurizer.feature_fields].apply(pd.to_numeric) X_shard = df.as_matrix(columns=featurizer.feature_fields) time2 = time.time() log("TIMING: user specified processing took %0.3f s" % (time2-time1), verbose) return X_shard def featurize_mol_df(df, featurizer, field, verbose=True, log_every_N=1000): """Featurize individual compounds in dataframe. Featurizes .sdf files, so the 3-D structure should be preserved so we use the rdkit "mol" object created from .sdf instead of smiles string. Some featurizers such as CoulombMatrix also require a 3-D structure. Featurizing from .sdf is currently the only way to perform CM feautization. """ sample_elems = df[field].tolist() features = [] for ind, mol in enumerate(sample_elems): if ind % log_every_N == 0: log("Featurizing sample %d" % ind, verbose) features.append(featurizer.featurize([mol])) return np.array(features) class DataLoader(object): """ Loading Loading @@ -57,110 +170,68 @@ class DataLoader(object): log("Loading raw samples now.", self.verbose) log("shard_size: %d" % shard_size, self.verbose) # Allow users to specify a single file for featurization if not isinstance(input_files, list): input_files = [input_files] if data_dir is not None: if not os.path.exists(data_dir): os.makedirs(data_dir) else: data_dir = tempfile.mkdtemp() if not len(input_files): return None input_type = get_input_type(input_files[0]) metadata_rows = [] for shard_num, elt in enumerate(load_data(input_files, shard_size)): time1 = time.time() metadata_row = self._featurize_shard(elt, data_dir, shard_num, input_type) if metadata_row is not None: metadata_rows.append(metadata_row) time2 = time.time() log("TIMING: shard %d took %0.3f s" % (shard_num, time2-time1), self.verbose) def shard_generator(): for shard_num, shard in enumerate(self.get_shards(input_files, shard_size)): time1 = time.time() dataset = DiskDataset(data_dir=data_dir, metadata_rows=metadata_rows, reload=True) X = self.featurize_shard(shard) ids, y, w = convert_df_to_numpy(shard, self.tasks, self.id_field) time2 = time.time() print("TIMING: dataset construction took %0.3f s" % (time2-time1), log("TIMING: featurizing shard %d took %0.3f s" % (shard_num, time2-time1), self.verbose) return dataset yield ids, X, y, w return DiskDataset(shard_generator(), data_dir, self.tasks) def _featurize_shard(self, df_shard, data_dir, shard_num, input_type): """Featurizes a shard of an input dataframe.""" field = self.mol_field if input_type == "sdf" else self.smiles_field field_type = "mol" if input_type == "sdf" else "smiles" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) 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, field_type=field_type) basename = "shard-%d" % shard_num time1 = time.time() metadata_row = DiskDataset.write_dataframe( basename, df_shard, data_dir=data_dir, featurizer=self.featurizer, tasks=self.tasks, mol_id_field=self.id_field) time2 = time.time() log("TIMING: writing metadata row took %0.3f s" % (time2-time1), self.verbose) return metadata_row def _featurize_mol(self, df, featurizer, parallel=True, field_type="mol", field=None): """Featurize individual compounds. def get_shards(self, input_files, shard_size): """Stub for children classes.""" raise NotImplementedError Given a featurizer that operates on individual chemical compounds or macromolecules, compute & add features for that compound to the features dataframe def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" raise NotImplementedError When featurizing a .sdf file, the 3-D structure should be preserved so we use the rdkit "mol" object created from .sdf instead of smiles string. Some featurizers such as CoulombMatrix also require a 3-D structure. Featurizing from .sdf is currently the only way to perform CM feautization. TODO(rbharath): Needs to be merged with _featurize_compounds class CSVLoader(DataLoader): """ assert field_type in ["mol", "smiles"] assert field is not None sample_elems = df[field].tolist() features = [] for ind, elem in enumerate(sample_elems): if field_type == "smiles": mol = Chem.MolFromSmiles(elem) else: mol = elem if ind % self.log_every_n == 0: log("Featurizing sample %d" % ind, self.verbose) features.append(featurizer.featurize([mol])) df[featurizer.__class__.__name__] = features Handles loading of CSV files. """ def get_shards(self, input_files, shard_size, verbose=True): """Defines a generator which returns data for each shard""" return load_csv_files(input_files, shard_size, verbose=verbose) def _add_user_specified_features(self, df, featurizer): """Merge user specified features. def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) return featurize_smiles_df(shard, self.featurizer, field=self.smiles_field) Merge features included in dataset provided by user into final features dataframe class UserCSVLoader(DataLoader): """ Handles loading of CSV files with user-defined featurizers. """ def get_shards(self, input_files, shard_size): """Defines a generator which returns data for each shard""" return load_csv_files(input_files, shard_size, verbose=verbose) Three types of featurization here: def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) assert isinstance(self.featurizer, UserDefinedFeaturizer) return get_user_specified_features(shard, self.featurizer) 1) Molecule featurization -) Smiles string featurization -) Rdkit MOL featurization 2) Complex featurization -) PDB files for interacting molecules. 3) User specified featurizations. class SDFLoader(DataLoader): """ time1 = time.time() df[featurizer.feature_fields] = df[featurizer.feature_fields].apply(pd.to_numeric) X_shard = df.as_matrix(columns=featurizer.feature_fields) df[featurizer.__class__.__name__] = [np.array(elt) for elt in X_shard.tolist()] time2 = time.time() log("TIMING: user specified processing took %0.3f s" % (time2-time1), self.verbose) Handles loading of SDF files. """ def get_shard(self, input_files, shard_size): """Defines a generator which returns data for each shard""" return load_sdf_files(input_files) def featurize_shard(self, shard): """Featurizes a shard of an input dataframe.""" log("Currently featurizing feature_type: %s" % self.featurizer.__class__.__name__, self.verbose) return featurize_mol_df(shard, self.featurizer, field=self.mol_field)
deepchem/data/datasets.py +85 −230 Original line number Diff line number Diff line Loading @@ -343,70 +343,43 @@ class NumpyDataset(Dataset): newx, newy, neww = fn(self._X, self._y, self._w) return NumpyDataset(newx, newy, neww, self._ids[:]) class DiskDataset(Dataset): """ A Dataset that is stored as a set of files on disk. """ def __init__(self, data_dir=None, tasks=[], metadata_rows=None, raw_data=None, reload=False, verbose=False): def __init__(self, shard_generator=None, data_dir=None, tasks=[], reload=False, verbose=True): """ Turns featurized dataframes into numpy files, writes them & metadata to disk. """ if data_dir is not None: if not os.path.exists(data_dir): os.makedirs(data_dir) else: data_dir = tempfile.mkdtemp() self.data_dir = data_dir self.verbose = verbose if not reload or not os.path.exists(self._get_metadata_filename()): if metadata_rows is not None: self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() elif raw_data is not None: metadata_rows = [] ids, X, y, w = raw_data metadata_rows.append( DiskDataset.write_data_to_disk( self.data_dir, "data", tasks, X, y, w, ids)) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() else: # Create an empty metadata dataframe to be filled at a later time basename = "metadata" metadata_rows = [DiskDataset.write_data_to_disk( self.data_dir, basename, tasks)] self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() else: log("Loading pre-existing metadata file.", self.verbose) if reload: log("Loading pre-existing dataset.", self.verbose) if os.path.exists(self._get_metadata_filename()): self.metadata_df = load_from_disk(self._get_metadata_filename()) else: raise ValueError("No metadata found.") return @staticmethod def write_dataframe(basename, df, data_dir, featurizer=None, tasks=None, raw_data=None, mol_id_field="mol_id", verbose=False): """Writes data from dataframe to disk.""" if featurizer is not None and tasks is not None: feature_type = featurizer.__class__.__name__ # TODO(rbharath): This is a hack. clean up. if not len(df): return None metadata_rows = [] time1 = time.time() ids, X, y, w = convert_df_to_numpy( df, feature_type, tasks, mol_id_field) for shard_num, (ids, X, y, w) in enumerate(shard_generator): basename = "shard-%d" % shard_num metadata_rows.append( DiskDataset.write_data_to_disk( self.data_dir, basename, tasks, X, y, w, ids)) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() time2 = time.time() log("TIMING: convert_df_to_numpy took %0.3f s" % (time2-time1), verbose) else: ids, X, y, w = raw_data assert X.shape[0] == y.shape[0] assert y.shape == w.shape assert len(ids) == X.shape[0] return DiskDataset.write_data_to_disk( data_dir, basename, tasks, X, y, w, ids) print("TIMING: dataset construction took %0.3f s" % (time2-time1), self.verbose) @staticmethod def construct_metadata(metadata_entries): Loading Loading @@ -457,7 +430,7 @@ class DiskDataset(Dataset): reshard_dir = tempfile.mkdtemp() new_metadata = [] # Write data in new shards ind = 0 def generator(): tasks = self.get_task_names() X_next = np.zeros((0,) + self.get_data_shape()) y_next = np.zeros((0,) + (len(tasks),)) Loading @@ -473,19 +446,10 @@ class DiskDataset(Dataset): y_batch, y_next = y_next[:shard_size], y_next[shard_size:] w_batch, w_next = w_next[:shard_size], w_next[shard_size:] ids_batch, ids_next = ids_next[:shard_size], ids_next[shard_size:] new_basename = "reshard-%d" % ind new_metadata.append(DiskDataset.write_data_to_disk( reshard_dir, new_basename, tasks, X_batch, y_batch, w_batch, ids_batch)) ind += 1 yield (ids_batch, X_batch, y_batch, w_batch) # Handle spillover from last shard new_basename = "reshard-%d" % ind new_metadata.append(DiskDataset.write_data_to_disk( reshard_dir, new_basename, tasks, X_next, y_next, w_next, ids_next)) ind += 1 # Get new metadata rows resharded_dataset = DiskDataset( data_dir=reshard_dir, tasks=tasks, metadata_rows=new_metadata, verbosity=self.verbosity) yield (ids_next, X_next, y_next, w_next) resharded_dataset = DiskDataset(generator(), data_dir=reshard_dir) shutil.rmtree(self.data_dir) shutil.move(reshard_dir, self.data_dir) self.metadata_df = resharded_dataset.metadata_df Loading Loading @@ -632,13 +596,12 @@ class DiskDataset(Dataset): out_dir = tempfile.mkdtemp() tasks = self.get_task_names() metadata_rows = [] def generator(): for shard_num, row in self.metadata_df.iterrows(): X, y, w, ids = self.get_shard(shard_num) newx, newy, neww = fn(X, y, w) basename = "dataset-%d" % shard_num metadata_rows.append(DiskDataset.write_data_to_disk( out_dir, basename, tasks, newx, newy, neww, ids)) return DiskDataset(data_dir=out_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=out_dir) @staticmethod def from_numpy(X, y, w=None, ids=None, tasks=None, data_dir=None): Loading Loading @@ -669,15 +632,11 @@ class DiskDataset(Dataset): os.makedirs(merge_dir) else: merge_dir = tempfile.mkdtemp() Xs, ys, ws, all_ids = [], [], [], [] metadata_rows = [] def generator(): for ind, dataset in enumerate(datasets): X, y, w, ids = (dataset.X, dataset.y, dataset.w, dataset.ids) basename = "dataset-%d" % ind tasks = dataset.get_task_names() metadata_rows.append( DiskDataset.write_data_to_disk(merge_dir, basename, tasks, X, y, w, ids)) return DiskDataset(data_dir=merge_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=merge_dir) def subset(self, shard_nums, subset_dir=None): """Creates a subset of the original dataset on disk.""" Loading @@ -688,14 +647,14 @@ class DiskDataset(Dataset): subset_dir = tempfile.mkdtemp() tasks = self.get_task_names() metadata_rows = [] def generator(): for shard_num, row in self.metadata_df.iterrows(): if shard_num not in shard_nums: continue X, y, w, ids = self.get_shard(shard_num) basename = "dataset-%d" % shard_num metadata_rows.append(DiskDataset.write_data_to_disk( subset_dir, basename, tasks, X, y, w, ids)) return DiskDataset(data_dir=subset_dir, metadata_rows=metadata_rows) yield (ids, X, y, w) return DiskDataset(generator(), data_dir=subset_dir) def sparse_shuffle(self): """Shuffling that exploits data sparsity to shuffle large datasets. Loading Loading @@ -739,51 +698,6 @@ class DiskDataset(Dataset): self.verbose) ######################################################### TIMING def shuffle(self, iterations=1): """Shuffles this dataset on disk to have random order.""" #np.random.seed(9452) for _ in range(iterations): metadata_rows = [] tasks = self.get_task_names() # Shuffle the arrays corresponding to each row in metadata_df n_rows = len(self.metadata_df.index) len_data = len(self) print("ABOUT TO SHUFFLE DATA ONCE") for i in range(n_rows): # Select random row to swap with j = np.random.randint(n_rows) row_i, row_j = self.metadata_df.iloc[i], self.metadata_df.iloc[j] metadata_rows.append(row_i) # Useful to avoid edge cases, but perhaps there's a better solution if i == j: continue basename_i, basename_j = row_i["basename"], row_j["basename"] X_i, y_i, w_i, ids_i = self.get_shard(i) X_j, y_j, w_j, ids_j = self.get_shard(j) n_i, n_j = X_i.shape[0], X_j.shape[0] # Join two shards and shuffle them at random. X = np.vstack([X_i, X_j]) y = np.vstack([y_i, y_j]) w = np.vstack([w_i, w_j]) ids = np.concatenate([ids_i, ids_j]) permutation = np.random.permutation(n_i + n_j) X, y, w, ids = (X[permutation], y[permutation], w[permutation], ids[permutation]) X_i, y_i, w_i, ids_i = X[:n_i], y[:n_i], w[:n_i], ids[:n_i] X_j, y_j, w_j, ids_j = X[n_i:], y[n_i:], w[n_i:], ids[n_i:] DiskDataset.write_data_to_disk( self.data_dir, basename_i, tasks, X_i, y_i, w_i, ids_i) DiskDataset.write_data_to_disk( self.data_dir, basename_j, tasks, X_j, y_j, w_j, ids_j) assert len(self) == len_data # Now shuffle order of rows in metadata_df random.shuffle(metadata_rows) self.metadata_df = DiskDataset.construct_metadata(metadata_rows) self.save_to_disk() def shuffle_each_shard(self): """Shuffles elements within each shard of the datset.""" tasks = self.get_task_names() Loading Loading @@ -827,8 +741,6 @@ class DiskDataset(Dataset): tasks = self.get_task_names() DiskDataset.write_data_to_disk(self.data_dir, basename, tasks, X, y, w, ids) # TODO(rbharath): This change for general object types seems a little # kludgey. Is there a more principled approach to support general objects? def select(self, indices, select_dir=None): """Creates a new dataset from a selection of indices from self. Loading @@ -846,12 +758,12 @@ class DiskDataset(Dataset): select_dir = tempfile.mkdtemp() # Handle edge case with empty indices if not len(indices): return DiskDataset( data_dir=select_dir, metadata_rows=[]) return DiskDataset([], data_dir=select_dir) indices = np.array(sorted(indices)).astype(int) count, indices_count = 0, 0 metadata_rows = [] tasks = self.get_task_names() def generator(): for shard_num, (X, y, w, ids) in enumerate(self.itershards()): shard_len = len(X) # Find indices which rest in this shard Loading @@ -866,17 +778,14 @@ class DiskDataset(Dataset): y_sel = y[shard_inds] w_sel = w[shard_inds] ids_sel = ids[shard_inds] basename = "dataset-%d" % shard_num metadata_rows.append( DiskDataset.write_data_to_disk( select_dir, basename, tasks, X_sel, y_sel, w_sel, ids_sel)) yield (ids_sel, X_sel, y_sel, w_sel) # Updating counts indices_count += num_shard_elts count += shard_len # Break when all indices have been used up already if indices_count >= len(indices): break return DiskDataset(data_dir=select_dir, metadata_rows=metadata_rows) return return DiskDataset(generator(), data_dir=select_dir) @property def ids(self): Loading Loading @@ -955,57 +864,3 @@ class DiskDataset(Dataset): def get_label_stds(self): """Return pandas series of label stds.""" return self.metadata_df["y_stds"] # The following are all associated with Dataset, but are separate functions to # make it easy to use multiprocessing. def convert_df_to_numpy(df, feature_type, tasks, mol_id_field, verbose=False): """Transforms a dataframe containing deepchem input into numpy arrays""" if feature_type not in df.keys(): raise ValueError( "Featurized data does not support requested feature_type %s." % feature_type) # perform common train/test split across all tasks n_samples = df.shape[0] n_tasks = len(tasks) time1 = time.time() y = np.hstack([ np.reshape(np.array(df[task].values), (n_samples, 1)) for task in tasks]) time2 = time.time() log("TIMING: convert_df_to_numpy y computation took %0.3f s" % (time2-time1), verbose) w = np.ones((n_samples, n_tasks)) missing = np.zeros_like(y).astype(int) feature_shape = None time1 = time.time() for ind in range(n_samples): for task in range(n_tasks): if y[ind, task] == "": missing[ind, task] = 1 x_list = list(df[feature_type].values) valid_inds = np.array([1 if elt.size > 0 else 0 for elt in x_list], dtype=bool) x_list = [elt for (is_valid, elt) in zip(valid_inds, x_list) if is_valid] x = np.squeeze(np.array(x_list)) time2 = time.time() log("TIMING: convert_df_to_numpy x computation took %0.3f s" % (time2-time1), verbose) sorted_ids = df[mol_id_field].values # Set missing data to have weight zero time1 = time.time() for ind in range(n_samples): for task in range(n_tasks): if missing[ind, task]: y[ind, task] = 0. w[ind, task] = 0. time2 = time.time() log("TIMING: convert_df_to_numpy missing elts computation took %0.3f s" % (time2-time1), verbose) sorted_ids = sorted_ids[valid_inds] y = y[valid_inds] w = w[valid_inds] # Adding this assertion in to avoid ill-formed outputs. assert len(sorted_ids) == len(x) == len(y) == len(w) return sorted_ids, x, y.astype(float), w.astype(float)
deepchem/data/tests/test_data_loader.py +8 −8 File changed.Preview size limit exceeded, changes collapsed. Show changes
deepchem/feat/base_classes.py +1 −3 Original line number Diff line number Diff line Loading @@ -68,9 +68,7 @@ class Featurizer(object): mols : iterable RDKit Mol objects. """ if self.conformers and isinstance(mols, types.GeneratorType): mols = list(mols) features = [] for i, mol in enumerate(mols): if verbose and i % log_every_n == 0: Loading
