exporter.py 文件源码

def process_stats(self,
                      stats: dict,
                      time_periods: Sequence[str] = ('last1min', )) -> None:
        ''' Process dump1090 statistics into exported metrics.

        :param stats: a dict containing dump1090 statistics data.
        '''
        metrics = self.metrics['stats']

        for time_period in time_periods:
            try:
                tp_stats = stats[time_period]
            except KeyError:
                logger.exception(
                    'Problem extracting time period: {}'.format(time_period))
                continue

            labels = dict(time_period=time_period)

            for key in metrics:
                d = tp_stats[key] if key else tp_stats
                for name, metric in metrics[key].items():
                    try:
                        value = d[name]
                        # 'accepted' values are in a list
                        if isinstance(value, list):
                            value = value[0]
                    except KeyError:
                        # 'signal' and 'peak_signal' are not present if
                        # there are no aircraft.
                        if name not in ['peak_signal', 'signal']:
                            logger.warning(
                                "Problem extracting{}item '{}' from: {}".format(
                                    ' {} '.format(key) if key else ' ', name, d))
                        value = math.nan
                    metric.set(labels, value)