Source code for pvp.common.utils

import signal
import time
import subprocess
import os

from contextlib import contextmanager
from pvp import prefs

import pvp

[docs]class TimeoutException(Exception): pass
_TIMEOUT = prefs.get_pref('TIMEOUT')
[docs]@contextmanager def time_limit(seconds): def signal_handler(signum, frame): raise TimeoutException("Timed out!") signal.signal(signal.SIGALRM, signal_handler) signal.setitimer(signal.ITIMER_REAL, seconds) try: yield finally: signal.alarm(0)
[docs]def timeout(func): """ Defines a decorator for a 50ms timeout. Usage/Test: @timeout def foo(sleeptime): time.sleep(sleeptime) print("hello") """ def wrapper(*args, **kwargs): try: with time_limit(globals()['_TIMEOUT']): ret = func(*args, **kwargs) return ret except TimeoutException as e: from pvp.common.loggers import init_logger log_str = f'Method call timed out - Method: {func}' logger = init_logger('timeouts') logger.exception(log_str) raise e return wrapper
[docs]def get_version(): """ Returns PVP version, and if available githash, as a string. """ version = pvp.__version__ try: # get git version git_version = subprocess.check_output(['git', 'describe', '--always'], cwd=os.path.dirname(__file__)).strip().decode('utf-8') version = " - ".join([version, git_version]) except Exception: # pragma: no cover # no problem, just use package version pass return version