Python psutil 模块，cpu_freq() 实例源码

def status(cmd, message, args):
    os_icon, os_color = get_os_icon()
    general_text = f'Latency: **{int(cmd.bot.latency * 1000)}ms**'
    general_text += f'\nPlatform: **{sys.platform.upper()}**'
    general_text += f'\nStarted: **{arrow.get(psutil.boot_time()).humanize()}**'
    cpu_clock = psutil.cpu_freq()
    if cpu_clock:
        cpu_clock = cpu_clock.current
    else:
        cpu_clock = 'Unknown'
    cpu_text = f'Count: **{psutil.cpu_count()} ({psutil.cpu_count(logical=False)})**'
    cpu_text += f'\nUsage: **{psutil.cpu_percent()}%**'
    cpu_text += f'\nClock: **{cpu_clock} MHz**'
    used_mem = humanfriendly.format_size(psutil.virtual_memory().available, binary=True)
    total_mem = humanfriendly.format_size(psutil.virtual_memory().total, binary=True)
    mem_text = f'Used: **{used_mem}**'
    mem_text += f'\nTotal: **{total_mem}**'
    mem_text += f'\nPercent: **{int(100 - psutil.virtual_memory().percent)}%**'
    response = discord.Embed(color=os_color)
    response.set_author(name=socket.gethostname(), icon_url=os_icon)
    response.add_field(name='General', value=general_text)
    response.add_field(name='CPU', value=cpu_text)
    response.add_field(name='Memory', value=mem_text)
    await message.channel.send(embed=response)

def test_cpu_freq(self):
        def check_ls(ls):
            for nt in ls:
                self.assertLessEqual(nt.current, nt.max)
                for name in nt._fields:
                    value = getattr(nt, name)
                    self.assertIsInstance(value, (int, long, float))
                    self.assertGreaterEqual(value, 0)

        ls = psutil.cpu_freq(percpu=True)
        if TRAVIS and not ls:
            return

        assert ls, ls
        check_ls([psutil.cpu_freq(percpu=False)])

        if LINUX:
            self.assertEqual(len(ls), psutil.cpu_count())

def test_cpu_freq_use_second_file(self):
        # https://github.com/giampaolo/psutil/issues/981
        def glob_mock(pattern):
            if pattern.startswith("/sys/devices/system/cpu/cpufreq/policy"):
                flags.append(None)
                return []
            else:
                flags.append(None)
                return orig_glob(pattern)

        flags = []
        orig_glob = glob.glob
        with mock.patch("psutil._pslinux.glob.glob", side_effect=glob_mock,
                        create=True):
            assert psutil.cpu_freq()
            self.assertEqual(len(flags), 2)


# =====================================================================
# --- system CPU stats
# =====================================================================

def status(cmd, message, args):
    os_icon, os_color = get_os_icon()
    general_text = f'Latency: **{int(cmd.bot.latency * 1000)}ms**'
    general_text += f'\nPlatform: **{sys.platform.upper()}**'
    general_text += f'\nStarted: **{arrow.get(psutil.boot_time()).humanize()}**'
    cpu_clock = psutil.cpu_freq()
    if cpu_clock:
        cpu_clock = cpu_clock.current
    else:
        cpu_clock = 'Unknown'
    cpu_text = f'Count: **{psutil.cpu_count()} ({psutil.cpu_count(logical=False)})**'
    cpu_text += f'\nUsage: **{psutil.cpu_percent()}%**'
    cpu_text += f'\nClock: **{cpu_clock} MHz**'
    avail_mem = psutil.virtual_memory().available
    total_mem = psutil.virtual_memory().total
    used_mem = humanfriendly.format_size(total_mem - avail_mem, binary=True)
    total_mem = humanfriendly.format_size(total_mem, binary=True)
    mem_text = f'Used: **{used_mem}**'
    mem_text += f'\nTotal: **{total_mem}**'
    mem_text += f'\nPercent: **{int(psutil.virtual_memory().percent)}%**'
    response = discord.Embed(color=os_color)
    response.set_author(name=socket.gethostname(), icon_url=os_icon)
    response.add_field(name='General', value=general_text)
    response.add_field(name='CPU', value=cpu_text)
    response.add_field(name='Memory', value=mem_text)
    if cmd.bot.cfg.dsc.bot:
        current_shard = message.guild.shard_id
        shard_latency = int(cmd.bot.latencies[current_shard][1] * 1000)
        verbose_description = f'Shard: #{current_shard} | '
        verbose_description += f'Latency: {shard_latency}ms | '
        verbose_description += f'Queue: {cmd.bot.queue.queue.qsize()}'
        response.description = verbose_description
    await message.channel.send(embed=response)

def test_cpu_freq(self):
        self.execute(psutil.cpu_freq)

    # --- mem

