Lucas Taylor avatar Lucas Taylor committed 54fce48

Initial checkin, works for simple cases

Comments (0)

Files changed (6)

+txGrowl
+=======
+
+protocol.py
+    UDP Growl Client for Twisted
+
+observer.py
+    A Twisted log observer that utilizes txGrowl to send a Growl message to registered daemons
+
+
+Usage
+*****
+
+..code:: python
+
+    from txGrowl import GrowlClient
+    from txGrowl import GrowlObserver
+
+    app_name = "My App"
+    password = "password"
+    destinations = ["127.0.0.1",]
+
+    # Setup observer
+    growler = GrowlObserver(app_name, password, destinations)
+
+    # Require this key to be present in the log message in order to Growl
+    growler.required_key = "growl"
+
+    # Register notification types that Growl should display
+    growler.addNotification("Significant Event")
+    growler.addNotification("Not Configurable", gui_enabled=False)
+
+    growler.register() # Register application with destination growl daemons
+
+    log.addObserver(growler)
+
+    log.msg("Boo", growl=True, notification="Default")
+    log.msg("Priority 1 Message", growl=True, notification="Special", priority=1)
+    log.err(Exception("Something Bad Happened"), growl=True)
+
+
+Notifications
+-------------
+Your application will have "Default" and "Error" available as Growl notifications that
+may be configured via the Growl System Preferences.
+Any notifications you add via addNotification will also be available
+
+
+Priorities
+----------
+Messages may have numeric Priorities:
+
+from txGrowl.protocol import growlPriority
+growlPriority = {
+    "Very Low": -2,
+    "Moderate": -1,
+    "Normal":    0,
+    "High":      1,
+    "Emergency" :2
+}
+
+The default is 0, (Normal)
+
+
+TODO
+****
+
+* Fix DirtyReactorAggregateErrors in tests
+"Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean."
+* Send messages to multiple growl daemons
+* Figure out how to register an icon
+* Is it a problem to register the app multiple times?
+from protocol import GrowlClient
+from observer import GrowlObserver
+"""observer.py
+
+Register an instance of the GrowlObserver to receive all log events:
+
+    twisted.python.log.addObserver(GrowlObserver())
+    
+"""
+from txGrowl import GrowlClient
+
+class GrowlObserver(GrowlClient):
+    """GrowlObserver"""
+    def __init__(self, app_name="txGrowl", password=None, destinations='127.0.0.1', required_key=None):
+        GrowlClient.__init__(self, app_name, password, destinations)
+        
+        self.required_key = required_key
+        
+        self.observed = []
+        self.defaultNotification = "Default"
+        self.addNotification("Default")
+        self.addNotification("Error")
+
+
+    def __call__(self, event):
+        if self.required_key not in event:
+            return
+            
+        isError = event['isError']
+        required_key = event[self.required_key]
+        title = event.get('title', self.app_name)
+        priority = event.get('priority', 0)
+        notification = event.get('notification', self.defaultNotification)
+
+        
+        if isError:
+            notification = "Error"
+            try:
+                msg = event['failure'].getErrorMessage()
+            except:
+                msg = '\n'.join(event['message'])
+        else:
+            msg = '\n'.join(event['message'])
+            
+        self.notify(notification, title, msg, priority)
+        self.observed.append((title, msg, priority))
+"""protocol.py
+Growl UDP protocol implementation
+
+Packet encoding and constants provided by the growl sdk
+See <http://growl.info/> for more information.
+
+"""
+import struct, md5
+
+from twisted.internet.protocol import DatagramProtocol
+from twisted.internet import reactor
+from twisted.python import log
+from twisted.python.failure import Failure
+
+# Growl constants from pygrowl
+GROWL_UDP_PORT = 9887
+GROWL_PROTOCOL_VERSION = 1
+
+GROWL_TYPE_REGISTRATION_MD5 = 0 #The packet type of registration packets with MD5 authentication.
+GROWL_TYPE_REGISTRATION_SHA256 = 2 #The packet type of registration packets with SHA-256 authentication.
+GROWL_TYPE_REGISTRATION_NOAUTH = 4 #The packet type of registration packets without authentication.
+GROWL_TYPE_NOTIFICATION_MD5 = 1 #The packet type of notification packets with MD5 authentication.
+GROWL_TYPE_NOTIFICATION_SHA256 = 3 #The packet type of notification packets with SHA-256 authentication.
+GROWL_TYPE_NOTIFICATION_NOAUTH = 5 #The packet type of notification packets without authentication.
+
+
+GROWL_TYPE_REGISTRATION = GROWL_TYPE_REGISTRATION_MD5
+GROWL_TYPE_NOTIFICATION = GROWL_TYPE_NOTIFICATION_MD5
+
+
+
+GROWL_APP_NAME="ApplicationName"
+GROWL_APP_ICON="ApplicationIcon"
+GROWL_NOTIFICATIONS_DEFAULT="DefaultNotifications"
+GROWL_NOTIFICATIONS_ALL="AllNotifications"
+GROWL_NOTIFICATIONS_USER_SET="AllowedUserNotifications"
+
+GROWL_NOTIFICATION_NAME="NotificationName"
+GROWL_NOTIFICATION_TITLE="NotificationTitle"
+GROWL_NOTIFICATION_DESCRIPTION="NotificationDescription"
+GROWL_NOTIFICATION_ICON="NotificationIcon"
+GROWL_NOTIFICATION_APP_ICON="NotificationAppIcon"
+GROWL_NOTIFICATION_PRIORITY="NotificationPriority"
+		
+GROWL_NOTIFICATION_STICKY="NotificationSticky"
+
+GROWL_APP_REGISTRATION="GrowlApplicationRegistrationNotification"
+GROWL_APP_REGISTRATION_CONF="GrowlApplicationRegistrationConfirmationNotification"
+GROWL_NOTIFICATION="GrowlNotification"
+GROWL_SHUTDOWN="GrowlShutdown"
+GROWL_PING="Growl Ping"
+GROWL_PONG="Growl Pong"
+GROWL_IS_READY="Lend Me Some Sugar; I Am Your Neighbor!"
+
+
+
+class GrowlProtocol(DatagramProtocol):
+    
+    def __init__(self, app_name="txGrowl", password=None, destinations=None):
+        self.app_name = app_name
+        self.password = password
+
+        if isinstance(destinations, str):
+            destinations = [destinations]
+        self.destinations = destinations or ['127.0.0.1']
+        
+    
+    def register(self, notifications=None, defaultNotifications=None):
+        """Register application with Growl server"""
+        notifications = notifications or []
+        defaultNotifications = defaultNotifications or []
+        if notifications:
+            data = self.encodeRegistration(notifications, defaultNotifications)
+            self.sendDatagram(data)
+    
+    
+    def encodeRegistration(self, notifications, defaultNotifications):
+        """Encode and return a Growl registration packet"""
+        data = struct.pack("!BBH",
+                           GROWL_PROTOCOL_VERSION,
+                           GROWL_TYPE_REGISTRATION,
+                           len(self.app_name) )
+        data += struct.pack("BB",
+                            len(notifications),
+                            len(defaultNotifications) )
+        data += self.app_name
+        for i in notifications:
+            encoded = i.encode("utf-8")
+            data += struct.pack("!H", len(encoded))
+            data += encoded
+        for i in defaultNotifications:
+            data += struct.pack("B", i)
+        return self.encodePassword(data)
+        
+    
+    def notify(self, notification, title, message, priority):
+        """Send Notification to Growl daemon"""
+        data = self.encodeNotify(notification, title, message, priority)
+        self.sendDatagram(data)
+        
+    
+    def encodeNotify(self, notification, title, message, priority=0, sticky=False):
+        """Encode and return Growl notification packet"""
+        application  = self.app_name.encode("utf-8")
+        notification = notification.encode("utf-8")
+        title       = title.encode("utf-8")
+        message  = message.encode("utf-8")
+        flags = (priority & 0x07) * 2
+        if priority < 0: 
+            flags |= 0x08
+        if sticky: 
+            flags = flags | 0x0001
+        data = struct.pack("!BBHHHHH",
+                           GROWL_PROTOCOL_VERSION,
+                           GROWL_TYPE_NOTIFICATION,
+                           flags,
+                           len(notification),
+                           len(title),
+                           len(message),
+                           len(application) )
+        data += notification
+        data += title
+        data += message
+        data += application
+        return self.encodePassword(data)
+        
+    
+    def encodePassword(self, data):
+        """Return md5 encoded data and password (if present)"""
+        checksum = md5.new()
+        checksum.update(data)
+        if self.password:
+           checksum.update(self.password)
+        data += checksum.digest()
+        return data
+        
+    
+    def sendDatagram(self, data):
+        """
+        Send any pending Growl notifications
+        """
+        self.transport.write(data)
+        
+    
+    def startProtocol(self):
+        for dest in self.destinations:
+            self.transport.connect(dest, GROWL_UDP_PORT)
+
+
+    def datagramReceived(self, data):
+        log.msg("Rx from Growl: %s" % data)
+
+
+    def connectionRefused(self):
+        log.msg("Growl connection refused")
+
+
+
+class GrowlClient(object):
+    """Basic protocol wrapper for Growl"""
+    priorities = {
+        "Very Low":-2,
+        "Moderate":-1,
+        "Normal":0,
+        "High":1,
+        "Emergency":2
+    }
+
+
+    def __init__(self, app_name=None, password=None, destinations=None):
+        self.app_name = app_name
+        self.password = password
+        self.destinations = destinations or []
+        self.notifications = []
+        self.defaults = [] # default in gui
+        
+        self.protocol = GrowlProtocol(self.app_name, self.password, self.destinations)
+        self.lport = reactor.listenUDP(0, self.protocol)
+        
+    
+    def addNotification(self, notification="Default", gui_enabled=True):
+        """Adds a notification type and sets whether it is enabled in Growl preferences"""
+        if notification not in self.notifications:
+            self.notifications.append(notification)
+            if gui_enabled:
+                self.defaults.append(len(self.notifications)-1)
+        
+    
+    def register(self):
+        """Register application with all destination Growl daemons"""
+        if self.notifications:
+            self.protocol.register(self.notifications, self.defaults)
+        
+    
+    def notify(self, notification=None, title=None, message=None, priority=0):
+        if message is not None:
+            self.protocol.notify(notification, title, message, priority)
+
+

