flowcraft/generator/inspect.py
def log_parser(self):
        """Method that parses the nextflow log file once and updates the
        submitted number of samples for each process
        """

        # Check the timestamp of the log file. Only proceed with the parsing
        # if it changed from the previous time.
        size_stamp = os.path.getsize(self.log_file)
        self.log_retry = 0
        if size_stamp and size_stamp == self.log_sizestamp:
            return
        else:
            logger.debug("Updating log size stamp to: {}".format(size_stamp))
            self.log_sizestamp = size_stamp

        # Regular expression to catch four groups:
        # 1. Start timestamp
        # 2. Work directory hash
        # 3. Process name
        # 4. Tag name
        r = ".* (.*) \[.*\].*\[(.*)\].*process > (.*) \((.*)\).*"

        with open(self.log_file) as fh:

            for line in fh:
                if "Submitted process >" in line or \
                        "Re-submitted process >" in line or \
                        "Cached process >" in line:
                    m = re.match(r, line)
                    if not m:
                        continue

                    time_start = m.group(1)
                    workdir = m.group(2)
                    process = m.group(3)
                    tag = m.group(4)

                    # Skip if this line has already been parsed
                    if time_start + tag not in self.stored_log_ids:
                        self.stored_log_ids.append(time_start + tag)
                    else:
                        continue

                    # For first time processes
                    if process not in self.processes:
                        continue
                    p = self.processes[process]

                    # Skip is process/tag combination has finished or is retrying
                    if tag in list(p["finished"]) + list(p["retry"]):
                        continue

                    # Update failed process/tags when they have been re-submitted
                    if tag in list(p["failed"]) and \
                            "Re-submitted process >" in line:
                        p["retry"].add(tag)
                        self.send = True
                        continue

                    # Set process barrier to running. Check for barrier status
                    # are performed at the end of the trace parsing in the
                    # _update_barrier_status method.
                    p["barrier"] = "R"
                    if tag not in p["submitted"]:
                        p["submitted"].add(tag)
                        # Update the process_tags attribute with the new tag.
                        # Update only when the tag does not exist. This may rarely
                        # occur when the tag is parsed first in the trace file
                        if tag not in self.process_tags[process]:
                            self.process_tags[process][tag] = {
                                "workdir": self._expand_path(workdir),
                                "start": time_start
                            }
                            self.send = True
                        # When the tag is filled in the trace file parsing,
                        # the timestamp may not be present in the trace. In
                        # those cases, fill that information here.
                        elif not self.process_tags[process][tag]["start"]:
                            self.process_tags[process][tag]["start"] = time_start
                            self.send = True

        self._update_pipeline_status()
def log_parser(self):
        """Method that parses the nextflow log file once and updates the
        submitted number of samples for each process
        """

        # Check the timestamp of the log file. Only proceed with the parsing
        # if it changed from the previous time.
        size_stamp = os.path.getsize(self.log_file)
        self.log_retry = 0
        if size_stamp and size_stamp == self.log_sizestamp:
            return
        else:
            logger.debug("Updating log size stamp to: {}".format(size_stamp))
            self.log_sizestamp = size_stamp

        # Regular expression to catch four groups:
        # 1. Start timestamp
        # 2. Work directory hash
        # 3. Process name
        # 4. Tag name
        r = ".* (.*) \[.*\].*\[(.*)\].*process > (.*) \((.*)\).*"

        with open(self.log_file) as fh:

            for line in fh:
                if "Submitted process >" in line or \
                        "Re-submitted process >" in line or \
                        "Cached process >" in line:
                    m = re.match(r, line)
                    if not m:
                        continue

                    time_start = m.group(1)
                    workdir = m.group(2)
                    process = m.group(3)
                    tag = m.group(4)

                    # Skip if this line has already been parsed
                    if time_start + tag not in self.stored_log_ids:
                        self.stored_log_ids.append(time_start + tag)
                    else:
                        continue

                    # For first time processes
                    if process not in self.processes:
                        continue
                    p = self.processes[process]

                    # Skip is process/tag combination has finished or is retrying
                    if tag in list(p["finished"]) + list(p["retry"]):
                        continue

                    # Update failed process/tags when they have been re-submitted
                    if tag in list(p["failed"]) and \
                            "Re-submitted process >" in line:
                        p["retry"].add(tag)
                        self.send = True
                        continue

                    # Set process barrier to running. Check for barrier status
                    # are performed at the end of the trace parsing in the
                    # _update_barrier_status method.
                    p["barrier"] = "R"
                    if tag not in p["submitted"]:
                        p["submitted"].add(tag)
                        # Update the process_tags attribute with the new tag.
                        # Update only when the tag does not exist. This may rarely
                        # occur when the tag is parsed first in the trace file
                        if tag not in self.process_tags[process]:
                            self.process_tags[process][tag] = {
                                "workdir": self._expand_path(workdir),
                                "start": time_start
                            }
                            self.send = True
                        # When the tag is filled in the trace file parsing,
                        # the timestamp may not be present in the trace. In
                        # those cases, fill that information here.
                        elif not self.process_tags[process][tag]["start"]:
                            self.process_tags[process][tag]["start"] = time_start
                            self.send = True

        self._update_pipeline_status()
Method that parses the nextflow log file once and updates the
        submitted number of samples for each process