~jconti/recent-notifications/trunk

"""

Lens.py

April 20, 2011

"""

import logging

from gi.repository import Dee, Gio, GLib, GObject, Unity

from Notification import Notification

import Timestamp

BUS_NAME = "net.launchpad.RecentNotifications"

OBJECT_PATH = "/net/launchpad/RecentNotifications"

DBUS_NAME_FLAG_DO_NOT_QUEUE = 0x4

DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER = 1

logging.basicConfig(level=logging.DEBUG)

logger = logging.getLogger("Lens")

class Group:

  RECENT_NOTIFICATIONS = 0

class LensException(Exception):

  pass

class Lens(object):

  def __init__(self):

    self._connection = Gio.bus_get_sync(Gio.BusType.SESSION, None)

    if not self.request_name(BUS_NAME):

      raise LensException("Failed to acquire bus name: {0}".format(BUS_NAME))

    self._messages = []

    self._entry = Unity.PlaceEntryInfo.new(OBJECT_PATH + "/mainentry")

    self._sections_model = Dee.SharedModel.new(BUS_NAME + ".SectionsModel")

    self._sections_model.set_schema("s", "s")

    self._entry.props.sections_model = self._sections_model

    self._groups_model = Dee.SharedModel.new(BUS_NAME + ".GroupsModel")

    self._groups_model.set_schema("s", "s", "s")

    self._entry.props.entry_renderer_info.props.groups_model = self._groups_model

    self._results_model = Dee.SharedModel.new(BUS_NAME + ".ResultsModel")

    self._results_model.set_schema("s", "s", "u", "s", "s", "s")

    self._entry.props.entry_renderer_info.props.results_model = self._results_model

    self._sections_model.connect("notify::synchronized", self.on_sections_synchronized)

    self._groups_model.connect("notify::synchronized", self.on_groups_synchronized)

    self._results_model.connect("notify::synchronized", self.on_results_synchronized)

    self._entry.connect("notify::active-search", self.on_search_changed)

    self._entry.connect("notify::active-section", self.on_section_changed)

    self._notify = Notification()

    self._notify.connect("message-received", self.on_message_received)

    self._controller = Unity.PlaceController.new(OBJECT_PATH)

    self._controller.add_entry(self._entry)

    self._controller.export()

  def append_message_to_results(self, message):

    if message.app_icon == "":

      icon = "file:///home/jconti/Projects/bzr/recent-notifications/icons/notification-normal.svg"

    else:

      icon = message.app_icon

    self._results_model.prepend("", icon, Group.RECENT_NOTIFICATIONS,

        "text/html", message.summary, "{0}\n{1} from {2}".format(message.body,

          Timestamp.locale_datetime(message.timestamp), message.app_name))

  def on_message_received(self, monitor, message):

    self._messages.append(message)

    if self._results_model.get_property("synchronized"):

      self.append_message_to_results(message)

      self._results_model.flush_revision_queue()

  def on_groups_synchronized(self, groups_model, *args):

    groups_model.clear()

    groups_model.append("UnityHorizontalTileRenderer",

      "Recent Notifications",

      "file:///home/jconti/Projects/bzr/recent-notifications/icons/notification-unread.svg")

  def on_results_synchronized(self, results_model, *args):

    results_model.clear()

    for message in self._messages:

      self.append_message_to_results(message)

    results_model.flush_revision_queue()

  def on_sections_synchronized(self, sections_model, *args):

    sections_model.clear()

    sections_model.append("All Applications",

        Gio.ThemedIcon.new("document").to_string())

  def on_search_changed(self, entry, *args):

    entry.freeze_notify()

    active_search = entry.get_property("active_search")

    search_string = active_search.get_search_string()

    logger.debug("Searching for: {0}".format(search_string))

    active_search.finished()

    GObject.idle_add(self.thaw, entry)

  def on_section_changed(self, entry, section):

    active_section = entry.get_property("active_section")

    logger.debug("Changed section: {0}".format(active_section))

  def request_name(self, name):

    proxy = Gio.DBusProxy.new_sync(self._connection, 0, None, 

        "org.freedesktop.DBus", "/org/freedesktop/DBus",

        "org.freedesktop.DBus", None)

    result = proxy.call_sync("RequestName", 

        GLib.Variant("(su)", (BUS_NAME, DBUS_NAME_FLAG_DO_NOT_QUEUE)), 

        Gio.DBusCallFlags.NONE, -1, None)

    value = result.unpack()

    if len(value) < 1 or value[0] != DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER:

      logger.debug("RequestName returned value: {0}".format(value))

      return False

    else:

      return True

  def thaw(self, entry):

    entry.thaw_notify()

    return False

def main():

  lens = Lens()

  loop = GObject.MainLoop()

  loop.run()

if __name__ == "__main__":

  main()