flowcraft/generator/report.py
def update_trace_watch(self):
        """Parses the nextflow trace file and retrieves the path of report JSON
        files that have not been sent to the service yet.
        """

        # Check the size stamp of the tracefile. Only proceed with the parsing
        # if it changed from the previous size.
        size_stamp = os.path.getsize(self.trace_file)
        self.trace_retry = 0
        if size_stamp and size_stamp == self.trace_sizestamp:
            return
        else:
            logger.debug("Updating trace size stamp to: {}".format(size_stamp))
            self.trace_sizestamp = size_stamp

        with open(self.trace_file) as fh:

            # Skip potential empty lines at the start of file
            header = next(fh).strip()
            while not header:
                header = next(fh).strip()

            # Get header mappings before parsing the file
            hm = self._header_mapping(header)

            for line in fh:
                # Skip empty lines
                if line.strip() == "":
                    continue

                fields = line.strip().split("\t")

                # Skip if task ID was already processes
                if fields[hm["task_id"]] in self.stored_ids:
                    continue

                if fields[hm["process"]] == "report":
                    self.report_queue.append(
                        self._expand_path(fields[hm["hash"]])
                    )
                    self.send = True

                # Add the processed trace line to the stored ids. It will be
                # skipped in future parsers
                self.stored_ids.append(fields[hm["task_id"]])
def update_trace_watch(self):
        # Check the size stamp of the tracefile. Only proceed with the parsing
        # if it changed from the previous size.
        size_stamp = os.path.getsize(self.trace_file)
        self.trace_retry = 0
        if size_stamp and size_stamp == self.trace_sizestamp:
            return
        else:
            logger.debug("Updating trace size stamp to: {}".format(size_stamp))
            self.trace_sizestamp = size_stamp

        with open(self.trace_file) as fh:

            # Skip potential empty lines at the start of file
            header = next(fh).strip()
            while not header:
                header = next(fh).strip()

            # Get header mappings before parsing the file
            hm = self._header_mapping(header)

            for line in fh:
                # Skip empty lines
                if line.strip() == "":
                    continue

                fields = line.strip().split("\t")

                # Skip if task ID was already processes
                if fields[hm["task_id"]] in self.stored_ids:
                    continue

                if fields[hm["process"]] == "report":
                    self.report_queue.append(
                        self._expand_path(fields[hm["hash"]])
                    )
                    self.send = True

                # Add the processed trace line to the stored ids. It will be
                # skipped in future parsers
                self.stored_ids.append(fields[hm["task_id"]])
def update_trace_watch ( self ) : # Check the size stamp of the tracefile. Only proceed with the parsing # if it changed from the previous size. size_stamp = os . path . getsize ( self . trace_file ) self . trace_retry = 0 if size_stamp and size_stamp == self . trace_sizestamp : return else : logger . debug ( "Updating trace size stamp to: {}" . format ( size_stamp ) ) self . trace_sizestamp = size_stamp with open ( self . trace_file ) as fh : # Skip potential empty lines at the start of file header = next ( fh ) . strip ( ) while not header : header = next ( fh ) . strip ( ) # Get header mappings before parsing the file hm = self . _header_mapping ( header ) for line in fh : # Skip empty lines if line . strip ( ) == "" : continue fields = line . strip ( ) . split ( "\t" ) # Skip if task ID was already processes if fields [ hm [ "task_id" ] ] in self . stored_ids : continue if fields [ hm [ "process" ] ] == "report" : self . report_queue . append ( self . _expand_path ( fields [ hm [ "hash" ] ] ) ) self . send = True # Add the processed trace line to the stored ids. It will be # skipped in future parsers self . stored_ids . append ( fields [ hm [ "task_id" ] ] )
Parses the nextflow trace file and retrieves the path of report JSON
        files that have not been sent to the service yet.
Parses the nextflow trace file and retrieves the path of report JSON files that have not been sent to the service yet .