"""
Access to system information on machine, which could be Linux, Mac or Windows,
and could be local or remote. In general, the `REMOTE` and `LOCAL`
attributes of this module should be used. They contain `_SysInfo` objects
with information about, respectively, the remote and local host. The
jobcontrol interface is used to determine the remote host - it is pulled from
the HOST argument, and is equivalent to the localhost if no remote host is
requested. Because of the reporting that this module is used for, the remote
host only cares about the FIRST host supplied to the HOST argument.
Contains the `_SysInfo` class, which determines a variety of system information
and stores it. Information includes: 64/32 bit, OS, mmshare #, and processor
details.
Use like this::
>>> import sysinfo
>>> sysinfo.LOCAL.host
'localhost'
or::
$SCHRODINGER/run python3 -m schrodinger.test.stu.sysinfo -h
Can be called directly to determine system information for all command line
arguments.
@copyright: Schrodinger, Inc. All rights reserved.
"""
import ast
import enum
import glob
import locale
import os
import shutil
import socket
import subprocess
import sys
import schrodinger.infra.mm
from schrodinger import gpgpu
from schrodinger.job import jobcontrol
from schrodinger.job import remote_command
from schrodinger.utils import fileutils
from schrodinger.utils import mmutil
from schrodinger.utils import sysinfo
from schrodinger.utils.featureflags.featureflags import get_nondefault_features
#The list of allowed platforms.
_PLATFORMS = ('Linux-x86_64', 'Linux-x86', 'Windows-x64', 'WIN32-x86',
'Darwin-x86_64')
TABLE_FORMAT = '{:<18}{}'.format
PYTHON_EXE = "python3"
class _FakeLocal:
"""
Lazily evaluated local host. Will be defined as a _SysInfo object as soon as it is
accessed.
"""
def _replace(self):
global LOCAL
# if the variable has been set once, don't reset
if LOCAL is not self:
return
LOCAL = _SysInfo('localhost')
def __getattribute__(self, name):
_FakeLocal._replace(self)
return getattr(LOCAL, name)
def __repr__(self):
_FakeLocal._replace(self)
return str(LOCAL)
LOCAL = _FakeLocal()
"""Information about the localhost in a `_SysInfo` object."""
class _FakeRemote:
"""
Lazily evaluated remote host. Will be defined as a _SysInfo object as soon as it is
accessed.
"""
def _replace(self):
global REMOTE
# if the variable has been set once, don't reset
if REMOTE is not self:
return
hosts = jobcontrol.get_backend_host_list()
if not hosts:
hosts = jobcontrol.get_command_line_host_list()
host = jobcontrol.get_host(hosts[0][0]).name
if host == 'localhost':
REMOTE = LOCAL
else:
REMOTE = _SysInfo(host)
def __getattribute__(self, name):
_FakeRemote._replace(self)
return getattr(REMOTE, name)
def __repr__(self):
_FakeRemote._replace(self)
return str(REMOTE)
REMOTE = _FakeRemote()
"""
Information about the first remote host supplied to -HOST in a `_SysInfo`
object.
"""
class _SysInfo:
"""
Determines and saves the system information.
@TODO: Does not support remote hosts from windows
"""
def __init__(self, host="localhost"):
self._host_entry_name = None
"""'Title from schrodinger.hosts"""
self.platform = 'Unknown'
self.jobserver = False
self.mmshare = 0
self.bit32_64 = 32
self.os_version = 'Unknown'
self.processor = 'Unknown'
self.schrodinger = 'Unknown'
self.release = None
self.name = None
# can't use set since doesn't work with ast.literal_eval
self.available_resources = []
self.update(host)
#Host methods, preserves other info relative to the value of _host
@property
def host(self):
return self._host_entry_name
def update(self, new_host='localhost'):
"""
Fills in the information based on the host name.
:rtype: None
"""
if new_host != 'localhost':
self._update_remote(new_host)
return
if not sys.platform.startswith('win32'):
lang = _get_environment_locale()
verify_lang(lang)
self._host_entry_name = 'localhost'
self.name = socket.getfqdn()
self.user = None
self.schrodinger = os.path.realpath(os.getenv("SCHRODINGER"))
self.mmshare = schrodinger.infra.mm.mmfile_get_product_version(
"mmshare")
self.platform = self._get_platform()
self.processor = sysinfo.get_cpu()
try:
self.bit32_64 = int(self.platform[-2:])
except:
self.bit32_64 = "Unknown"
if self.bit32_64 == 86:
self.bit32_64 = 32
self.release = schrodinger.infra.mm.mmfile_get_release_name()
self.os_version = sysinfo.get_osname()
if not self.is_dev_env_locale():
self.os_version += f' (LANG={guess_lang()})'
if not self._are_display_libs_available():
self.os_version += ' (no X)'
else:
self.available_resources.append('display_libs')
if gpgpu.is_any_gpu_available():
self.available_resources.append('gpgpu')
self.available_resources.append('gpu')
if self.platform != 'Darwin-x86_64' or not is_scipy_pff_broken():
self.available_resources.append('scipy_pff')
if sysinfo.is_display_present() or shutil.which('xvfb-run'):
self.available_resources.append('display')
if fileutils.locate_pymol():
self.available_resources.append('pymol')
if self.is_quantum_espresso_present():
self.available_resources.append('quantum_espresso')
if mmutil.feature_flag_is_enabled(mmutil.JOB_SERVER):
self.available_resources.append(ResourceTags.JOB_SERVER.value)
self.jobserver = True
else:
self.available_resources.append(
ResourceTags.LEGACY_JOBCONTROL.value)
self.jobserver = False
def _update_remote(self, new_host):
host = jobcontrol.get_host(new_host)
self._host_entry_name = host.name
self.name = host.name
ssh_host_name = host.host
self.user = host.user
self.schrodinger = host.schrodinger
cmd = " ".join([
"env", f"SCHRODINGER_FEATURE_FLAGS=\"{get_nondefault_features()}\"",
f"'{self.schrodinger}/run'", sysinfo.PYTHON_EXE, "-m",
"schrodinger.test.stu.sysinfo", "--serialize"
])
data = remote_command.remote_command(cmd, ssh_host_name, user=self.user)
data = ast.literal_eval(data)
for k, v in data.items():
if k in ('name', 'host', 'user'):
continue
setattr(self, k, v)
def _get_platform(self):
"""Finds the platform based on the execute host."""
mmshare_path = os.getenv('MMSHARE_EXEC')
myplatform = os.path.basename(mmshare_path)
if myplatform not in _PLATFORMS:
raise UnrecognizedPlatform(
'Found platform="%s" for host %s (using MMSHARE_EXEC=%s)' %
(myplatform, self.host, mmshare_path))
return myplatform
#The following make it easy to check which OS group I belong to.
@property
def isLinux(self):
"""
Is this a Linux platform?
:rtype: bool
"""
return self.platform in _PLATFORMS[0:2]
@property
def isWindows(self):
"""
Is this a Windows platform?
:rtype: bool
"""
return self.platform in _PLATFORMS[2:4]
@property
def isDarwin(self):
"""
Is this a Darwin/Mac platform?
:rtype: bool
"""
return self.platform == _PLATFORMS[4]
def is_dev_env_locale(self):
"""
Current only valid for Linux. Checks if the current locale matches the
development locale (en_US.UTF-8)
:rtype: bool
"""
if not self.isLinux:
return True
lang, encoding = locale.getdefaultlocale()
return (lang == 'en_US') and (encoding == 'UTF-8')
def _are_display_libs_available(self):
"""
Is libQTGUI usable on this machine?
Basically, the answer is true unless this is a Linux machine without
X installed.
"""
if not self.isLinux:
return True
qtgui = glob.glob(os.environ['MMSHARE_EXEC'] +
'/../../lib/Linux-x86_64/libQt*Gui.so.?')[0]
cmd = [self.schrodinger + '/run', 'ldd', qtgui]
output = subprocess.check_output(cmd,
stderr=subprocess.STDOUT,
universal_newlines=True)
return 'not found' not in output
def is_quantum_espresso_present(self):
"""Is quantum espresso installed on this machine?"""
if not self.isLinux:
return False
else:
return os.path.isdir(os.path.join(self.schrodinger, 'qe-bin'))
def toDict(self):
"""
Dump sysinfo object as a dict.
Used in upload to STU server as well as in reading information from a
remote host.
"""
rdict = self.__dict__.copy()
rdict['host'] = self.host
rdict.pop('_host_entry_name')
return rdict
def __str__(self):
"""
Allows direct printing of the _SysInfo object.
:rtype: str
"""
output = TABLE_FORMAT("platform:", self.platform) + '\n'
output += TABLE_FORMAT("os version:", self.os_version) + '\n'
output += TABLE_FORMAT("mmshare version:", self.mmshare) + '\n'
output += TABLE_FORMAT("release:", self.release) + '\n'
processor = f"{self.bit32_64} bit, {self.processor}"
output += TABLE_FORMAT("processor:", processor) + '\n'
output += TABLE_FORMAT("SCHRODINGER:", self.schrodinger) + '\n'
return output
[docs]def is_scipy_pff_broken():
"""
PYTHON-3168: some scipy interpolate functions are broken on certain Darwin
platforms.
"""
try:
cmd = [
sysinfo.PYTHON_EXE, '-c',
'import scipy.stats; print(scipy.stats.t.ppf(0.975, 1116))'
]
subprocess.check_output(cmd, universal_newlines=True)
except subprocess.CalledProcessError:
return True
return False
[docs]def verify_lang(lang):
"""
Assert the LANG used is properly installed and configured for this machine
"""
if sys.platform.startswith('linux'):
cmd = ['env', f'LANG={lang}', 'locale']
check_locale = subprocess.run(cmd, capture_output=True, text=True)
if check_locale.returncode != 0:
raise RuntimeError('Unable to check the locale of this machine.')
if check_locale.stderr:
error_msg = (
f'LANG "{lang}" is not recognized by the operating system. '
'Please verify that the locales have been properly configured '
'on this machine.')
raise RuntimeError(error_msg)
[docs]def guess_lang():
"""
Return a best guess of a LANG variable from the encoding observed by the
running process. Raise RuntimeError if the LANG is not recognized
by the system (only check on linux).
"""
lang_code, encoding = locale.getdefaultlocale()
lang = f'{lang_code}.{encoding}'
if not (lang_code or encoding):
lang = 'C'
return lang
verify_lang(lang)
return lang
def _get_environment_locale() -> str:
"""
Return locale as set by the environment.
"""
for var in ("LC_ALL", "LANG"):
if var in os.environ:
return os.environ[var]
return "C"
def _main(args=None):
import argparse
parser = argparse.ArgumentParser('Find information about a host')
parser.add_argument(
'-HOST', help='Hostname from schrodinger.hosts (default: localhost)')
parser.add_argument('--serialize',
action='store_true',
help='Print host data as a machine readable string')
opts = parser.parse_args(args)
if opts.serialize:
print(LOCAL.toDict())
else:
print(REMOTE.name)
print(REMOTE)
if __name__ == "__main__":
_main()