flowcraft/generator/inspect.py
def _update_process_stats(self):
        """Updates the process stats with the information from the processes

        This method is called at the end of each static parsing of the nextflow
        trace file. It re-populates the :attr:`process_stats` dictionary
        with the new stat metrics.
        """

        good_status = ["COMPLETED", "CACHED"]

        for process, vals in self.trace_info.items():

            # Update submission status of tags for each process
            vals = self._update_tag_status(process, vals)

            # Update process resources
            self._update_process_resources(process, vals)

            self.process_stats[process] = {}

            inst = self.process_stats[process]

            # Get number of completed samples
            inst["completed"] = "{}".format(
                len([x for x in vals if x["status"] in good_status]))

            # Get average time
            try:
                time_array = [self._hms(x["realtime"]) for x in vals]
                mean_time = round(sum(time_array) / len(time_array), 1)
                mean_time_str = strftime('%H:%M:%S', gmtime(mean_time))
                inst["realtime"] = mean_time_str
            # When the realtime column is not present
            except KeyError:
                inst["realtime"] = "-"

            # Get cumulative cpu/hours
            try:
                cpu_hours = [self._cpu_load_parser(
                    x["cpus"], x["%cpu"], x["realtime"]) for x in vals]
                inst["cpuhour"] = round(sum(cpu_hours), 2)
            # When the realtime, cpus or %cpus column are not present
            except KeyError:
                inst["cpuhour"] = "-"

            # Assess resource warnings
            inst["cpu_warnings"], inst["mem_warnings"] = \
                self._assess_resource_warnings(process, vals)

            # Get maximum memory
            try:
                rss_values = [self._size_coverter(x["rss"]) for x in vals
                              if x["rss"] != "-"]
                if rss_values:
                    max_rss = round(max(rss_values))
                    rss_str = self._size_compress(max_rss)
                else:
                    rss_str = "-"
                inst["maxmem"] = rss_str
            except KeyError:
                inst["maxmem"] = "-"

            # Get read size
            try:
                rchar_values = [self._size_coverter(x["rchar"]) for x in vals
                                if x["rchar"] != "-"]
                if rchar_values:
                    avg_rchar = round(sum(rchar_values) / len(rchar_values))
                    rchar_str = self._size_compress(avg_rchar)
                else:
                    rchar_str = "-"
            except KeyError:
                rchar_str = "-"
            inst["avgread"] = rchar_str

            # Get write size
            try:
                wchar_values = [self._size_coverter(x["wchar"]) for x in vals
                                if x["wchar"] != "-"]
                if wchar_values:
                    avg_wchar = round(sum(wchar_values) / len(wchar_values))
                    wchar_str = self._size_compress(avg_wchar)
                else:
                    wchar_str = "-"
            except KeyError:
                wchar_str = "-"
            inst["avgwrite"] = wchar_str
def _update_process_stats(self):
        """Updates the process stats with the information from the processes

        This method is called at the end of each static parsing of the nextflow
        trace file. It re-populates the :attr:`process_stats` dictionary
        with the new stat metrics.
        """

        good_status = ["COMPLETED", "CACHED"]

        for process, vals in self.trace_info.items():

            # Update submission status of tags for each process
            vals = self._update_tag_status(process, vals)

            # Update process resources
            self._update_process_resources(process, vals)

            self.process_stats[process] = {}

            inst = self.process_stats[process]

            # Get number of completed samples
            inst["completed"] = "{}".format(
                len([x for x in vals if x["status"] in good_status]))

            # Get average time
            try:
                time_array = [self._hms(x["realtime"]) for x in vals]
                mean_time = round(sum(time_array) / len(time_array), 1)
                mean_time_str = strftime('%H:%M:%S', gmtime(mean_time))
                inst["realtime"] = mean_time_str
            # When the realtime column is not present
            except KeyError:
                inst["realtime"] = "-"

            # Get cumulative cpu/hours
            try:
                cpu_hours = [self._cpu_load_parser(
                    x["cpus"], x["%cpu"], x["realtime"]) for x in vals]
                inst["cpuhour"] = round(sum(cpu_hours), 2)
            # When the realtime, cpus or %cpus column are not present
            except KeyError:
                inst["cpuhour"] = "-"

            # Assess resource warnings
            inst["cpu_warnings"], inst["mem_warnings"] = \
                self._assess_resource_warnings(process, vals)

            # Get maximum memory
            try:
                rss_values = [self._size_coverter(x["rss"]) for x in vals
                              if x["rss"] != "-"]
                if rss_values:
                    max_rss = round(max(rss_values))
                    rss_str = self._size_compress(max_rss)
                else:
                    rss_str = "-"
                inst["maxmem"] = rss_str
            except KeyError:
                inst["maxmem"] = "-"

            # Get read size
            try:
                rchar_values = [self._size_coverter(x["rchar"]) for x in vals
                                if x["rchar"] != "-"]
                if rchar_values:
                    avg_rchar = round(sum(rchar_values) / len(rchar_values))
                    rchar_str = self._size_compress(avg_rchar)
                else:
                    rchar_str = "-"
            except KeyError:
                rchar_str = "-"
            inst["avgread"] = rchar_str

            # Get write size
            try:
                wchar_values = [self._size_coverter(x["wchar"]) for x in vals
                                if x["wchar"] != "-"]
                if wchar_values:
                    avg_wchar = round(sum(wchar_values) / len(wchar_values))
                    wchar_str = self._size_compress(avg_wchar)
                else:
                    wchar_str = "-"
            except KeyError:
                wchar_str = "-"
            inst["avgwrite"] = wchar_str
