path: root/GstInspector/main.py

# -*- coding: utf-8; mode: python; -*-
#
#  GStreamer Inspector - Multimedia system plugin introspection
#
#  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 Inspector main module."""

GETTEXT_DOMAIN = "gst-inspector"

import sys
import os
from operator import attrgetter
import logging
from gettext import gettext as _

import pygtk
pygtk.require ("2.0")
del pygtk

import gobject

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 Paths (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."""

        cls.data_dir = os.path.join (data_prefix, "share", "gst-inspector")
        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")

        cls.icon_dir = os.path.join (source_dir, "data", "icons")

        # The locale data might be missing if "setup.py build" wasn't run.
        cls.locale_dir = os.path.join (source_dir, "build", "mo")

    @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")

def main_dump_data (*a, **kw):

    """Main function that is invoked for the --data-dump-acquisition command
    line option."""

    from GstInspector import Data

    try:

        Data._dump_cache ()

    except IOError, exc:
        import errno
        if exc.errno == errno.EPIPE:
            print >> sys.stderr, "Broken pipe, exiting."
        else:
            raise

def main_update_cache (*a, **kw):

    """Main function that is invoked for the --update-cache command line
    option."""

    from GstInspector import Data

    class Consumer (Data.Consumer):

        def handle_data_error (self, error):

            print >> sys.stderr, "Exception in child process:"
            print >> sys.stderr, error.traceback,
            sys.exit (5)

    policy = Data.Policy (update = True)
    producer = Data.Producer (policy = policy)
    producer.consumers.append (Consumer ())
    producer.start ()

def main_version (*a, **kw):

    """Main function that is invoked for the --version command line option."""

    from GstInspector import version

    print "GStreamer Inspector %s" % (version,)

class InspectorOptions (object):

    def __init__ (self):

        self.data_dump_acquisition = False
        self.log_level = None
        self.update_cache = False
        self.version = False

        self.main = None

class OptionError (Exception):

    pass

class OptionParser (object):

    def __init__ (self, options):

        self.__entries = []
        self.__parsers = {}

        self.options = options

    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):

        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

    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:
            context.parse (argv)
        except gobject.GError, exc:
            raise OptionError (exc.message)

        self.handle_parse_complete ()

    def get_parameter_string (self):

        raise NotImplementedError ("derived classes must override this method")

    def handle_parse_complete (self):

        pass

class InspectorOptionParser (OptionParser):

    def __init__ (self):

        options = InspectorOptions ()
        OptionParser.__init__ (self, options.__dict__)
        self.inspector_options = options

        # TODO: Re-evaluate usage of log levels to use less of them.  Like
        # unifying warning, error and critical.

        self.add_option ("version", None,
                         _("Print version information and exit"))
        self.add_option ("update-cache", "u",
                         _("Update data cache and exit"))
        self.add_option ("log-level", "l",
                         "%s (debug, info, warning, error, critical)"
                         % (_("Enable logging"),),
                         "LEVEL", self.parse_log_level)

        # Secret command line option used for out-of-process data acquisition
        # (i.e. GstInspector.Data.SpawnAcquisition).
        self.add_option ("data-dump-acquisition", hidden = True)

    @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 get_parameter_string (self):

        return _("- Introspect multimedia system plugins")

    def handle_parse_complete (self):

        options = self.inspector_options

        if options.update_cache:
            options.main = main_update_cache
        if options.data_dump_acquisition:
            options.main = main_dump_data
        if options.version:
            options.main = main_version
        if options.main is None:
            from GstInspector import GUI
            options.main = GUI.main_gui

def _init_excepthooks ():

    ExceptHookManager.setup ()
    ExceptHookManager.register_handler (ExitOnInterruptExceptionHandler ())

def _init_paths ():

    Paths.ensure_setup ()

def _init_locale ():

    if Paths.locale_dir:
        import locale
        try:
            locale.setlocale (locale.LC_ALL, "")
        except locale.Error, exc:
            from warnings import warn
            warn ("locale error: %s" % (exc,),
                  RuntimeWarning,
                  stacklevel = 2)
            Paths.locale_dir = None
        else:
            import gettext
            gettext.bindtextdomain (GETTEXT_DOMAIN, Paths.locale_dir)
            gettext.textdomain (GETTEXT_DOMAIN)
            gettext.bind_textdomain_codeset (GETTEXT_DOMAIN, "UTF-8")

def _init_options ():

    parser = InspectorOptionParser ()
    options = parser.inspector_options
    try:
        parser.parse (sys.argv)
    except OptionError, exc:
        print >> sys.stderr, exc.args[0]
        sys.exit (1)

    return 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 ():

    _init_excepthooks ()
    _init_paths ()
    _init_locale ()
    options = _init_options ()
    _init_logging (options.log_level)

    try:
        options.main (options)
    finally:
        logging.shutdown ()