# -*- coding: utf-8 -*-

# Licensed under a 3-clause BSD style license - see LICENSE.rst

"""

A "grab bag" of relatively small general-purpose utilities that don't have

a clear module/package to live in.

"""

from __future__ import (absolute_import, division, print_function,

unicode_literals)

import contextlib

import difflib

import functools

import inspect

import json

import os

import signal

import sys

import textwrap

import traceback

import unicodedata

import warnings

from .exceptions import AstropyDeprecationWarning, AstropyPendingDeprecationWarning

from ..extern import six

from ..extern.six.moves import urllib

__all__ = ['find_current_module', 'isiterable', 'deprecated', 'lazyproperty',

'deprecated_attribute', 'silence', 'format_exception',

'NumpyRNGContext', 'find_api_page', 'is_path_hidden',

'walk_skip_hidden', 'JsonCustomEncoder', 'indent',

'InheritDocstrings']

__doctest_skip__ = ['find_current_module']

def find_current_module(depth=1, finddiff=False):

""" Determines the module/package from which this function is called.

This function has two modes, determined by the ``finddiff`` option. it

will either simply go the requested number of frames up the call

stack (if ``finddiff`` is False), or it will go up the call stack until

it reaches a module that is *not* in a specified set.

Parameters

----------

depth : int

Specifies how far back to go in the call stack (0-indexed, so that

passing in 0 gives back `astropy.utils.misc`).

finddiff : bool or list

If False, the returned ``mod`` will just be ``depth`` frames up from

the current frame. Otherwise, the function will start at a frame

``depth`` up from current, and continue up the call stack to the

first module that is *different* from those in the provided list.

In this case, ``finddiff`` can be a list of modules or modules

names. Alternatively, it can be True, which will use the module

``depth`` call stack frames up as the module the returned module

most be different from.

Returns

-------

mod : module or None

The module object or None if the package cannot be found. The name of

the module is available as the ``__name__`` attribute of the returned

object (if it isn't None).

Raises

------

ValueError

If ``finddiff`` is a list with an invalid entry.

Examples

--------

The examples below assume that there are two modules in a package named

``pkg``. ``mod1.py``::

def find1():

from astropy.utils import find_current_module

print find_current_module(1).__name__

def find2():

from astropy.utils import find_current_module

cmod = find_current_module(2)

if cmod is None:

print 'None'

else:

print cmod.__name__

def find_diff():

from astropy.utils import find_current_module

print find_current_module(0,True).__name__

``mod2.py``::

def find():

from .mod1 import find2

find2()

With these modules in place, the following occurs::

>>> from pkg import mod1, mod2

>>> from astropy.utils import find_current_module

>>> mod1.find1()

pkg.mod1

>>> mod1.find2()

None

>>> mod2.find()

pkg.mod2

>>> find_current_module(0)

<module 'astropy.utils.misc' from 'astropy/utils/misc.py'>

>>> mod1.find_diff()

pkg.mod1

"""

# using a patched version of getmodule because the py 3.1 and 3.2 stdlib

# is broken if the list of modules changes during import

from .compat import inspect_getmodule

frm = inspect.currentframe()

for i in range(depth):

frm = frm.f_back

if frm is None:

return None

if finddiff:

currmod = inspect_getmodule(frm)

if finddiff is True:

diffmods = [currmod]

else:

diffmods = []

for fd in finddiff:

if inspect.ismodule(fd):

diffmods.append(fd)

elif isinstance(fd, six.string_types):

diffmods.append(__import__(fd))

elif fd is True:

diffmods.append(currmod)

else:

raise ValueError('invalid entry in finddiff')

while frm:

frmb = frm.f_back

modb = inspect_getmodule(frmb)

if modb not in diffmods:

return modb

frm = frmb

else:

return inspect_getmodule(frm)

def find_mod_objs(modname, onlylocals=False):

