Method set_raw_inputs

flowcraft/generator/process.py:687–737

Sets the main input channels of the pipeline and their forks. The ``raw_input`` dictionary input should contain one entry for each input type (fastq, fasta, etc). The corresponding value should be a dictionary/json with the following key:values: - ``channel``: Name

(self, raw_input)

        self.status_channels = []

    def set_raw_inputs(self, raw_input):
        """Sets the main input channels of the pipeline and their forks.

        The ``raw_input`` dictionary input should contain one entry for each
        input type (fastq, fasta, etc). The corresponding value should be a
        dictionary/json with the following key:values:

        - ``channel``: Name of the raw input channel (e.g.: channel1)
        - ``channel_str``: The nextflow definition of the channel and
           eventual checks (e.g.: channel1 = Channel.fromPath(param))
        - ``raw_forks``: A list of channels to which the channel name will
          for to.

        Each new type of input parameter is automatically added to the
        :attr:`params` attribute, so that they are automatically collected
        for the pipeline description and help.

        Parameters
        ----------
        raw_input : dict
            Contains an entry for each input type with the channel name,
            channel string and forks.
        """

        logger.debug("Setting raw inputs using raw input dict: {}".format(
            raw_input))

        primary_inputs = []

        for input_type, el in raw_input.items():

            primary_inputs.append(el["channel_str"])

            # Update the process' parameters with the raw input
            raw_channel = self.RAW_MAPPING[input_type]
            self.params[input_type] = {
                "default": raw_channel["default_value"],
                "description": raw_channel["description"]
            }

            op = "set" if len(el["raw_forks"]) == 1 else "into"

            self.forks.append("\n{}.{}{{ {} }}\n".format(
                el["channel"], op, ";".join(el["raw_forks"])
            ))

        logger.debug("Setting raw inputs: {}".format(primary_inputs))
        logger.debug("Setting forks attribute to: {}".format(self.forks))
        self._context = {**self._context,
                         **{"forks": "\n".join(self.forks),
                            "main_inputs": "\n".join(primary_inputs)}}

    def set_secondary_inputs(self, channel_dict):
        """ Adds secondary inputs to the start of the pipeline.