flowcraft/generator/inspect.py
def _get_pipeline_processes(self):
        """Parses the .nextflow.log file and retrieves the complete list
        of processes

        This method searches for specific signatures at the beginning of the
        .nextflow.log file::

             Apr-19 19:07:32.660 [main] DEBUG nextflow.processor
             TaskProcessor - Creating operator > report_corrupt_1_1 --
             maxForks: 4

        When a line with the .*Creating operator.* signature is found, the
        process name is retrieved and populates the :attr:`processes` attribute
        """

        with open(self.log_file) as fh:

            for line in fh:
                if re.match(".*Creating operator.*", line):
                    # Retrieves the process name from the string
                    match = re.match(".*Creating operator > (.*) --", line)
                    process = match.group(1)

                    if any([process.startswith(x) for x in self._blacklist]):
                        continue

                    if process not in self.skip_processes:
                        self.processes[match.group(1)] = {
                            "barrier": "W",
                            "submitted": set(),
                            "finished": set(),
                            "failed": set(),
                            "retry": set(),
                            "cpus": None,
                            "memory": None
                        }
                        self.process_tags[process] = {}

                # Retrieves the pipeline name from the string
                if re.match(".*Launching `.*` \[.*\] ", line):
                    tag_match = re.match(".*Launching `.*` \[(.*)\] ", line)
                    self.pipeline_tag = tag_match.group(1) if tag_match else \
                        "?"
                    name_match = re.match(".*Launching `(.*)` \[.*\] ", line)
                    self.pipeline_name = name_match.group(1) if name_match \
                        else "?"

        self.content_lines = len(self.processes)
def _get_pipeline_processes(self):
        """Parses the .nextflow.log file and retrieves the complete list
        of processes

        This method searches for specific signatures at the beginning of the
        .nextflow.log file::

             Apr-19 19:07:32.660 [main] DEBUG nextflow.processor
             TaskProcessor - Creating operator > report_corrupt_1_1 --
             maxForks: 4

        When a line with the .*Creating operator.* signature is found, the
        process name is retrieved and populates the :attr:`processes` attribute
        """

        with open(self.log_file) as fh:

            for line in fh:
                if re.match(".*Creating operator.*", line):
                    # Retrieves the process name from the string
                    match = re.match(".*Creating operator > (.*) --", line)
                    process = match.group(1)

                    if any([process.startswith(x) for x in self._blacklist]):
                        continue

                    if process not in self.skip_processes:
                        self.processes[match.group(1)] = {
                            "barrier": "W",
                            "submitted": set(),
                            "finished": set(),
                            "failed": set(),
                            "retry": set(),
                            "cpus": None,
                            "memory": None
                        }
                        self.process_tags[process] = {}

                # Retrieves the pipeline name from the string
                if re.match(".*Launching `.*` \[.*\] ", line):
                    tag_match = re.match(".*Launching `.*` \[(.*)\] ", line)
                    self.pipeline_tag = tag_match.group(1) if tag_match else \
                        "?"
                    name_match = re.match(".*Launching `(.*)` \[.*\] ", line)
                    self.pipeline_name = name_match.group(1) if name_match \
                        else "?"

        self.content_lines = len(self.processes)
def _get_pipeline_processes ( self ) : with open ( self . log_file ) as fh : for line in fh : if re . match ( ".*Creating operator.*" , line ) : # Retrieves the process name from the string match = re . match ( ".*Creating operator > (.*) --" , line ) process = match . group ( 1 ) if any ( [ process . startswith ( x ) for x in self . _blacklist ] ) : continue if process not in self . skip_processes : self . processes [ match . group ( 1 ) ] = { "barrier" : "W" , "submitted" : set ( ) , "finished" : set ( ) , "failed" : set ( ) , "retry" : set ( ) , "cpus" : None , "memory" : None } self . process_tags [ process ] = { } # Retrieves the pipeline name from the string if re . match ( ".*Launching `.*` \[.*\] " , line ) : tag_match = re . match ( ".*Launching `.*` \[(.*)\] " , line ) self . pipeline_tag = tag_match . group ( 1 ) if tag_match else "?" name_match = re . match ( ".*Launching `(.*)` \[.*\] " , line ) self . pipeline_name = name_match . group ( 1 ) if name_match else "?" self . content_lines = len ( self . processes )
Parses the .nextflow.log file and retrieves the complete list
        of processes

        This method searches for specific signatures at the beginning of the
        .nextflow.log file::

             Apr-19 19:07:32.660 [main] DEBUG nextflow.processor
             TaskProcessor - Creating operator > report_corrupt_1_1 --
             maxForks: 4

        When a line with the .*Creating operator.* signature is found, the
        process name is retrieved and populates the :attr:`processes` attribute
Parses the . nextflow . log file and retrieves the complete list of processes