""" Returns all the public attributes of a module referenced by name.

.. note::

The returned list *not* include subpackages or modules of

``modname``, nor does it include private attributes (those that

beginwith '_' or are not in `__all__`).

Parameters

----------

modname : str

The name of the module to search.

onlylocals : bool or list of str

If `True`, only attributes that are either members of ``modname`` OR

one of its modules or subpackages will be included. If it is a list

of strings, those specify the possible packages that will be

considered "local".

Returns

-------

localnames : list of str

A list of the names of the attributes as they are named in the

module ``modname`` .

fqnames : list of str

A list of the full qualified names of the attributes (e.g.,

``astropy.utils.misc.find_mod_objs``). For attributes that are

simple variables, this is based on the local name, but for

functions or classes it can be different if they are actually

defined elsewhere and just referenced in ``modname``.

objs : list of objects

A list of the actual attributes themselves (in the same order as

the other arguments)

"""

__import__(modname)

mod = sys.modules[modname]

if hasattr(mod, '__all__'):

pkgitems = [(k, mod.__dict__[k]) for k in mod.__all__]

else:

pkgitems = [(k, mod.__dict__[k]) for k in dir(mod) if k[0] != '_']

# filter out modules and pull the names and objs out

ismodule = inspect.ismodule

localnames = [k for k, v in pkgitems if not ismodule(v)]

objs = [v for k, v in pkgitems if not ismodule(v)]

# fully qualified names can be determined from the object's module

fqnames = []

for obj, lnm in zip(objs, localnames):

if hasattr(obj, '__module__') and hasattr(obj, '__name__'):

fqnames.append(obj.__module__ + '.' + obj.__name__)

else:

fqnames.append(modname + '.' + lnm)

if onlylocals:

if onlylocals is True:

onlylocals = [modname]

valids = [any([fqn.startswith(nm) for nm in onlylocals]) for fqn in fqnames]

localnames = [e for i, e in enumerate(localnames) if valids[i]]

fqnames = [e for i, e in enumerate(fqnames) if valids[i]]

objs = [e for i, e in enumerate(objs) if valids[i]]

return localnames, fqnames, objs

def isiterable(obj):

"""Returns `True` if the given object is iterable."""

try:

iter(obj)

return True

except TypeError:

return False

def indent(s, shift=1, width=4):

"""Indent a block of text. The indentation is applied to each line."""

indented = '\n'.join(' ' * (width * shift) + l if l else ''

for l in s.splitlines())

if s[-1] == '\n':

indented += '\n'

return indented

class lazyproperty(object):

"""

Works similarly to property(), but computes the value only once.

This essentially memoizes the value of the property by storing the result

of its computation in the ``__dict__`` of the object instance. This is

useful for computing the value of some property that should otherwise be

invariant. For example::

>>> class LazyTest(object):

... @lazyproperty

... def complicated_property(self):

... print('Computing the value for complicated_property...')

... return 42

...

>>> lt = LazyTest()

>>> lt.complicated_property

Computing the value for complicated_property...

42

>>> lt.complicated_property

42

If a setter for this property is defined, it will still be possible to

manually update the value of the property, if that capability is desired.

Adapted from the recipe at

http://code.activestate.com/recipes/363602-lazy-property-evaluation

"""

def __init__(self, fget, fset=None, fdel=None, doc=None):

self._fget = fget

self._fset = fset

self._fdel = fdel

if doc is None:

self.__doc__ = fget.__doc__

else:

self.__doc__ = doc

self._key = self._fget.__name__

def __get__(self, obj, owner=None):

if obj is None:

return self

try:

return obj.__dict__[self._key]

except KeyError:

val = self._fget(obj)

obj.__dict__[self._key] = val

return val

def __set__(self, obj, val):

obj_dict = obj.__dict__

if self._fset:

ret = self._fset(obj, val)

if ret is not None and obj_dict.get(self._key) is ret:

# By returning the value set the setter signals that it took

# over setting the value in obj.__dict__; this mechanism allows

# it to override the input value

return

obj_dict[self._key] = val

def __delete__(self, obj):

if self._fdel:

self._fdel(obj)

if self._key in obj.__dict__:

del obj.__dict__[self._key]

def getter(self, fget):

return self.__ter(fget, 0)

def setter(self, fset):

return self.__ter(fset, 1)

def deleter(self, fdel):

return self.__ter(fdel, 2)

def __ter(self, f, arg):

args = [self._fget, self._fset, self._fdel, self.__doc__]

args[arg] = f

cls_ns = sys._getframe(1).f_locals

for k, v in six.iteritems(cls_ns):

if v is self:

property_name = k

break

cls_ns[property_name] = lazyproperty(*args)

return cls_ns[property_name]

