Source code for monty.dev

"""
This module implements several useful functions and decorators that can be
particularly useful for developers. E.g., deprecating methods / classes, etc.
"""

from __future__ import absolute_import

import re
import logging
import warnings
import os
import subprocess
import multiprocessing
import functools

__author__ = 'Shyue Ping Ong'
__copyright__ = "Copyright 2014, The Materials Virtual Lab"
__version__ = '0.1'
__maintainer__ = 'Shyue Ping Ong'
__email__ = 'ongsp@ucsd.edu'
__date__ = '1/24/14'


logger = logging.getLogger(__name__)


[docs]def deprecated(replacement=None, message=None): """ Decorator to mark classes or functions as deprecated, with a possible replacement. Args: replacement (callable): A replacement class or method. message (str): A warning message to be displayed. Returns: Original function, but with a warning to use the updated class. """ def wrap(old): def wrapped(*args, **kwargs): msg = "%s is deprecated" % old.__name__ if replacement is not None: if isinstance(replacement, property): r = replacement.fget elif isinstance(replacement, (classmethod, staticmethod)): r = replacement.__func__ else: r = replacement msg += "; use %s in %s instead." % (r.__name__, r.__module__) if message is not None: msg += "\n" + message warnings.simplefilter('default') warnings.warn(msg, DeprecationWarning, stacklevel=2) return old(*args, **kwargs) return wrapped return wrap
[docs]class requires(object): """ Decorator to mark classes or functions as requiring a specified condition to be true. This can be used to present useful error messages for optional dependencies. For example, decorating the following code will check if scipy is present and if not, a runtime error will be raised if someone attempts to call the use_scipy function:: try: import scipy except ImportError: scipy = None @requires(scipy is not None, "scipy is not present.") def use_scipy(): print(scipy.majver) Args: condition: Condition necessary to use the class or function. message: A message to be displayed if the condition is not True. """ def __init__(self, condition, message): self.condition = condition self.message = message def __call__(self, _callable): @functools.wraps(_callable) def decorated(*args, **kwargs): if not self.condition: raise RuntimeError(self.message) return _callable(*args, **kwargs) return decorated
[docs]def get_ncpus(): """ .. note:: If you are using Python >= 2.7, multiprocessing.cpu_count() already provides the number of CPUs. In fact, this is the first method tried. The purpose of this function is to cater to old Python versions that still exist on many Linux style clusters. Number of virtual or physical CPUs on this system, i.e. user/real as output by time(1) when called with an optimally scaling userspace-only program. Return -1 if ncpus cannot be detected. Taken from: http://stackoverflow.com/questions/1006289/how-to-find-out-the-number-of- cpus-in-python """ # Python 2.6+ # May raise NonImplementedError try: return multiprocessing.cpu_count() except (ImportError, NotImplementedError): pass # POSIX try: res = int(os.sysconf('SC_NPROCESSORS_ONLN')) if res > 0: return res except (AttributeError, ValueError): pass # Windows try: res = int(os.environ['NUMBER_OF_PROCESSORS']) if res > 0: return res except (KeyError, ValueError): pass # jython try: from java.lang import Runtime runtime = Runtime.getRuntime() res = runtime.availableProcessors() if res > 0: return res except ImportError: pass # BSD try: sysctl = subprocess.Popen(['sysctl', '-n', 'hw.ncpu'], stdout=subprocess.PIPE) scstdout = sysctl.communicate()[0] res = int(scstdout) if res > 0: return res except (OSError, ValueError): pass # Linux try: res = open('/proc/cpuinfo').read().count('processor\t:') if res > 0: return res except IOError: pass # Solaris try: pseudo_devices = os.listdir('/devices/pseudo/') expr = re.compile('^cpuid@[0-9]+$') res = 0 for pd in pseudo_devices: if expr.match(pd) is not None: res += 1 if res > 0: return res except OSError: pass # Other UNIXes (heuristic) try: try: dmesg = open('/var/run/dmesg.boot').read() except IOError: dmesg_process = subprocess.Popen(['dmesg'], stdout=subprocess.PIPE) dmesg = dmesg_process.communicate()[0] res = 0 while '\ncpu' + str(res) + ':' in dmesg: res += 1 if res > 0: return res except OSError: pass logger.warning('Cannot determine number of CPUs on this system!') return -1
[docs]def install_excepthook(hook_type="color", **kwargs): """ This function replaces the original python traceback with an improved version from Ipython. Use `color` for colourful traceback formatting, `verbose` for Ka-Ping Yee's "cgitb.py" version kwargs are the keyword arguments passed to the constructor. See IPython.core.ultratb.py for more info. Return: 0 if hook is installed successfully. """ try: from IPython.core import ultratb except ImportError: import warnings warnings.warn("Cannot install excepthook, IPyhon.core.ultratb not available") return 1 # Select the hook. hook = dict( color=ultratb.ColorTB, verbose=ultratb.VerboseTB, ).get(hook_type.lower(), None) if hook is None: return 2 import sys sys.excepthook = hook(**kwargs) return 0