gstreamer/debug-viewer/GstDebugViewer/Common/utils.py

# -*- 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 utils module."""

import os
import logging
import subprocess as _subprocess

class SingletonMeta (type):

    def __init__ (cls, name, bases, dict_):

        from weakref import WeakValueDictionary

        super (SingletonMeta, cls).__init__ (name, bases, dict_)

        cls._singleton_instances = WeakValueDictionary ()

    def __call__ (cls, *a, **kw):

        kw_key = tuple (sorted (kw.iteritems ()))

        try:
            obj = cls._singleton_instances[a + kw_key]
        except KeyError:
            obj = super (SingletonMeta, cls).__call__ (*a, **kw)
            cls._singleton_instances[a + kw_key] = obj
        return obj

def gettext_cache ():

    """Return a callable object that operates like gettext.gettext, but is much
    faster when a string is looked up more than once.  This is very useful in
    loops, where calling gettext.gettext can quickly become a major performance
    bottleneck."""

    from gettext import gettext

    d = {}

    def gettext_cache_access (s):

        if not s in d:
            d[s] = gettext (s)
        return d[s]

    return gettext_cache_access

class ClassProperty (property):

    "Like the property class, but also invokes the getter for class access."

    def __init__ (self, fget = None, fset = None, fdel = None, doc = None):

        property.__init__ (self, fget, fset, fdel, doc)

        self.__fget = fget

    def __get__ (self, obj, obj_class = None):

        ret = property.__get__ (self, obj, obj_class)
        if ret == self:
            return self.__fget (None)
        else:
            return ret

class _XDGClass (object):

    """Partial implementation of the XDG Base Directory specification v0.6.

    http://standards.freedesktop.org/basedir-spec/basedir-spec-0.6.html"""

    def __init__ (self):

        self._add_base_dir ("DATA_HOME", "~/.local/share")
        self._add_base_dir ("CONFIG_HOME", "~/.config")
        self._add_base_dir ("CACHE_HOME", "~/.cache")

    def _add_base_dir (self, name, default):

        dir = os.environ.get ("XDG_%s" % (name,))
        if not dir:
            dir = os.path.expanduser (os.path.join (*default.split ("/")))

        setattr (self, name, dir)

XDG = _XDGClass ()

class SaveWriteFile (object):

    def __init__ (self, filename, mode = "wt"):

        from tempfile import mkstemp

        self.logger = logging.getLogger ("tempfile")

        dir = os.path.dirname (filename)
        base_name = os.path.basename (filename)
        temp_prefix = "%s-tmp" % (base_name,)

        if dir:
            # Destination dir differs from current directory, ensure that it
            # exists:
            try:
                os.makedirs (dir)
            except OSError:
                pass

            self.clean_stale (dir, temp_prefix)

        fd, temp_name = mkstemp (dir = dir, prefix = temp_prefix)

        self.target_name = filename
        self.temp_name = temp_name
        self.real_file = os.fdopen (fd, mode)

    def __enter__ (self):

        return self

    def __exit__ (self, *exc_args):

        if exc_args == (None, None, None,):
            self.close ()
        else:
            self.discard ()

    def __del__ (self):

        try:
            self.discard ()
        except AttributeError:
            # If __init__ failed, self has no real_file attribute.
            pass

    def __close_real (self):

        if self.real_file:
            self.real_file.close ()
            self.real_file = None

    def clean_stale (self, dir, temp_prefix):

        from time import time
        from glob import glob

        now = time ()
        pattern = os.path.join (dir, "%s*" % (temp_prefix,))

        for temp_filename in glob (pattern):
            mtime = os.stat (temp_filename).st_mtime
            if now - mtime > 3600:
                self.logger.info ("deleting stale temporary file %s",
                                  temp_filename)
                try:
                    os.unlink (temp_filename)
                except EnvironmentError, exc:
                    self.logger.warning ("deleting stale temporary file "
                                         "failed: %s", exc)

    def tell (self, *a, **kw):

        return self.real_file.tell (*a, **kw)

    def write (self, *a, **kw):

        return self.real_file.write (*a, **kw)

    def close (self):

        self.__close_real ()

        if self.temp_name:
            try:
                os.rename (self.temp_name, self.target_name)
            except OSError, exc:
                import errno
                if exc.errno == errno.EEXIST:
                    # We are probably on windows.
                    os.unlink (self.target_name)
                    os.rename (self.temp_name, self.target_name)
            self.temp_name = None

    def discard (self):

        self.__close_real ()

        if self.temp_name:

            try:
                os.unlink (self.temp_name)
            except EnvironmentError, exc:
                self.logger.warning ("deleting temporary file failed: %s", exc)
            self.temp_name = None

class TeeWriteFile (object):

    # TODO Py2.5: Add context manager methods.

    def __init__ (self, *file_objects):

        self.files = list (file_objects)

    def close (self):

        for file in self.files:
            file.close ()

    def flush (self):

        for file in self.files:
            file.flush ()

    def write (self, string):

        for file in self.files:
            file.write (string)

    def writelines (self, lines):

        for file in self.files:
            file.writelines (lines)

class FixedPopen (_subprocess.Popen):

    def __init__ (self, args, **kw):

        # Unconditionally specify all descriptors as redirected, to
        # work around Python bug #1358527 (which is triggered for
        # console-less applications on Windows).

        close = []

        for name in ("stdin", "stdout", "stderr",):
            target = kw.get (name)
            if not target:
                kw[name] = _subprocess.PIPE
                close.append (name)

        _subprocess.Popen.__init__ (self, args, **kw)

        for name in close:
            fp = getattr (self, name)
            fp.close ()
            setattr (self, name, None)

class DevhelpError (EnvironmentError):

    pass

class DevhelpUnavailableError (DevhelpError):

    pass

class DevhelpClient (object):

    def available (self):

        try:
            self.version ()
        except DevhelpUnavailableError:
            return False
        else:
            return True

    def version (self):

        return self._invoke ("--version")

    def search (self, entry):

        self._invoke_no_interact ("-s", entry)

    def _check_os_error (self, exc):

        import errno
        if exc.errno == errno.ENOENT:
            raise DevhelpUnavailableError ()

    def _invoke (self, *args):

        from subprocess import PIPE

        try:
            proc = FixedPopen (("devhelp",) + args,
                               stdout = PIPE)
        except OSError, exc:
            self._check_os_error (exc)
            raise

        out, err = proc.communicate ()

        if proc.returncode is not None and proc.returncode != 0:
            raise DevhelpError ("devhelp exited with status %i"
                                % (proc.returncode,))
        return out

    def _invoke_no_interact (self, *args):

        from subprocess import PIPE

        try:
            proc = FixedPopen (("devhelp",) + args)
        except OSError, exc:
            self._check_os_error (exc)
            raise