def deprecated(since, message='', name='', alternative='', pending=False,

obj_type=None):

"""

Used to mark a function or class as deprecated.

To mark an attribute as deprecated, use `deprecated_attribute`.

Parameters

------------

since : str

The release at which this API became deprecated. This is

required.

message : str, optional

Override the default deprecation message. The format

specifier ``func`` may be used for the name of the function,

and ``alternative`` may be used in the deprecation message

to insert the name of an alternative to the deprecated

function. ``obj_type`` may be used to insert a friendly name

for the type of object being deprecated.

name : str, optional

The name of the deprecated function or class; if not provided

the name is automatically determined from the passed in

function or class, though this is useful in the case of

renamed functions, where the new function is just assigned to

the name of the deprecated function. For example::

def new_function():

...

oldFunction = new_function

alternative : str, optional

An alternative function or class name that the user may use in

place of the deprecated object. The deprecation warning will

tell the user about this alternative if provided.

pending : bool, optional

If True, uses a AstropyPendingDeprecationWarning instead of a

AstropyDeprecationWarning.

obj_type : str, optional

The type of this object, if the automatically determined one

needs to be overridden.

"""

def deprecate_doc(old_doc, message):

"""

Returns a given docstring with a deprecation message prepended

to it.

"""

if not old_doc:

old_doc = ''

old_doc = textwrap.dedent(old_doc).strip('\n')

new_doc = (('\n.. deprecated:: %(since)s'

'\n %(message)s\n\n' %

{'since': since, 'message': message.strip()}) + old_doc)

if not old_doc:

# This is to prevent a spurious 'unexected unindent' warning from

# docutils when the original docstring was blank.

new_doc += r'\ '

return new_doc

def get_function(func):

"""

Given a function or classmethod, get the function object.

"""

if isinstance(func, classmethod):

try:

func = func.__func__

except AttributeError:

# classmethods in Python2.6 and below lack the __func__

# attribute so we need to hack around to get it

method = func.__get__(None, object)

if hasattr(method, '__func__'):

func = method.__func__

elif hasattr(method, 'im_func'):

func = method.im_func

else:

# Nothing we can do really... just return the original

# classmethod

return func

return func

def deprecate_function(func, message):

"""

Returns a wrapped function that displays an

``AstropyDeprecationWarning`` when it is called.

"""

is_classmethod = isinstance(func, classmethod)

func = get_function(func)

def deprecated_func(*args, **kwargs):

if pending:

category = AstropyPendingDeprecationWarning

else:

category = AstropyDeprecationWarning

warnings.warn(message, category, stacklevel=2)

return func(*args, **kwargs)

# If this is an extension function, we can't call

# functools.wraps on it, but we normally don't care.

# This crazy way to get the type of a wrapper descriptor is

# straight out of the Python 3.3 inspect module docs.

if type(func) != type(str.__dict__['__add__']):

deprecated_func = functools.wraps(func)(deprecated_func)

deprecated_func.__doc__ = deprecate_doc(

deprecated_func.__doc__, message)

if is_classmethod:

return classmethod(deprecated_func)

return deprecated_func

def deprecate_class(cls, message):

"""

Returns a wrapper class with the docstrings updated and an

__init__ function that will raise an

``AstropyDeprectationWarning`` warning when called.

"""

# Creates a new class with the same name and bases as the

# original class, but updates the dictionary with a new

# docstring and a wrapped __init__ method. __module__ needs

# to be manually copied over, since otherwise it will be set

# to *this* module (astropy.utils.misc).

# This approach seems to make Sphinx happy (the new class

# looks enough like the original class), and works with

# extension classes (which functools.wraps does not, since

# it tries to modify the original class).

# We need to add a custom pickler or you'll get

# Can't pickle <class ..>: it's not found as ...

# errors. Picklability is required for any class that is

# documented by Sphinx.

def __getstate__(self):

return super(cls, self).__getstate__()

def __setstate__(self, state):

return super(cls, self).__setstate__(state)

d = {

'__doc__': deprecate_doc(cls.__doc__, message),

'__init__': deprecate_function(cls.__init__, message),

'__module__': cls.__module__,

'__getstate__': __getstate__,

'__setstate__': __setstate__

}

return type(cls.__name__, (cls,), d)

def deprecate(obj, message=message, name=name, alternative=alternative,

pending=pending):

