Tornado硬件管理平台中的设备信息采集技术深入解析（三）

#-*- coding: utf-8 -*-
import psutil
import time,datetime
from pprint import pprint
#定义一个专门用于获取系统信息的类
class Moniter(object):
#转换方法
def byte_to_gb(self,data,key=""):
if key=="percent":
return data
else:
return round(data/(1024**3),2)
def cpu(self):
data = dict(
percent_avg=psutil.cpu_percent(interval=0, percpu=False),
percent_per = psutil.cpu_percent(interval=0, percpu=True),
num_p = psutil.cpu_count(logical=False),
num_l = psutil.cpu_count(logical=True),
)
return data
def men(self):
info = psutil.virtual_memory()
data = dict(
total = self.byte_to_gb(info.total),
used = self.byte_to_gb(info.used),
free = self.byte_to_gb(info.free),
percent = self.byte_to_gb(info.percent),
)
return data
def swap(self):
info = psutil.swap_memory()
data = dict(
total = self.byte_to_gb(info.total),
used = self.byte_to_gb(info.used),
free = self.byte_to_gb(info.free),
percent = self.byte_to_gb(info.percent)
)
return data
#获取磁盘信息
def disk(self):
info = psutil.disk_partitions()
data =[
dict(
device = v.device,
mountpoint = v.mountpoint,
fstype = v.fstype,
opts = v.opts,
used = {
k: self.byte_to_gb(v,k)
for k,v in psutil.disk_usage(v.mountpoint)._asdict().items()
}
)
for v in info
]
return data
#获取网络信息
def net(self):
addrs = psutil.net_if_addrs()
addrs_info = {
k:[
dict(
family = val.family.name,
address = val.address,
netmask = val.netmask,
broadcast = val.broadcast,
)
for val in v if val.family.name == "AF_INET"
][0]
for k,v in addrs.items()
}
#收集io的输入输出信息
io = psutil.net_io_counters(pernic=True)
data = [
dict(
name = k,
bytes_sent = v.bytes_sent,
bytes_recv = v.bytes_recv,
packets_sent = v.packets_sent,
packets_recv = v.packets_recv,
**addrs_info[k],
)
for k,v in io.items()
]
return data
#时间戳转换
def td(self,tm):
dt = datetime.datetime.fromtimestamp(tm)
return dt.strftime("%Y-%m-%d %H:%M:%S")
#获取开机时间
def lastest_start_time(self):
return self.td(psutil.boot_time())
#登录用户
def logined_user(self):
users = psutil.users()
data =[
dict(
name=v.name,
terminal=v.terminal,
host=v.host,
started=self.td(v.started),
pid=v.pid
)
for v in users
]
return data
if __name__ == "__main__":
m = Moniter()
pprint(m.logined_user())
# for i in range(10):
# print(m.cpu())
# time.sleep(1)