# Licensed under a 3-clause BSD style license - see LICENSE.rst
import multiprocessing as mp
import os
import platform
import sys
import textwrap
from asv_runner.console import color_print, get_answer_default
from . import console, util
from .console import log
[docs]
def iter_machine_files(results_dir):
"""
Iterate over all of the machine.json files in the results_dir
"""
for root, dirs, files in os.walk(results_dir):
for filename in files:
if filename == 'machine.json':
path = os.path.join(root, filename)
yield path
[docs]
def _get_unique_machine_name():
(system, node, release, version, machine, processor) = platform.uname()
return node
[docs]
class MachineCollection:
"""
Stores information about 1 or more machines in the
~/.asv-machine.json file.
"""
@staticmethod
[docs]
def get_machine_file_path():
return os.path.expanduser('~/.asv-machine.json')
@classmethod
[docs]
def load(cls, machine_name, _path=None):
if _path is None:
path = cls.get_machine_file_path()
else:
path = _path
d = {}
if os.path.isfile(path):
d = util.load_json(path, cls.api_version)
if machine_name in d:
return d[machine_name]
elif len(d) == 1 and machine_name == _get_unique_machine_name():
return d[next(iter(d.keys()))]
raise util.UserError(
f"No information stored about machine '{machine_name}'. "
f"I know about {util.human_list(d.keys())}."
)
@classmethod
[docs]
def save(cls, machine_name, machine_info, _path=None):
if _path is None:
path = cls.get_machine_file_path()
else:
path = _path
if os.path.isfile(path):
d = util.load_json(path)
else:
d = {}
d[machine_name] = machine_info
util.write_json(path, d, cls.api_version)
@classmethod
[docs]
def update(cls, _path=None):
if _path is None:
path = cls.get_machine_file_path()
else:
path = _path
if os.path.isfile(path):
util.update_json(cls, path, cls.api_version)
[docs]
class Machine:
"""
Stores information about a particular machine.
"""
[docs]
fields = [
(
"machine",
"""
A unique name to identify this machine in the results. May
be anything, as long as it is unique across all the machines used
to benchmark this project.
NOTE: If changed from the default, it will no longer match
the hostname of this machine, and you may need to explicitly
use the --machine argument to asv.
""",
),
(
"os",
"""
The OS type and version of this machine. For example,
'Macintosh OS-X 10.8'.""",
),
(
"arch",
"""
The generic CPU architecture of this machine. For
example, 'i386' or 'x86_64'.""",
),
(
"cpu",
"""
A specific description of the CPU of this machine,
including its speed and class. For example, 'Intel(R)
Core(TM) i5-2520M CPU @ 2.50GHz (4 cores)'.""",
),
(
"num_cpu",
"""
The number of CPUs in the system. For example,
'4'.""",
),
(
"ram",
"""
The amount of physical RAM on this machine. For example,
'4GB'.""",
),
]
[docs]
hardcoded_machine_name = None
@classmethod
[docs]
def get_unique_machine_name(cls):
if cls.hardcoded_machine_name:
return cls.hardcoded_machine_name
return _get_unique_machine_name()
@staticmethod
[docs]
def get_defaults():
(system, node, release, version, machine, processor) = platform.uname()
cpu = util.get_cpu_info()
ram = str(util.get_memsize())
num_cpu = str(mp.cpu_count())
return {
'machine': node,
'os': f"{system} {release}",
'num_cpu': num_cpu,
'arch': platform.machine(),
'cpu': cpu,
'ram': ram,
}
@staticmethod
[docs]
def generate_machine_file(use_defaults=False):
if not sys.stdout.isatty() and not use_defaults and not log._colorama:
raise util.UserError(
"Run asv at the console the first time to generate "
"one, or run `asv machine --yes`."
)
log.flush()
color_print(
"I will now ask you some questions about this machine to "
"identify it in the benchmarks."
)
color_print("")
defaults = Machine.get_defaults()
values = {}
for i, (name, description) in enumerate(Machine.fields):
print(
textwrap.fill(
f'{i + 1}. {name}: {textwrap.dedent(description)}', subsequent_indent=' '
)
)
values[name] = get_answer_default(name, defaults[name], use_defaults=use_defaults)
return values
@classmethod
[docs]
def load(
cls,
interactive=False,
force_interactive=False,
_path=None,
machine_name=None,
use_defaults=False,
**kwargs,
):
self = Machine()
if machine_name is None:
machine_name = cls.get_unique_machine_name()
try:
d = MachineCollection.load(machine_name, _path=_path)
except util.UserError as e:
console.log.error(str(e) + '\n')
d = {}
d.update(kwargs)
if (not len(d) and interactive) or force_interactive:
d.update(self.generate_machine_file(use_defaults=use_defaults))
machine_name = d['machine']
self.__dict__.update(d)
MachineCollection.save(machine_name, self.__dict__, _path=_path)
return self
[docs]
def save(self, results_dir):
path = os.path.join(results_dir, self.machine, 'machine.json')
util.write_json(path, self.__dict__, self.api_version)
@classmethod
[docs]
def update(cls, path):
util.update_json(cls, path, cls.api_version)