gnome-osd 0.12.2-1.1 / usr / share / pyshared / gnomeosd / eventbridge.py

# -*- Mode: python; encoding: utf-8 -*-

## XChat OSD song notification
## (C) 2004 Gustavo J. A. M. Carneiro <gustavo@users.sf.net>
## Some code copied from xchatrb.py (xchat plugin), (c) Gustavo
## Carneiro and G-LiTe / <stephan@wilkogazu.nl>
## GPL license applies, blah blah

## More recent changes in main ChangeLog...
## Changes in version 0.3
## - Ditch gosd, require gnome-osd

## Changes in version 0.2
## - Thinner and less ugly border (Gustavo)

## Version: 0.1 (first release)

## Configuration variables:

SHOW_FULL_INFO = True

import pygtk; pygtk.require("2.0")
import ORBit
import CORBA
import bonobo.activation
import string
import bonobo
import gobject
import xml.sax.saxutils
from gnomeosd import gnome_osd_conf
import gettext
import gnome.ui
import sys
import os.path
import atexit
import gconfsync
from xscreensaver import XScreenSaverMonitor

try:
    import dbus
    import dbus.service
    if getattr(dbus, "version", (0,0,0)) >= (0,41,0):
        import dbus.glib
    HAVE_DBUS = True
except:
    HAVE_DBUS = False

# missing consts from the typelib
_ACTIVATION_FLAG_NO_LOCAL = 1<<0;      # No shared libraries
_ACTIVATION_FLAG_PRIVATE = 1<<1;       # start a new server and don't register it
_ACTIVATION_FLAG_EXISTING_ONLY = 1<<2; # don't start the server if not started


song_change_prefs = gconfsync.GConfSync("/apps/gnome-osd/plugins/song_change")
xchat_prefs = gconfsync.GConfSync("/apps/gnome-osd/plugins/xchat")
email_prefs = gconfsync.GConfSync("/apps/gnome-osd/plugins/email")
gaim_prefs = gconfsync.GConfSync("/apps/gnome-osd/plugins/gaim")
pidgin_prefs = gconfsync.GConfSync("/apps/gnome-osd/plugins/pidgin")
listener = None # Bonobo/ObjectDirectory:activation:register listener
screensaver_active = False

class OSD(object):
    def __init__(self):
        self.server = None
        self.client = None
        self.timeout = None
        self.last_message = None

    def _get_osd(self):
        if self.server is not None:
            try:
                # "ping" it
                self.server.ref()
                self.server.unref()
            except:
                print "OSD server died!"
                self.server = None
                self.client = None
        if self.server is None:
            self.server = bonobo.get_object("OAFIID:GNOME_OSD", "IDL:Bonobo/Application:1.0")
            self.server.ref() # AppClient steals one reference
            self.client = bonobo.AppClient(self.server)

    def _filter_timeout(self):
        self.timeout = None

    def send(self, message):
        if screensaver_active:
            return
        self._get_osd()
        if isinstance(message, unicode):
            message = message.encode("utf-8")
        if self.timeout is None:
            self.timeout = gobject.timeout_add(200, self._filter_timeout)
        else:
            if message == self.last_message:
                #print >> sys.stderr, "filtered duplicate message", message
                return
        err = self.client.msg_send("show-full", (message,))
        if err:
            print >> sys.stderr, err
        self.last_message = message

osd = OSD()

_rhythmbox = None
def get_rhythmbox():
    global _rhythmbox

    def get_new(): return bonobo.activation.activate(
	"repo_ids.has('IDL:GNOME/Rhythmbox:1.0')",
	[], _ACTIVATION_FLAG_EXISTING_ONLY)

    if _rhythmbox is None:
	_rhythmbox = get_new()
    else:
	# Check if the cached reference points to a server that is
	# still alive, otherwise ask bonobo-activation for a new fresh
	# reference.
	try:
	    # basically this means: _rhythmbox.ping()
	    _rhythmbox.ref()
	    _rhythmbox.unref()
	except CORBA.COMM_FAILURE:
	    _rhythmbox = get_new()
    return _rhythmbox

class SongInfo(object):
    __slots__ = ['title', 'album', 'artist', 'duration', 'track_number']


