Asterisk Call Notifications to PushBullet

Pushbullet Notification for an Incoming Call - not my real number!

Pushbullet Notification for an Incoming Call – no, it’s not my real phone number 🙂

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.

#!/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()

3 Thoughts on “Asterisk Call Notifications to PushBullet

  1. Hi,
    Can you point me out in the right directions i`m testing your script but for some reason is not working
    now i have like this

    curl –header ‘Access-Token:xxxxxxxxxxxxxxxxxxxxxx’ https://api.pushbullet.com/v2/pushes -d type=note -d title=”Asterisk Call” -d body=”Aici o sa scriem ce trebuie”

    Wich from my asterisk box is going well but from your script nope 🙁

    Asterisk box 1.8

    • Hi Robert,
      If you replace the os.system by a print and then execute the output, does it work? If so, it might be a problem with dependencies (twisted, starpy, etc.). If not, it might be a problem with curl or pushbullet. I’ve stopped using Asterisk daily so I’m sorry I can’t be of any help than this.

  2. Yes try it with print so i so was a missing s from Access-Token 🙂
    Now is it working well trying to figure out other purpose for it

    Manny thnks

Leave a Reply

Your email address will not be published. Required fields are marked *

Post Navigation