Asterisk Call Notifications to PushBullet

Several years ago, I wrote a python script to notify me when my asterisk server got an incoming call. I programmed this out of pure laziness: I wanted to know who was calling me without having to look at my phone.
This script would show a pop-up on my laptop whenever I got a call and the notification would show who was calling and for whom. This was very useful since I was traveling a lot during that time, so I would know if somebody was trying to reach me at home.
I’m still using this script, but I’m finding myself using my cellphone more often than my laptop when I’m out of the house. So, I modified my script to send a notification to my smart phone instead of my laptop.

Even though it would have been very good learning experience, I didn’t want to program a fully fledged app for this. First of all, I would have needed to setup a new development environment, learn a whole new API stack, deal with deployment, etc. Fortunately, I heard of Pushbullet. This app can receive notifications – and do much more – from an web API so it seemed like the perfect way to resolve my problem.
Simply, I only needed to send an web request using curl to the pushbullet API using the provided API key (found on Pushbullet’s setting page):

curl -u $API_KEY: https://api.pushbullet.com/v2/pushes -d type=note -d title="Alert" -d body="Notification Body"

This simple request would pop-up a notification on my smartphone – see the screenshot above. The only thing remaining to do was to cleanup my old script, remove everything related to the GUI and add the curl command. After testing the following script, I added it on my asterisk server and made sure it was launched on startup using upstart.
Here’s the script. You can use it and adapt it has you wish, but a short credit would be nice.
[sourcecode language=”python”]
#!/usr/bin/env python
”’
Created on 2015-05-18
@author: olivier@olihb.com
”’
from twisted.internet import reactor
from starpy.manager import AMIFactory
from twisted.internet import task
import logging
from starpy import manager
import os
#asterisk AMI login
server="asterisk_server.host.com"
port=5038
username="ami_user"
secret="ami_pass"
extensions={"SIP/6002":"Olivier",
"SIP/6000":"Olivier – 2"}
API_KEY = "XXXXXXXXXXXXXXXXX"
log = logging.getLogger("pyCalledMe-PB")
timeouttask=None
timeoutping=5
timeoutloop=30
class callMeFactory(AMIFactory):
cbconnect=None
def __init__(self):
AMIFactory.__init__(self, username, secret)
def connect(self):
df = self.login(server, port)
if self.cbconnect!=None:
df.addCallback(self.cbconnect)
def clientConnectionLost(self,connector,reason):
log.info("connection lost – connecting again")
reactor.callLater(1,self.connect)
def clientConnectionFailed(self,connector,reason):
log.info("connection failed – connecting again")
reactor.callLater(1,self.connect)
def onDial(protocol,event):
if ‘destination’ in event:
destination=event[‘destination’]
for s in extensions.keys():
if destination.startswith(s):
cid=event[‘calleridnum’]
cidname=event[‘calleridname’]
extname=extensions[s]
os.system("curl -silent -u "+API_KEY+": https://api.pushbullet.com/v2/pushes -d type=note -d title=\"Call for "+extname+"\" -d body=\""+cidname+" – "+cid+"\" > /dev/null")
def checknetlink(protocol):
def ontimeout():
log.info("timeout")
if dc.active():
dc.cancel()
timeouttask.stop()
protocol.transport.loseConnection()
def canceltimeout(*val):
if dc.active():
dc.cancel()
log.info("cancel timeout")
#log.info(val)
def success(val):
log.info("ping")
pass
log.info("setting timeout")
dc = reactor.callLater(timeoutping,ontimeout)
df = protocol.ping()
df.addBoth(canceltimeout)
df.addCallback(success)
df.addErrback(ontimeout)
def onLogin(protocol):
df = protocol.registerEvent("Dial",onDial)
global timeouttask
timeouttask = task.LoopingCall(checknetlink,protocol)
timeouttask.start(timeoutloop);
return df
def main():
cm = callMeFactory()
cm.cbconnect=onLogin
cm.connect()
def killapp(*args):
reactor.stop()
return True
if __name__ == ‘__main__’:
#log.setLevel(logging.DEBUG)
#manager.log.setLevel( logging.DEBUG )
logging.basicConfig()
reactor.callWhenRunning(main)
reactor.run()
[/sourcecode]