clockutils.py

#!/usr/bin/python
#

r'''
A framework to present system clocks by feature, intended to avoid
the library-as-policy pitfalls of the discussion around PEP 418.

My 2c:
* http://www.gossamer-threads.com/lists/python/dev/977474#977474
* http://www.gossamer-threads.com/lists/python/dev/977495#977495
or:
* http://www.mail-archive.com/python-dev@python.org/msg66174.html
* http://www.mail-archive.com/python-dev@python.org/msg66179.html

- Cameron Simpson <cs@cskk.id.au> 02apr2012
'''

from __future__ import print_function
from collections import namedtuple
import os
from time import time

DISTINFO = {
    'description': "implementation of PEP0418 with the \"Choosing the clock from a list of constraints\" get_clock() and get_clocks() functions",
    'keywords': ["python2", "python3"],
    'classifiers': [
        "Programming Language :: Python",
        "Programming Language :: Python :: 2",
        "Programming Language :: Python :: 3",
    ],
    'install_requires': [],
}

# the module exposing OS clock features
_time = os

HIGHRES = 0x01    # high resolution
MONOTONIC = 0x02  # never goes backwards
STEADY = 0x04     # never steps; implies MONOTONIC
ADJUSTED = 0x08   # may be adjusted, for example by NTP
WALLCLOCK = 0x10  # tracks real world time, will usually be ADJUSTED too
RUNTIME = 0x20    # track system run time - stops when system suspended
SYNTHETIC = 0x40  # a synthetic clock, computed from other clocks

def get_clock(flags=0, clocklist=None):
    ''' Return a clock based on the supplied `flags`.
        The returned clock shall have all the requested flags.
        If no clock matches, return None.
    '''
    for clock in get_clocks(flags=flags, clocklist=clocklist):
        return clock
    return None

def get_clocks(flags=0, clocklist=None):
    ''' Yield all clocks matching the supplied `flags`.
        The returned clocks shall have all the requested flags.
    '''
    if clocklist is None:
        clocklist = ALL_CLOCKS
    for clock in clocklist:
        if clock.flags & flags == flags:
            yield clock.factory()

def monotonic_clock(other_flags=0):
    ''' Return a monotonic clock, preferably high resolution.
    '''
    return get_clock(MONOTONIC | HIGHRES | other_flags, MONOTONIC_CLOCKS) \
        or get_clock(MONOTONIC | other_flags, MONOTONIC_CLOCKS)

def steady_clock(other_flags=0):
    ''' Return a steady clock, preferably high resolution.
    '''
    return get_clock(STEADY | HIGHRES | other_flags, STEADY_CLOCKS) \
        or get_clock(STEADY | other_flags, STEADY_CLOCKS)

def highres_clock(other_flags=0):
    ''' Return a high resolution clock, preferably steady.
    '''
    return get_clock(HIGHRES | STEADY | other_flags, HIGHRES_CLOCKS) \
        or get_clock(HIGHRES | other_flags, HIGHRES_CLOCKS)

_global_monotonic = None

def monotonic():
    ''' Return the current time according to the default monotonic clock.
    '''
    global _global_monotonic
    if _global_monotonic is None:
        _global_monotonic = monotonic_clock()
        if _global_monotonic is None:
            raise RuntimeError("no monotonic clock available")
    return _global_monotonic.now()

_global_hires = None

def highres():
    ''' Return the current time according to the default high resolution clock.
    '''
    global _global_hires
    if _global_hires is None:
        _global_hires = highres()
        if _global_hires is None:
            raise RuntimeError("no highres clock available")
    return _global_hires.now()

_global_steady = None

def steady():
    ''' Return the current time according to the default steady clock.
    '''
    global _global_steady
    if _global_steady is None:
        _global_steady = steady()
        if _global_steady is None:
            raise RuntimeError("no steady clock available")
    return _global_steady.now()