def get_psong():
    rb = get_rhythmbox()
    if rb is None:
	print "Rhythmbox is not running"
	return None
    pb = rb.getPlayerProperties()
    info = pb.getValue("song").value()
    pb.unref()
    if info is None:
        print "Rhythmbox is not playing anything"
        return None
    if not info.title:
        return "(untitled song)"
    return format_psong(info)

def format_psong(info):
    #  --- duration ---
    duration = info.duration
    if duration is None or duration == 0:
	duration_str = None
    else:
	if duration == -1:
	    print "Rhythmbox is not playing anything"
	    return None
	else:
	    hours    = duration // 3600
	    duration = duration % 3600
	    minutes  = duration // 60
	    duration = duration % 60
	    duration_str = str(minutes) + ":" + str(duration).zfill(2)
	    if hours == 0:
		duration_str = '%s:%s' % (str(minutes), str(duration).zfill(2))
	    else:
		duration_str = '%s:%s:%s' % (str(hours), str(minutes).zfill(2),
					     str(duration).zfill(2))
    # --- title ---
    title = escape(info.title)
    if title.endswith('.mp3') or title.endswith('.ogg'):
        title = title[:-4]
    msg = "♪ <span foreground='yellow'>%s</span>" % title
    if duration_str is not None:
        msg += " <span size='smaller'>(%s)</span> ♪" % duration_str
    else:
        msg += " ♪"

    extra_info = []
    # --- artist ---
    if info.artist:
        extra_info.append("<span size='smaller' style='italic'>%s</span>"
                          " <span foreground='yellow'>%s</span>" %
                          (_("Artist"), escape(info.artist)))

    # --- album ---
    if info.album:
        extra_info.append("<span size='smaller' style='italic'>%s</span>"
                          " <span foreground='yellow'>%s</span>" %
                          (_("Album"), escape(info.album)))
        if info.track_number != -1:
            extra_info.append("<span size='smaller' style='italic'>%s</span>"
                              " <span foreground='yellow'>%i</span>" %
                              (_("Track"), info.track_number))
    if SHOW_FULL_INFO and extra_info:
        return "\n".join((msg, "<span size='smaller'>%s</span>" % "   ".join(extra_info)))
    else:
        return msg


def rb_properties_cb(listener, event, foo, *ev):
    global song_change_prefs, osd
    if not song_change_prefs['enabled']:
        return
    song = get_psong()
    osd.send("<message id='rhythmbox' hide_timeout='10000'>" + song + "</message>")

def monitor_rhythmbox(*args):
    global listener
    rb = get_rhythmbox()
    if rb is not None:
        if listener is not None:
            bonobo.event_source_client_remove_listener(
                rb.getPlayerProperties(), listener)
        listener = bonobo.event_source_client_add_listener(
            rb.getPlayerProperties(), rb_properties_cb,
            "Bonobo/Property:change:song")

def die_cb(cli):
    print "Die!!"
    global listener
    rb = get_rhythmbox()
    if rb is not None:
        if listener is not None:
            bonobo.event_source_client_remove_listener(
                rb.getPlayerProperties(), listener)
    bonobo.main_quit()
    print "OK, quitting..."

def evolution_new_mail_callback(uri, folder=None):
    global email_prefs, osd
    if not email_prefs['enabled']:
        return
    if folder is None:
        if not uri.upper().endswith("INBOX"):
            return
    else:
        if folder.upper() != "INBOX":
            return
    osd.send("<message id='newmail' hide_timeout='3000'>You Have Mail</message>")

def muine_song_change_cb(arg):
    global song_change_prefs, osd
    if not song_change_prefs['enabled']:
        return
    d = dict((x.strip() for x in  s1.split(":", 1)) for s1 in arg.split("\n"))
    info = SongInfo()
    info.title = d['title']
    info.album = d['album']
    info.artist = d['artist']
    info.duration = int(d['duration'])
    info.track_number = int(d['track_number'])
    song = format_psong(info)
    osd.send("<message id='muine' hide_timeout='10000'>" + song + "</message>")

def escape(text):
    return xml.sax.saxutils.escape(text, {'\\': '&apos;',
                                          '"': '&quot;'})

XCHAT_EAT_NONE	= 0        # pass it on through!
XCHAT_EAT_XCHAT = 1        # don't let xchat see this event
XCHAT_EAT_PLUGIN = 2       # don't let other plugins see this event
XCHAT_EAT_ALL =	(XCHAT_EAT_XCHAT|XCHAT_EAT_PLUGIN) # don't let anything see this event