if obj_type is None:

if isinstance(obj, type):

obj_type_name = 'class'

elif inspect.isfunction(obj):

obj_type_name = 'function'

elif inspect.ismethod(obj):

obj_type_name = 'method'

else:

obj_type_name = 'object'

else:

obj_type_name = obj_type

if not name:

name = get_function(obj).__name__

altmessage = ''

if not message or type(message) == type(deprecate):

if pending:

message = ('The %(func)s %(obj_type)s will be deprecated in a '

'future version.')

else:

message = ('The %(func)s %(obj_type)s is deprecated and may '

'be removed in a future version.')

if alternative:

altmessage = '\n Use %s instead.' % alternative

message = ((message % {

'func': name,

'name': name,

'alternative': alternative,

'obj_type': obj_type_name}) +

altmessage)

if not altmessage:

altmessage = message

if isinstance(obj, type):

return deprecate_class(obj, altmessage)

else:

return deprecate_function(obj, altmessage)

if type(message) == type(deprecate):

return deprecate(message)

return deprecate

def deprecated_attribute(name, since, message=None, alternative=None,

pending=False):

"""

Used to mark a public attribute as deprecated. This creates a

property that will warn when the given attribute name is accessed.

To prevent the warning (i.e. for internal code), use the private

name for the attribute by prepending an underscore

(i.e. ``self._name``).

Parameters

----------

name : str

The name of the deprecated attribute.

since : str

The release at which this API became deprecated. This is

required.

message : str, optional

Override the default deprecation message. The format

specifier ``name`` may be used for the name of the attribute,

and ``alternative`` may be used in the deprecation message

to insert the name of an alternative to the deprecated

function.

alternative : str, optional

An alternative attribute that the user may use in place of the

deprecated attribute. The deprecation warning will tell the

user about this alternative if provided.

pending : bool, optional

If True, uses a AstropyPendingDeprecationWarning instead of a

AstropyDeprecationWarning.

Examples

--------

::

class MyClass:

# Mark the old_name as deprecated

old_name = misc.deprecated_attribute('old_name', '0.1')

def method(self):

self._old_name = 42

"""

private_name = '_' + name

@deprecated(since, name=name, obj_type='attribute')

def get(self):

return getattr(self, private_name)

@deprecated(since, name=name, obj_type='attribute')

def set(self, val):

setattr(self, private_name, val)

@deprecated(since, name=name, obj_type='attribute')

def delete(self):

delattr(self, private_name)

return property(get, set, delete)

class _DummyFile(object):

"""A noop writeable object."""

def write(self, s):

pass

@contextlib.contextmanager

def silence():

"""A context manager that silences sys.stdout and sys.stderr."""

old_stdout = sys.stdout

old_stderr = sys.stderr

sys.stdout = _DummyFile()

sys.stderr = _DummyFile()

yield

sys.stdout = old_stdout

sys.stderr = old_stderr

def format_exception(msg, *args, **kwargs):

"""

Given an exception message string, uses new-style formatting arguments

``{filename}``, ``{lineno}``, ``{func}`` and/or ``{text}`` to fill in

information about the exception that occurred. For example:

try:

1/0

except:

raise ZeroDivisionError(

format_except('A divide by zero occurred in {filename} at '

'line {lineno} of function {func}.'))

Any additional positional or keyword arguments passed to this function are

also used to format the message.

.. note::

This uses `sys.exc_info` to gather up the information needed to fill

in the formatting arguments. Python 2.x and 3.x have slightly

different behavior regarding `sys.exc_info` (the latter will not

carry it outside a handled exception), so it's not wise to use this

outside of an ``except`` clause - if it is, this will substitute

'<unkown>' for the 4 formatting arguments.

"""

tb = traceback.extract_tb(sys.exc_info()[2], limit=1)

if len(tb) > 0:

filename, lineno, func, text = tb[0]

else:

filename = lineno = func = text = '<unknown>'

return msg.format(*args, filename=filename, lineno=lineno, func=func,

text=text, **kwargs)

class NumpyRNGContext(object):

