fairness_instru.py
def pd_to_dataflow_graph(pipeline_func, log_list, parent_vertices=[]):
    """
    Function translating pandas operations to DAGs.

    args:
        pipeline_func: raw pipeline codes. String.
        log_list: log_list storing all operation identifiers. List.
        parent_vertices: parent nodes. default to be None. No operations before pandas. List.

    return:
        graph: list of nodes in the eligible format from graphviz.
        previous: last node used for parent node of sklearn part.
        log_list: log_list storing all operation identifiers.
    """
    executable_list = find_pd_lines(pipeline_func)
    graph = []
    previous = []

    for line in executable_list:
        line.replace('{','').replace('}', '')
        if 'inplace' in line and '#' not in line:
            log_list.append(line_cleansing(line))

            df_name = line.split('.')[0]
            func_name = line.split('.')[1].split('(')[0].strip()
            col_effect = line.split('[')[1].split(']')[0].strip()
            if len(previous) > 1:
                for node in previous:
                    if node.name == df_name:
                        vertex = DataFlowVertex([node], df_name+'_drop', func_name+' '+col_effect, col_effect)
                        previous.append(vertex)
                        previous.remove(node)
            else:
                vertex = DataFlowVertex(previous, df_name+'_drop', func_name+' '+col_effect, col_effect)
                previous = [vertex]
        else:
            var_name = line.split('=')[0].strip()

            # match ".func_name(...)"
            pd_func = re.search('\.\s*([_a-z]*)\s*\(',line)
            if pd_func:
                func_name = pd_func.group(1)
                params = re.search('\(([^\)]*)\)',line)  #"(...)"

                if params:
                    params = params.group(1).strip()

                    if func_name == 'read_csv': #df = pd.read_csv(path)
                        vertex = DataFlowVertex(parent_vertices,var_name, func_name, params)
                        previous.append(vertex)
                        log_list.append(line_cleansing(line))
                    elif func_name in ['join','merge','concat']:
                        log_list.append(line_cleansing(line))
                        if func_name == 'concat': #df_new = pd.concat([df1,df2],keys=[])
                            df_names = [item.strip() for item in params.split(']')[0].strip().strip('[]').split(',')]

                        else: # df_new = df1.join/merge(df2,on='...',how='...')
                            df_names = [line.split('=')[1].strip().split('.')[0], params.split(',')[0].strip()]
                        parent_vertices = search_vertex_by_names(df_names, graph) #search in graph by df_names
                        vertex = DataFlowVertex(parent_vertices, var_name, func_name, params) #TODO vertex name?
                        previous = [vertex] + list(set(previous) - set(parent_vertices))
                    elif 'lambda' in params:
                        log_list.append(line_cleansing(line))
                        cols = var_name.split('[')[1].split(']')[0].strip()
                        vertex = DataFlowVertex(previous, func_name+' '+cols, func_name, params)
                        previous = [vertex]
                    elif '[' in var_name:
                        log_list.append(line_cleansing(line))
                        cols = var_name.split('[')[1].split(']')[0].strip()
                        vertex = DataFlowVertex(previous, func_name+' '+cols+' '+params, func_name, params)
                        previous = [vertex]
                    else:
                        log_list.append(line_cleansing(line))
                        vertex = DataFlowVertex(previous, func_name+' '+params, func_name, params)
                        previous = [vertex]


            # filter operation: "df[[cols]]", "df[condition]","df.loc[]","df.iloc[]"
            else:
                if '[[' in line:
                    is_filter = re.search('\[([^\]]*)\]',line) #"[...]"
                else:
                    is_filter = re.search('\(([^\)]*)\)',line) #"[...]"
                if is_filter:
                    log_list.append(line_cleansing(line))
                    filter_cond = is_filter.group(1).strip('[').strip(']')
                    vertex = DataFlowVertex(previous, 'select '+filter_cond, 'filter', filter_cond)
                    previous = [vertex]

        graph.append(vertex)

    return graph, previous, log_list

    Function translating pandas operations to DAGs.

    args:
        pipeline_func: raw pipeline codes. String.
        log_list: log_list storing all operation identifiers. List.
        parent_vertices: parent nodes. default to be None. No operations before pandas. List.

    return:
        graph: list of nodes in the eligible format from graphviz.
        previous: last node used for parent node of sklearn part.
        log_list: log_list storing all operation identifiers.
    
