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