try:
    _xchat_ver = [int(x) for x in os.popen("xchat --version 2> /dev/null").read().split()[1].split('.')]
except IndexError:
    _xchat_ver = None

XCHAT_DBUS_SERVICE = "org.xchat.service"
if _xchat_ver >= [2,8]:
    XCHAT_DBUS_OBJECT = "/org/xchat/Remote"
    XCHAT_DBUS_INTERFACE = "org.xchat.plugin"
else:
    XCHAT_DBUS_OBJECT = "/org/xchat/RemoteObject"
    XCHAT_DBUS_INTERFACE = "org.xchat.interface"
del _xchat_ver

def monitor_xchat(bus):
    remote_object = bus.get_object(XCHAT_DBUS_SERVICE, XCHAT_DBUS_OBJECT)
    xchat = dbus.Interface(remote_object, XCHAT_DBUS_INTERFACE)
    event_ids = []
    for event in ('Channel Action Hilight', 'Channel Msg Hilight',
                  'Private Message', 'Private Message to Dialog'):
        id_ = xchat.HookPrint(event, 0, XCHAT_EAT_NONE)
        event_ids.append(id_)
        atexit.register(xchat.unhook, id_)
        
    def print_cb(word, id_, context_id=None):
        global xchat_prefs, osd
        if id_ not in event_ids:
            return
        if not xchat_prefs['enabled']:
            return
        sender = word[0]
        channel= xchat.GetInfo('channel')
        text = word[1]
        message = "<span foreground='red'>%s</span> "\
                  "<span weight='bold' foreground='white'>&lt;</span>%s"\
                  "<span weight='bold' foreground='white'>&gt;</span>" % \
                  (escape(channel), escape(sender))
        if xchat_prefs['full']:
            message += ' ' + escape(text)
        message = "<message id='xchat' ellipsize='end'>%s</message>" % message
        osd.send(message)
    
    remote_object.connect_to_signal("PrintSignal", print_cb,
                                    dbus_interface=XCHAT_DBUS_INTERFACE)


def xchat_registered_cb(*args):
    try:
        monitor_xchat(dbus.SessionBus())
    except dbus.DBusException, ex:
        print "XChat monitoring is not working:", ex

def screensaver_active_cb(active):
    global screensaver_active
    screensaver_active = active


def xscreensaver_state_changed(mon, state):
    global screensaver_active
    if state == 'UNBLANK':
        screensaver_active = False
    elif state == 'LOCK':
        screensaver_active = True

def monitor_xscreensaver():
    try:
        mon = XScreenSaverMonitor()
    except gobject.GError:
        print "xscreensaver not available"
    else:
        mon.connect("state-changed", xscreensaver_state_changed)


def rhythmbox_dbus_song_change_cb(uri):
    if not uri:
        return
    global song_change_prefs, osd
    if not song_change_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    rbshellobj = bus.get_object('org.gnome.Rhythmbox', '/org/gnome/Rhythmbox/Shell')
    rbshell = dbus.Interface(rbshellobj, 'org.gnome.Rhythmbox.Shell')
    props = rbshell.getSongProperties(uri)
    info = SongInfo()
    info.title = props['title']
    info.album = props['album']
    info.artist = props['artist']
    info.duration = int(props['duration'])
    info.track_number = int(props['track-number'])
    song = format_psong(info)
    osd.send("<message id='rhythmbox' hide_timeout='10000'>" + song + "</message>")


## --------- pidgin
def format_ppidgin(buddy, action):
    return "<span foreground='yellow'>%s</span> " \
           "<span size='smaller' style='italic'>%s</span>" % (buddy, action)

def pidgin_event_buddyon_cb(buddyid):
    global pidgin_prefs, osd
    if not pidgin_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    pidginobj = bus.get_object("im.pidgin.purple.PurpleService", "/im/pidgin/purple/PurpleObject")
    pidgin = dbus.Interface(pidginobj, "im.pidgin.purple.PurpleInterface")
    alias = pidgin.PurpleBuddyGetAlias(buddyid)
    message = format_ppidgin(alias, "logs in")
    osd.send("<message id='pidgin' hide_timeout='3000'>" + message + "</message>")

def pidgin_event_buddyoff_cb(buddyid):
    global pidgin_prefs, osd
    if not pidgin_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    pidginobj = bus.get_object("im.pidgin.purple.PurpleService", "/im/pidgin/purple/PurpleObject")
    pidgin = dbus.Interface(pidginobj, "im.pidgin.purple.PurpleInterface")
    alias = pidgin.PurpleBuddyGetAlias(buddyid)
    message = format_ppidgin(alias, "logs out")
    osd.send("<message id='pidgin' hide_timeout='3000'>" + message + "</message>")

def pidgin_event_message_cb(account, name):
    global pidgin_prefs, osd
    if not pidgin_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    pidginobj = bus.get_object("im.pidgin.purple.PurpleService", "/im/pidgin/purple/PurpleObject")
    pidgin = dbus.Interface(pidginobj, "im.pidgin.purple.PurpleInterface")
    buddy = pidgin.PurpleFindBuddy(account, name)
    if buddy != 0:
        alias = pidgin.PurpleBuddyGetAlias(buddy)
    else:
        alias = name
    message = format_ppidgin(alias, "is writing a message")
    osd.send("<message id='pidgin' hide_timeout='3000'>" + message + "</message>")

def pidgin_event_messagereceived_cb(account, name, message, conversation, flags):
    global pidgin_prefs, osd
    if not pidgin_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    pidginobj = bus.get_object("im.pidgin.purple.PurpleService", "/im/pidgin/purple/PurpleObject")
    pidgin = dbus.Interface(pidginobj, "im.pidgin.purple.PurpleInterface")
    buddy = pidgin.PurpleFindBuddy(account, name)
    if buddy != 0:
        alias = pidgin.PurpleBuddyGetAlias(buddy)
    else:
        alias = name
    message = format_ppidgin(alias, "sent a message")
    osd.send("<message id='pidgin' hide_timeout='3000'>" + message + "</message>")

def monitor_pidgin(bus):
    bus.add_signal_receiver(pidgin_event_buddyon_cb, "BuddySignedOn",
                            "im.pidgin.purple.PurpleInterface")
    bus.add_signal_receiver(pidgin_event_buddyoff_cb, "BuddySignedOff",
                            "im.pidgin.purple.PurpleInterface")
    bus.add_signal_receiver(pidgin_event_message_cb, "BuddyTyping",
                            "im.pidgin.purple.PurpleInterface")
    bus.add_signal_receiver(pidgin_event_messagereceived_cb, "ReceivedImMsg",
                            "im.pidgin.purple.PurpleInterface")
## --------- /pidgin

## --------- gaim

def format_pgaim(buddy, action):
    return "<span foreground='yellow'>%s</span> " \
           "<span size='smaller' style='italic'>%s</span>" % (buddy, action)

def gaim_event_buddyon_cb(buddyid):
    global gaim_prefs, osd
    if not gaim_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    gaimobj = bus.get_object("net.sf.gaim.GaimService", "/net/sf/gaim/GaimObject")
    gaim = dbus.Interface(gaimobj, "net.sf.gaim.GaimInterface")
    alias = gaim.GaimBuddyGetAlias(buddyid)
    message = format_pgaim(alias, "logs in")
    osd.send("<message id='gaim' hide_timeout='3000'>" + message + "</message>")

def gaim_event_buddyoff_cb(buddyid):
    global gaim_prefs, osd
    if not gaim_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    gaimobj = bus.get_object("net.sf.gaim.GaimService", "/net/sf/gaim/GaimObject")
    gaim = dbus.Interface(gaimobj, "net.sf.gaim.GaimInterface")
    alias = gaim.GaimBuddyGetAlias(buddyid)
    message = format_pgaim(alias, "logs out")
    osd.send("<message id='gaim' hide_timeout='3000'>" + message + "</message>")

def gaim_event_message_cb(account, name):
    global gaim_prefs, osd
    if not gaim_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    gaimobj = bus.get_object("net.sf.gaim.GaimService", "/net/sf/gaim/GaimObject")
    gaim = dbus.Interface(gaimobj, "net.sf.gaim.GaimInterface")
    buddy = gaim.GaimFindBuddy(account, name)
    if buddy != 0:
        alias = gaim.GaimBuddyGetAlias(buddy)
    else:
        alias = name
    message = format_pgaim(alias, "is writing a message")
    osd.send("<message id='gaim' hide_timeout='3000'>" + message + "</message>")

