Skip to content

Instantly share code, notes, and snippets.

@gyglim
Created March 17, 2017 09:16
Show Gist options
  • Save gyglim/340c9d276d7f53578eb9673156e556df to your computer and use it in GitHub Desktop.
Save gyglim/340c9d276d7f53578eb9673156e556df to your computer and use it in GitHub Desktop.
Wrapper around subprocess to debug moviepy OSErrors
"""Wrapper around subprocess that logs the calls to Popen and tracks how many pipes are open"""
import inspect
import os
import subprocess as sp
import logging
import sys
logger = logging.getLogger('subprocess')
logger.setLevel(logging.DEBUG)
def get_log_handler():
"""Returns logging handler."""
logging_handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging_handler.setFormatter(formatter)
return logging_handler
logger.addHandler(get_log_handler())
PIPE = sp.PIPE
PY3 = sys.version_info.major >= 3
if PY3:
from sp import DEVNULL # py3k
else:
DEVNULL = open(os.devnull, 'wb')
current_callers = []
STACK_FMT = "%s, line %d in function %s."
def clear_all():
for c in current_callers:
current_callers.remove(c)
c.__del__()
class Popen(sp.Popen):
def __init__(self, *args, **kwargs):
# The call stack.
stack = inspect.stack()
# The index of the first frame to print.
begin = 1
limit=5
# The index of the last frame to print.
end = min(begin + limit, len(stack))
# Print the next frames up to the limit.
self.caller ='\n'.join([STACK_FMT % frame[1:4] for frame in stack[begin:end]])
logger.info("Open subprocess pipe for \n%s\n%s" % (stack[begin:begin+1], ' '.join(args[0])))
try:
super(Popen,self).__init__(*args, **kwargs)
except OSError as e:
logger.error(e,exc_info=True)
logger.info(
'Remaining callers: %d\n %s' % (len(current_callers), '\n\n'.join([c for c in current_callers])))
raise e
current_callers.append(self.caller)
def __del__(self):
logger.info("Close subprocess pipe for %s" % self.caller)
if self.caller in current_callers:
current_callers.remove(self.caller)
logger.info(
'Remaining callers: %d\n %s' % (len(current_callers), '\n\n'.join([c for c in current_callers])))
super(Popen,self).__del__()
def wait(self):
logger.info("Start waiting for suprocess to finish (caller %s)" % self.caller)
data = super(Popen,self).wait()
logger.info("Done waiting for suprocess to finish (caller %s)" % self.caller)
return data
def kill(self):
data = super(Popen, self).kill()
return data
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment