flowcraft/generator/pipeline_parser.py
def get_source_lane(fork_process, pipeline_list):
    """Returns the lane of the last process that matches fork_process

    Parameters
    ----------
    fork_process : list
        List of processes before the fork.
    pipeline_list : list
        List with the pipeline connection dictionaries.

    Returns
    -------
    int
        Lane of the last process that matches fork_process
    """

    fork_source = fork_process[-1]
    fork_sig = [x for x in fork_process if x != "__init__"]

    for position, p in enumerate(pipeline_list[::-1]):

        if p["output"]["process"] == fork_source:

            lane = p["output"]["lane"]
            logger.debug("Possible source match found in position {} in lane"
                         " {}".format(position, lane))
            lane_sequence = [x["output"]["process"] for x in pipeline_list
                             if x["output"]["lane"] == lane]
            logger.debug("Testing lane sequence '{}' against fork signature"
                         " '{}'".format(lane_sequence, fork_sig))
            if lane_sequence == fork_sig:
                return p["output"]["lane"]

    return 0
def get_source_lane(fork_process, pipeline_list):
    fork_source = fork_process[-1]
    fork_sig = [x for x in fork_process if x != "__init__"]

    for position, p in enumerate(pipeline_list[::-1]):

        if p["output"]["process"] == fork_source:

            lane = p["output"]["lane"]
            logger.debug("Possible source match found in position {} in lane"
                         " {}".format(position, lane))
            lane_sequence = [x["output"]["process"] for x in pipeline_list
                             if x["output"]["lane"] == lane]
            logger.debug("Testing lane sequence '{}' against fork signature"
                         " '{}'".format(lane_sequence, fork_sig))
            if lane_sequence == fork_sig:
                return p["output"]["lane"]

    return 0
def get_source_lane ( fork_process , pipeline_list ) : fork_source = fork_process [ - 1 ] fork_sig = [ x for x in fork_process if x != "__init__" ] for position , p in enumerate ( pipeline_list [ : : - 1 ] ) : if p [ "output" ] [ "process" ] == fork_source : lane = p [ "output" ] [ "lane" ] logger . debug ( "Possible source match found in position {} in lane" " {}" . format ( position , lane ) ) lane_sequence = [ x [ "output" ] [ "process" ] for x in pipeline_list if x [ "output" ] [ "lane" ] == lane ] logger . debug ( "Testing lane sequence '{}' against fork signature" " '{}'" . format ( lane_sequence , fork_sig ) ) if lane_sequence == fork_sig : return p [ "output" ] [ "lane" ] return 0
Returns the lane of the last process that matches fork_process

    Parameters
    ----------
    fork_process : list
        List of processes before the fork.
    pipeline_list : list
        List with the pipeline connection dictionaries.

    Returns
    -------
    int
        Lane of the last process that matches fork_process
Returns the lane of the last process that matches fork_process