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)
python类cpu_freq()的实例源码
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