"""

A context manager (for use with the ``with`` statement) that will seed the

numpy random number generator (RNG) to a specific value, and then restore

the RNG state back to whatever it was before.

This is primarily intended for use in the astropy testing suit, but it

may be useful in ensuring reproducibility of Monte Carlo simulations in a

science context.

Parameters

----------

seed : int

The value to use to seed the numpy RNG

Examples

--------

A typical use case might be::

with NumpyRNGContext(<some seed value you pick>):

from numpy import random

randarr = random.randn(100)

... run your test using `randarr` ...

#Any code using numpy.random at this indent level will act just as it

#would have if it had been before the with statement - e.g. whatever

#the default seed is.

"""

def __init__(self, seed):

self.seed = seed

def __enter__(self):

from numpy import random

self.startstate = random.get_state()

random.seed(self.seed)

def __exit__(self, exc_type, exc_value, traceback):

from numpy import random

random.set_state(self.startstate)

def find_api_page(obj, version=None, openinbrowser=True, timeout=None):

"""

Determines the URL of the API page for the specified object, and

optionally open that page in a web browser.

.. note::

You must be connected to the internet for this to function even if

``openinbrowser`` is `False`, unless you provide a local version of

the documentation to ``version`` (e.g., ``file:///path/to/docs``).

Parameters

----------

obj

The object to open the docs for or its fully-qualified name

(as a str).

version : str

The doc version - either a version number like '0.1', 'dev' for

the development/latest docs, or a URL to point to a specific

location that should be the *base* of the documentation. Defaults to

latest if you are on aren't on a release, otherwise, the version you

are on.

openinbrowser : bool

If `True`, the `webbrowser` package will be used to open the doc

page in a new web browser window.

timeout : number, optional

The number of seconds to wait before timing-out the query to

the astropy documentation. If not given, the default python

stdlib timeout will be used.

Returns

-------

url : str

The loaded URL

Raises

------

ValueError

If the documentation can't be found

"""

import webbrowser

from zlib import decompress

if (not isinstance(obj, six.string_types) and

hasattr(obj, '__module__') and

hasattr(obj, '__name__')):

obj = obj.__module__ + '.' + obj.__name__

elif inspect.ismodule(obj):

obj = obj.__name__

if version is None:

from .. import version

if version.release:

version = 'v' + version.version

else:

version = 'dev'

if '://' in version:

if version.endswith('index.html'):

baseurl = version[:-10]

elif version.endswith('/'):

baseurl = version

else:

baseurl = version + '/'

elif version == 'dev' or version == 'latest':

baseurl = 'http://devdocs.astropy.org/'

else:

baseurl = 'http://docs.astropy.org/en/{vers}/'.format(vers=version)

if timeout is None:

uf = urllib.request.urlopen(baseurl + 'objects.inv')

else:

uf = urllib.request.urlopen(baseurl + 'objects.inv', timeout=timeout)

try:

# we read these lines so that `oistr` only gets the compressed

# contents, not the header information

isvers = uf.readline().rstrip().decode('utf-8') # intersphinx version line

proj = uf.readline().rstrip().decode('utf-8') # project name

vers = uf.readline().rstrip().decode('utf-8') # project version

uf.readline().rstrip().decode('utf-8')

oistr = uf.read()

finally:

uf.close()

oistr = decompress(oistr).decode('utf-8')

resurl = None

for l in oistr.strip().splitlines():

ls = l.split()

name = ls[0]

loc = ls[3]

if loc.endswith('$'):

loc = loc[:-1] + name

if name == obj:

resurl = baseurl + loc

break

if resurl is None:

raise ValueError('Could not find the docs for the object {obj}'.format(obj=obj))

elif openinbrowser:

webbrowser.open(resurl)

return resurl

def signal_number_to_name(signum):

"""

Given an OS signal number, returns a signal name. If the signal

number is unknown, returns ``'UNKNOWN'``.

"""

# Since these numbers and names are platform specific, we use the

# builtin signal module and build a reverse mapping.

signal_to_name_map = dict(

(k, v) for v, k in signal.__dict__.iteritems() if v.startswith('SIG'))

return signal_to_name_map.get(signum, 'UNKNOWN')

if sys.platform == 'win32':

import ctypes

def _has_hidden_attribute(filepath):

"""

Returns True if the given filepath has the hidden attribute on

MS-Windows. Based on a post here:

http://stackoverflow.com/questions/284115/cross-platform-hidden-file-detection

"""

if isinstance(filepath, bytes):

