Coverage for taksonomia/machine.py: 100.00%

1"""Taxonomy (Finnish: taksonomia) guided by conventions of a folder tree (implementation).""" 

2 

3import datetime as dti 

4import os 

5import platform 

6import uuid 

7from typing import no_type_check 

8 

9import psutil # type: ignore 

10from cpuinfo import get_cpu_info # type: ignore 

11 

12from taksonomia import TS_FORMAT 

13 

14 

15@no_type_check 

16class Machine: 

17 """Auxiliary context and performance information provider.""" 

18 

19 def __init__(self, path_selector: str, pid: int) -> None: 

20 """Construct a machine instance pinned on a process id and a path.""" 

21 self.path_selector = path_selector 

22 self.pid = pid 

23 

24 @no_type_check 

25 def perf(self): 

26 """Some random information from machine performance measures.""" 

27 p = psutil.Process(self.pid) 

28 cpts = p.cpu_times() 

29 cpts_union_attrs = ('children_system', 'children_user', 'system', 'user') 

30 mfi = p.memory_full_info() 

31 mfi_union_attrs = ('pageins', 'pfaults', 'rss', 'uss', 'vms') 

32 threads = p.threads() 

33 thr_union_attrs = ('id', 'system_time', 'user_time') 

34 ctxs = p.num_ctx_switches() 

35 ctxs_union_attrs = ('involuntary', 'voluntary') 

36 return { 

37 'name': p.name() if hasattr(p, 'name') else None, 

38 'status': p.status() if hasattr(p, 'status') else None, 

39 'create_time': dti.datetime.fromtimestamp(p.create_time(), dti.timezone.utc).strftime(TS_FORMAT), 

40 'username': p.username() if hasattr(p, 'username') else None, 

41 'cpu_times': {aspect: getattr(cpts, aspect, None) for aspect in cpts_union_attrs}, 

42 'memory_full_info': {aspect: getattr(mfi, aspect, None) for aspect in mfi_union_attrs}, 

43 'memory_percent': p.memory_percent() if hasattr(p, 'memory_percent') else None, 

44 'num_threads': p.num_threads() if hasattr(p, 'num_threads') else None, 

45 'threads': [{aspect: getattr(thr, aspect, None) for aspect in thr_union_attrs} for thr in threads], 

46 'num_fds': p.num_fds() if hasattr(p, 'num_fds') else None, 

47 'num_ctx_switches': {aspect: getattr(ctxs, aspect, None) for aspect in ctxs_union_attrs}, 

48 } 

49 

50 @no_type_check 

51 def context(self): 

52 """Some random information from machine context.""" 

53 swa = psutil.swap_memory() 

54 swa_union_attrs = ('free', 'percent', 'sin', 'sout', 'total', 'used') 

55 vim = psutil.virtual_memory() 

56 vim_union_attrs = ( 

57 'active', 

58 'available', 

59 'buffers', 

60 'cached', 

61 'free', 

62 'inactive', 

63 'percent', 

64 'shared', 

65 'slab', 

66 'total', 

67 'used', 

68 'wired', 

69 ) 

70 dic = psutil.disk_io_counters(perdisk=False) 

71 dic_union_attrs = ('read_bytes', 'read_count', 'read_time', 'write_bytes', 'write_count', 'write_time') 

72 du = psutil.disk_usage(self.path_selector) 

73 du_union_attrs = ('free', 'percent', 'total', 'used') 

74 dpas = psutil.disk_partitions() 

75 dpa_union_attrs = ('device', 'fstype', 'maxfile', 'maxpath', 'mountpoint') 

76 partitions = [] 

77 for dpa in dpas: 

78 partitions.append( 

79 { 

80 **{aspect: getattr(dpa, aspect, None) for aspect in dpa_union_attrs}, 

81 'opts': dpa.opts.split(',') if hasattr(dpa, 'opts') else None, 

82 } 

83 ) 

84 

85 return { 

86 'node_identifier': str(uuid.uuid3(uuid.NAMESPACE_DNS, platform.node())), 

87 'pid': os.getpid(), 

88 'ppid': os.getppid(), 

89 'boottime': dti.datetime.fromtimestamp(psutil.boot_time(), dti.timezone.utc).strftime(TS_FORMAT), 

90 'cpu_info': {**get_cpu_info()}, 

91 'memory': { 

92 'swap': {aspect: getattr(swa, aspect, None) for aspect in swa_union_attrs}, 

93 'virtual': {aspect: getattr(vim, aspect, None) for aspect in vim_union_attrs}, 

94 }, 

95 'disks': { 

96 'counters_combined': {aspect: getattr(dic, aspect, None) for aspect in dic_union_attrs}, 

97 'partitions': partitions, 

98 'path_selector': self.path_selector, 

99 'usage': {aspect: getattr(du, aspect, None) for aspect in du_union_attrs}, 

100 }, 

101 }