Source code for fastr.helpers.sysinfo

# Copyright 2011-2014 Biomedical Imaging Group Rotterdam, Departments of
# Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

This module contains function to help gather system information use for the
provenance of the Job execution.

import getpass
import os
import platform
import sys

    import psutil
except ImportError:

[docs]def namedtuple_to_dict(ntuple): """ Helper function to convert a named tuple into a dict :param namedtuple ntuple: the namedtuple to convert :return: named tuple as a dict :rtype: dict """ # Pylint does not play nice with namedtuples (does not recognize their methods) # pylint: disable=no-member,maybe-no-member,protected-access return dict(ntuple._asdict())
[docs]def get_cpu_usage(): """ Get the current CPU usage :return: CPU usage info :rtype: dict """ cpu_usage = namedtuple_to_dict(psutil.cpu_times()) cpu_usage['percent'] = psutil.cpu_percent() return cpu_usage
[docs]def get_memory_usage(): """ Get the current memory usage :return: memory usage info :rtype: dict """ virtual_memory = namedtuple_to_dict(psutil.virtual_memory()) swap_memory = namedtuple_to_dict(psutil.swap_memory()) return { 'virtual': virtual_memory, 'swap': swap_memory}
[docs]def get_mounts(): """ Get the current mounts known on the system :return: mount info :rtype: dict """ mounts = [] for mount in psutil.disk_partitions(): mount = namedtuple_to_dict(mount) try: usage = psutil.disk_usage(mount['mountpoint']) mount["disk_total"] = mount["disk_used"] = usage.used mount["disk_free"] = mount["disk_use_percent"] = usage.percent except OSError: pass mounts.append(mount) return mounts
[docs]def get_users(): """ Get current users on the system :return: user info :rtype: dict """ users = [] for user in psutil.users(): user = namedtuple_to_dict(user) users.append(user) return users
[docs]def get_processes(): """ Get a list of all currently running processes :return: process information :rtype: list """ # List all running processes processes = [] for process in psutil.process_iter(): proc = {'pid':, 'name':, 'status': process.status, 'memoryuse': process.get_memory_percent(), 'cpu_user': process.get_cpu_times()[0], 'cpu_system': process.get_cpu_times()[1], 'nrthreads': process.get_num_threads()} try: proc["user"] = process.username except psutil.AccessDenied: pass try: proc["workdir"] = process.getcwd() except psutil.AccessDenied: pass try: proc["nice"] = process.get_nice() except psutil.AccessDenied: pass processes.append(proc) return processes
[docs]def get_sysinfo(): """ Get system information (cpu, memory, mounts and users) :return: system information :rtype: dict """ if _PSUTIL_AVAILABLE: sysinfo = {'cpu_usage': get_cpu_usage(), 'memory_usage': get_memory_usage(), 'mounts': get_mounts(), 'users': get_users()} else: sysinfo = {} return sysinfo
[docs]def get_os(): """ Get information about the OS :return: OS information :rtype: dict """ os_ = {'system': platform.system(), 'is64bits': int(sys.maxsize > 2 ** 32), 'maxsize': sys.maxsize} version = {} # Save version depending on platform if platform.system() == 'Linux': linux_distribution = platform.linux_distribution() version["distname"] = linux_distribution[0] version["version"] = linux_distribution[1] version["id"] = linux_distribution[2] libc = dict(zip(['libname', 'version'], platform.libc_ver())) version["libc"] = libc elif platform.system() == 'Windows': win32_version = platform.win32_ver() version["release"] = win32_version[0] version["version"] = win32_version[1] version["servicepack"] = win32_version[2] version["ptype"] = win32_version[3] elif platform.system() == 'Java': java_ver = platform.java_ver() version["release"] = java_ver[0] version["vendor"] = java_ver[1] version["vm_name"] = java_ver[2][0] version["vm_release"] = java_ver[2][1] version["vm_vendor"] = java_ver[2][2] version["os_name"] = java_ver[3][0] version["os_version"] = java_ver[3][1] version["os_arch"] = java_ver[3][2] elif platform.system() == 'Darwin': mac_ver = platform.mac_ver() version["release"] = mac_ver[0] version["version"] = mac_ver[1][0] version["dev_stage"] = mac_ver[1][1] version["non_release_version"] = mac_ver[1][2] version["machine"] = mac_ver[2] os_['version'] = version return os_
[docs]def get_python(): """ Get information about the currently used Python implementation :return: python info :rtype: dict """ return {"version": platform.python_version(), "branch": platform.python_branch(), "compiler": platform.python_compiler(), "implementation": platform.python_implementation(), "revision": platform.python_revision()}
[docs]def get_drmaa_info(): """ Get information about the SGE cluster (if applicable) :return: cluster info :rtype: dict """ drmaa_info = {} drmaa_info['jobid'] = os.environ.get('JOB_ID', None) drmaa_info['taskid'] = os.environ.get('SGE_TASK_ID', None) drmaa_info['jobname'] = os.environ.get('JOB_NAME', None) return drmaa_info
[docs]def get_hostinfo(): """ Get all information about the current host machine :return: host info :rtype: dict """ hostenv = {} hostenv['os'] = get_os() hostenv['uname'] = platform.uname() hostenv['username'] = getpass.getuser() hostenv['cwd'] = os.getcwd() hostenv['python'] = get_python() hostenv['envvar'] = dict(os.environ) hostenv['drmaa'] = get_drmaa_info() return hostenv