1# -*- coding: utf-8 -*-
2"""
3    celery.five
4    ~~~~~~~~~~~
5
6    Compatibility implementations of features
7    only available in newer Python versions.
8
9
10"""
11from __future__ import absolute_import
12
13# ############# py3k #########################################################
14import sys
15PY3 = sys.version_info[0] == 3
16
17try:
18    reload = reload                         # noqa
19except NameError:                           # pragma: no cover
20    try:
21        from importlib import reload        # noqa
22    except ImportError:                     # pragma: no cover
23        from imp import reload              # noqa
24
25try:
26    from UserList import UserList           # noqa
27except ImportError:                         # pragma: no cover
28    from collections import UserList        # noqa
29
30try:
31    from UserDict import UserDict           # noqa
32except ImportError:                         # pragma: no cover
33    from collections import UserDict        # noqa
34
35# ############# time.monotonic ###############################################
36
37if sys.version_info < (3, 3):
38
39    import platform
40    SYSTEM = platform.system()
41
42    try:
43        import ctypes
44    except ImportError:  # pragma: no cover
45        ctypes = None  # noqa
46
47    if SYSTEM == 'Darwin' and ctypes is not None:
48        from ctypes.util import find_library
49        libSystem = ctypes.CDLL(find_library('libSystem.dylib'))
50        CoreServices = ctypes.CDLL(find_library('CoreServices'),
51                                   use_errno=True)
52        mach_absolute_time = libSystem.mach_absolute_time
53        mach_absolute_time.restype = ctypes.c_uint64
54        absolute_to_nanoseconds = CoreServices.AbsoluteToNanoseconds
55        absolute_to_nanoseconds.restype = ctypes.c_uint64
56        absolute_to_nanoseconds.argtypes = [ctypes.c_uint64]
57
58        def _monotonic():
59            return absolute_to_nanoseconds(mach_absolute_time()) * 1e-9
60
61    elif SYSTEM == 'Linux' and ctypes is not None:
62        # from stackoverflow:
63        # questions/1205722/how-do-i-get-monotonic-time-durations-in-python
64        import ctypes
65        import os
66
67        CLOCK_MONOTONIC = 1  # see <linux/time.h>
68
69        class timespec(ctypes.Structure):
70            _fields_ = [
71                ('tv_sec', ctypes.c_long),
72                ('tv_nsec', ctypes.c_long),
73            ]
74
75        librt = ctypes.CDLL('librt.so.1', use_errno=True)
76        clock_gettime = librt.clock_gettime
77        clock_gettime.argtypes = [
78            ctypes.c_int, ctypes.POINTER(timespec),
79        ]
80
81        def _monotonic():  # noqa
82            t = timespec()
83            if clock_gettime(CLOCK_MONOTONIC, ctypes.pointer(t)) != 0:
84                errno_ = ctypes.get_errno()
85                raise OSError(errno_, os.strerror(errno_))
86            return t.tv_sec + t.tv_nsec * 1e-9
87    else:
88        from time import time as _monotonic
89
90try:
91    from time import monotonic
92except ImportError:
93    monotonic = _monotonic  # noqa
94
95if PY3:
96    import builtins
97
98    from queue import Queue, Empty, Full
99    from itertools import zip_longest
100    from io import StringIO, BytesIO
101
102    map = map
103    string = str
104    string_t = str
105    long_t = int
106    text_t = str
107    range = range
108    int_types = (int, )
109
110    def items(d):
111        return d.items()
112
113    def keys(d):
114        return d.keys()
115
116    def values(d):
117        return d.values()
118
119    def nextfun(it):
120        return it.__next__
121
122    exec_ = getattr(builtins, 'exec')
123
124    def reraise(tp, value, tb=None):
125        if value.__traceback__ is not tb:
126            raise value.with_traceback(tb)
127        raise value
128
129    class WhateverIO(StringIO):
130
131        def write(self, data):
132            if isinstance(data, bytes):
133                data = data.encode()
134            StringIO.write(self, data)
135
136else:
137    import __builtin__ as builtins  # noqa
138    from Queue import Queue, Empty, Full  # noqa
139    from itertools import imap as map, izip_longest as zip_longest  # noqa
140    from StringIO import StringIO   # noqa
141    string = unicode                # noqa
142    string_t = basestring           # noqa
143    text_t = unicode
144    long_t = long                   # noqa
145    range = xrange
146    int_types = (int, long)
147
148    def items(d):                   # noqa
149        return d.iteritems()
150
151    def keys(d):                    # noqa
152        return d.iterkeys()
153
154    def values(d):                  # noqa
155        return d.itervalues()
156
157    def nextfun(it):                # noqa
158        return it.next
159
160    def exec_(code, globs=None, locs=None):
161        """Execute code in a namespace."""
162        if globs is None:
163            frame = sys._getframe(1)
164            globs = frame.f_globals
165            if locs is None:
166                locs = frame.f_locals
167            del frame
168        elif locs is None:
169            locs = globs
170        exec("""exec code in globs, locs""")
171
172    exec_("""def reraise(tp, value, tb=None): raise tp, value, tb""")
173
174    BytesIO = WhateverIO = StringIO         # noqa
175
176
177def with_metaclass(Type, skip_attrs=set(['__dict__', '__weakref__'])):
178    """Class decorator to set metaclass.
179
180    Works with both Python 2 and Python 3 and it does not add
181    an extra class in the lookup order like ``six.with_metaclass`` does
182    (that is -- it copies the original class instead of using inheritance).
183
184    """
185
186    def _clone_with_metaclass(Class):
187        attrs = dict((key, value) for key, value in items(vars(Class))
188                     if key not in skip_attrs)
189        return Type(Class.__name__, Class.__bases__, attrs)
190
191    return _clone_with_metaclass
192