Coverage for moptipy / utils / sys_info.py: 79%
194 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-14 07:09 +0000
1"""A tool for writing a section with system information into log files."""
2import contextlib
3import importlib.metadata as ilm
4import os
5import platform
6import re
7import socket
8import sys
9from contextlib import suppress
10from datetime import UTC, datetime
11from typing import Final, Iterable
13import psutil # type: ignore
14from pycommons.io.csv import CSV_SEPARATOR, SCOPE_SEPARATOR
15from pycommons.io.path import Path
16from pycommons.processes.caller import is_build
18import moptipy.version as ver
19from moptipy.api import logging
20from moptipy.utils.logger import (
21 KEY_VALUE_SEPARATOR,
22 InMemoryLogger,
23 KeyValueLogSection,
24 Logger,
25)
28def __cpu_affinity(proc: psutil.Process | None = None) -> str | None:
29 """
30 Get the CPU affinity.
32 :param proc: the process handle
33 :return: the CPU affinity string.
34 """
35 if proc is None:
36 proc = psutil.Process()
37 if proc is None:
38 return None
39 cpua = proc.cpu_affinity()
40 if cpua is not None:
41 cpua = CSV_SEPARATOR.join(map(str, cpua))
42 if len(cpua) > 0:
43 return cpua
44 return None
47#: the dependencies
48__DEPENDENCIES: set[str] | None = {
49 "cmaes", "contourpy", "cycler", "fonttools", "intel-cmplr-lib-rt",
50 "joblib", "kiwisolver", "llvmlite", "matplotlib", "moptipy", "numba",
51 "numpy", "packaging", "pillow", "psutil", "pycommons", "pyparsing",
52 "python-dateutil", "scipy", "setuptools", "six", "threadpoolctl"}
55def add_dependency(dependency: str,
56 ignore_if_make_build: bool = False) -> None:
57 """
58 Add a dependency so that its version can be stored in log files.
60 Warning: You must add all dependencies *before* the first log file is
61 written. As soon as the :func:`log_sys_info` is invoked for the first
62 time, adding new dependencies will cause an error. And writing a log
63 file via the :mod:`~moptipy.api.experiment` API or to a file specified in
64 the :mod:`~moptipy.api.execution` API will invoke this function.
66 :param dependency: the basic name of the library, exactly as you would
67 `import` it in a Python module. For example, to include the version of
68 `numpy` in the log files, you would do `add_dependency("numpy")` (of
69 course, the version of `numpy` is already automatically included
70 anyway).
71 :param ignore_if_make_build: should this dependency be ignored if this
72 method is invoked during a `make` build? This makes sense if the
73 dependency itself is a package which is uninstalled and then
74 re-installed during a `make` build process. In such a situation, the
75 dependency version may be unavailable and cause an exception.
76 :raises TypeError: if `dependency` is not a string
77 :raises ValueError: if `dependency` is an invalid string or the log
78 information has already been accessed before and modifying it now is
79 not permissible.
80 """
81 if (str.__len__(dependency) <= 0) or (dependency != str.strip(dependency))\
82 or (" " in dependency):
83 raise ValueError(f"Invalid dependency string {dependency!r}.")
84 if __DEPENDENCIES is None:
85 raise ValueError(
86 f"Too late. Cannot add dependency {dependency!r} anymore.")
87 if ignore_if_make_build and is_build():
88 return
89 __DEPENDENCIES.add(dependency)
92# noinspection PyBroadException
93def __make_sys_info() -> str:
94 """
95 Build the system info string.
97 This method is only used once and then deleted.
99 :returns: the system info string.
100 """
101 global __DEPENDENCIES # noqa: PLW0603 # pylint: disable=W0603
102 if __DEPENDENCIES is None:
103 raise ValueError("Cannot re-create log info.")
104 dep: Final[set[str]] = __DEPENDENCIES
105 __DEPENDENCIES = None # noqa: PLW0603 # pylint: disable=W0603
107 def __v(sec: KeyValueLogSection, key: str, value) -> None:
108 """
109 Create a key-value pair if value is not empty.
111 :param sec: the section to write to.
112 :param key: the key
113 :param value: an arbitrary value, maybe consisting of multiple lines.
114 """
115 if value is None:
116 return
117 value = str.strip(" ".join([str.strip(ts) for ts in
118 str.strip(str(value)).split("\n")]))
119 if str.__len__(value) <= 0:
120 return
121 sec.key_value(key, value)
123 def __get_processor_name() -> str | None:
124 """
125 Get the processor name.
127 :returns: a string if there is any processor information
128 """
129 with contextlib.suppress(Exception):
130 if platform.system() == "Windows":
131 return platform.processor()
132 if platform.system() == "Linux":
133 with Path("/proc/cpuinfo").open_for_read() as rd:
134 for line in rd:
135 if "model name" in line:
136 return re.sub(
137 pattern=".*model name.*:",
138 repl="", string=line, count=1).strip()
139 return None
141 def __get_mem_size_sysconf() -> int | None:
142 """
143 Get the memory size information from sysconf.
145 :returns: an integer with the memory size if available
146 """
147 with contextlib.suppress(Exception):
148 k1 = "SC_PAGESIZE"
149 if k1 not in os.sysconf_names:
150 k1 = "SC_PAGE_SIZE"
151 if k1 not in os.sysconf_names:
152 return None
153 v1 = os.sysconf(k1)
154 if not isinstance(v1, int):
155 return None
156 k2 = "SC_PHYS_PAGES"
157 if k2 not in os.sysconf_names:
158 k2 = "_SC_PHYS_PAGES"
159 if k2 not in os.sysconf_names:
160 return None
161 v2 = os.sysconf(k2)
162 if not isinstance(v1, int):
163 return None
165 if (v1 > 0) and (v2 > 0):
166 return v1 * v2
167 return None
169 def __get_mem_size_meminfo() -> int | None:
170 """
171 Get the memory size information from meminfo.
173 :returns: an integer with the memory size if available
174 """
175 with contextlib.suppress(Exception):
176 with Path("/proc/meminfo").open_for_read() as rd:
177 meminfo = {i.split()[0].rstrip(":"): int(i.split()[1])
178 for i in rd}
179 mem_kib = meminfo["MemTotal"] # e.g. 3921852
180 mem_kib = int(mem_kib)
181 if mem_kib > 0:
182 return 1024 * mem_kib
183 return None
185 def __get_mem_size() -> int | None:
186 """
187 Get the memory size information from any available source.
189 :returns: an integer with the memory size if available
190 """
191 vs = __get_mem_size_sysconf()
192 if vs is None:
193 vs = __get_mem_size_meminfo()
194 if vs is None:
195 return psutil.virtual_memory().total
196 return vs
198 # log all information in memory to convert it to one constant string.
199 with InMemoryLogger() as imr:
200 with imr.key_values(logging.SECTION_SYS_INFO) as kv:
201 with kv.scope(logging.SCOPE_SESSION) as k:
202 __v(k, logging.KEY_SESSION_START,
203 datetime.now(tz=UTC))
204 __v(k, logging.KEY_NODE_NAME, platform.node())
205 proc = psutil.Process()
206 __v(k, logging.KEY_PROCESS_ID, hex(proc.pid))
207 cpua = __cpu_affinity(proc)
208 if cpua is not None:
209 __v(k, logging.KEY_CPU_AFFINITY, cpua)
210 del proc, cpua
212 # get the command line and working directory of the process
213 with contextlib.suppress(Exception):
214 proc = psutil.Process(os.getpid())
215 cmd = proc.cmdline()
216 if isinstance(cmd, Iterable):
217 __v(k, logging.KEY_COMMAND_LINE,
218 " ".join(map(repr, proc.cmdline())))
219 cwd = proc.cwd()
220 if isinstance(cwd, str):
221 __v(k, logging.KEY_WORKING_DIRECTORY,
222 repr(proc.cwd()))
224 # see https://stackoverflow.com/questions/166506/.
225 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
226 try:
227 # doesn't even have to be reachable
228 s.connect(("10.255.255.255", 1))
229 ip = s.getsockname()[0]
230 except Exception: # noqa
231 ip = "127.0.0.1"
232 finally:
233 s.close()
234 __v(k, logging.KEY_NODE_IP, ip)
236 with kv.scope(logging.SCOPE_VERSIONS) as k:
237 for package in sorted(dep):
238 if package == "moptipy":
239 __v(k, "moptipy", ver.__version__)
240 else:
241 use_v: str | None = None
242 with suppress(BaseException):
243 use_v = str.strip(ilm.version(package))
244 if (use_v is None) or (str.__len__(use_v) <= 0):
245 use_v = "notFound"
246 __v(k, package.replace("-", ""), use_v)
248 with kv.scope(logging.SCOPE_HARDWARE) as k:
249 __v(k, logging.KEY_HW_MACHINE, platform.machine())
250 __v(k, logging.KEY_HW_N_PHYSICAL_CPUS,
251 psutil.cpu_count(logical=False))
252 __v(k, logging.KEY_HW_N_LOGICAL_CPUS,
253 psutil.cpu_count(logical=True))
255 # store the CPU speed information
256 cpuf: dict[tuple[int, int], int] = {}
257 total: int = 0
258 for cf in psutil.cpu_freq(True):
259 t = (int(cf.min), int(cf.max))
260 cpuf[t] = cpuf.get(t, 0) + 1
261 total += 1
262 memlst: list[tuple[int, ...]]
263 if total > 1:
264 memlst = [(key[0], key[1], value) for
265 key, value in cpuf.items()]
266 memlst.sort()
267 else:
268 memlst = list(cpuf)
270 def __make_mhz_str(tpl: tuple[int, ...]) -> str:
271 """Convert a MHz tuple to a string."""
272 base: str = f"({tpl[0]}MHz..{tpl[1]}MHz)" \
273 if tpl[1] > tpl[0] else f"{tpl[0]}MHz"
274 return base if (len(tpl) < 3) or (tpl[2] <= 1) \
275 else f"{base}*{tpl[2]}"
276 __v(k, logging.KEY_HW_CPU_MHZ,
277 "+".join([__make_mhz_str(t) for t in memlst]))
279 __v(k, logging.KEY_HW_BYTE_ORDER, sys.byteorder)
280 __v(k, logging.KEY_HW_CPU_NAME, __get_processor_name())
281 __v(k, logging.KEY_HW_MEM_SIZE, __get_mem_size())
283 with kv.scope(logging.SCOPE_PYTHON) as k:
284 __v(k, logging.KEY_PYTHON_VERSION, sys.version)
285 __v(k, logging.KEY_PYTHON_IMPLEMENTATION,
286 platform.python_implementation())
288 with kv.scope(logging.SCOPE_OS) as k:
289 __v(k, logging.KEY_OS_NAME, platform.system())
290 __v(k, logging.KEY_OS_RELEASE, platform.release())
291 __v(k, logging.KEY_OS_VERSION, platform.version())
293 lst = imr.get_log()
295 if len(lst) < 3:
296 raise ValueError("sys info turned out to be empty?")
297 __DEPENDENCIES = None
298 return "\n".join(lst[1:(len(lst) - 1)])
301def get_sys_info() -> str:
302 r"""
303 Get the system information as string.
305 :returns: the system information as string
307 >>> raw_infos = get_sys_info()
308 >>> raw_infos is get_sys_info() # caching!
309 True
310 >>> for k in raw_infos.split("\n"):
311 ... print(k[:k.find(": ")])
312 session.start
313 session.node
314 session.processId
315 session.cpuAffinity
316 session.commandLine
317 session.workingDirectory
318 session.ipAddress
319 version.cmaes
320 version.contourpy
321 version.cycler
322 version.fonttools
323 version.intelcmplrlibrt
324 version.joblib
325 version.kiwisolver
326 version.llvmlite
327 version.matplotlib
328 version.moptipy
329 version.numba
330 version.numpy
331 version.packaging
332 version.pillow
333 version.psutil
334 version.pycommons
335 version.pyparsing
336 version.pythondateutil
337 version.scipy
338 version.setuptools
339 version.six
340 version.threadpoolctl
341 hardware.machine
342 hardware.nPhysicalCpus
343 hardware.nLogicalCpus
344 hardware.cpuMhz
345 hardware.byteOrder
346 hardware.cpu
347 hardware.memSize
348 python.version
349 python.implementation
350 os.name
351 os.release
352 os.version
353 """
354 the_object: Final[object] = get_sys_info
355 the_attr: Final[str] = "__the_sysinfo"
356 if hasattr(the_object, the_attr):
357 return getattr(the_object, the_attr)
358 sys_info: Final[str] = __make_sys_info()
359 setattr(the_object, the_attr, sys_info)
360 return sys_info
363def update_sys_info_cpu_affinity() -> None:
364 """Update the CPU affinity of the system information."""
365 sys_info_str = get_sys_info()
366 start = (f"\n{logging.SCOPE_SESSION}{SCOPE_SEPARATOR}"
367 f"{logging.KEY_CPU_AFFINITY}{KEY_VALUE_SEPARATOR}")
368 start_i = sys_info_str.find(start)
369 if start_i < 0:
370 return # no affinity, don't need to update
371 start_i += len(start)
372 end_i = sys_info_str.find("\n", start_i)
373 if end_i <= start_i:
374 raise ValueError(f"Empty {logging.KEY_CPU_AFFINITY}?")
375 affinity = __cpu_affinity()
376 if affinity is None:
377 raise ValueError(
378 f"first affinity query is {sys_info_str[start_i:end_i]},"
379 f" but second one is None?")
380 sys_info_str = f"{sys_info_str[:start_i]}{affinity}{sys_info_str[end_i:]}"
382 the_object: Final[object] = get_sys_info
383 the_attr: Final[str] = "__the_sysinfo"
384 setattr(the_object, the_attr, sys_info_str)
387def log_sys_info(logger: Logger) -> None:
388 r"""
389 Write the system information section to a logger.
391 The concept of this method is that we only construct the whole system
392 configuration exactly once in memory and then always directly flush it
393 as a string to the logger. This is much more efficient than querying it
394 every single time.
396 :param logger: the logger
398 >>> from moptipy.utils.logger import InMemoryLogger
399 >>> with InMemoryLogger() as l:
400 ... log_sys_info(l)
401 ... log = l.get_log()
402 >>> print(log[0])
403 BEGIN_SYS_INFO
404 >>> print(log[-1])
405 END_SYS_INFO
406 >>> log[1:-1] == get_sys_info().split('\n')
407 True
408 """
409 with logger.text(logging.SECTION_SYS_INFO) as txt:
410 txt.write(get_sys_info())