filepath = filepath.decode(sys.getfilesystemencoding())

try:

attrs = ctypes.windll.kernel32.GetFileAttributesW(filepath)

assert attrs != -1

result = bool(attrs & 2)

except (AttributeError, AssertionError):

result = False

return result

else:

def _has_hidden_attribute(filepath):

return False

def is_path_hidden(filepath):

"""

Determines if a given file or directory is hidden.

Parameters

----------

filepath : str

The path to a file or directory

Returns

-------

hidden : bool

Returns `True` if the file is hidden

"""

name = os.path.basename(os.path.abspath(filepath))

if isinstance(name, bytes):

is_dotted = name.startswith(b'.')

else:

is_dotted = name.startswith('.')

return is_dotted or _has_hidden_attribute(filepath)

def walk_skip_hidden(top, onerror=None, followlinks=False):

"""

A wrapper for `os.walk` that skips hidden files and directories.

This function does not have the parameter ``topdown`` from

`os.walk`: the directories must always be recursed top-down when

using this function.

See also

--------

os.walk : For a description of the parameters

"""

for root, dirs, files in os.walk(

top, topdown=True, onerror=onerror,

followlinks=followlinks):

# These lists must be updated in-place so os.walk will skip

# hidden directories

dirs[:] = [d for d in dirs if not is_path_hidden(d)]

files[:] = [f for f in files if not is_path_hidden(f)]

yield root, dirs, files

class JsonCustomEncoder(json.JSONEncoder):

"""Support for data types that JSON default encoder

does not do.

This includes:

* Numpy array or number

* Complex number

* Set

* Bytes (Python 3)

Examples

--------

>>> import json

>>> import numpy as np

>>> from astropy.utils.misc import JsonCustomEncoder

>>> json.dumps(np.arange(3), cls=JsonCustomEncoder)

'[0, 1, 2]'

"""

def default(self, obj):

import numpy as np

if isinstance(obj, (np.ndarray, np.number)):

return obj.tolist()

elif isinstance(obj, (complex, np.complex)):

return [obj.real, obj.imag]

elif isinstance(obj, set):

return list(obj)

elif isinstance(obj, bytes): # pragma: py3

return obj.decode()

return json.JSONEncoder.default(self, obj)

def strip_accents(s):

"""

Remove accents from a Unicode string.

This helps with matching "ångström" to "angstrom", for example.

"""

return ''.join(

c for c in unicodedata.normalize('NFD', s)

if unicodedata.category(c) != 'Mn')

def did_you_mean(s, candidates, n=3, cutoff=0.8):

"""

When a string isn't found in a set of candidates, we can be nice

to provide a list of alternatives in the exception. This

convenience function helps to format that part of the exception.

Parameters

----------

s : str

candidates : sequence of str or dict of str keys

n : int

The maximum number of results to include. See

`difflib.get_close_matches`.

cutoff : float

In the range [0, 1]. Possibilities that don't score at least

that similar to word are ignored. See

`difflib.get_close_matches`.

Returns

-------

message : str

Returns the string "Did you mean X, Y, or Z?", or the empty

string if no alternatives were found.

"""

if isinstance(s, six.text_type):

s = strip_accents(s)

s_lower = s.lower()

# Create a mapping from the lower case name to all capitalization

# variants of that name.

candidates_lower = {}

for candidate in candidates:

candidate_lower = candidate.lower()

candidates_lower.setdefault(candidate_lower, [])

candidates_lower[candidate_lower].append(candidate)

# The heuristic here is to first try "singularizing" the word. If

# that doesn't match anything use difflib to find close matches in

# original, lower and upper case.

if s_lower.endswith('s') and s_lower[:-1] in candidates_lower:

matches = [s_lower[:-1]]

else:

matches = difflib.get_close_matches(

s_lower, candidates_lower, n=n, cutoff=cutoff)

if len(matches):

capitalized_matches = set()

for match in matches:

capitalized_matches.update(candidates_lower[match])

matches = sorted(capitalized_matches)

if len(matches) == 1:

matches = matches[0]

else:

matches = ', '.join(matches[:-1]) + ' or ' + matches[-1]

return 'Did you mean {0}?'.format(matches)

return ''

class InheritDocstrings(type):

"""

This metaclass makes methods of a class automatically have their

docstrings filled in from the methods they override in the base

class.