Empty file added.

tests/test_growl.py

+# -*- coding: utf-8 -*-
+from twisted.trial.unittest import TestCase
+from twisted.python import log
+
+from txGrowl import GrowlObserver
+
+
+
+class GrowlTestCase(TestCase):
+    """Tests Growl Observer"""
+    
+    def setUp(self):
+
+        app_name = "GrowlTestCase"
+        password = "password"
+        destinations = ["127.0.0.1",]
+
+        growler = GrowlObserver(app_name, password, destinations)
+        growler.required_key = "growl"
+
+        # Register notification types that Growl should display
+        # These can be passed in logging calls: log.msg("Boo", growl=True, notification="Special Notification")
+        growler.addNotification("Extreme Error")
+        #growler.addNotification("Not Configurable", gui_enabled=False)
+        
+        growler.register() # Register application with destination growl daemons
+
+        self.growler = growler
+
+        log.addObserver(self.growler)
+        self.addCleanup(log.removeObserver, self.growler)
+
+
+    def tearDown(self):
+        def finalize(_):
+            log.msg("Finished", growl=True)
+        del self.growler
+        
+    def test_growler(self):
+        log.msg("Significant Event", growl=True)
+        
+        growled = self.growler.observed.pop()
+        title, message, priority = growled
+        self.assertEqual(title, "GrowlTestCase")
+        self.assertEqual(message, "Significant Event")
+        self.assertEqual(priority, 0)
+
+        log.err("Problem", growl=True)
+        self.flushLoggedErrors()
+
+    def test_ignore(self):
+        # Ignore messages w/o required key
+        self.growler.required_key = None
+        log.msg("This should be ignored by Growl observer", growl=False, notification="Default")
+
+        self.assertRaises(IndexError, self.growler.observed.pop)
+        self.flushLoggedErrors()        
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.