Function translating pandas operations to DAGs.
args:
pipeline_func: raw pipeline codes. String.

list of nodes in the eligible format from graphviz.
previous: last node used for parent node of sklearn part.
Function translating pandas operations to DAGs . args : pipeline_func : raw pipeline codes . String . list of nodes in the eligible format from graphviz . previous : last node used for parent node of sklearn part .
def pd_to_dataflow_graph(pipeline_func, log_list, parent_vertices=[]):
    executable_list = find_pd_lines(pipeline_func)
    graph = []
    previous = []
    for line in executable_list:
        line.replace('{','').replace('}', '')
        if 'inplace' in line and '#' not in line:
            log_list.append(line_cleansing(line))
            df_name = line.split('.')[0]
            func_name = line.split('.')[1].split('(')[0].strip()
            col_effect = line.split('[')[1].split(']')[0].strip()
            if len(previous) > 1:
                for node in previous:
                    if node.name == df_name:
                        vertex = DataFlowVertex([node], df_name+'_drop', func_name+' '+col_effect, col_effect)
                        previous.append(vertex)
                        previous.remove(node)
            else:
                vertex = DataFlowVertex(previous, df_name+'_drop', func_name+' '+col_effect, col_effect)
                previous = [vertex]
        else:
            var_name = line.split('=')[0].strip()
            pd_func = re.search('\.\s*([_a-z]*)\s*\(',line)
            if pd_func:
                func_name = pd_func.group(1)
                params = re.search('\(([^\)]*)\)',line)  
                if params:
                    params = params.group(1).strip()
                    if func_name == 'read_csv': 
                        vertex = DataFlowVertex(parent_vertices,var_name, func_name, params)
                        previous.append(vertex)
                        log_list.append(line_cleansing(line))
                    elif func_name in ['join','merge','concat']:
                        log_list.append(line_cleansing(line))
                        if func_name == 'concat': 
                            df_names = [item.strip() for item in params.split(']')[0].strip().strip('[]').split(',')]
                        else: 
                            df_names = [line.split('=')[1].strip().split('.')[0], params.split(',')[0].strip()]
                        parent_vertices = search_vertex_by_names(df_names, graph) 
                        vertex = DataFlowVertex(parent_vertices, var_name, func_name, params) 
                        previous = [vertex] + list(set(previous) - set(parent_vertices))
                    elif 'lambda' in params:
                        log_list.append(line_cleansing(line))
                        cols = var_name.split('[')[1].split(']')[0].strip()
                        vertex = DataFlowVertex(previous, func_name+' '+cols, func_name, params)
                        previous = [vertex]
                    elif '[' in var_name:
                        log_list.append(line_cleansing(line))
                        cols = var_name.split('[')[1].split(']')[0].strip()
                        vertex = DataFlowVertex(previous, func_name+' '+cols+' '+params, func_name, params)
                        previous = [vertex]
                    else:
                        log_list.append(line_cleansing(line))
                        vertex = DataFlowVertex(previous, func_name+' '+params, func_name, params)
                        previous = [vertex]
            else:
                if '[[' in line:
                    is_filter = re.search('\[([^\]]*)\]',line) 
                else:
                    is_filter = re.search('\(([^\)]*)\)',line) 
                if is_filter:
                    log_list.append(line_cleansing(line))
                    filter_cond = is_filter.group(1).strip('[').strip(']')
                    vertex = DataFlowVertex(previous, 'select '+filter_cond, 'filter', filter_cond)
                    previous = [vertex]
        graph.append(vertex)
    return graph, previous, log_list