class ClockFlags(int):
    ''' An int with human friendly str() and repr() for clock flags.

        The flag names are:
        * `HIGHRES`: clock with the highest resolution.
        * `MONOTONIC`: clock does not go backwards.
        * `STEADY`: clock with high *stability*
          and relatively high *accuracy* and *precision*.
        * `ADJUSTED`: clock whose value may be changed
          to the correct time.
        * `WALLCLOCK`: what the clock on the wall shows.
        * `RUNTIME`: clock based on the process running time.
        * `SYNTHETIC`: clock computed from other clocks,
          such as a monotonic clock computed from other
          nonmonotonic clocks.
    '''

    _flag_names = (
        'HIGHRES',
        'MONOTONIC',
        'STEADY',
        'ADJUSTED',
        'WALLCLOCK',
        'RUNTIME',
        'SYNTHETIC',
    )

    def __str__(self):
        f = self
        G = globals()
        names = []
        for name in ClockFlags._flag_names:
            n = G[name]
            if f & n:
                names.append(name)
                f &= ~n
        if f:
            names.append('0x%02x' % f)
        return '|'.join(names) if names else '0'

    def __repr__(self):
        return '<%s %02x %s>' % (self.__class__.__name__, self, self)

# now assemble the list of platform clocks

class BaseClock(object):
    ''' A BaseClock is the private base class of clock objects.

        A clock has the following mandatory attributes:
        * `.flags`: Feature flags describing the clock.

        A clock may have the following optional attributes:
        * `.epoch`:
          If present, the offset from time.time()'s epoch of this
          clock's epoch(). Not all clocks have epochs; some measure
          elapsed time from an unknown point and only the  difference
          in two measurements is useful.
        * `.resolution`:
          The resolution of the underlying clock facility's
          reporting units. The clock can never be more precise than
          this value. The actual accuracy of the reported time may
          vary with adjustments and the real accuracy of the
          underlying OS clock facility (which in turn may be
          dependent on the precision of some hardware clock).

        A clock must also supply the following methods:
        * `.now()`: Report the current time in seconds, a float.
    '''

    def __init__(self):
        ''' Set instance attributes from class attributes, suitable to
            singleton clocks.
        '''
        klass = self.__class__
        self.flags = ClockFlags(klass.flags)
        for attr in 'epoch', 'resolution':
            try:
                attrval = getattr(klass, attr)
            except AttributeError:
                pass
            else:
                setattr(self, attr, attrval)

    def __repr__(self):
        props = [self.__class__.__name__]
        for attr in sorted([attr for attr in dir(self)
                            if attr
                            and attr[0].isalpha()
                            and attr not in ('now',)]):
            props.append("%s=%s" % (attr, getattr(self, attr)))
        return "<" + " ".join(props) + ">"

ClockEntry = namedtuple('ClockEntry', 'flags factory')

ALL_CLOCKS = []

def _SingletonClockEntry(klass):
    ''' Construct a ClockEntry for a Singleton class, typical for system clocks.
    '''
    klock = klass()
    return ClockEntry(klass.flags, lambda: klock)

# always provide good old time.time()
# provide no flags - this is a fallback - totally implementation agnostic
class _TimeDotTimeClock(BaseClock):
    ''' A clock made from time.time().
    '''

    flags = 0
    epoch = 0

    def now(self):
        ''' The current time from time.time().
        '''
        return time()

ALL_CLOCKS.append(_SingletonClockEntry(_TimeDotTimeClock))

# load system specific clocks here
# in notional order of preference

if os.name == "nt":

    class _NT_GetSystemTimeAsFileTimeClock(BaseClock):
        ''' A clock made from GetSystemTimeAsFileTime().
        '''

        flags = WALLCLOCK
        epoch = EPOCH_16010101T000000   # 01jan1601
        # a negative value wrt 01jan1970
        resolution = 0.0000001          # 100 nanosecond units
        # accuracy HW dependent?

        def now(self):
            ''' The current time from _GetSystemTimeAsFileTime.
            '''
            # convert 100-nanosecond intervals since 1601 to UNIX style seconds
            return (_time._GetSystemTimeAsFileTime() / 10000000
                    + NT_GetSystemTimeAsFileTimeClock.epoch
                    )
    ALL_CLOCKS.append(_SingletonClockEntry(_NT_GetSystemTimeAsFileTimeClock))

    class _NT_GetTickCount64(BaseClock):
        ''' Based on
                http://msdn.microsoft.com/en-us/library/windows/desktop/ms724411%28v=vs.85%29.aspx
            Note this this specificly disavows high resolution.
        '''

        flags = RUNTIME | MONOTONIC
        resolution = 0.001

        def now(self):
            ''' The current time from GetTickCount64.
            '''
            msecs = _time.GetTickCount64()
            return msecs / 1000
    ALL_CLOCKS.append(_SingletonClockEntry(_NT_GetTickCount64))

