cengal.hardware.info.cpu.versions.v_1.cpu
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18 19__all__ = ['CpuInfo', 'cpu_info'] 20 21 22# from operator import is_ 23from cengal.modules_management.alternative_import import alt_import 24with alt_import('cpuinfo') as cpuinfo: 25 if cpuinfo is None: 26 CPUINFO_PRESENT: bool = False 27 else: 28 CPUINFO_PRESENT = True 29 30from cengal.modules_management.ignore_in_build_mode import ignore_in_build_mode 31PSUTIL_PRESENT: bool = False 32with ignore_in_build_mode(): 33 with alt_import('psutil') as psutil: 34 if psutil is not None: 35 PSUTIL_PRESENT = True 36 37import sys 38from typing import Dict, Set, Optional 39 40 41""" 42Module Docstring 43Docstrings: http://www.python.org/dev/peps/pep-0257/ 44""" 45 46__author__ = "ButenkoMS <gtalk@butenkoms.space>" 47__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 48__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 49__license__ = "Apache License, Version 2.0" 50__version__ = "4.4.1" 51__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 52__email__ = "gtalk@butenkoms.space" 53# __status__ = "Prototype" 54__status__ = "Development" 55# __status__ = "Production" 56 57 58if hasattr(cpuinfo.cpuinfo, '_check_arch'): 59 check_arch = getattr(cpuinfo.cpuinfo, '_check_arch') 60 try: 61 check_arch() 62 except: 63 CPUINFO_PRESENT = False 64 65 66_known_gathering_methods_names = ( 67 '_get_cpu_info_from_wmic', 68 '_get_cpu_info_from_registry', 69 '_get_cpu_info_from_proc_cpuinfo', 70 '_get_cpu_info_from_cpufreq_info', 71 '_get_cpu_info_from_lscpu', 72 '_get_cpu_info_from_sysctl', 73 '_get_cpu_info_from_kstat', 74 '_get_cpu_info_from_dmesg', 75 '_get_cpu_info_from_cat_var_run_dmesg_boot', 76 '_get_cpu_info_from_ibm_pa_features', 77 '_get_cpu_info_from_sysinfo', 78 '_get_cpu_info_from_cpuid', 79 '_get_cpu_info_from_platform_uname', 80) 81last_stage = len(_known_gathering_methods_names) - 1 82stage_after_last = last_stage + 1 83 84 85_can_use_lazy_gathering = True 86if hasattr(cpuinfo.cpuinfo, '_copy_new_fields'): 87 copy_new_fields = getattr(cpuinfo.cpuinfo, '_copy_new_fields') 88elif hasattr(cpuinfo.cpuinfo, 'CopyNewFields'): 89 copy_new_fields = getattr(cpuinfo.cpuinfo, 'CopyNewFields') 90else: 91 _can_use_lazy_gathering = False 92 93 94def get_cpu_info_lazy(desired_keys = None, stage = 0, info = None): 95 stage = stage if stage is not None else 0 96 stage = stage if stage >= 0 else last_stage 97 98 arch, bits = cpuinfo.cpuinfo._parse_arch(cpuinfo.cpuinfo.DataSource.arch_string_raw) 99 100 friendly_maxsize = { 2**31-1: '32 bit', 2**63-1: '64 bit' }.get(sys.maxsize) or 'unknown bits' 101 friendly_version = "{0}.{1}.{2}.{3}.{4}".format(*sys.version_info) 102 PYTHON_VERSION = "{0} ({1})".format(friendly_version, friendly_maxsize) 103 104 info = { 105 'python_version' : PYTHON_VERSION, 106 'cpuinfo_version' : cpuinfo.cpuinfo.CPUINFO_VERSION, 107 'cpuinfo_version_string' : cpuinfo.cpuinfo.CPUINFO_VERSION_STRING, 108 'arch' : arch, 109 'bits' : bits, 110 'count' : cpuinfo.cpuinfo.DataSource.cpu_count, 111 'arch_string_raw' : cpuinfo.cpuinfo.DataSource.arch_string_raw, 112 } if info is None else info 113 114 if desired_keys is None: 115 return info, stage 116 117 desired_keys = desired_keys if isinstance(desired_keys, set) else set(desired_keys) 118 flags: bool = 'flags' in desired_keys 119 120 if (not flags) & (not (desired_keys - set(info.keys()))): 121 return info, stage 122 123 current_stage = -1 124 for method_name in _known_gathering_methods_names: 125 current_stage += 1 126 if current_stage < stage: 127 continue 128 129 if hasattr(cpuinfo.cpuinfo, method_name): 130 method = getattr(cpuinfo.cpuinfo, method_name) 131 copy_new_fields(info, method()) 132 if (not flags) & (not (desired_keys - set(info.keys()))): 133 break 134 135 next_stage = current_stage + 1 136 if next_stage >= stage_after_last: 137 next_stage = None 138 139 return info, next_stage 140 141 142class CpuInfo: 143 _cache: Dict = None 144 _stage: int = None 145 _cache_friendly: Dict = None 146 _cores_num: int = None 147 _virtual_cores_num: int = None 148 149 def __init__(self, desired_keys: Optional[Set[str]] = None, stage: Optional[int] = 0, info: Dict = None): 150 if CpuInfo._cores_num is None: 151 CpuInfo._cores_num = psutil.cpu_count(logical=False) if PSUTIL_PRESENT else 0 152 153 if CpuInfo._virtual_cores_num is None: 154 CpuInfo._virtual_cores_num = psutil.cpu_count(logical=True) if PSUTIL_PRESENT else 0 155 156 if CpuInfo._cache is None: 157 if _can_use_lazy_gathering: 158 CpuInfo._cache_friendly, CpuInfo._stage = get_cpu_info_lazy(desired_keys, stage, info) if CPUINFO_PRESENT else ((dict(), stage) if info is None else (info, stage)) 159 else: 160 CpuInfo._cache_friendly = cpuinfo.get_cpu_info() if CPUINFO_PRESENT else (dict() if info is None else info) 161 162 CpuInfo._cache = self._normalize_cpu_info_values(CpuInfo._cache_friendly) 163 164 def _normalize_cpu_info_values(self, cpu_info_dict: Dict) -> Dict: 165 result = dict() 166 frequency: Dict[str, int] = { 167 'ghz': 10 ** 9, 'mhz': 10 ** 6, 'khz': 10 ** 3, 'hz': 1 168 } 169 memory_size: Dict[str, int] = { 170 'tib': 10 ** 12, 'gib': 10 ** 9, 'mib': 10 ** 6, 'kib': 10 ** 3, 'b': 1, 171 'tb': 10 ** 12, 'gb': 10 ** 9, 'mb': 10 ** 6, 'kb': 10 ** 3, 172 't': 10 ** 12, 'g': 10 ** 9, 'm': 10 ** 6, 'k': 10 ** 3, 173 } 174 modif_per_key: Dict[str, Dict[str, int]] = { 175 'l1_data_cache_size': memory_size, 176 'l1_instruction_cache_size': memory_size, 177 'l2_cache_size': memory_size, 178 'l3_cache_size': memory_size, 179 } 180 for key, text_value in cpu_info_dict.items(): 181 value = text_value 182 if isinstance(text_value, str): 183 if key in modif_per_key: 184 conversion_required = False 185 try: 186 value = int(text_value) 187 except ValueError: 188 conversion_required = True 189 190 if conversion_required: 191 modif: Dict[str, int] = modif_per_key[key] 192 is_ok = False 193 minus_data_len = -len(text_value) 194 index = -1 195 while index > minus_data_len: 196 try: 197 value = int(text_value[:index]) 198 is_ok = True 199 break 200 except ValueError: 201 pass 202 203 index -= 1 204 205 if not is_ok: 206 index = -1 207 while index > minus_data_len: 208 try: 209 value = float(text_value[:index]) 210 is_ok = True 211 break 212 except ValueError: 213 pass 214 215 index -= 1 216 217 if is_ok: 218 modif_text: str = text_value[index:] 219 modif_text = modif_text.strip().casefold() 220 modif_value = modif.get(modif_text, 1) 221 value = int(value * modif_value) 222 else: 223 value = 0 224 else: 225 try_with_float: bool = False 226 try: 227 value = int(text_value) 228 except ValueError: 229 try_with_float = True 230 231 if try_with_float: 232 try: 233 value = float(text_value) 234 except ValueError: 235 value = text_value 236 237 result[key] = value 238 239 return result 240 241 def _ensure_field(self, field_name): 242 if not CPUINFO_PRESENT: 243 return field_name 244 245 if not _can_use_lazy_gathering: 246 return field_name 247 248 if ('flags' != field_name) & (field_name in self._cache): 249 return field_name 250 251 CpuInfo._cache_friendly, CpuInfo._stage = get_cpu_info_lazy({field_name,}, CpuInfo._stage, CpuInfo._cache_friendly) 252 CpuInfo._cache.clear() 253 CpuInfo._cache.update(self._normalize_cpu_info_values(CpuInfo._cache_friendly)) 254 return field_name 255 256 @property 257 def python_version(self): 258 return self._cache.get(self._ensure_field('python_version'), str()) 259 260 @property 261 def cpuinfo_version(self): 262 return self._cache.get(self._ensure_field('cpuinfo_version'), (0, 0, 0)) 263 264 @property 265 def cpuinfo_version_string(self): 266 return self._cache.get(self._ensure_field('cpuinfo_version_string'), str()) 267 268 @property 269 def python_hz_advertised_friendlyversion(self): 270 return self._cache.get(self._ensure_field('hz_advertised_friendly'), str()) 271 272 @property 273 def hz_actual_friendly(self): 274 return self._cache.get(self._ensure_field('hz_actual_friendly'), str()) 275 276 @property 277 def hz_advertised(self): 278 return self._cache.get(self._ensure_field('hz_advertised'), (0, 0)) 279 280 @property 281 def hz_actual(self): 282 return self._cache.get(self._ensure_field('hz_actual'), (0, 0)) 283 284 @property 285 def arch(self): 286 return self._cache.get(self._ensure_field('arch'), str()) 287 288 @property 289 def bits(self): 290 return self._cache.get(self._ensure_field('bits'), 0) 291 292 @property 293 def count(self): 294 return self._cache.get(self._ensure_field('count'), 0) 295 296 @property 297 def l1_data_cache_size(self): 298 return self._cache.get(self._ensure_field('l1_data_cache_size'), 0) 299 300 @property 301 def l1_instruction_cache_size(self): 302 return self._cache.get(self._ensure_field('l1_instruction_cache_size'), 0) 303 304 @property 305 def l2_cache_size(self): 306 return self._cache.get(self._ensure_field('l2_cache_size'), 0) 307 308 @property 309 def l2_cache_line_size(self): 310 return self._cache.get(self._ensure_field('l2_cache_line_size'), 0) 311 312 @property 313 def l2_cache_associativity(self): 314 return self._cache.get(self._ensure_field('l2_cache_associativity'), 0) 315 316 @property 317 def l3_cache_size(self): 318 return self._cache.get(self._ensure_field('l3_cache_size'), 0) 319 320 @property 321 def stepping(self): 322 return self._cache.get(self._ensure_field('stepping'), 0) 323 324 @property 325 def model(self): 326 return self._cache.get(self._ensure_field('model'), 0) 327 328 @property 329 def family(self): 330 return self._cache.get(self._ensure_field('family'), 0) 331 332 @property 333 def processor_type(self): 334 return self._cache.get(self._ensure_field('processor_type'), 0) 335 336 @property 337 def flags(self): 338 return self._cache.get(self._ensure_field('flags'), list()) 339 340 @property 341 def vendor_id_raw(self): 342 return self._cache.get(self._ensure_field('vendor_id_raw'), str()) 343 344 @property 345 def hardware_raw(self): 346 return self._cache.get(self._ensure_field('hardware_raw'), str()) 347 348 @property 349 def brand_raw(self): 350 return self._cache.get(self._ensure_field('brand_raw'), str()) 351 352 @property 353 def arch_string_raw(self): 354 return self._cache.get(self._ensure_field('arch_string_raw'), str()) 355 356 @property 357 def cores_num(self): 358 return CpuInfo._cores_num if CpuInfo._cores_num else self.count 359 360 @property 361 def virtual_cores_num(self): 362 return CpuInfo._virtual_cores_num if CpuInfo._virtual_cores_num else CpuInfo._cores_num 363 364 @property 365 def l2_cache_size_per_core(self) -> int: 366 if 0 == self.count: 367 return self.l2_cache_size 368 369 return int(self.l2_cache_size / self.cores_num) 370 371 @property 372 def l2_cache_size_per_virtual_core(self) -> int: 373 if 0 == self.count: 374 return self.l2_cache_size 375 376 return int(self.l2_cache_size / self.virtual_cores_num) 377 378 @property 379 def l3_cache_size_per_core(self) -> int: 380 if 0 == self.count: 381 return self.l3_cache_size 382 383 return int(self.l3_cache_size / self.cores_num) 384 385 @property 386 def l3_cache_size_per_virtual_core(self) -> int: 387 if 0 == self.count: 388 return self.l3_cache_size 389 390 return int(self.l3_cache_size / self.virtual_cores_num) 391 392 @property 393 def is_x86(self) -> bool: 394 return self.arch.lower().startswith('x86') 395 396 @property 397 def is_arm(self) -> bool: 398 return self.arch.lower().startswith('arm') 399 400 def gather_all(self): 401 self.flags 402 403 404_CPU_INFO = None 405 406 407def cpu_info() -> CpuInfo: 408 global _CPU_INFO 409 if _CPU_INFO is None: 410 _CPU_INFO = CpuInfo() 411 412 return _CPU_INFO
class
CpuInfo:
143class CpuInfo: 144 _cache: Dict = None 145 _stage: int = None 146 _cache_friendly: Dict = None 147 _cores_num: int = None 148 _virtual_cores_num: int = None 149 150 def __init__(self, desired_keys: Optional[Set[str]] = None, stage: Optional[int] = 0, info: Dict = None): 151 if CpuInfo._cores_num is None: 152 CpuInfo._cores_num = psutil.cpu_count(logical=False) if PSUTIL_PRESENT else 0 153 154 if CpuInfo._virtual_cores_num is None: 155 CpuInfo._virtual_cores_num = psutil.cpu_count(logical=True) if PSUTIL_PRESENT else 0 156 157 if CpuInfo._cache is None: 158 if _can_use_lazy_gathering: 159 CpuInfo._cache_friendly, CpuInfo._stage = get_cpu_info_lazy(desired_keys, stage, info) if CPUINFO_PRESENT else ((dict(), stage) if info is None else (info, stage)) 160 else: 161 CpuInfo._cache_friendly = cpuinfo.get_cpu_info() if CPUINFO_PRESENT else (dict() if info is None else info) 162 163 CpuInfo._cache = self._normalize_cpu_info_values(CpuInfo._cache_friendly) 164 165 def _normalize_cpu_info_values(self, cpu_info_dict: Dict) -> Dict: 166 result = dict() 167 frequency: Dict[str, int] = { 168 'ghz': 10 ** 9, 'mhz': 10 ** 6, 'khz': 10 ** 3, 'hz': 1 169 } 170 memory_size: Dict[str, int] = { 171 'tib': 10 ** 12, 'gib': 10 ** 9, 'mib': 10 ** 6, 'kib': 10 ** 3, 'b': 1, 172 'tb': 10 ** 12, 'gb': 10 ** 9, 'mb': 10 ** 6, 'kb': 10 ** 3, 173 't': 10 ** 12, 'g': 10 ** 9, 'm': 10 ** 6, 'k': 10 ** 3, 174 } 175 modif_per_key: Dict[str, Dict[str, int]] = { 176 'l1_data_cache_size': memory_size, 177 'l1_instruction_cache_size': memory_size, 178 'l2_cache_size': memory_size, 179 'l3_cache_size': memory_size, 180 } 181 for key, text_value in cpu_info_dict.items(): 182 value = text_value 183 if isinstance(text_value, str): 184 if key in modif_per_key: 185 conversion_required = False 186 try: 187 value = int(text_value) 188 except ValueError: 189 conversion_required = True 190 191 if conversion_required: 192 modif: Dict[str, int] = modif_per_key[key] 193 is_ok = False 194 minus_data_len = -len(text_value) 195 index = -1 196 while index > minus_data_len: 197 try: 198 value = int(text_value[:index]) 199 is_ok = True 200 break 201 except ValueError: 202 pass 203 204 index -= 1 205 206 if not is_ok: 207 index = -1 208 while index > minus_data_len: 209 try: 210 value = float(text_value[:index]) 211 is_ok = True 212 break 213 except ValueError: 214 pass 215 216 index -= 1 217 218 if is_ok: 219 modif_text: str = text_value[index:] 220 modif_text = modif_text.strip().casefold() 221 modif_value = modif.get(modif_text, 1) 222 value = int(value * modif_value) 223 else: 224 value = 0 225 else: 226 try_with_float: bool = False 227 try: 228 value = int(text_value) 229 except ValueError: 230 try_with_float = True 231 232 if try_with_float: 233 try: 234 value = float(text_value) 235 except ValueError: 236 value = text_value 237 238 result[key] = value 239 240 return result 241 242 def _ensure_field(self, field_name): 243 if not CPUINFO_PRESENT: 244 return field_name 245 246 if not _can_use_lazy_gathering: 247 return field_name 248 249 if ('flags' != field_name) & (field_name in self._cache): 250 return field_name 251 252 CpuInfo._cache_friendly, CpuInfo._stage = get_cpu_info_lazy({field_name,}, CpuInfo._stage, CpuInfo._cache_friendly) 253 CpuInfo._cache.clear() 254 CpuInfo._cache.update(self._normalize_cpu_info_values(CpuInfo._cache_friendly)) 255 return field_name 256 257 @property 258 def python_version(self): 259 return self._cache.get(self._ensure_field('python_version'), str()) 260 261 @property 262 def cpuinfo_version(self): 263 return self._cache.get(self._ensure_field('cpuinfo_version'), (0, 0, 0)) 264 265 @property 266 def cpuinfo_version_string(self): 267 return self._cache.get(self._ensure_field('cpuinfo_version_string'), str()) 268 269 @property 270 def python_hz_advertised_friendlyversion(self): 271 return self._cache.get(self._ensure_field('hz_advertised_friendly'), str()) 272 273 @property 274 def hz_actual_friendly(self): 275 return self._cache.get(self._ensure_field('hz_actual_friendly'), str()) 276 277 @property 278 def hz_advertised(self): 279 return self._cache.get(self._ensure_field('hz_advertised'), (0, 0)) 280 281 @property 282 def hz_actual(self): 283 return self._cache.get(self._ensure_field('hz_actual'), (0, 0)) 284 285 @property 286 def arch(self): 287 return self._cache.get(self._ensure_field('arch'), str()) 288 289 @property 290 def bits(self): 291 return self._cache.get(self._ensure_field('bits'), 0) 292 293 @property 294 def count(self): 295 return self._cache.get(self._ensure_field('count'), 0) 296 297 @property 298 def l1_data_cache_size(self): 299 return self._cache.get(self._ensure_field('l1_data_cache_size'), 0) 300 301 @property 302 def l1_instruction_cache_size(self): 303 return self._cache.get(self._ensure_field('l1_instruction_cache_size'), 0) 304 305 @property 306 def l2_cache_size(self): 307 return self._cache.get(self._ensure_field('l2_cache_size'), 0) 308 309 @property 310 def l2_cache_line_size(self): 311 return self._cache.get(self._ensure_field('l2_cache_line_size'), 0) 312 313 @property 314 def l2_cache_associativity(self): 315 return self._cache.get(self._ensure_field('l2_cache_associativity'), 0) 316 317 @property 318 def l3_cache_size(self): 319 return self._cache.get(self._ensure_field('l3_cache_size'), 0) 320 321 @property 322 def stepping(self): 323 return self._cache.get(self._ensure_field('stepping'), 0) 324 325 @property 326 def model(self): 327 return self._cache.get(self._ensure_field('model'), 0) 328 329 @property 330 def family(self): 331 return self._cache.get(self._ensure_field('family'), 0) 332 333 @property 334 def processor_type(self): 335 return self._cache.get(self._ensure_field('processor_type'), 0) 336 337 @property 338 def flags(self): 339 return self._cache.get(self._ensure_field('flags'), list()) 340 341 @property 342 def vendor_id_raw(self): 343 return self._cache.get(self._ensure_field('vendor_id_raw'), str()) 344 345 @property 346 def hardware_raw(self): 347 return self._cache.get(self._ensure_field('hardware_raw'), str()) 348 349 @property 350 def brand_raw(self): 351 return self._cache.get(self._ensure_field('brand_raw'), str()) 352 353 @property 354 def arch_string_raw(self): 355 return self._cache.get(self._ensure_field('arch_string_raw'), str()) 356 357 @property 358 def cores_num(self): 359 return CpuInfo._cores_num if CpuInfo._cores_num else self.count 360 361 @property 362 def virtual_cores_num(self): 363 return CpuInfo._virtual_cores_num if CpuInfo._virtual_cores_num else CpuInfo._cores_num 364 365 @property 366 def l2_cache_size_per_core(self) -> int: 367 if 0 == self.count: 368 return self.l2_cache_size 369 370 return int(self.l2_cache_size / self.cores_num) 371 372 @property 373 def l2_cache_size_per_virtual_core(self) -> int: 374 if 0 == self.count: 375 return self.l2_cache_size 376 377 return int(self.l2_cache_size / self.virtual_cores_num) 378 379 @property 380 def l3_cache_size_per_core(self) -> int: 381 if 0 == self.count: 382 return self.l3_cache_size 383 384 return int(self.l3_cache_size / self.cores_num) 385 386 @property 387 def l3_cache_size_per_virtual_core(self) -> int: 388 if 0 == self.count: 389 return self.l3_cache_size 390 391 return int(self.l3_cache_size / self.virtual_cores_num) 392 393 @property 394 def is_x86(self) -> bool: 395 return self.arch.lower().startswith('x86') 396 397 @property 398 def is_arm(self) -> bool: 399 return self.arch.lower().startswith('arm') 400 401 def gather_all(self): 402 self.flags
CpuInfo( desired_keys: Union[Set[str], NoneType] = None, stage: Union[int, NoneType] = 0, info: Dict = None)
150 def __init__(self, desired_keys: Optional[Set[str]] = None, stage: Optional[int] = 0, info: Dict = None): 151 if CpuInfo._cores_num is None: 152 CpuInfo._cores_num = psutil.cpu_count(logical=False) if PSUTIL_PRESENT else 0 153 154 if CpuInfo._virtual_cores_num is None: 155 CpuInfo._virtual_cores_num = psutil.cpu_count(logical=True) if PSUTIL_PRESENT else 0 156 157 if CpuInfo._cache is None: 158 if _can_use_lazy_gathering: 159 CpuInfo._cache_friendly, CpuInfo._stage = get_cpu_info_lazy(desired_keys, stage, info) if CPUINFO_PRESENT else ((dict(), stage) if info is None else (info, stage)) 160 else: 161 CpuInfo._cache_friendly = cpuinfo.get_cpu_info() if CPUINFO_PRESENT else (dict() if info is None else info) 162 163 CpuInfo._cache = self._normalize_cpu_info_values(CpuInfo._cache_friendly)