Coverage for versioalueet/env.py: 94.25%

1"""Report facts from the environment.""" 

2 

3import argparse 

4import importlib.util 

5import json 

6import os 

7import platform 

8 

9if importlib.util.find_spec('resource'): 

10 import resource 

11else: # pragma: no cover 

12 pass 

13import sys 

14import uuid 

15from typing import Union 

16from versioalueet import ENCODING, ENCODING_ERRORS_POLICY, VERSION 

17 

18VolatileDictType = dict[str, Union[str, bool, float, int]] 

19EnvType = dict[str, dict[str, VolatileDictType]] 

20FormatType = str 

21FORMATS = ('text', 'dict', 'json') 

22 

23 

24def assess(options: argparse.Namespace) -> EnvType: 

25 """Assess process environment with standard library functions.""" 

26 if not platform.platform(aliased=True, terse=True).lower().startswith('windows'): 26 ↛ 32line 26 didn't jump to line 32 because the condition on line 26 was always true

27 os_uname = os.uname() 

28 os_sysname = os_uname.sysname 

29 os_nodename = os_uname.nodename 

30 os_version = os_uname.version 

31 else: 

32 pf_uname = platform.uname() 

33 os_sysname = pf_uname.system 

34 os_nodename = pf_uname.node 

35 os_version = pf_uname.version 

36 os_cpu_present = os.cpu_count() 

37 os_cpu_available = len(os.sched_getaffinity(0)) if 'sched_getaffinity' in dir(os) else -1 # type: ignore 

38 

39 names = sorted( 

40 name 

41 for name in dir(sys.flags) 

42 if not name.startswith('_') and not name.startswith('n_') and name not in ('count', 'index') 

43 ) 

44 flags = {name: getattr(sys.flags, name) for name in names if getattr(sys.flags, name)} 

45 

46 if 'resource' in sys.modules: 

47 res_self = resource.getrusage(resource.RUSAGE_SELF) 

48 ru_utime_msec = res_self.ru_utime # time spent executing in user mode in seconds (usec resolution) 

49 ru_utime_msec *= 1e3 

50 ru_utime_msec_usec_precision = round(ru_utime_msec, 3) 

51 ru_stime_msec = res_self.ru_stime # time spent executing in kernel mode in seconds (usec resolution) 

52 ru_stime_msec *= 1e3 

53 ru_stime_msec_usec_precision = round(ru_stime_msec, 3) 

54 ru_maxrss = float(res_self.ru_maxrss) # maximum resident set size used (linux in kilobytes) 

55 if platform.platform(aliased=True, terse=True).lower().startswith('macos'): # pragma: no cover 

56 ru_maxrss /= 1024 # "man 2 getrusage" on MacOS indicates unit is bytes 

57 ru_maxrss_mbytes_kbytes_precision = round(ru_maxrss / 1024, 3) 

58 ru_minflt = res_self.ru_minflt # number of page faults serviced without any I/O activity 

59 ru_majflt = res_self.ru_majflt # number of page faults serviced that required I/O activity 

60 ru_inblock = res_self.ru_inblock # number of times the filesystem had to perform input 

61 ru_oublock = res_self.ru_oublock # number of times the filesystem had to perform output 

62 ru_nvcsw = res_self.ru_nvcsw # number of times a context switch resulted due to a process 

63 # voluntarily giving up the processor before its time slice 

64 # was completed 

65 ru_nivcsw = res_self.ru_nivcsw # number of times a context switch resulted due to a higher 

66 # priority process becoming runnable or because the current 

67 # process exceeded its time slice 

68 else: 

69 ru_utime_msec_usec_precision = -1 

70 ru_stime_msec_usec_precision = -1 

71 ru_maxrss_mbytes_kbytes_precision = -1 

72 ru_minflt = -1 

73 ru_majflt = -1 

74 ru_inblock = -1 

75 ru_oublock = -1 

76 ru_nvcsw = -1 

77 ru_nivcsw = -1 

78 

79 data = { 

80 'library-env': { 

81 'debug-mode': options.debug, 

82 'quiet-mode': options.quiet, 

83 'verbose-mode': options.verbose, 

84 'version': VERSION, 

85 'encoding': ENCODING, 

86 'encoding-errors-policy': ENCODING_ERRORS_POLICY, 

87 }, 

88 'interpreter-env': { 

89 'exec-prefix': sys.exec_prefix, 

90 'exec-path': sys.executable, 

91 }, 

92 'interpreter-impl': { 

93 'impl-name': sys.implementation.name, 

94 'version': { 

95 'major': sys.implementation.version.major, 

96 'minor': sys.implementation.version.minor, 

97 'micro': sys.implementation.version.micro, 

98 'releaselevel': sys.implementation.version.releaselevel, 

99 'serial': sys.implementation.version.serial, 

100 }, 

101 }, 

102 'interpreter-flags': { 

103 **flags, 

104 }, 

105 'os-env': { 

106 'node-id': str(uuid.uuid3(uuid.NAMESPACE_DNS, platform.node())), 

107 'machine-type': platform.machine(), 

108 'platform-code': platform.platform(aliased=True, terse=True), 

109 'platform_release': platform.release(), 

110 }, 

111 'os-uname': { 

112 'os-sysname': os_sysname, 

113 'os-nodename': os_nodename, 

114 'os-version': os_version, 

115 }, 

116 'os-resource-usage': { 

117 'ru-maxrss-mbytes-kbytes-precision': ru_maxrss_mbytes_kbytes_precision, 

118 'ru-utime-msec-usec-precision': ru_utime_msec_usec_precision, 

119 'ru-stime-msec-usec-precision': ru_stime_msec_usec_precision, 

120 'ru-minflt': ru_minflt, 

121 'ru-majflt': ru_majflt, 

122 'ru-inblock': ru_inblock, 

123 'ru-outblock': ru_oublock, 

124 'ru_nvcsw': ru_nvcsw, 

125 'ru_nivcsw': ru_nivcsw, 

126 }, 

127 'os-cpu-resources': { 

128 'os-cpu-present': os_cpu_present, 

129 'os-cpu-available': os_cpu_available, 

130 }, 

131 } 

132 

133 return data 

134 

135 

136def report(options: argparse.Namespace, format: FormatType = 'text') -> Union[str, EnvType]: 

137 """Assess process environment and provide report in JSON or text format.""" 

138 data = assess(options) 

139 

140 if format == 'dict': 

141 return data 

142 

143 if format == 'json': 

144 return json.dumps(data, indent=2) 

145 

146 # Format is text 

147 text = [] 

148 for pk, pv in data.items(): 

149 line = f'{pk}: ' 

150 values = [] 

151 for sk, sv in pv.items(): 

152 if not isinstance(sv, dict): 

153 values.append(f'{sk}={sv}') 

154 else: 

155 tvs = ', '.join([f'{tk}={tv}' for tk, tv in sv.items()]) 

156 values.append(f'{sk}({tvs})') 

157 text.append(line + ', '.join(values)) 

158 

159 return '\n'.join(text)