def test_cpu_freq(self):
        freq = psutil.cpu_freq()
        self.assertEqual(
            freq.current * 1000 * 1000, sysctl("sysctl hw.cpufrequency"))
        self.assertEqual(
            freq.min * 1000 * 1000, sysctl("sysctl hw.cpufrequency_min"))
        self.assertEqual(
            freq.max * 1000 * 1000, sysctl("sysctl hw.cpufrequency_max"))

    # --- virtual mem

def test_cpu_freq(self):
        w = wmi.WMI()
        proc = w.Win32_Processor()[0]
        self.assertEqual(proc.CurrentClockSpeed, psutil.cpu_freq().current)
        self.assertEqual(proc.MaxClockSpeed, psutil.cpu_freq().max)

def test_cpu_freq(self):
        self.execute(psutil.cpu_freq)

    # --- mem

def test_cpu_freq_no_result(self):
        with mock.patch("psutil._pslinux.glob.glob", return_value=[]):
            self.assertIsNone(psutil.cpu_freq())

def test_cpu_freq(self):
        freq = psutil.cpu_freq()
        self.assertEqual(
            freq.current * 1000 * 1000, sysctl("sysctl hw.cpufrequency"))
        self.assertEqual(
            freq.min * 1000 * 1000, sysctl("sysctl hw.cpufrequency_min"))
        self.assertEqual(
            freq.max * 1000 * 1000, sysctl("sysctl hw.cpufrequency_max"))

    # --- virtual mem

def test_cpu_freq(self):
        w = wmi.WMI()
        proc = w.Win32_Processor()[0]
        self.assertEqual(proc.CurrentClockSpeed, psutil.cpu_freq().current)
        self.assertEqual(proc.MaxClockSpeed, psutil.cpu_freq().max)

def update(self):
        """Update CPU frequency data"""
        def get_avarage_cpu_freq():
            with open("/proc/cpuinfo") as cpuinfo:
                cores_freq = []
                for line in cpuinfo:
                    if "cpu MHz" in line:
                        core_freq = re.findall('\d+\.\d+', line)
                        cores_freq += core_freq
            return round(sum(float(x) for x in cores_freq) / len(cores_freq), 1)

        try:
            cur_freq = int(psutil.cpu_freq().current)
        except:
            cur_freq = 0
            try:
                cur_freq = get_avarage_cpu_freq()
            except:
                cur_freq = 0
                logging.debug("Frequency unavailable")

        if self.stress_started:
            self.samples_taken += 1

        # Here is where we need to generate the max frequency lost

        if self.is_admin and self.samples_taken > self.WAIT_SAMPLES:
            self.perf_lost = int(self.top_freq) - int(cur_freq)
            if self.top_freq != 0:
                self.perf_lost = (round(float(self.perf_lost) / float(self.top_freq) * 100, 1))
            else:
                self.perf_lost = 0
            if self.perf_lost > self.max_perf_lost:
                self.max_perf_lost = self.perf_lost
        elif not self.is_admin:
            self.max_perf_lost = 0

        self.last_freq = cur_freq

def __init__(self, is_admin):
        self.is_admin = is_admin
        self.is_avaiable = True

        self.top_freq = 0
        self.turbo_freq = False
        self.last_freq = 0
        self.samples_taken = 0
        self.WAIT_SAMPLES = 5
        self.perf_lost = 0
        self.max_perf_lost = 0
        self.stress_started = False
        # Top frequency in case using Intel Turbo Boost
        if self.is_admin:
            try:
                num_cpus = psutil.cpu_count()
                logging.debug("num cpus " + str(num_cpus))
                available_freq = read_msr(TURBO_MSR, 0)
                logging.debug(available_freq)
                max_turbo_msr = num_cpus
                # The MSR only holds 8 values. Number of cores could be higher
                if num_cpus > 8:
                    max_turbo_msr = 8
                freq = float(available_freq[max_turbo_msr - 1] * 100)
                if freq > 0:
                    self.top_freq = freq
                    self.turbo_freq = True
            except (Exception) as e:
                logging.debug(e.message)

        if self.turbo_freq == False:
            try:
                self.top_freq = psutil.cpu_freq().max

            except:
                logging.debug("Max freq from psutil not available")
                try:
                    cmd = "lscpu | grep 'CPU max MHz'"
                    ps = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
                    output = ps.communicate()[0]
                    self.top_freq = float(re.findall(b'\d+\.\d+', output)[0])
                    logging.debug("Top freq " + str(self.top_freq))
                    if self.top_freq <= 0:
                        cmd = "lscpu | grep 'CPU * MHz'"
                        ps = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
                        output = ps.communicate()[0]
                        self.top_freq = float(re.findall(b'\d+\.\d+', output)[0])
                except:
                    logging.debug("Max frequency from lscpu not available")
                    logging.debug("CPU top freqency N/A")

        self.update()
        # If top freq not available, take the current as top
        if self.last_freq >= 0 and self.top_freq <=0:
            self.top_freq = self.last_freq
        if self.last_freq <= 0:
            self.is_avaiable = False