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):
    """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
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