pyledriver/auxilary.py

100 lines
2.8 KiB
Python
Raw Normal View History

2017-05-29 17:43:03 -04:00
import time, psutil, yaml, os
2016-12-30 02:51:56 -05:00
from subprocess import check_output, DEVNULL, CalledProcessError
from threading import Thread, Event
class ConfigFile():
def __init__(self, path):
self._path = path
with open(self._path, 'r') as f:
self._dict = yaml.safe_load(f)
def __getitem__(self, key):
return self._dict[key]
def __setitem__(self, key, value):
self._dict[key] = value
def sync(self):
with open(self._path, 'w') as f:
yaml.dump(self._dict, f, default_flow_style=False)
2017-05-22 02:32:19 -04:00
class async:
def __init__(self, daemon=False):
self._daemon = daemon
def __call__(self, f):
def wrapper(*args, **kwargs):
t = Thread(target=f, daemon=self._daemon, args=args, kwargs=kwargs)
t.start()
return wrapper
class CountdownTimer(Thread):
def __init__(self, countdownSeconds, action, sound=None):
self._stopper = Event()
def countdown():
for i in range(countdownSeconds, 0, -1):
if self._stopper.isSet():
return None
if sound and i < countdownSeconds:
sound.play()
time.sleep(1)
action()
super().__init__(target=countdown, daemon=True)
self.start()
def stop(self):
self._stopper.set()
def __del__(self):
self.stop()
2017-05-30 01:00:23 -04:00
def waitForPath(path, timeout=30):
for i in range(0, timeout):
if os.path.exists(path):
return
time.sleep(1)
raise FileNotFoundError('Could not find {} after {} seconds'.format(path, timeout))
2017-05-22 02:32:19 -04:00
2017-05-30 01:00:23 -04:00
# we shouldn't need this if we are disabling the usb hub globally and nuking any past connections
#~ def freeBusyPath(path, logger=None):
#~ # check if any other processes are using file path
#~ # if found, politely ask them to exit, else nuke them
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
#~ # NOTE: fuser sends diagnostic info (eg filenames and modes...which we
#~ # don't want) to stderr. This is weird, but let's me route to /dev/null
#~ # so I don't have to parse it later
#~ try:
#~ stdout = check_output(['fuser', path], universal_newlines=True, stderr=DEVNULL)
#~ except CalledProcessError:
#~ logger.debug('%s not in use. Execution may continue', path)
#~ else:
#~ # assume stdout is one PID first
#~ try:
#~ processes = [psutil.Process(int(stdout))]
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
#~ # else assume we have multiple PIDs separated by arbitrary space
#~ except ValueError:
#~ processes = [psutil.Process(int(s)) for s in stdout.split()]
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
#~ for p in processes:
#~ if logger:
#~ logger.warning('%s in use by PID %s. Sending SIGTERM', path, p.pid)
#~ p.terminate()
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
#~ dead, alive = psutil.wait_procs(processes, timeout=10)
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
#~ for p in alive:
#~ if logger:
#~ logger.warning('Failed to terminate PID %s. Sending SIGKILL', p.pid)
#~ p.kill()
2016-12-30 02:51:56 -05:00
2017-05-30 01:00:23 -04:00
# crude way to reset USB device, pretty rough but works
2017-05-29 17:43:03 -04:00
def resetUSBDevice(device):
devpath = os.path.join('/sys/bus/usb/devices/' + device + '/authorized')
with open(devpath, 'w') as f:
f.write('0')
with open(devpath, 'w') as f:
f.write('1')