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):
        # 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 ) : # 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
Method that parses the nextflow log file once and updates the submitted number of samples for each process