<commit_before>'''
a class providing convenient default functions for processing
'''

import os
import sys
import pandas as pd
from datetime import datetime
import traceback
import cloudpickle


class rt1_processing_config(object):
    def __init__(self, save_path=None, dumpfolder=None,
                 error_dumpfolder=None, finalout_name=None):

        self.save_path = save_path
        self.dumpfolder = dumpfolder
        self.error_dumpfolder = error_dumpfolder
        self.finalout_name = finalout_name


    def get_names_ids(self, reader_arg):
        '''
        A function that returns the file-name based on the passed reader_args

        - the filenames are generated from the reader-argument `'gpi'`

        Parameters
        ----------
        reader_arg : dict
            the arguments passed to the reader function.

        Returns
        -------
        filename : str
            the file-name that will be used to save the fit dump-files in case the
            processing was successful
        error_filename : str
            the file-name that will be used to save the error files in case an
            error occured
        '''

        # the ID used for indexing the processed sites
        feature_id = reader_arg['gpi']

        # the filename of the dump-file
        filename = f"{feature_id}.dump"

        # the filename of the error-dumpfile
        error_filename = f"{feature_id}_error.dump"

        return feature_id, filename, error_filename


    def preprocess(self, reader_arg):
        '''
        a function that is called PRIOR to processing that does the following:
            - create the folder-structure if it does not yet exist
            - check if a dumpfile of the site already exists and if yes,
              raise a 'rt1_file_already_exists' error

        Parameters
        ----------
        reader_arg : dict
            the arguments passed to the reader.

        '''
        if self.save_path is not None and self.dumpfolder is not None:
            # generate "save_path" directory if it does not exist
            if not os.path.exists(self.save_path):
                print(self.save_path, 'does not exist... creating directory')
                os.mkdir(self.save_path)
            # generate "dumpfolder" directory if it does not exist
            dumppath = os.path.join(self.save_path, self.dumpfolder)
            if not os.path.exists(dumppath):
                print(dumppath, 'does not exist... creating directory')
                os.mkdir(dumppath)

            # obtain the path where the dump-file would be stored
            _, filename, _ = self.get_names_ids(reader_arg)
            dumppath = os.path.join(self.save_path, self.dumpfolder, filename)

            # raise a file already exists error
            if os.path.exists(dumppath):
                raise Exception('rt1_file_already_exists')


    def postprocess(self, fit, reader_arg):
        '''
        A function that is called AFTER processing of each site:

        - a pandas.DataFrame with the obtained parameters is returned.
          The columns are multiindexes corresponding to::

              columns = [feature_id,
                         [param_1, param_2, ...]]

        - in case "save_path" is provided:
            - a dump of the fit object will be stored in the folder
            - if the file already exists, a 'rt1_file_already_exists' error
              will be raised


        Parameters
        ----------
        fit: rt1.rtfits.Fits object
            The fits objec.
        reader_arg: dict
            the arguments passed to the reader function.

        Returns
        -------
        df: pandas.DataFrame
            a pandas dataframe containing the fitted parameterss.

        '''

        # set filenames
        feature_id, fname, error_fname = self.get_names_ids(reader_arg)

        if self.save_path is not None and self.dumpfolder is not None:
            dumppath = os.path.join(self.save_path, self.dumpfolder, fname)

            # if no dump exists, dump it, else load the existing dump
            if not os.path.exists(dumppath):
                fit.dump(dumppath, mini=True)
                print(datetime.now().strftime('%d-%b-%Y %H:%M:%S'),
                      'finished', feature_id)

        # get resulting parameter DataFrame
        df = fit.res_df
        # add the feature_id as first column-level
        df.columns = pd.MultiIndex.from_product([[feature_id], df.columns])
        df.columns.names = ['feature_id', 'param']

        # flush stdout to see output of child-processes
        sys.stdout.flush()

        return df


    def finaloutput(self, res, hdf_key=None):
        '''
        A function that is called after ALL sites are processed:

        - first, the obtained parameter-dataframes returned by the
          `postprocess()` function are concatenated
            - if `"save_path"` is defined, the resulting dataframe will be
              saved (or appended) to a hdf-store.
              the used key will be either the value of `"hdf_key"` or
              `"dumpfolder`" or in case both are `None`, the key `"result`"
              will be used
            - if `"save_path"` is None, the DataFrame of the concatenated
              results will be returned

        Parameters
        ----------
        res: list
            A list of return-values from the "postprocess()" function.
        save_path: str, optional
            The path where the finalout-file will be stored.
            The default is None.
        hdf_key: str, optional
            The key used in the hdf-file. If None, the dumpfolder name will
            be used. If dumpfolder is not provided, 'result' is used.
            The default is None.

        Returns
        -------
        res: pandas.DataFrame
             the concatenated results (ONLY if `"save_path"` is `None`)
        '''

        # concatenate the results
        res = pd.concat([i for i in res if i is not None], axis=1)

        if self.save_path is None or self.finalout_name is None:
            print('both save_path and finalout_name must be specified... ',
                  'results have NOT been saved!')
            return res
        else:

            if hdf_key is None:
                if self.dumpfolder is not None:
                    hdf_key = str(self.dumpfolder)
                else:
                    hdf_key = 'result'

            # create (or append) results to a HDF-store
            with pd.HDFStore(os.path.join(self.save_path, self.finalout_name),
                             'a') as hdfstore:
                res.to_hdf(hdfstore, key=hdf_key)

        # flush stdout to see output of child-processes
        sys.stdout.flush()


    def exceptfunc(self, ex, reader_arg):
        '''
        a error-catch function that handles the following errors:

        - 'rt1_skip'
            - exceptions are ignored and the next site is processed
        - 'rt1_data_error'
            - exceptions are ignored and the next site is processed
        - 'rt1_file_already_exists'
            - the already existing dump-file is loaded, the postprocess()
            function is applied and the result is returned
        - for any other exceptions:
            - if 'save_path' and 'error_dumpfolder' are specified
              a dump of the exception-message is saved and the exception
              is ignored. otherwise the exception will be raised.
<commit_msg>use pd.to_hdf instead of HDF_store directly<commit_after>'''
a class providing convenient default functions for processing
'''

import os
import sys
import pandas as pd
from datetime import datetime
import traceback
import cloudpickle


class rt1_processing_config(object):
    def __init__(self, save_path=None, dumpfolder=None,
                 error_dumpfolder=None, finalout_name=None):

        self.save_path = save_path
        self.dumpfolder = dumpfolder
        self.error_dumpfolder = error_dumpfolder
        self.finalout_name = finalout_name


    def get_names_ids(self, reader_arg):
        '''
        A function that returns the file-name based on the passed reader_args

        - the filenames are generated from the reader-argument `'gpi'`

        Parameters
        ----------
        reader_arg : dict
            the arguments passed to the reader function.

        Returns
        -------
        filename : str
            the file-name that will be used to save the fit dump-files in case the
            processing was successful
        error_filename : str
            the file-name that will be used to save the error files in case an
            error occured
        '''

        # the ID used for indexing the processed sites
        feature_id = reader_arg['gpi']

        # the filename of the dump-file
        filename = f"{feature_id}.dump"

        # the filename of the error-dumpfile
        error_filename = f"{feature_id}_error.dump"

        return feature_id, filename, error_filename


    def preprocess(self, reader_arg):
        '''
        a function that is called PRIOR to processing that does the following:
            - create the folder-structure if it does not yet exist
            - check if a dumpfile of the site already exists and if yes,
              raise a 'rt1_file_already_exists' error

        Parameters
        ----------
        reader_arg : dict
            the arguments passed to the reader.

        '''
        if self.save_path is not None and self.dumpfolder is not None:
            # generate "save_path" directory if it does not exist
            if not os.path.exists(self.save_path):
                print(self.save_path, 'does not exist... creating directory')
                os.mkdir(self.save_path)
            # generate "dumpfolder" directory if it does not exist
            dumppath = os.path.join(self.save_path, self.dumpfolder)
            if not os.path.exists(dumppath):
                print(dumppath, 'does not exist... creating directory')
                os.mkdir(dumppath)

            # obtain the path where the dump-file would be stored
            _, filename, _ = self.get_names_ids(reader_arg)
            dumppath = os.path.join(self.save_path, self.dumpfolder, filename)

            # raise a file already exists error
            if os.path.exists(dumppath):
                raise Exception('rt1_file_already_exists')


    def postprocess(self, fit, reader_arg):
        '''
        A function that is called AFTER processing of each site:

        - a pandas.DataFrame with the obtained parameters is returned.
          The columns are multiindexes corresponding to::

              columns = [feature_id,
                         [param_1, param_2, ...]]

        - in case "save_path" is provided:
            - a dump of the fit object will be stored in the folder
            - if the file already exists, a 'rt1_file_already_exists' error
              will be raised


        Parameters
        ----------
        fit: rt1.rtfits.Fits object
            The fits objec.
        reader_arg: dict
            the arguments passed to the reader function.

        Returns
        -------
        df: pandas.DataFrame
            a pandas dataframe containing the fitted parameterss.

        '''

        # set filenames
        feature_id, fname, error_fname = self.get_names_ids(reader_arg)

        if self.save_path is not None and self.dumpfolder is not None:
            dumppath = os.path.join(self.save_path, self.dumpfolder, fname)

            # if no dump exists, dump it, else load the existing dump
            if not os.path.exists(dumppath):
                fit.dump(dumppath, mini=True)
                print(datetime.now().strftime('%d-%b-%Y %H:%M:%S'),
                      'finished', feature_id)

        # get resulting parameter DataFrame
        df = fit.res_df
        # add the feature_id as first column-level
        df.columns = pd.MultiIndex.from_product([[feature_id], df.columns])
        df.columns.names = ['feature_id', 'param']

        # flush stdout to see output of child-processes
        sys.stdout.flush()

        return df


    def finaloutput(self, res, hdf_key=None):
        '''
        A function that is called after ALL sites are processed:

        - first, the obtained parameter-dataframes returned by the
          `postprocess()` function are concatenated
            - if `"save_path"` is defined, the resulting dataframe will be
              saved (or appended) to a hdf-store.
              the used key will be either the value of `"hdf_key"` or
              `"dumpfolder`" or in case both are `None`, the key `"result`"
              will be used
            - if `"save_path"` is None, the DataFrame of the concatenated
              results will be returned

        Parameters
        ----------
        res: list
            A list of return-values from the "postprocess()" function.
        save_path: str, optional
            The path where the finalout-file will be stored.
            The default is None.
        hdf_key: str, optional
            The key used in the hdf-file. If None, the dumpfolder name will
            be used. If dumpfolder is not provided, 'result' is used.
            The default is None.

        Returns
        -------
        res: pandas.DataFrame
             the concatenated results (ONLY if `"save_path"` is `None`)
        '''

        # concatenate the results
        res = pd.concat([i for i in res if i is not None], axis=1)

        if self.save_path is None or self.finalout_name is None:
            print('both save_path and finalout_name must be specified... ',
                  'results have NOT been saved!')
            return res
        else:

            if hdf_key is None:
                if self.dumpfolder is not None:
                    hdf_key = str(self.dumpfolder)
                else:
                    hdf_key = 'result'

            # create (or append) results to a HDF-store
            res.to_hdf(os.path.join(self.save_path, self.finalout_name),
                       key=hdf_key, format='table', complevel=5)

        # flush stdout to see output of child-processes
        sys.stdout.flush()


    def exceptfunc(self, ex, reader_arg):
        '''
        a error-catch function that handles the following errors:

        - 'rt1_skip'
            - exceptions are ignored and the next site is processed
        - 'rt1_data_error'
            - exceptions are ignored and the next site is processed
        - 'rt1_file_already_exists'
            - the already existing dump-file is loaded, the postprocess()
            function is applied and the result is returned
        - for any other exceptions:
            - if 'save_path' and 'error_dumpfolder' are specified
              a dump of the exception-message is saved and the exception
              is ignored. otherwise the exception will be raised.

{'fraction_non_alphanumeric': 0.07444533547179899, 'fraction_numerical': 0.001537022186581128, 'mean_word_length': 1.8061128820551284, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '205648', 'n_tokens_mistral': 3967, 'n_tokens_neox': 3802, 'n_words': 1424}