coprinet/pricePrediction/selectFromRawData/selectFromRawDataNonVirtualAll_base.py

import os
import dask.dataframe as dd
import pandas as pd

from pricePrediction import config
from pricePrediction.ArgParser_base import ArgParseable
from pricePrediction.config import TEST_SIZE


class RawDataSelector(ArgParseable):
    COLNAMES = "Mcule ID,SMILES,price 1 (USD),amount 1 (mg),delivery time 1 (w.days),available amount 1 (mg)".split(",")

    DESIRED_PARAMS_TO_ASK= ['full_dataset_fname', 'bb_fname', 'computed_datadir', 'test_size', 'nrows']
    def __init__(self, full_dataset_fname: str, bb_fname: str= config.BUILDING_BLOCKS_FNAME,
                 computed_datadir: str= config.DATASET_DIRNAME, test_size:int=TEST_SIZE, nrows:int=None):
        '''

        :param str full_dataset_fname: The fname for the whole Mcule csv file (csv.gz)
        :param str bb_fname: The fname for the building blocks csv file (csv.gz)
        :param str computed_datadir: The directory where the selected records will be saved
        :param int test_size: The number of entries to include in the test set
        :param int nrows: The number of rows to process in the full_dataset_fname. If None, process all rows
        '''
        self.full_dataset_fname = full_dataset_fname
        self.bb_fname = bb_fname
        self.computed_datadir = computed_datadir
        self.test_size = test_size
        self.nrows = nrows

    def operate_data(self, data):
        raise NotImplementedError()

    def compute(self):
        if self.full_dataset_fname.endswith(".gz"):
            data = pd.read_csv(self.full_dataset_fname, usecols=self.COLNAMES, nrows=self.nrows)
            data = dd.from_pandas(data, npartitions=30)
        else:
            data = dd.read_csv(self.full_dataset_fname, usecols=self.COLNAMES,blocksize=None)
            if self.nrows is not None:
                data = data.head(n=self.nrows, compute=False)

        data = self.operate_data(data)
        self.saveDataSplits(data)


    def saveDataSplits(self, data, computed_datadir=None):
        if computed_datadir is None:
            computed_datadir = self.computed_datadir
        print("Saving results to %s"%computed_datadir)
        if isinstance(data, dd.DataFrame):
            n_elems = data.shape[0].compute()
        print("Total size: ", n_elems)
        test_frac = self.test_size/n_elems
        data_train, data_test = data.random_split([1-test_frac, test_frac])

        del data

        data_test = data_test.repartition(npartitions= int(n_elems*test_frac//self.test_size)+1)
        data_test.to_csv(os.path.join(computed_datadir, "mcule_full_test.csv_split_*.csv"), index=False)
        del data_test

        n_elems = data_train.shape[0].compute()
        val_frac = self.test_size/n_elems
        data_train, data_val = data_train.random_split([1-val_frac, val_frac])
        data_val= data_val.repartition(npartitions= int(n_elems*val_frac//self.test_size)+1)

        data_val.to_csv(os.path.join(computed_datadir, "mcule_full_val.csv_split_*.csv"), index=False)
        del data_val

        data_train= data_train.repartition(npartitions= int(n_elems*(1-val_frac)//self.test_size)+1)
        data_train.to_csv(os.path.join(computed_datadir, "mcule_full_train.csv_split_*.csv"), index=False)
        with open(os.path.join(computed_datadir,"params.txt"), "w") as f:
            f.write(str(type(self))+"\n" )
            f.write(str(self.full_dataset_fname)+"\n" )
            f.write(str(self.bb_fname)+"\n" )
            f.write(str(self.computed_datadir))