def _update_process_stats ( self ) : good_status = [ "COMPLETED" , "CACHED" ] for process , vals in self . trace_info . items ( ) : # Update submission status of tags for each process vals = self . _update_tag_status ( process , vals ) # Update process resources self . _update_process_resources ( process , vals ) self . process_stats [ process ] = { } inst = self . process_stats [ process ] # Get number of completed samples inst [ "completed" ] = "{}" . format ( len ( [ x for x in vals if x [ "status" ] in good_status ] ) ) # Get average time try : time_array = [ self . _hms ( x [ "realtime" ] ) for x in vals ] mean_time = round ( sum ( time_array ) / len ( time_array ) , 1 ) mean_time_str = strftime ( '%H:%M:%S' , gmtime ( mean_time ) ) inst [ "realtime" ] = mean_time_str # When the realtime column is not present except KeyError : inst [ "realtime" ] = "-" # Get cumulative cpu/hours try : cpu_hours = [ self . _cpu_load_parser ( x [ "cpus" ] , x [ "%cpu" ] , x [ "realtime" ] ) for x in vals ] inst [ "cpuhour" ] = round ( sum ( cpu_hours ) , 2 ) # When the realtime, cpus or %cpus column are not present except KeyError : inst [ "cpuhour" ] = "-" # Assess resource warnings inst [ "cpu_warnings" ] , inst [ "mem_warnings" ] = self . _assess_resource_warnings ( process , vals ) # Get maximum memory try : rss_values = [ self . _size_coverter ( x [ "rss" ] ) for x in vals if x [ "rss" ] != "-" ] if rss_values : max_rss = round ( max ( rss_values ) ) rss_str = self . _size_compress ( max_rss ) else : rss_str = "-" inst [ "maxmem" ] = rss_str except KeyError : inst [ "maxmem" ] = "-" # Get read size try : rchar_values = [ self . _size_coverter ( x [ "rchar" ] ) for x in vals if x [ "rchar" ] != "-" ] if rchar_values : avg_rchar = round ( sum ( rchar_values ) / len ( rchar_values ) ) rchar_str = self . _size_compress ( avg_rchar ) else : rchar_str = "-" except KeyError : rchar_str = "-" inst [ "avgread" ] = rchar_str # Get write size try : wchar_values = [ self . _size_coverter ( x [ "wchar" ] ) for x in vals if x [ "wchar" ] != "-" ] if wchar_values : avg_wchar = round ( sum ( wchar_values ) / len ( wchar_values ) ) wchar_str = self . _size_compress ( avg_wchar ) else : wchar_str = "-" except KeyError : wchar_str = "-" inst [ "avgwrite" ] = wchar_str
Updates the process stats with the information from the processes

        This method is called at the end of each static parsing of the nextflow
        trace file. It re-populates the :attr:`process_stats` dictionary
        with the new stat metrics.
Updates the process stats with the information from the processes