def gaim_event_messagereceived_cb(account, name, message, conversation, flags):
    global gaim_prefs, osd
    if not gaim_prefs['enabled']:
        return
    bus = dbus.SessionBus()
    gaimobj = bus.get_object("net.sf.gaim.GaimService", "/net/sf/gaim/GaimObject")
    gaim = dbus.Interface(gaimobj, "net.sf.gaim.GaimInterface")
    buddy = gaim.GaimFindBuddy(account, name)
    if buddy != 0:
        alias = gaim.GaimBuddyGetAlias(buddy)
    else:
        alias = name
    message = format_pgaim(alias, "sent a message")
    osd.send("<message id='gaim' hide_timeout='3000'>" + message + "</message>")

def monitor_gaim(bus):
    bus.add_signal_receiver(gaim_event_buddyon_cb, "BuddySignedOn",
                            "net.sf.gaim.GaimInterface")
    bus.add_signal_receiver(gaim_event_buddyoff_cb, "BuddySignedOff",
                            "net.sf.gaim.GaimInterface")
    bus.add_signal_receiver(gaim_event_message_cb, "BuddyTyping",
                            "net.sf.gaim.GaimInterface")
    bus.add_signal_receiver(gaim_event_messagereceived_cb, "ReceivedImMsg",
                            "net.sf.gaim.GaimInterface")
## --------- /gaim

def monitor_dbus_apps():
    bus = dbus.SessionBus()
    bus.add_signal_receiver(evolution_new_mail_callback, "Newmail",
                            "org.gnome.evolution.mail.dbus.Signal")
    bus.add_signal_receiver(muine_song_change_cb, "SongChanged",
                            "org.gnome.Muine.Player")
    bus.add_signal_receiver(rhythmbox_dbus_song_change_cb, "playingUriChanged",
                            "org.gnome.Rhythmbox.Player")
    bus.add_signal_receiver(xchat_registered_cb, "NameOwnerChanged",
                            "org.freedesktop.DBus", arg0=XCHAT_DBUS_SERVICE)
    bus.add_signal_receiver(screensaver_active_cb, "ActiveChanged",
                            "org.gnome.ScreenSaver")
    monitor_pidgin(bus)
    try:
        monitor_xchat(bus)
    except dbus.DBusException, ex:
        print "XChat monitoring is not working:", ex

if HAVE_DBUS:
    _our_bus_name = None

    def start_dbus_interface():
        global _our_bus_name
        session_bus = dbus.SessionBus()
        bus = dbus.SessionBus()
        if dbus.version >= (0,80):
            _our_bus_name = None
            # need to keep the BusName referenced in order to hang on to the name
            try:
                _our_bus_name = dbus.service.BusName("pt.inescporto.telecom.GnomeOSD.EventBridge",
                                                 bus, do_not_queue=True)
            except dbus.NameExistsException:
                raise SystemExit("already running")
        else:
            retval = dbus.dbus_bindings.bus_request_name(session_bus.get_connection(),
                                                         "pt.inescporto.telecom.GnomeOSD.EventBridge",
                                                         dbus.dbus_bindings.NAME_FLAG_DO_NOT_QUEUE)
            if retval in (dbus.dbus_bindings.REQUEST_NAME_REPLY_PRIMARY_OWNER,
                          dbus.dbus_bindings.REQUEST_NAME_REPLY_ALREADY_OWNER):
                pass
            elif retval in (dbus.dbus_bindings.REQUEST_NAME_REPLY_EXISTS,
                            dbus.dbus_bindings.REQUEST_NAME_REPLY_IN_QUEUE):
                raise SystemExit("already running")


def main():
    gettext.install("gnome-osd", gnome_osd_conf.datadir + "/locale")
    gnome.program_init(os.path.basename(sys.argv[0]), gnome_osd_conf.VERSION)
    cli = gnome.ui.master_client()
    cli.connect("die", die_cb)
    cli.set_restart_command(sys.argv[:1])
    global listener
    listener = bonobo.event_source_client_add_listener(
        bonobo.activation.activate("iid == 'OAFIID:Bonobo_Activation_EventSource'"),
        monitor_rhythmbox, "Bonobo/ObjectDirectory:activation:register")

    monitor_rhythmbox()
    monitor_xscreensaver()

    if HAVE_DBUS:
        monitor_dbus_apps()
        start_dbus_interface()
    else:
        print >> sys.stderr, "Unable to monitor D-BUS applications"

    bonobo.main()