else:

    # presuming clock_gettime() and clock_getres() exposed in the os
    # module, along with the clock id names
    if hasattr(_time, "clock_gettime"):

        try:
            clk_id = _time.CLOCK_REALTIME
        except AttributeError:
            pass
        else:
            try:
                timespec = _time.clock_getres(clk_id)
            except OSError:
                pass
            else:
                class _UNIX_CLOCK_REALTIME(BaseClock):
                    ''' A clock made from clock_gettime(CLOCK_REALTIME).
                    '''

                    epoch = 0
                    flags = WALLCLOCK
                    resolution = timespec.tv_sec + \
                        timespec.tv_nsec / 1000000000

                    def now(self):
                        ''' The current time from clock_gettime.
                        '''
                        timespec = _time.clock_gettime(_time.CLOCK_REALTIME)
                        return timespec.tv_sec + timespec.tv_nsec / 1000000000
                ALL_CLOCKS.append(_SingletonClockEntry(_UNIX_CLOCK_REALTIME))

        try:
            clk_id = _time.CLOCK_MONOTONIC
        except AttributeError:
            pass
        else:
            try:
                timespec = _time.clock_getres(clk_id)
            except OSError:
                pass
            else:
                class _UNIX_CLOCK_MONOTONIC(BaseClock):
                    ''' A clock made from clock_gettime(CLOCK_MONOTONIC).
                    '''

                    flags = MONOTONIC | STEADY | ADJUSTED
                    resolution = timespec.tv_sec + \
                        timespec.tv_nsec / 1000000000

                    def now(self):
                        ''' The current time from clock_gettime.
                        '''
                        timespec = _time.clock_gettime(_time.CLOCK_MONOTONIC)
                        return timespec.tv_sec + timespec.tv_nsec / 1000000000
                ALL_CLOCKS.append(_SingletonClockEntry(_UNIX_CLOCK_MONOTONIC))

        try:
            clk_id = _time.CLOCK_MONOTONIC_RAW
        except AttributeError:
            pass
        else:
            try:
                timespec = _time.clock_getres(clk_id)
            except OSError:
                pass
            else:
                class _UNIX_CLOCK_MONOTONIC_RAW(BaseClock):
                    ''' A clock made from clock_gettime(CLOCK_MONOTONIC_RAW).
                    '''

                    flags = MONOTONIC | STEADY
                    resolution = timespec.tv_sec + \
                        timespec.tv_nsec / 1000000000

                    def now(self):
                        timespec = _time.clock_gettime(
                            _time.CLOCK_MONOTONIC_RAW)
                        return timespec.tv_sec + timespec.tv_nsec / 1000000000
                ALL_CLOCKS.append(
                    _SingletonClockEntry(_UNIX_CLOCK_MONOTONIC_RAW))

        try:
            clk_id = _time.CLOCK_PROCESS_CPUTIME_ID
        except AttributeError:
            pass
        else:
            try:
                timespec = _time.clock_getres(clk_id)
            except OSError:
                pass
            else:
                class _UNIX_CLOCK_PROCESS_CPUTIME_ID(BaseClock):
                    ''' A clock made from clock_gettime(CLOCK_PROCESS_CPUTIME_ID).
                    '''

                    flags = MONOTONIC
                    resolution = timespec.tv_sec + \
                        timespec.tv_nsec / 1000000000

                    def now(self):
                        timespec = _time.clock_gettime(
                            _time.CLOCK_PROCESS_CPUTIME_ID)
                        return timespec.tv_sec + timespec.tv_nsec / 1000000000
                ALL_CLOCKS.append(
                    _SingletonClockEntry(_CLOCK_PROCESS_CPUTIME_ID))

        try:
            clk_id = _time.CLOCK_THREAD_CPUTIME_ID
        except AttributeError:
            pass
        else:
            try:
                timespec = _time.clock_getres(clk_id)
            except OSError:
                pass
            else:
                class _UNIX_CLOCK_THREAD_CPUTIME_ID(BaseClock):
                    ''' A clock made from clock_gettime(CLOCK_THREAD_CPUTIME_ID).
                    '''

                    flags = MONOTONIC
                    resolution = timespec.tv_sec + \
                        timespec.tv_nsec / 1000000000

                    def now(self):
                        ''' The clock from clock_gettime.
                        '''
                        timespec = _time.clock_gettime(
                            _time.CLOCK_THREAD_CPUTIME_ID)
                        return timespec.tv_sec + timespec.tv_nsec / 1000000000
                ALL_CLOCKS.append(
                    _SingletonClockEntry(_CLOCK_CLOCK_THREAD_CPUTIME_ID))

    if hasattr(_time, "gettimeofday"):
        class _UNIX_gettimeofday(BaseClock):
            ''' A clock made from gettimeofday().
            '''

            epoch = 0
            flags = WALLCLOCK
            resolution = 0.000001

            def now(self):
                ''' The current time from gettimeofday.
                '''
                timeval = _time.gettimeofday()
                return timeval.tv_sec + timeval.tv_usec / 1000000
        ALL_CLOCKS.append(_SingletonClockEntry(_UNIX_gettimeofday))

    if hasattr(_time, "ftime"):
        class _UNIX_ftime(BaseClock):
            ''' A clock made from ftime().
            '''

            epoch = 0
            flags = WALLCLOCK | ADJUSTED
            resolution = 0.001

            def now(self):
                ''' Return the current time.
                '''
                timeb = _time.ftime()
                return timeb.time + timeb.millitm / 1000
        ALL_CLOCKS.append(_SingletonClockEntry(_UNIX_ftime))

