gstreamer/debug-viewer/GstDebugViewer/Common/Main.py
Stefan Sauer d783c9cf36 formatting: run autopep8 over all files
We have a commit hook on the repo. Get all files to match the pep8 guidelines.
2016-09-28 20:38:55 +02:00

508 lines
13 KiB
Python

# -*- coding: utf-8; mode: python; -*-
#
# GStreamer Development Utilities
#
# Copyright (C) 2007 René Stadler <mail@renestadler.de>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 3 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program. If not, see <http://www.gnu.org/licenses/>.
"""GStreamer Development Utilities Common Main module."""
import sys
import os
import traceback
from operator import attrgetter
import logging
import locale
import gettext
from gettext import gettext as _, ngettext
import gi
from gi.repository import GObject
from gi.repository import Gtk
class ExceptionHandler (object):
exc_types = (Exception,)
priority = 50
inherit_fork = True
_handling_exception = False
def __call__(self, exc_type, exc_value, exc_traceback):
raise NotImplementedError(
"derived classes need to override this method")
class DefaultExceptionHandler (ExceptionHandler):
# TODO Py2.5: In Python 2.5, this succeeds. Remove the try...except block
# once we depend on 2.5.
try:
exc_types = (BaseException,)
except NameError:
# Python < 2.5.
exc_types = (Exception,)
priority = 0
inherit_fork = True
def __init__(self, excepthook):
ExceptionHandler.__init__(self)
self.excepthook = excepthook
def __call__(self, *exc_info):
return self.excepthook(*exc_info)
class ExitOnInterruptExceptionHandler (ExceptionHandler):
exc_types = (KeyboardInterrupt,)
priority = 100
inherit_fork = False
exit_status = 2
def __call__(self, *args):
print >> sys.stderr, "Interrupt caught, exiting."
sys.exit(self.exit_status)
class MainLoopWrapper (ExceptionHandler):
priority = 95
inherit_fork = False
def __init__(self, enter, exit):
ExceptionHandler.__init__(self)
self.exc_info = (None,) * 3
self.enter = enter
self.exit = exit
def __call__(self, *exc_info):
self.exc_info = exc_info
self.exit()
def run(self):
ExceptHookManager.register_handler(self)
try:
self.enter()
finally:
ExceptHookManager.unregister_handler(self)
if self.exc_info != (None,) * 3:
# Re-raise unhandled exception that occured while running the loop.
exc_type, exc_value, exc_tb = self.exc_info
raise exc_type, exc_value, exc_tb
class ExceptHookManagerClass (object):
def __init__(self):
self._in_forked_child = False
self.handlers = []
def setup(self):
if sys.excepthook == self.__excepthook:
raise ValueError("already set up")
hook = sys.excepthook
self.__instrument_excepthook()
self.__instrument_fork()
self.register_handler(DefaultExceptionHandler(hook))
def shutdown(self):
if sys.excepthook != self.__excepthook:
raise ValueError("not set up")
self.__restore_excepthook()
self.__restore_fork()
def __instrument_excepthook(self):
hook = sys.excepthook
self._original_excepthook = hook
sys.excepthook = self.__excepthook
def __restore_excepthook(self):
sys.excepthook = self._original_excepthook
def __instrument_fork(self):
try:
fork = os.fork
except AttributeError:
# System has no fork() system call.
self._original_fork = None
else:
self._original_fork = fork
os.fork = self.__fork
def __restore_fork(self):
if not hasattr(os, "fork"):
return
os.fork = self._original_fork
def entered_forked_child(self):
self._in_forked_child = True
for handler in tuple(self.handlers):
if not handler.inherit_fork:
self.handlers.remove(handler)
def register_handler(self, handler):
if self._in_forked_child and not handler.inherit_fork:
return
self.handlers.append(handler)
def unregister_handler(self, handler):
self.handlers.remove(handler)
def __fork(self):
pid = self._original_fork()
if pid == 0:
# Child process.
self.entered_forked_child()
return pid
def __excepthook(self, exc_type, exc_value, exc_traceback):
for handler in sorted(self.handlers,
key=attrgetter("priority"),
reverse=True):
if handler._handling_exception:
continue
for type_ in handler.exc_types:
if issubclass(exc_type, type_):
break
else:
continue
handler._handling_exception = True
handler(exc_type, exc_value, exc_traceback)
# Not using try...finally on purpose here. If the handler itself
# fails with an exception, this prevents recursing into it again.
handler._handling_exception = False
return
else:
from warnings import warn
warn("ExceptHookManager: unhandled %r" % (exc_value,),
RuntimeWarning,
stacklevel=2)
ExceptHookManager = ExceptHookManagerClass()
class PathsBase (object):
data_dir = None
icon_dir = None
locale_dir = None
@classmethod
def setup_installed(cls, data_prefix):
"""Set up paths for running from a regular installation."""
pass
@classmethod
def setup_uninstalled(cls, source_dir):
"""Set up paths for running 'uninstalled' (i.e. directly from the
source dist)."""
pass
@classmethod
def ensure_setup(cls):
"""If paths are still not set up, try to set from a fallback."""
if cls.data_dir is None:
source_dir = os.path.dirname(
os.path.dirname(os.path.abspath(__file__)))
cls.setup_uninstalled(source_dir)
def __new__(cls):
raise RuntimeError("do not create instances of this class -- "
"use the class object directly")
class PathsProgramBase (PathsBase):
program_name = None
@classmethod
def setup_installed(cls, data_prefix):
if cls.program_name is None:
raise NotImplementedError(
"derived classes need to set program_name attribute")
cls.data_dir = os.path.join(data_prefix, "share", cls.program_name)
cls.icon_dir = os.path.join(data_prefix, "share", "icons")
cls.locale_dir = os.path.join(data_prefix, "share", "locale")
@classmethod
def setup_uninstalled(cls, source_dir):
"""Set up paths for running 'uninstalled' (i.e. directly from the
source dist)."""
# This is essential: The GUI module needs to find the .glade file.
cls.data_dir = os.path.join(source_dir, "data")
# The locale data might be missing if "setup.py build" wasn't run.
cls.locale_dir = os.path.join(source_dir, "build", "mo")
# Not setting icon_dir. It is not useful since we don't employ the
# needed directory structure in the source dist.
class OptionError (Exception):
pass
class OptionParser (object):
def __init__(self, options):
self.__entries = []
self.__parsers = {}
self.options = options
self.__remaining_args = []
# Remaining args parsing with pygobject does not work with glib before
# 2.13.2 (e.g. Ubuntu Feisty).
# if GObject.glib_version >= (2, 13, 2,):
# self.__entries.append ((GObject.OPTION_REMAINING, "\0", 0, "", "",))
def add_option(self, long_name, short_name=None, description=None,
arg_name=None, arg_parser=None, hidden=False):
flags = 0
if not short_name:
# A deficiency of pygobject:
short_name = "\0"
if not description:
description = ""
if arg_name is None:
flags |= GObject.OPTION_FLAG_NO_ARG
elif arg_parser is not None:
self.__parsers[long_name] = arg_parser
if hidden:
flags |= GObject.OPTION_FLAG_HIDDEN
self.__entries.append((long_name, short_name, flags, description,
arg_name,))
def __handle_option(self, option, arg, group):
# See __init__ for glib requirement.
# if option == GObject.OPTION_REMAINING:
# self.__remaining_args.append (arg)
# return
for entry in self.__entries:
long_name, short_name = entry[:2]
arg_name = entry[-1]
if (option != "--%s" % (long_name,) and
option != "-%s" % (short_name,)):
continue
attr = long_name.replace("-", "_")
if arg_name is None:
value = True
elif long_name in self.__parsers:
value = self.__parsers[long_name](arg)
else:
value = arg
self.options[attr] = value
break
def parse(self, argv):
context = GObject.OptionContext(self.get_parameter_string())
group = GObject.OptionGroup(None, None, None, self.__handle_option)
context.set_main_group(group)
group.add_entries(self.__entries)
try:
result_argv = context.parse(argv)
except GObject.GError as exc:
raise OptionError(exc.message)
self.__remaining_args = result_argv[1:]
self.handle_parse_complete(self.__remaining_args)
def get_parameter_string(self):
raise NotImplementedError("derived classes must override this method")
def handle_parse_complete(self, remaining_args):
pass
class LogOptionParser (OptionParser):
"""Like OptionParser, but adds a --log-level option."""
def __init__(self, *a, **kw):
OptionParser.__init__(self, *a, **kw)
# TODO: Re-evaluate usage of log levels to use less of them. Like
# unifying warning, error and critical.
self.add_option("log-level", "l",
"%s (debug, info, warning, error, critical)"
% (_("Enable logging"),),
"LEVEL", self.parse_log_level)
@staticmethod
def parse_log_level(arg):
try:
level = int(arg)
except ValueError:
level = {"off": None,
"none": None,
"debug": logging.DEBUG,
"info": logging.INFO,
"warning": logging.WARNING,
"error": logging.ERROR,
"critical": logging.CRITICAL}.get(arg.strip().lower())
if level is None:
return None
else:
return level
else:
if level < 0:
level = 0
elif level > 5:
level = 5
return {0: None,
1: logging.DEBUG,
2: logging.INFO,
3: logging.WARNING,
4: logging.ERROR,
5: logging.CRITICAL}[level]
def _init_excepthooks():
ExceptHookManager.setup()
ExceptHookManager.register_handler(ExitOnInterruptExceptionHandler())
def _init_paths(paths):
paths.ensure_setup()
def _init_locale(gettext_domain=None):
if Paths.locale_dir and gettext_domain is not None:
try:
locale.setlocale(locale.LC_ALL, "")
except locale.Error as exc:
from warnings import warn
warn("locale error: %s" % (exc,),
RuntimeWarning,
stacklevel=2)
Paths.locale_dir = None
else:
gettext.bindtextdomain(gettext_domain, Paths.locale_dir)
gettext.textdomain(gettext_domain)
gettext.bind_textdomain_codeset(gettext_domain, "UTF-8")
def _init_options(option_parser=None):
if option_parser is None:
return {}
try:
option_parser.parse(sys.argv)
except OptionError as exc:
print >> sys.stderr, exc.args[0]
sys.exit(1)
return option_parser.options
def _init_logging(level=None):
logging.basicConfig(level=level,
format='%(asctime)s.%(msecs)03d %(levelname)8s %(name)20s: %(message)s',
datefmt='%H:%M:%S')
logger = logging.getLogger("main")
logger.debug("logging at level %s", logging.getLevelName(level))
logger.info("using Python %i.%i.%i %s %i", *sys.version_info)
def main(option_parser=None, gettext_domain=None, paths=None):
# FIXME:
global Paths
Paths = paths
_init_excepthooks()
_init_paths(paths)
_init_locale(gettext_domain)
options = _init_options(option_parser)
try:
log_level = options["log_level"]
except KeyError:
_init_logging()
else:
_init_logging(log_level)
try:
options["main"](options)
finally:
logging.shutdown()