flowcraft/generator/engine.py
def _build_connections(self, process_list, ignore_dependencies,
                           auto_dependency):
        """Parses the process connections dictionaries into a process list

        This method is called upon instantiation of the NextflowGenerator
        class. Essentially, it sets the main input/output channel names of the
        processes so that they can be linked correctly.

        If a connection between two consecutive process is not possible due
        to a mismatch in the input/output types, it exits with an error.

        Returns
        -------

        """

        logger.debug("=============================")
        logger.debug("Building pipeline connections")
        logger.debug("=============================")

        logger.debug("Processing connections: {}".format(process_list))

        for p, con in enumerate(process_list):

            logger.debug("Processing connection '{}': {}".format(p, con))

            # Get lanes
            in_lane = con["input"]["lane"]
            out_lane = con["output"]["lane"]
            logger.debug("[{}] Input lane: {}".format(p, in_lane))
            logger.debug("[{}] Output lane: {}".format(p, out_lane))

            # Update the total number of lines of the pipeline
            if out_lane > self.lanes:
                self.lanes = out_lane

            # Get process names and directives for the output process
            p_in_name, p_out_name, out_directives = self._get_process_names(
                con, p)

            # Check if process is available or correctly named
            if p_out_name not in self.process_map:
                logger.error(colored_print(
                    "\nThe process '{}' is not available."
                        .format(p_out_name), "red_bold"))
                guess_process(p_out_name, self.process_map)
                sys.exit(1)

            # Instance output process
            out_process = self.process_map[p_out_name](template=p_out_name)

            # Update directives, if provided
            if out_directives:
                out_process.update_attributes(out_directives)

            # Set suffix strings for main input/output channels. Suffixes are
            # based on the lane and the arbitrary and unique process id
            # e.g.: 'process_1_1'
            input_suf = "{}_{}".format(in_lane, p)
            output_suf = "{}_{}".format(out_lane, p)
            logger.debug("[{}] Setting main channels with input suffix '{}'"
                         " and output suffix '{}'".format(
                            p, input_suf, output_suf))
            out_process.set_main_channel_names(input_suf, output_suf, out_lane)

            # Instance input process, if it exists. In case of init, the
            # output process forks from the raw input user data
            if p_in_name != "__init__":
                # Create instance of input process
                in_process = self.process_map[p_in_name](template=p_in_name)
                # Test if two processes can be connected by input/output types
                logger.debug("[{}] Testing connection between input and "
                             "output processes".format(p))
                self._test_connection(in_process, out_process)
                out_process.parent_lane = in_lane
            else:
                # When the input process is __init__, set the parent_lane
                # to None. This will tell the engine that this process
                # will receive the main input from the raw user input.
                out_process.parent_lane = None
            logger.debug("[{}] Parent lane: {}".format(
                p, out_process.parent_lane))

            # If the current connection is a fork, add it to the fork tree
            if in_lane != out_lane:
                logger.debug("[{}] Connection is a fork. Adding lanes to "
                             "fork list".format(p))
                self._fork_tree[in_lane].append(out_lane)
                # Update main output fork of parent process
                try:
                    parent_process = [
                        x for x in self.processes if x.lane == in_lane and
                        x.template == p_in_name
                    ][0]
                    logger.debug(
                        "[{}] Updating main forks of parent fork '{}' with"
                        " '{}'".format(p, parent_process,
                                       out_process.input_channel))
                    parent_process.update_main_forks(out_process.input_channel)
                except IndexError:
                    pass
            else:
                # Get parent process, naive version
                parent_process = self.processes[-1]

                # Check if the last process' lane matches the lane of the
                # current output process. If not, get the last process
                # in the same lane
                if parent_process.lane and parent_process.lane != out_lane:
                    parent_process = [x for x in self.processes[::-1]
                                      if x.lane == out_lane][0]

                if parent_process.output_channel:
                    logger.debug(
                        "[{}] Updating input channel of output process"
                        " with '{}'".format(
                            p, parent_process.output_channel))
                    out_process.input_channel = parent_process.output_channel

            # Check for process dependencies
            if out_process.dependencies and not ignore_dependencies:
                logger.debug("[{}] Dependencies found for process '{}': "
                             "{}".format(p, p_out_name,
                                         out_process.dependencies))
                parent_lanes = self._get_fork_tree(out_lane)
                for dep in out_process.dependencies:
                    if not self._search_tree_backwards(dep, parent_lanes):
                        if auto_dependency:
                            self._add_dependency(
                                out_process, dep, in_lane, out_lane, p)
                        elif not self.export_parameters:
                            logger.error(colored_print(
                                "\nThe following dependency of the process"
                                " '{}' is missing: {}".format(p_out_name, dep),
                                "red_bold"))
                            sys.exit(1)

            self.processes.append(out_process)

        logger.debug("Completed connections: {}".format(self.processes))
        logger.debug("Fork tree: {}".format(self._fork_tree))
def _build_connections(self, process_list, ignore_dependencies,
                           auto_dependency):
        """Parses the process connections dictionaries into a process list

        This method is called upon instantiation of the NextflowGenerator
        class. Essentially, it sets the main input/output channel names of the
        processes so that they can be linked correctly.

        If a connection between two consecutive process is not possible due
        to a mismatch in the input/output types, it exits with an error.

        Returns
        -------

        """

        logger.debug("=============================")
        logger.debug("Building pipeline connections")
        logger.debug("=============================")

        logger.debug("Processing connections: {}".format(process_list))

        for p, con in enumerate(process_list):

            logger.debug("Processing connection '{}': {}".format(p, con))

            # Get lanes
            in_lane = con["input"]["lane"]
            out_lane = con["output"]["lane"]
            logger.debug("[{}] Input lane: {}".format(p, in_lane))
            logger.debug("[{}] Output lane: {}".format(p, out_lane))

            # Update the total number of lines of the pipeline
            if out_lane > self.lanes:
                self.lanes = out_lane

            # Get process names and directives for the output process
            p_in_name, p_out_name, out_directives = self._get_process_names(
                con, p)

            # Check if process is available or correctly named
            if p_out_name not in self.process_map:
                logger.error(colored_print(
                    "\nThe process '{}' is not available."
                        .format(p_out_name), "red_bold"))
                guess_process(p_out_name, self.process_map)
                sys.exit(1)

            # Instance output process
            out_process = self.process_map[p_out_name](template=p_out_name)

            # Update directives, if provided
            if out_directives:
                out_process.update_attributes(out_directives)

            # Set suffix strings for main input/output channels. Suffixes are
            # based on the lane and the arbitrary and unique process id
            # e.g.: 'process_1_1'
            input_suf = "{}_{}".format(in_lane, p)
            output_suf = "{}_{}".format(out_lane, p)
            logger.debug("[{}] Setting main channels with input suffix '{}'"
                         " and output suffix '{}'".format(
                            p, input_suf, output_suf))
            out_process.set_main_channel_names(input_suf, output_suf, out_lane)

            # Instance input process, if it exists. In case of init, the
            # output process forks from the raw input user data
            if p_in_name != "__init__":
                # Create instance of input process
                in_process = self.process_map[p_in_name](template=p_in_name)
                # Test if two processes can be connected by input/output types
                logger.debug("[{}] Testing connection between input and "
                             "output processes".format(p))
                self._test_connection(in_process, out_process)
                out_process.parent_lane = in_lane
            else:
                # When the input process is __init__, set the parent_lane
                # to None. This will tell the engine that this process
                # will receive the main input from the raw user input.
                out_process.parent_lane = None
            logger.debug("[{}] Parent lane: {}".format(
                p, out_process.parent_lane))

            # If the current connection is a fork, add it to the fork tree
            if in_lane != out_lane:
                logger.debug("[{}] Connection is a fork. Adding lanes to "
                             "fork list".format(p))
                self._fork_tree[in_lane].append(out_lane)
                # Update main output fork of parent process
                try:
                    parent_process = [
                        x for x in self.processes if x.lane == in_lane and
                        x.template == p_in_name
                    ][0]
                    logger.debug(
                        "[{}] Updating main forks of parent fork '{}' with"
                        " '{}'".format(p, parent_process,
                                       out_process.input_channel))
                    parent_process.update_main_forks(out_process.input_channel)
                except IndexError:
                    pass
            else:
                # Get parent process, naive version
                parent_process = self.processes[-1]

                # Check if the last process' lane matches the lane of the
                # current output process. If not, get the last process
                # in the same lane
                if parent_process.lane and parent_process.lane != out_lane:
                    parent_process = [x for x in self.processes[::-1]
                                      if x.lane == out_lane][0]

                if parent_process.output_channel:
                    logger.debug(
                        "[{}] Updating input channel of output process"
                        " with '{}'".format(
                            p, parent_process.output_channel))
                    out_process.input_channel = parent_process.output_channel

            # Check for process dependencies
            if out_process.dependencies and not ignore_dependencies:
                logger.debug("[{}] Dependencies found for process '{}': "
                             "{}".format(p, p_out_name,
                                         out_process.dependencies))
                parent_lanes = self._get_fork_tree(out_lane)
                for dep in out_process.dependencies:
                    if not self._search_tree_backwards(dep, parent_lanes):
                        if auto_dependency:
                            self._add_dependency(
                                out_process, dep, in_lane, out_lane, p)
                        elif not self.export_parameters:
                            logger.error(colored_print(
                                "\nThe following dependency of the process"
                                " '{}' is missing: {}".format(p_out_name, dep),
                                "red_bold"))
                            sys.exit(1)

            self.processes.append(out_process)

        logger.debug("Completed connections: {}".format(self.processes))
        logger.debug("Fork tree: {}".format(self._fork_tree))
