Utility functions
import inspect
def _get_caller_module_name():
"""Return the name of the caller's module by inspecting call stack."""
caller_frame = inspect.stack()[1][0]
caller_module = inspect.getmodule(caller_frame)
caller_module_name = None
if caller_module is not None:
caller_module_name = caller_module.__name__
return caller_module_name
import os, fcntl
def read_non_blocking(file_handler):
"""A iterator that does a non-blocking read over given file handler.
:param file_handler: file_handler to read from
:returns: line read from the given file
"""
# set non-blocking mode
current_flags = fcntl.fcntl(file_handler, fcntl.F_GETFL)
fcntl.fcntl(file_handler, fcntl.F_SETFL, current_flags | os.O_NONBLOCK)
output_buffer = ''
while True:
try:
output = output_buffer + os.read(file_handler.fileno(), 1024)
if output == '': # EOF
raise StopIteration
# Since there is no non-blocking os.readline(), parsing output and creating an equivalent
output_buffer = ''
for line in output.splitlines(True):
if line.endswith('\n'):
yield line
else: # Incomplete
output_buffer += line
except OSError:
sleep(1)
def get_installed_pkgs():
""" Get list of installed pkgs at sys level or in a venv. """
# http://stackoverflow.com/a/23885252
import pip, pprint
installed_packages = pip.get_installed_distributions()
installed_packages_list = sorted(["%s==%s" % (i.key, i.version)
for i in installed_packages])
pprint.pprint(installed_packages_list)
with open('/tmp/pkgs.list', 'w') as f:
f.write('\n'.join(installed_packages_list))