codex-storage
/
deluge
mirror of https://github.com/codex-storage/deluge.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Moved all notifications into their own classes from which the Core, GtkUi, and possibly latter, Web, plugins inherit from. 

The plugins now just handle configuration stuff.
This commit is contained in:
Pedro Algarvio 2009-11-22 02:35:47 +00:00
parent c4f0920c18
commit e327d87ebc
4 changed files with 362 additions and 390 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

438
deluge/plugins/notifications/notifications/common.py
View File

@ -36,10 +36,28 @@
# this exception statement from your version. If you delete this exception
# statement from all source files in the program, then also delete it here.
#
from twisted.internet import defer
import gtk
import smtplib
from twisted.internet import defer, threads
from deluge import component
from deluge.event import known_events
from deluge.log import LOG as log
from deluge.ui.client import client
import deluge.common
try:
import pygame
SOUND_AVAILABLE = True
except ImportError:
SOUND_AVAILABLE = False
try:
import pynotify
POPUP_AVAILABLE = True
except ImportError:
POPUP_AVAILABLE = False
def get_resource(filename):
import pkg_resources, os
@ -48,9 +66,10 @@ def get_resource(filename):
class CustomNotifications(component.Component):
def __init__(self, impl):
config = None # shut-up pylint
def __init__(self, plugin_name=None):
component.Component.__init__(self, "Notifications")
self.__impl = impl
self.custom_notifications = {
"email": {},
"popup": {},
@ -65,105 +84,9 @@ class CustomNotifications(component.Component):
for kind in self.custom_notifications.iterkeys():
for eventtype in self.custom_notifications[kind].copy().iterkeys():
wrapper, handler = self.custom_notifications[kind][eventtype]
self.__deregister_custom_provider(kind, eventtype)
self._deregister_custom_provider(kind, eventtype)
def handle_custom_email_notification(self, result, eventtype):
raise NotImplementedError("%s does not implement this function" %
self.__class__.__name__)
def handle_custom_popup_notification(self, result, eventtype):
raise NotImplementedError("%s does not implement this function" %
self.__class__.__name__)
def handle_custom_blink_notification(self, result, eventtype):
raise NotImplementedError("%s does not implement this function" %
self.__class__.__name__)
def handle_custom_sound_notification(self, result, eventtype):
raise NotImplementedError("%s does not implement this function" %
self.__class__.__name__)
def register_custom_email_notification(self, eventtype, handler):
"""This is used to register email notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return a tuple of (email_subject, email_contents).
"""
if self.__impl != 'core':
return defer.fail("This function can only be called from a "
"CorePlugin")
self.__register_custom_provider('email', eventtype, handler)
def deregister_custom_email_notification(self, eventtype):
if self.__impl != 'core':
return defer.fail("This function can only be called from a "
"CorePlugin")
self.__deregister_custom_provider('email', eventtype)
def register_custom_popup_notification(self, eventtype, handler):
"""This is used to register popup notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return a tuple of (popup_title, popup_contents).
"""
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__register_custom_provider('popup', eventtype, handler)
def deregister_custom_popup_notification(self, eventtype):
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__deregister_custom_provider('popup', eventtype)
def register_custom_blink_notification(self, eventtype, handler):
"""This is used to register blink notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return `True` or `False` to blink or not the
trayicon.
"""
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__register_custom_provider('blink', eventtype, handler)
def deregister_custom_blink_notification(self, eventtype):
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__deregister_custom_provider('blink', eventtype)
def register_custom_sound_notification(self, eventtype, handler):
"""This is used to register sound notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return either '' to use the sound defined on the
notification preferences, the path to a sound file, which will then be
played or None, where no sound will be played at all.
"""
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__register_custom_provider('sound', eventtype, handler)
def deregister_custom_sound_notification(self, eventtype):
if self.__impl != 'gtk':
return defer.fail("This function can only be called from a "
"GtkPlugin")
self.__deregister_custom_provider('sound', eventtype)
def __handle_custom_providers(self, kind, eventtype, *args, **kwargs):
def _handle_custom_providers(self, kind, eventtype, *args, **kwargs):
log.debug("\n\nCalling CORE's custom %s providers for %s: %s %s",
kind, eventtype, args, kwargs)
if eventtype in self.config["subscriptions"][kind]:
@ -177,12 +100,12 @@ class CustomNotifications(component.Component):
d.addErrback(self._on_notify_failure, kind)
return d
def __register_custom_provider(self, kind, eventtype, handler):
if not self.__handled_eventtype(eventtype):
def _register_custom_provider(self, kind, eventtype, handler):
if not self._handled_eventtype(eventtype, handler):
return defer.succeed("Event not handled")
if eventtype not in self.custom_notifications:
def wrapper(*args, **kwargs):
return self.__handle_custom_providers(kind, eventtype,
return self._handle_custom_providers(kind, eventtype,
*args, **kwargs)
self.custom_notifications[kind][eventtype] = (wrapper, handler)
else:
@ -195,7 +118,7 @@ class CustomNotifications(component.Component):
from deluge.ui.client import client
client.register_event_handler(eventtype, wrapper)
def __deregister_custom_provider(self, kind, eventtype):
def _deregister_custom_provider(self, kind, eventtype):
wrapper, handler = self.custom_notifications[kind][eventtype]
try:
component.get("EventManager").deregister_event_handler(
@ -206,11 +129,21 @@ class CustomNotifications(component.Component):
client.deregister_event_handler(eventtype, wrapper)
self.custom_notifications[kind].pop(eventtype)
def __handled_eventtype(self, eventtype):
def _handled_eventtype(self, eventtype, handler):
if eventtype not in known_events:
log.error("The event \"%s\" is not known" % eventtype)
return False
log.debug('\n\nKLASSS %s: %s: %s',
handler.__class__,
isinstance(handler.__class__, self.__class__),
handler in dir(self))
log.debug(dir(handler))
log.debug(handler.im_self)
log.debug(self)
log.debug("IM SELF: %s", handler.im_self is self)
if known_events[eventtype].__module__.startswith('deluge.event'):
if handler.im_self is self:
return True
log.error("You cannot register custom notification providers "
"for built-in event types.")
return False
@ -223,3 +156,296 @@ class CustomNotifications(component.Component):
def _on_notify_failure(self, failure, kind):
log.debug("\n\nNotification failure using %s: %s", kind, failure)
return failure
class CoreNotifications(CustomNotifications):
def enable(self):
CustomNotifications.enable(self)
self.register_custom_email_notification('TorrentFinishedEvent',
self._on_torrent_finished_event)
def disable(self):
self.deregister_custom_email_notification('TorrentFinishedEvent')
CustomNotifications.disable(self)
def register_custom_email_notification(self, eventtype, handler):
"""This is used to register email notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return a tuple of (email_subject, email_contents).
"""
self._register_custom_provider('email', eventtype, handler)
def deregister_custom_email_notification(self, eventtype):
self._deregister_custom_provider('email', eventtype)
def handle_custom_email_notification(self, result, eventtype):
if not self.config['smtp_enabled']:
return defer.succeed("SMTP notification not enabled.")
subject, message = result
log.debug("\n\nSpawning new thread to send email with subject: %s: %s",
subject, message)
# Spawn thread because we don't want Deluge to lock up while we send the
# email.
return threads.deferToThread(self._notify_email, subject, message)
def get_handled_events(self):
handled_events = []
for evt in sorted(known_events.keys()):
if known_events[evt].__module__.startswith('deluge.event'):
if evt not in ('TorrentFinishedEvent',):
# Skip all un-handled built-in events
continue
classdoc = known_events[evt].__doc__.strip()
handled_events.append((evt, classdoc))
log.debug("Handled Notification Events: %s", handled_events)
return handled_events
def _notify_email(self, subject='', message=''):
log.debug("Email prepared")
to_addrs = '; '.join(self.config['smtp_recipients'])
headers = """\
From: %(smtp_from)s
To: %(smtp_recipients)s
Subject: %(subject)s
""" % {'smtp_from': self.config['smtp_from'],
'subject': subject,
'smtp_recipients': to_addrs}
message = '\r\n'.join((headers + message).splitlines())
try:
server = smtplib.SMTP(self.config["smtp_host"],
self.config["smtp_port"])
except Exception, err:
err_msg = _("There was an error sending the notification email:"
" %s") % err
log.error(err_msg)
return err
security_enabled = self.config['smtp_tls']
if security_enabled:
server.ehlo()
if not server.esmtp_features.has_key('starttls'):
log.warning("TLS/SSL enabled but server does not support it")
else:
server.starttls()
server.ehlo()
if self.config['smtp_user'] and self.config['smtp_pass']:
try:
server.login(self.config['smtp_user'], self.config['smtp_pass'])
except smtplib.SMTPHeloError, err:
err_msg = _("The server didn't reply properly to the helo "
"greeting: %s") % err
log.error(err_msg)
return err
except smtplib.SMTPAuthenticationError, err:
err_msg = _("The server didn't accept the username/password "
"combination: %s") % err
log.error(err_msg)
return err
try:
try:
server.sendmail(self.config['smtp_from'], to_addrs, message)
except smtplib.SMTPException, err:
err_msg = _("There was an error sending the notification email:"
" %s") % err
log.error(err_msg)
return err
finally:
if security_enabled:
# avoid false failure detection when the server closes
# the SMTP connection with TLS enabled
import socket
try:
server.quit()
except socket.sslerror:
pass
else:
server.quit()
return _("Notification email sent.")
def _on_torrent_finished_event(self, torrent_id):
log.debug("\n\nHandler for TorrentFinishedEvent called for CORE")
torrent = component.get("TorrentManager")[torrent_id]
torrent_status = torrent.get_status({})
# Email
subject = _("Finished Torrent \"%(name)s\"") % torrent_status
message = _(
"This email is to inform you that Deluge has finished "
"downloading \"%(name)s\", which includes %(num_files)i files."
"\nTo stop receiving these alerts, simply turn off email "
"notification in Deluge's preferences.\n\n"
"Thank you,\nDeluge."
) % torrent_status
return subject, message
d = defer.maybeDeferred(self.handle_custom_email_notification,
[subject, message],
"TorrentFinishedEvent")
d.addCallback(self._on_notify_sucess, 'email')
d.addErrback(self._on_notify_failure, 'email')
return d
class GtkUiNotifications(CustomNotifications):
def enable(self):
CustomNotifications.enable(self)
self.register_custom_blink_notification(
"TorrentFinishedEvent", self._on_torrent_finished_event_blink
)
self.register_custom_sound_notification(
"TorrentFinishedEvent", self._on_torrent_finished_event_sound
)
self.register_custom_popup_notification(
"TorrentFinishedEvent", self._on_torrent_finished_event_popup
)
def disable(self):
self.deregister_custom_blink_notification("TorrentFinishedEvent")
self.deregister_custom_sound_notification("TorrentFinishedEvent")
self.deregister_custom_popup_notification("TorrentFinishedEvent")
CustomNotifications.disable(self)
def register_custom_popup_notification(self, eventtype, handler):
"""This is used to register popup notifications for custom event types.
:param event: the event name
:param type: string
:param handler: function, to be called when `:param:event` is emitted
Your handler should return a tuple of (popup_title, popup_contents).
"""
self._register_custom_provider('popup', eventtype, handler)
def deregister_custom_popup_notification(self, eventtype):
self._deregister_custom_provider('popup', eventtype)
def register_custom_blink_notification(self, eventtype, handler):
"""This is used to register blink notifications for custom event types.
:param event: str, the event name
:param handler: function, to be called when `:param:event` is emitted
Your handler should return `True` or `False` to blink or not the
trayicon.
"""
self._register_custom_provider('blink', eventtype, handler)
def deregister_custom_blink_notification(self, eventtype):
self._deregister_custom_provider('blink', eventtype)
def register_custom_sound_notification(self, eventtype, handler):
"""This is used to register sound notifications for custom event types.
:param event: the event name
:type event: string
:param handler: function to be called when `:param:event` is emitted
Your handler should return either '' to use the sound defined on the
notification preferences, the path to a sound file, which will then be
played or None, where no sound will be played at all.
"""
self._register_custom_provider('sound', eventtype, handler)
def deregister_custom_sound_notification(self, eventtype):
self._deregister_custom_provider('sound', eventtype)
def handle_custom_popup_notification(self, result, eventtype):
title, message = result
return defer.maybeDeferred(self.__popup, title, message)
def handle_custom_blink_notification(self, result, eventtype):
if result:
return defer.maybeDeferred(self.__blink)
return defer.succeed("Won't blink. The returned value from the custom "
"handler was: %s", result)
def handle_custom_sound_notification(self, result, eventtype):
if isinstance(result, basestring):
if not result and eventtype in self.config['custom_sounds']:
return defer.maybeDeferred(
self.__play_sound, self.config['custom_sounds'][eventtype])
return defer.maybeDeferred(self.__play_sound, result)
return defer.succeed("Won't play sound. The returned value from the "
"custom handler was: %s", result)
def __blink(self):
self.systray.blink(True)
return defer.succeed(_("Notification Blink shown"))
def __popup(self, title='', message=''):
if not self.config['popup_enabled']:
return defer.succeed(_("Popup notification is not enabled."))
if not POPUP_AVAILABLE:
return defer.fail(_("pynotify is not installed"))
if pynotify.init("Deluge"):
icon = gtk.gdk.pixbuf_new_from_file_at_size(
deluge.common.get_pixmap("deluge.svg"), 48, 48)
self.note = pynotify.Notification(title, message)
self.note.set_icon_from_pixbuf(icon)
if not self.note.show():
err_msg = _("pynotify failed to show notification")
log.warning(err_msg)
return defer.fail(err_msg)
return defer.succeed(_("Notification popup shown"))
def __play_sound(self, sound_path=''):
if not self.config['sound_enabled']:
return defer.succeed(_("Sound notification not enabled"))
if not SOUND_AVAILABLE:
err_msg = _("pygame is not installed")
log.warning(err_msg)
return defer.fail(err_msg)
pygame.init()
try:
if not sound_path:
sound_path = self.config['sound_path']
alert_sound = pygame.mixer.music
alert_sound.load(sound_path)
alert_sound.play()
except pygame.error, message:
err_msg = _("Sound notification failed %s") % (message)
log.warning(err_msg)
return defer.fail(err_msg)
else:
msg = _("Sound notification Success")
log.info(msg)
return defer.succeed(msg)
def _on_torrent_finished_event_blink(self, torrent_id):
return True # Yes, Blink
def _on_torrent_finished_event_sound(self, torrent_id):
# Since there's no custom sound hardcoded, just return ''
return ''
def _on_torrent_finished_event_popup(self, torrent_id):
d = client.core.get_torrent_status(torrent_id, ["name", "num_files"])
d.addCallback(self._on_torrent_finished_event_got_torrent_status)
d.addErrback(self._on_torrent_finished_event_torrent_status_failure)
return d
def _on_torrent_finished_event_torrent_status_failure(self, failure):
log.debug("Failed to get torrent status to be able to show the popup")
def _on_torrent_finished_event_got_torrent_status(self, torrent_status):
log.debug("\n\nhandler for TorrentFinishedEvent GTKUI called. Torrent Status")
title = _("Finished Torrent")
message = _("The torrent \"%(name)s\" including %(num_files)i "
"has finished downloading.") % torrent_status
return title, message

133
deluge/plugins/notifications/notifications/core.py
View File

@ -37,17 +37,12 @@
# statement from all source files in the program, then also delete it here.
#
import smtplib
from twisted.internet import defer, threads
from deluge.event import known_events
from deluge.log import LOG as log
from deluge.plugins.pluginbase import CorePluginBase
import deluge.component as component
import deluge.configmanager
from deluge.core.rpcserver import export
from test import TestEmailNotifications
from common import CustomNotifications
from notifications.common import CoreNotifications
DEFAULT_PREFS = {
"smtp_enabled": False,
@ -64,28 +59,20 @@ DEFAULT_PREFS = {
}
}
class Core(CorePluginBase, CustomNotifications):
class Core(CorePluginBase, CoreNotifications):
def __init__(self, plugin_name):
CorePluginBase.__init__(self, plugin_name)
CustomNotifications.__init__(self, 'core')
self.tn = TestEmailNotifications('core')
CoreNotifications.__init__(self)
def enable(self):
CoreNotifications.enable(self)
self.config = deluge.configmanager.ConfigManager(
"notifications-core.conf", DEFAULT_PREFS)
component.get("EventManager").register_event_handler(
"TorrentFinishedEvent", self._on_torrent_finished_event
)
log.debug("\n\nENABLING CORE NOTIFICATIONS\n\n")
self.tn.enable()
def disable(self):
self.tn.disable()
log.debug("\n\nDISABLING CORE NOTIFICATIONS\n\n")
CustomNotifications.disable(self)
def update(self):
pass
CoreNotifications.disable(self)
@export
def set_config(self, config):
@ -101,112 +88,4 @@ class Core(CorePluginBase, CustomNotifications):
@export
def get_handled_events(self):
handled_events = []
for evt in sorted(known_events.keys()):
if known_events[evt].__module__.startswith('deluge.event'):
if evt not in ('TorrentFinishedEvent',):
# Skip the base class for all events
continue
classdoc = known_events[evt].__doc__.strip()
handled_events.append((evt, classdoc))
log.debug("Handled Notification Events: %s", handled_events)
return handled_events
def handle_custom_email_notification(self, result, eventtype):
if not self.config['smtp_enabled']:
return defer.succeed("SMTP notification not enabled.")
subject, message = result
log.debug("\n\nSending email with subject: %s: %s", subject, message)
return threads.deferToThread(self._notify_email, subject, message)
def _notify_email(self, subject='', message=''):
log.debug("Email prepared")
config = self.config
to_addrs = '; '.join(config['smtp_recipients'])
headers = """\
From: %(smtp_from)s
To: %(smtp_recipients)s
Subject: %(subject)s
""" % {'smtp_from': config['smtp_from'],
'subject': subject,
'smtp_recipients': to_addrs}
message = '\r\n'.join((headers + message).splitlines())
try:
server = smtplib.SMTP(config["smtp_host"], config["smtp_port"])
except Exception, err:
err_msg = _("There was an error sending the notification email:"
" %s") % err
log.error(err_msg)
return err
security_enabled = config['smtp_tls']
if security_enabled:
server.ehlo()
if not server.esmtp_features.has_key('starttls'):
log.warning("TLS/SSL enabled but server does not support it")
else:
server.starttls()
server.ehlo()
if config['smtp_user'] and config['smtp_pass']:
try:
server.login(config['smtp_user'], config['smtp_pass'])
except smtplib.SMTPHeloError, err:
err_msg = _("The server didn't reply properly to the helo "
"greeting: %s") % err
log.error(err_msg)
return err
except smtplib.SMTPAuthenticationError, err:
err_msg = _("The server didn't accept the username/password "
"combination: %s") % err
log.error(err_msg)
return err
try:
try:
server.sendmail(config['smtp_from'], to_addrs, message)
except smtplib.SMTPException, err:
err_msg = _("There was an error sending the notification email:"
" %s") % err
log.error(err_msg)
return err
finally:
if security_enabled:
# avoid false failure detection when the server closes
# the SMTP connection with TLS enabled
import socket
try:
server.quit()
except socket.sslerror:
pass
else:
server.quit()
return _("Notification email sent.")
def _on_torrent_finished_event(self, torrent_id):
log.debug("\n\nHandler for TorrentFinishedEvent called for CORE")
torrent = component.get("TorrentManager")[torrent_id]
torrent_status = torrent.get_status({})
# Email
subject = _("Finished Torrent \"%(name)s\"") % torrent_status
message = _(
"This email is to inform you that Deluge has finished "
"downloading \"%(name)s\", which includes %(num_files)i files."
"\nTo stop receiving these alerts, simply turn off email "
"notification in Deluge's preferences.\n\n"
"Thank you,\nDeluge."
) % torrent_status
d = defer.maybeDeferred(self.handle_custom_email_notification,
[subject, message],
"TorrentFinishedEvent")
d.addCallback(self._on_notify_sucess, 'email')
d.addErrback(self._on_notify_failure, 'email')
return d
return CoreNotifications.get_handled_events(self)

171
deluge/plugins/notifications/notifications/gtkui.py
View File

@ -41,7 +41,6 @@ from os.path import basename
import gtk
from twisted.internet import defer
from deluge.event import known_events, DelugeEvent
from deluge.log import LOG as log
from deluge.ui.client import client
from deluge.plugins.pluginbase import GtkPluginBase
@ -49,21 +48,9 @@ import deluge.component as component
import deluge.common
import deluge.configmanager
try:
import pygame
SOUND_AVAILABLE = True
except ImportError:
SOUND_AVAILABLE = False
try:
import pynotify
POPUP_AVAILABLE = True
except ImportError:
POPUP_AVAILABLE = False
# Relative imports
from common import get_resource, CustomNotifications
from test import TestEmailNotifications
from notifications.common import (get_resource, GtkUiNotifications,
SOUND_AVAILABLE, POPUP_AVAILABLE)
DEFAULT_PREFS = {
# BLINK
@ -90,11 +77,10 @@ RECIPIENT_FIELD, RECIPIENT_EDIT = range(2)
SND_EVENT, SND_EVENT_DOC, SND_NAME, SND_PATH = range(4)
class GtkUI(GtkPluginBase, CustomNotifications):
class GtkUI(GtkPluginBase, GtkUiNotifications):
def __init__(self, plugin_name):
GtkPluginBase.__init__(self, plugin_name)
CustomNotifications.__init__(self, 'gtk')
self.tn = TestEmailNotifications('gtk')
GtkUiNotifications.__init__(self)
def enable(self):
self.config = deluge.configmanager.ConfigManager(
@ -127,7 +113,6 @@ class GtkUI(GtkPluginBase, CustomNotifications):
'on_sound_path_update_preview': self.on_sound_path_update_preview
})
# component.get("Preferences").add_page("Notifications", self.prefs)
prefs = component.get("Preferences")
parent = self.prefs.get_parent()
if parent:
@ -154,10 +139,15 @@ class GtkUI(GtkPluginBase, CustomNotifications):
self.glade.get_widget('blink_enabled').set_property('sensitive',
False)
client.register_event_handler("TorrentFinishedEvent",
self._on_torrent_finished_event)
GtkUiNotifications.enable(self)
self.tn.enable()
def disable(self):
GtkUiNotifications.disable(self)
component.get("Preferences").remove_page("Notifications")
component.get("PluginManager").deregister_hook("on_apply_prefs",
self.on_apply_prefs)
component.get("PluginManager").deregister_hook("on_show_prefs",
self.on_show_prefs)
def build_recipients_model_populate_treeview(self):
# SMTP Recipients treeview/model
@ -238,59 +228,35 @@ class GtkUI(GtkPluginBase, CustomNotifications):
column.set_property('visible', False)
self.subscriptions_treeview.append_column(column)
renderer = gtk.CellRendererToggle()
renderer.set_property('activatable', True)
renderer.connect('toggled', self._on_email_col_toggled)
column = gtk.TreeViewColumn("Email", renderer, active=SUB_NOT_EMAIL)
column.set_clickable(True)
# column.add_attribute(renderer, "active", False)
# column.set_expand(True)
self.subscriptions_treeview.append_column(column)
renderer = gtk.CellRendererToggle()
# renderer.connect("edited", self.on_cell_edited, self.recipients_model)
# renderer.set_data("popup", SUB_NOT_POPUP)
renderer.set_property('activatable', True)
renderer.connect( 'toggled', self._on_popup_col_toggled)
column = gtk.TreeViewColumn("Popup", renderer, active=SUB_NOT_POPUP)
column.set_clickable(True)
# column.add_attribute(renderer, "active", False)
# column.set_expand(True)
self.subscriptions_treeview.append_column(column)
renderer = gtk.CellRendererToggle()
# renderer.connect("edited", self.on_cell_edited, self.recipients_model)
# renderer.set_data("blink", SUB_NOT_BLINK)
renderer.set_property('activatable', True)
renderer.connect( 'toggled', self._on_blink_col_toggled)
column = gtk.TreeViewColumn("Blink", renderer, active=SUB_NOT_BLINK)
column.set_clickable(True)
# column.add_attribute(renderer, "active", False)
# column.set_expand(True)
self.subscriptions_treeview.append_column(column)
renderer = gtk.CellRendererToggle()
# renderer.connect("edited", self.on_cell_edited, self.recipients_model)
renderer.set_property('activatable', True)
renderer.connect('toggled', self._on_sound_col_toggled)
# renderer.set_data("sound", SUB_NOT_SOUND)
column = gtk.TreeViewColumn("Sound", renderer, active=SUB_NOT_SOUND)
column.set_clickable(True)
# column.add_attribute(renderer, "active", False)
# column.set_expand(True)
self.subscriptions_treeview.append_column(column)
self.subscriptions_treeview.set_model(self.subscriptions_model)
def disable(self):
self.tn.disable()
CustomNotifications.disable(self)
component.get("Preferences").remove_page("Notifications")
component.get("PluginManager").deregister_hook("on_apply_prefs",
self.on_apply_prefs)
component.get("PluginManager").deregister_hook("on_show_prefs",
self.on_show_prefs)
def popuplate_what_needs_handled_events(self, handled_events,
email_subscriptions=[]):
self.populate_subscriptions(handled_events, email_subscriptions)
@ -344,12 +310,14 @@ class GtkUI(GtkPluginBase, CustomNotifications):
if sound:
current_sound_subscriptions.append(event)
default_sound_file = self.glade.get_widget("sound_path").get_filename()
log.debug("Default sound file: %s", default_sound_file)
old_sound_file = self.config['sound_path']
new_sound_file = self.glade.get_widget("sound_path").get_filename()
log.debug("Old Default sound file: %s New one: %s",
old_sound_file, new_sound_file)
custom_sounds = {}
for event_name, event_doc, filename, filepath in self.sounds_model:
log.debug("Custom sound for event \"%s\": %s", event_name, filename)
if filepath == default_sound_file:
if filepath == old_sound_file:
continue
custom_sounds[event_name] = filepath
log.debug(custom_sounds)
@ -358,7 +326,7 @@ class GtkUI(GtkPluginBase, CustomNotifications):
"popup_enabled": self.glade.get_widget("popup_enabled").get_active(),
"blink_enabled": self.glade.get_widget("blink_enabled").get_active(),
"sound_enabled": self.glade.get_widget("sound_enabled").get_active(),
"sound_path": default_sound_file,
"sound_path": new_sound_file,
"subscriptions": {
"popup": current_popup_subscriptions,
"blink": current_blink_subscriptions,
@ -542,90 +510,6 @@ class GtkUI(GtkPluginBase, CustomNotifications):
else:
self.glade.get_widget('sound_path').set_property('sensitive', False)
def on_sounds_treeview_clicked(self, widget, event):
if event.button == 1 and event.type == gtk.gdk._2BUTTON_PRESS:
selection = self.sounds_treeview.get_selection()
print "Double clicked on treeview", selection
def __blink(self):
self.systray.blink(True)
return defer.succeed(_("Notification Blink shown"))
def __popup(self, title='', message=''):
if not self.config['popup_enabled']:
return defer.succeed(_("Popup notification is not enabled."))
if not POPUP_AVAILABLE:
return defer.fail(_("pynotify is not installed"))
if pynotify.init("Deluge"):
icon = gtk.gdk.pixbuf_new_from_file_at_size(
deluge.common.get_pixmap("deluge.svg"), 48, 48)
self.note = pynotify.Notification(title, message)
self.note.set_icon_from_pixbuf(icon)
if not self.note.show():
err_msg = _("pynotify failed to show notification")
log.warning(err_msg)
return defer.fail(err_msg)
return defer.succeed(_("Notification popup shown"))
def __play_sound(self, sound_path=''):
if not self.config['sound_enabled']:
return defer.succeed(_("Sound notification not enabled"))
if not SOUND_AVAILABLE:
err_msg = _("pygame is not installed")
log.warning(err_msg)
return defer.fail(err_msg)
pygame.init()
try:
if not sound_path:
sound_path = self.config['sound_path']
alert_sound = pygame.mixer.music
alert_sound.load(sound_path)
alert_sound.play()
except pygame.error, message:
err_msg = _("Sound notification failed %s") % (message)
log.warning(err_msg)
return defer.fail(err_msg)
else:
msg = _("Sound notification Success")
log.info(msg)
return defer.succeed(msg)
# Internal methods
def _on_torrent_finished_event(self, torrent_id):
log.debug("\n\nhandler for TorrentFinishedEvent GTKUI called")
# Blink
d0 = defer.maybeDeferred(self.__blink)
d0.addCallback(self._on_notify_sucess, 'blink')
d0.addErrback(self._on_notify_failure, 'blink')
log.debug("Blink notification callback yielded")
# Sound
d1 = defer.maybeDeferred(self.__play_sound)
d1.addCallback(self._on_notify_sucess, 'sound')
d1.addErrback(self._on_notify_failure, 'sound')
log.debug("Sound notification callback yielded")
# Popup
d2 = client.core.get_torrent_status(torrent_id, ["name", "num_files"])
d2.addCallback(self._on_torrent_finished_event_got_torrent_status)
d2.addErrback(self._on_torrent_finished_event_torrent_status_failure)
return defer.succeed("\n\nGtkUI on torrent finished")
def _on_torrent_finished_event_torrent_status_failure(self, failure):
log.debug("Failed to get torrent status to be able to show the popup")
def _on_torrent_finished_event_got_torrent_status(self, torrent_status):
log.debug("\n\nhandler for TorrentFinishedEvent GTKUI called. Torrent Status")
title = _("Finished Torrent")
message = _("The torrent \"%(name)s\" including %(num_files)i "
"has finished downloading.") % torrent_status
d = defer.maybeDeferred(self.__popup, title, message)
d.addCallback(self._on_notify_sucess, 'popup')
d.addErrback(self._on_notify_failure, 'popup')
return d
def _on_email_col_toggled(self, cell, path):
self.subscriptions_model[path][SUB_NOT_EMAIL] = \
not self.subscriptions_model[path][SUB_NOT_EMAIL]
@ -646,22 +530,3 @@ class GtkUI(GtkPluginBase, CustomNotifications):
not self.subscriptions_model[path][SUB_NOT_SOUND]
return
def handle_custom_popup_notification(self, result, eventtype):
title, message = result
return defer.maybeDeferred(self.__popup, title, message)
def handle_custom_blink_notification(self, result, eventtype):
if result:
return defer.maybeDeferred(self.__blink)
return defer.succeed("Won't blink. The returned value from the custom "
"handler was: %s", result)
def handle_custom_sound_notification(self, result, eventtype):
if isinstance(result, basestring):
if not result and eventtype in self.config['custom_sounds']:
return defer.maybeDeferred(
self.__play_sound, self.config['custom_sounds'][eventtype])
return defer.maybeDeferred(self.__play_sound, result)
return defer.succeed("Won't play sound. The returned value from the "
"custom handler was: %s", result)

10
deluge/plugins/notifications/notifications/test.py
View File

@ -6,7 +6,7 @@
# License: BSD - Please view the LICENSE file for additional information.
# ==============================================================================
from twisted.internet import task
from twisted.internet import reactor, task
from deluge import component
from deluge.event import DelugeEvent
from deluge.log import LOG as log
@ -32,20 +32,22 @@ class TestEmailNotifications(component.Component):
log.debug("\n\nEnabling %s", self.__class__.__name__)
for event in self.events:
if self.__imp == 'core':
# component.get("CorePlugin.Notifications").register_custom_email_notification(
component.get("Notifications").register_custom_email_notification(
event.__class__.__name__,
self.custom_email_message_provider
)
elif self.__imp == 'gtk':
component.get("Notifications").register_custom_popup_notification(
notifications_component = component.get("Notifications")
notifications_component.register_custom_popup_notification(
event.__class__.__name__,
self.custom_popup_message_provider
)
component.get("Notifications").register_custom_blink_notification(
notifications_component.register_custom_blink_notification(
event.__class__.__name__,
self.custom_blink_message_provider
)
component.get("Notifications").register_custom_sound_notification(
notifications_component.register_custom_sound_notification(
event.__class__.__name__,
self.custom_sound_message_provider
)