def pd_to_dataflow_graph ( pipeline_func , log_list , parent_vertices = [ ] ) : executable_list = find_pd_lines ( pipeline_func ) graph = [ ] previous = [ ] for line in executable_list : line . replace ( '{' , '' ) . replace ( '}' , '' ) if 'inplace' in line and '#' not in line : log_list . append ( line_cleansing ( line ) ) df_name = line . split ( '.' ) [ 0 ] func_name = line . split ( '.' ) [ 1 ] . split ( '(' ) [ 0 ] . strip ( ) col_effect = line . split ( '[' ) [ 1 ] . split ( ']' ) [ 0 ] . strip ( ) if len ( previous ) > 1 : for node in previous : if node . name == df_name : vertex = DataFlowVertex ( [ node ] , df_name + '_drop' , func_name + ' ' + col_effect , col_effect ) previous . append ( vertex ) previous . remove ( node ) else : vertex = DataFlowVertex ( previous , df_name + '_drop' , func_name + ' ' + col_effect , col_effect ) previous = [ vertex ] else : var_name = line . split ( '=' ) [ 0 ] . strip ( ) pd_func = re . search ( '\.\s*([_a-z]*)\s*\(' , line ) if pd_func : func_name = pd_func . group ( 1 ) params = re . search ( '\(([^\)]*)\)' , line ) if params : params = params . group ( 1 ) . strip ( ) if func_name == 'read_csv' : vertex = DataFlowVertex ( parent_vertices , var_name , func_name , params ) previous . append ( vertex ) log_list . append ( line_cleansing ( line ) ) elif func_name in [ 'join' , 'merge' , 'concat' ] : log_list . append ( line_cleansing ( line ) ) if func_name == 'concat' : df_names = [ item . strip ( ) for item in params . split ( ']' ) [ 0 ] . strip ( ) . strip ( '[]' ) . split ( ',' ) ] else : df_names = [ line . split ( '=' ) [ 1 ] . strip ( ) . split ( '.' ) [ 0 ] , params . split ( ',' ) [ 0 ] . strip ( ) ] parent_vertices = search_vertex_by_names ( df_names , graph ) vertex = DataFlowVertex ( parent_vertices , var_name , func_name , params ) previous = [ vertex ] + list ( set ( previous ) - set ( parent_vertices ) ) elif 'lambda' in params : log_list . append ( line_cleansing ( line ) ) cols = var_name . split ( '[' ) [ 1 ] . split ( ']' ) [ 0 ] . strip ( ) vertex = DataFlowVertex ( previous , func_name + ' ' + cols , func_name , params ) previous = [ vertex ] elif '[' in var_name : log_list . append ( line_cleansing ( line ) ) cols = var_name . split ( '[' ) [ 1 ] . split ( ']' ) [ 0 ] . strip ( ) vertex = DataFlowVertex ( previous , func_name + ' ' + cols + ' ' + params , func_name , params ) previous = [ vertex ] else : log_list . append ( line_cleansing ( line ) ) vertex = DataFlowVertex ( previous , func_name + ' ' + params , func_name , params ) previous = [ vertex ] else : if '[[' in line : is_filter = re . search ( '\[([^\]]*)\]' , line ) else : is_filter = re . search ( '\(([^\)]*)\)' , line ) if is_filter : log_list . append ( line_cleansing ( line ) ) filter_cond = is_filter . group ( 1 ) . strip ( '[' ) . strip ( ']' ) vertex = DataFlowVertex ( previous , 'select ' + filter_cond , 'filter' , filter_cond ) previous = [ vertex ] graph . append ( vertex ) return graph , previous , log_list
Function translating pandas operations to DAGs.
Function translating pandas operations to DAGs .
"""
    Function translating pandas operations to DAGs.

    args:
        pipeline_func: raw pipeline codes. String.
        log_list: log_list storing all operation identifiers. List.
        parent_vertices: parent nodes. default to be None. No operations before pandas. List.

    return:
        graph: list of nodes in the eligible format from graphviz.
        previous: last node used for parent node of sklearn part.
        log_list: log_list storing all operation identifiers.
    """ # match ".func_name(...)" #"(...)" #df = pd.read_csv(path) #df_new = pd.concat([df1,df2],keys=[]) # df_new = df1.join/merge(df2,on='...',how='...') #search in graph by df_names #TODO vertex name? # filter operation: "df[[cols]]", "df[condition]","df.loc[]","df.iloc[]" #"[...]" #"[...]"
pipeline_func
log_list
parent_vertices