class SyntheticMonotonic(BaseClock):
    ''' An example synthetic clock.

        This class comes after time.time() because I think synthetic
        clocks should be less desired - they tend to have side
        effects; but perhaps offered anyway because they can offer
        flag combinations not always presented by the system clocks.

        A simple synthetic monotonic clock may skew with respect to other
        instances.
        Steven D'Aprano wrote a better one.
    '''
    flags = SYNTHETIC | MONOTONIC

    def __init__(self, base_clock=None):
        BaseClock.__init__(self)
        if base_clock is None:
            base_clock = _TimeDotTimeClock()
        self.base_clock = base_clock
        for attr in 'epoch', 'resolution':
            try:
                attrval = getattr(base_clock, attr)
            except AttributeError:
                pass
            else:
                setattr(self, attr, attrval)
        self.__last = None
        self.__base = base_clock

    def now(self):
        ''' The current time.
        '''
        last = self.__last
        t = self.__base.now()
        if last is None or last < t:
            self.__last = t
        else:
            t = last
        return t

ALL_CLOCKS.append(ClockEntry(SyntheticMonotonic.flags, SyntheticMonotonic))

# With more clocks, these will be ALL_CLOCKS listed in order of preference
# for these types i.e. MONOTONIC_CLOCKS will list only monotonic clocks
# in order of quality (an arbitrary measure, perhaps).
MONOTONIC_CLOCKS = ALL_CLOCKS
HIGHRES_CLOCKS = ALL_CLOCKS
STEADY_CLOCKS = ALL_CLOCKS

if __name__ == '__main__':
    print("ALL_CLOCKS =", repr(ALL_CLOCKS))
    for a_clock in get_clocks():
        print("clock = %r" % (a_clock,))
        print(a_clock.__class__.__doc__)