Coverage for moptipy/evaluation/machine_spec.py: 54%
293 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-03 10:46 +0000
1"""Load machine specs from log files."""
3from contextlib import suppress
4from dataclasses import dataclass, field
5from typing import Callable, Final, Iterable
7from pycommons.io.csv import COMMENT_START
8from pycommons.io.path import Path
9from pycommons.types import type_error
11from moptipy.api.logging import SECTION_SYS_INFO
12from moptipy.utils.logger import SECTION_END, SECTION_START, InMemoryLogger
13from moptipy.utils.sys_info import log_sys_info
16@dataclass(frozen=True, init=True, order=True, eq=True)
17class Machine:
18 """
19 An immutable record of machine information.
21 >>> m = Machine(
22 ... machine_id="A",
23 ... ram_bytes=16853479424,
24 ... os="Ubuntu Linux",
25 ... cpu="Intel64 Family 6 Model 151",
26 ... cpu_mhz=2100,
27 ... python="3.12.13")
28 >>> m.setup_str()
29 'Python 3.12.13 on an Intel64 Family 6 Model 151 CPU at 2.1 GHz with \
3015.7 GiB RAM and Ubuntu Linux'
31 """
33 #: The machine id
34 machine_id: str
36 #: the amount of memory
37 ram_bytes: int | None = field(default=None)
39 #: the operating system
40 os: str | None = field(default=None)
42 #: the CPU
43 cpu: str | None = field(default=None)
45 #: the mhz
46 cpu_mhz: int | None = field(default=None)
48 #: the python version
49 python: str | None = field(default=None)
51 def setup_str(self) -> str:
52 """
53 Get the system setup as string.
55 :return: the setup string
56 """
57 result = ""
58 if self.python is not None:
59 result = f"Python {self.python}"
61 if self.cpu is not None:
62 if str.__len__(result) > 0:
63 schr = "an" if str.lower(self.cpu[0]) in "aeiou" else "a"
64 result = f"{result} on {schr} "
65 result = f"{result}{self.cpu} CPU"
66 if self.cpu_mhz is not None:
67 speed = f"{self.cpu_mhz / 1000:.1f}".removesuffix(".0")
68 result = f"{result} at {speed} GHz"
69 if self.ram_bytes is not None:
70 ram = f"{self.ram_bytes / 1073741824:.1f}".removesuffix(".0")
71 if str.__len__(result) > 0:
72 result = f"{result} with "
73 result = f"{result}{ram} GiB RAM"
74 if self.os is not None:
75 if str.__len__(result) > 0:
76 result = f"{result} and "
77 result = f"{result}{self.os}"
78 return result
81#: the architecture key
82__ARCH_KEY: Final[str] = "hardware.machine"
83#: the machine key
84__MACHINE_KEY: Final[str] = "session.node"
85#: the CPU key
86__CPU_KEY: Final[str] = "hardware.cpu"
87#: the MHz key
88__MHZ_KEY: Final[str] = "hardware.cpuMhz"
89#: the ram key
90__RAM_KEY: Final[str] = "hardware.memSize"
91#: the os name
92__PYTHON_KEY: Final[str] = "python.version"
93#: the os name key
94__OS_NAME_KEY: Final[str] = "os.name"
95#: the os release key
96__OS_RELEASE_KEY: Final[str] = "os.release"
97#: the os version key
98__OS_VERSION_KEY: Final[str] = "os.version"
100#: the keys
101__KEYS: Final[set[str]] = {
102 __ARCH_KEY, __MACHINE_KEY, __CPU_KEY, __MHZ_KEY, __RAM_KEY, __PYTHON_KEY,
103 __OS_NAME_KEY, __OS_RELEASE_KEY, __OS_VERSION_KEY}
105#: the section start
106__SEC_START: Final[str] = f"{SECTION_START}{SECTION_SYS_INFO}"
107#: the section end
108__SEC_END: Final[str] = f"{SECTION_END}{SECTION_SYS_INFO}"
110#: drop this text
111__CPU_DROP: Final[tuple[str, ...]] = (
112 "genuineintel", "stepping", "authenticamd")
113#: drop this text
114__CPU_DROP_CORES: Final[tuple[str, ...]] = ("-core processor", )
117def __load_machine(file: Path, result: dict[str, Machine]) -> None:
118 """
119 Load a machine record from a log file.
121 :param file: the path to load from
122 """
123 state: int = 0
124 data: dict[str, str] = {}
125 with file.open_for_read() as stream:
126 for srow in stream:
127 row = str.strip(srow)
128 if (str.__len__(row) <= 0) or row.startswith(COMMENT_START):
129 continue
130 if str.__eq__(row, __SEC_START):
131 if state != 0:
132 raise ValueError(f"{__SEC_START!r} appears twice?")
133 state = 1
134 continue
135 if str.__eq__(row, __SEC_END):
136 if state != 1:
137 raise ValueError(f"{__SEC_END!r} before {__SEC_START!r}?")
138 state = 2
139 continue
141 if state == 1:
142 dot = str.find(row, ":")
143 if dot > 0:
144 key = str.strip(row[:dot])
145 if key in __KEYS:
146 val = str.strip(row[dot + 1:])
147 if str.__len__(val) > 0:
148 data[key] = val
150 if __MACHINE_KEY not in data:
151 return
152 machine_id = data[__MACHINE_KEY]
153 if machine_id in result:
154 return
156 try:
157 ram = int(data[__RAM_KEY])
158 except (KeyError, ValueError):
159 ram = None
161 cpu: str | None = data.get(__CPU_KEY)
162 if cpu is None:
163 cpu = data.get(__ARCH_KEY)
164 if cpu is not None:
165 ncpu = cpu
166 new_len = str.__len__(ncpu)
167 old_len = new_len + 1
168 while old_len > new_len:
169 old_len = new_len
170 ncpu_l = str.lower(ncpu)
171 for d in __CPU_DROP:
172 di = str.rfind(ncpu_l, d)
173 if di > 0:
174 ncpu = str.strip(ncpu[:di])
175 ncpu_l = str.lower(ncpu)
177 for d in __CPU_DROP_CORES:
178 if ncpu_l.endswith(d):
179 ncpux = str.strip(ncpu[:-str.__len__(d)])
180 v = str.rfind(ncpux, " ")
181 if 0 < v < (str.__len__(ncpux) - 1):
182 ncpu = str.strip(ncpu[:v])
184 ncpu = str.removesuffix(ncpu, ",")
185 new_len = str.__len__(ncpu)
186 if str.__len__(ncpu) > 0:
187 cpu = ncpu
189 mhz: int | None = None
190 mhz_str: str | None = data.get(__MHZ_KEY)
191 if mhz_str is not None:
192 di = str.rfind(mhz_str, "*")
193 if 0 < di < (str.__len__(mhz_str) - 1):
194 mhz_str = str.strip(mhz_str[:di])
195 mhz_str = str.strip(str.removesuffix(str.removeprefix(
196 mhz_str, "("), ")"))
197 di = str.rfind(mhz_str, ".")
198 if 0 < di < (str.__len__(mhz_str) - 1):
199 mhz_str = str.strip(mhz_str[di + 1:])
200 mhz_str = mhz_str.removesuffix("MHz")
201 with suppress(ValueError):
202 mhz = int(mhz_str)
204 python: str | None = data.get(__PYTHON_KEY)
205 if python is not None:
206 di = str.find(python, "|")
207 if di > 0:
208 python = str.strip(python[:di])
209 di = str.find(python, " ")
210 if di > 0:
211 python = str.strip(python[:di])
212 di = str.find(python, ".")
213 if di > 0:
214 si = str.find(python, ".", di + 1)
215 if si > di:
216 python = str.strip(python[:si])
218 os: str | None = data.get(__OS_NAME_KEY)
219 if os is not None:
220 os_release: str | None = data.get(__OS_RELEASE_KEY)
221 os_version: str | None = data.get(__OS_VERSION_KEY)
223 if str.lower(os) == "linux":
224 if os_version is not None:
225 di = os_version.find(" ")
226 if di > 0:
227 os_version = str.strip(os_version[:di])
228 os = f"{os_version} Linux"
229 if os_release is not None:
230 di = os_release.find("-")
231 if di > 0:
232 os_release = str.strip(os_release[:di])
233 os = f"{os}, {os_release} Kernel"
234 elif os_release is not None:
235 os = f"{os} {os_release}"
236 elif os_version is not None:
237 os = f"{os} {os_version}"
239 result[machine_id] = Machine(
240 machine_id=machine_id,
241 ram_bytes=ram,
242 os=os,
243 cpu=cpu,
244 cpu_mhz=mhz,
245 python=python)
248def load_machines(source: str,
249 result: dict[str, Machine]) -> None:
250 """
251 Load the machine data from logs and store them under the given dictionary.
253 :param source: the path to load from
254 :param result: the dictionary of machine-IDs and machine records.
255 """
256 path: Final[Path] = Path(source)
257 if not isinstance(result, dict):
258 raise type_error(result, "result", dict)
259 if path.is_dir():
260 for sub in path.list_dir():
261 load_machines(sub, result)
262 elif path.is_file():
263 with path.open_for_read() as stream:
264 machine = get_machine(stream, result.__contains__)
265 if machine is not None:
266 result[machine.machine_id] = machine
267 else:
268 raise ValueError(
269 f"{path!r} identifies neither a file nor a directory.")
272def get_machine(
273 stream: Iterable[str],
274 can_skip: Callable[[str], bool] = lambda _: False) -> Machine | None:
275 """
276 Load a machine data record from a stream of strings.
278 This function parses a `stream` of strings, which could be from a log
279 file, and extracts the core data of a machine.
280 It then returns this data as a machine record.
282 Optionally, a function `can_skip` can be provided.
283 Each machine record has as ID usually the session node, i.e., the name of
284 the corresponding computer.
285 The function `can_skip` receives this as parameter and is called once.
286 It can then decide whether the data should be fully parsed and a record
287 should be returned (by returning `False`) or whether parsing can be
288 aborted and `None` shall be returned.
290 :param stream: the stream of strings
291 :param can_skip: a function receiving a machine ID and returning `True` if
292 `None` should be returned for this machine ID, i.e., if the record does
293 not need to be parsed, and `False` otherwise
294 """
295 if not callable(can_skip):
296 raise type_error(can_skip, "can_skip", call=True)
297 if not isinstance(stream, Iterable):
298 raise type_error(stream, "stream", Iterable)
300 state: int = 0
301 data: dict[str, str] = {}
302 for srow in stream:
303 row = str.strip(srow)
304 if (str.__len__(row) <= 0) or row.startswith(COMMENT_START):
305 continue
306 if str.__eq__(row, __SEC_START):
307 if state != 0:
308 raise ValueError(f"{__SEC_START!r} appears twice?")
309 state = 1
310 continue
311 if str.__eq__(row, __SEC_END):
312 if state != 1:
313 raise ValueError(f"{__SEC_END!r} before {__SEC_START!r}?")
314 state = 2
315 continue
317 if state == 1:
318 dot = str.find(row, ":")
319 if dot > 0:
320 key = str.strip(row[:dot])
321 if key in __KEYS:
322 val = str.strip(row[dot + 1:])
323 if (key == __MACHINE_KEY) and can_skip(__MACHINE_KEY):
324 return None
325 if str.__len__(val) > 0:
326 data[key] = val
328 machine_id = data.get(__MACHINE_KEY)
329 if machine_id is None:
330 return None
332 try:
333 ram = int(data[__RAM_KEY])
334 except (KeyError, ValueError):
335 ram = None
337 cpu: str | None = data.get(__CPU_KEY)
338 if cpu is None:
339 cpu = data.get(__ARCH_KEY)
340 if cpu is not None:
341 ncpu = cpu
342 new_len = str.__len__(ncpu)
343 old_len = new_len + 1
344 while old_len > new_len:
345 old_len = new_len
346 ncpu_l = str.lower(ncpu)
347 for d in __CPU_DROP:
348 di = str.rfind(ncpu_l, d)
349 if di > 0:
350 ncpu = str.strip(ncpu[:di])
351 ncpu_l = str.lower(ncpu)
353 for d in __CPU_DROP_CORES:
354 if ncpu_l.endswith(d):
355 ncpux = str.strip(ncpu[:-str.__len__(d)])
356 v = str.rfind(ncpux, " ")
357 if 0 < v < (str.__len__(ncpux) - 1):
358 ncpu = str.strip(ncpu[:v])
360 ncpu = str.removesuffix(ncpu, ",")
361 new_len = str.__len__(ncpu)
362 if str.__len__(ncpu) > 0:
363 cpu = ncpu
365 mhz: int | None = None
366 mhz_str: str | None = data.get(__MHZ_KEY)
367 if mhz_str is not None:
368 di = str.rfind(mhz_str, "*")
369 if 0 < di < (str.__len__(mhz_str) - 1):
370 mhz_str = str.strip(mhz_str[:di])
371 mhz_str = str.strip(str.removesuffix(str.removeprefix(
372 mhz_str, "("), ")"))
373 di = str.rfind(mhz_str, ".")
374 if 0 < di < (str.__len__(mhz_str) - 1):
375 mhz_str = str.strip(mhz_str[di + 1:])
376 mhz_str = mhz_str.removesuffix("MHz")
377 with suppress(ValueError):
378 mhz = int(mhz_str)
380 python: str | None = data.get(__PYTHON_KEY)
381 if python is not None:
382 di = str.find(python, "|")
383 if di > 0:
384 python = str.strip(python[:di])
385 di = str.find(python, " ")
386 if di > 0:
387 python = str.strip(python[:di])
388 di = str.find(python, ".")
389 if di > 0:
390 si = str.find(python, ".", di + 1)
391 if si > di:
392 python = str.strip(python[:si])
394 os: str | None = data.get(__OS_NAME_KEY)
395 if os is not None:
396 os_release: str | None = data.get(__OS_RELEASE_KEY)
397 os_version: str | None = data.get(__OS_VERSION_KEY)
399 if str.lower(os) == "linux":
400 if os_version is not None:
401 di = os_version.find(" ")
402 if di > 0:
403 os_version = str.strip(os_version[:di])
404 os = f"{os_version} Linux"
405 if os_release is not None:
406 di = os_release.find("-")
407 if di > 0:
408 os_release = str.strip(os_release[:di])
409 os = f"{os}, {os_release} Kernel"
410 elif os_release is not None:
411 os = f"{os} {os_release}"
412 elif os_version is not None:
413 os = f"{os} {os_version}"
415 return Machine(
416 machine_id=machine_id,
417 ram_bytes=ram,
418 os=os,
419 cpu=cpu,
420 cpu_mhz=mhz,
421 python=python)
424def current_machine_spec() -> Machine:
425 """
426 Get the current machine specification.
428 :returns: the specification of the current machine
430 >>> a = current_machine_spec()
431 >>> a is not None
432 True
433 >>> a is current_machine_spec()
434 True
435 """
436 the_object = current_machine_spec
437 the_property = "_machine_spec"
438 if hasattr(the_object, the_property):
439 return getattr(the_object, the_property)
441 with InMemoryLogger() as ml:
442 log_sys_info(ml)
443 result = get_machine(ml.get_log())
445 if result is None:
446 raise ValueError("Did not get machine?")
448 setattr(the_object, the_property, result)
449 return result