我们从Python开源项目中，提取了以下13个代码示例，用于说明如何使用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_no_result(self): with mock.patch("psutil._pslinux.glob.glob", return_value=[]): self.assertIsNone(psutil.cpu_freq())
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