def _build_connections ( self , process_list , ignore_dependencies , auto_dependency ) : logger . debug ( "=============================" ) logger . debug ( "Building pipeline connections" ) logger . debug ( "=============================" ) logger . debug ( "Processing connections: {}" . format ( process_list ) ) for p , con in enumerate ( process_list ) : logger . debug ( "Processing connection '{}': {}" . format ( p , con ) ) # Get lanes in_lane = con [ "input" ] [ "lane" ] out_lane = con [ "output" ] [ "lane" ] logger . debug ( "[{}] Input lane: {}" . format ( p , in_lane ) ) logger . debug ( "[{}] Output lane: {}" . format ( p , out_lane ) ) # Update the total number of lines of the pipeline if out_lane > self . lanes : self . lanes = out_lane # Get process names and directives for the output process p_in_name , p_out_name , out_directives = self . _get_process_names ( con , p ) # Check if process is available or correctly named if p_out_name not in self . process_map : logger . error ( colored_print ( "\nThe process '{}' is not available." . format ( p_out_name ) , "red_bold" ) ) guess_process ( p_out_name , self . process_map ) sys . exit ( 1 ) # Instance output process out_process = self . process_map [ p_out_name ] ( template = p_out_name ) # Update directives, if provided if out_directives : out_process . update_attributes ( out_directives ) # Set suffix strings for main input/output channels. Suffixes are # based on the lane and the arbitrary and unique process id # e.g.: 'process_1_1' input_suf = "{}_{}" . format ( in_lane , p ) output_suf = "{}_{}" . format ( out_lane , p ) logger . debug ( "[{}] Setting main channels with input suffix '{}'" " and output suffix '{}'" . format ( p , input_suf , output_suf ) ) out_process . set_main_channel_names ( input_suf , output_suf , out_lane ) # Instance input process, if it exists. In case of init, the # output process forks from the raw input user data if p_in_name != "__init__" : # Create instance of input process in_process = self . process_map [ p_in_name ] ( template = p_in_name ) # Test if two processes can be connected by input/output types logger . debug ( "[{}] Testing connection between input and " "output processes" . format ( p ) ) self . _test_connection ( in_process , out_process ) out_process . parent_lane = in_lane else : # When the input process is __init__, set the parent_lane # to None. This will tell the engine that this process # will receive the main input from the raw user input. out_process . parent_lane = None logger . debug ( "[{}] Parent lane: {}" . format ( p , out_process . parent_lane ) ) # If the current connection is a fork, add it to the fork tree if in_lane != out_lane : logger . debug ( "[{}] Connection is a fork. Adding lanes to " "fork list" . format ( p ) ) self . _fork_tree [ in_lane ] . append ( out_lane ) # Update main output fork of parent process try : parent_process = [ x for x in self . processes if x . lane == in_lane and x . template == p_in_name ] [ 0 ] logger . debug ( "[{}] Updating main forks of parent fork '{}' with" " '{}'" . format ( p , parent_process , out_process . input_channel ) ) parent_process . update_main_forks ( out_process . input_channel ) except IndexError : pass else : # Get parent process, naive version parent_process = self . processes [ - 1 ] # Check if the last process' lane matches the lane of the # current output process. If not, get the last process # in the same lane if parent_process . lane and parent_process . lane != out_lane : parent_process = [ x for x in self . processes [ : : - 1 ] if x . lane == out_lane ] [ 0 ] if parent_process . output_channel : logger . debug ( "[{}] Updating input channel of output process" " with '{}'" . format ( p , parent_process . output_channel ) ) out_process . input_channel = parent_process . output_channel # Check for process dependencies if out_process . dependencies and not ignore_dependencies : logger . debug ( "[{}] Dependencies found for process '{}': " "{}" . format ( p , p_out_name , out_process . dependencies ) ) parent_lanes = self . _get_fork_tree ( out_lane ) for dep in out_process . dependencies : if not self . _search_tree_backwards ( dep , parent_lanes ) : if auto_dependency : self . _add_dependency ( out_process , dep , in_lane , out_lane , p ) elif not self . export_parameters : logger . error ( colored_print ( "\nThe following dependency of the process" " '{}' is missing: {}" . format ( p_out_name , dep ) , "red_bold" ) ) sys . exit ( 1 ) self . processes . append ( out_process ) logger . debug ( "Completed connections: {}" . format ( self . processes ) ) logger . debug ( "Fork tree: {}" . format ( self . _fork_tree ) )
Parses the process connections dictionaries into a process list

        This method is called upon instantiation of the NextflowGenerator
        class. Essentially, it sets the main input/output channel names of the
        processes so that they can be linked correctly.

        If a connection between two consecutive process is not possible due
        to a mismatch in the input/output types, it exits with an error.

        Returns
        -------
Parses the process connections dictionaries into a process list