Python script to hangup Asterisk SIP call via Automation

Summary: I created a python script to disconnect (hangup) an Asterisk call that takes an extension number as an argument. It solves a specific issue I wanted to resolve and I’m posting it here because I couldn’t find anything else like this in any Asterisk or HA forums and I thought someone else might be able to use it for something. Disclaimer: I’m not a professional programmer. There are lots of failure modes that my script will not handle gracefully. Feel free to improve as needed.

My issue I wanted to solve:
I have a Grandstream Door Intercom that uses SIP and has a built in IP camera. Camera integration works well with HA and intercom works well with Asterisk. I have the intercom programmed to create a SIP call to an Asterisk Call Group that rings all my phones and wall-mounted tablets. If no one answers, it goes to an IVR with a “push 1 to leave a voice message, push 2 to call my cell phone, push 3 to call my wife’s cell phone” prompt. If no one answers the phone when the doorbell is rung and the user doesn’t push any button, then it just disconnects after 10 seconds. All of this is working fine. And it’s pretty cool. The voice messages get emailed to me along with all other voice mails so there is no place to go to check messages other than my Inbox which I like.

If we are home then sometimes it is just easier to open the door rather than answer one of the phones or tablets to talk to the person at the door. If I do this, then the intercom just keeps on ringing, then goes through the whole IVR prompt while I’m already talking to the person at the door. Kind of annoying. I knew I already had an HA binary sensor for my door (via my alarm panel) and I knew Asterisk has an API method (called Asterisk Manager Interface, AMI I learned) so I set out to find a way to cancel the call when the door opens. Here is how it works:

Asterisk Server:
edit /etc/asterisk/manager.conf to make sure the bindaddr is on an interface that you can reach with your HA server. Set to 0.0.0.0 for all interfaces. Use with caution if you have an interface on a public network.

Append the following to /etc/asterisk/manager_additional.conf to add a new restricted access user:

[homeassistant]
   secret = somegoodpassword
   deny=0.0.0.0/0.0.0.0
   permit=192.168.45.4/255.255.255.0
   read = call
   write = call
   writetimeout = 5000

Again, use with caution on a public network. User/pass will be transmitted in plain text.

The IP of my HA server is 192.168.45.4. Change to whatever you are running on. Restart Asterisk after adding this new user. Asterisk should now expose TCP port 5038 for the AMI interface. Asterisk is probably one of the most well documented projects there is so I’m not going to include any direction here. If you have trouble getting AMI to work, Google it.

Create the following Python Script hangup.py, save it on your HA server somewhere, and make sure your homeassistant user has read permissions on it:

import sys
import telnetlib
import re

host = "192.168.45.5" 				#IP Address of Asterisk Server
port = 5038 						#Port of Asterisk Server.  AMI default port is 5038.
user = "homeassistant" 				#username for Asterisk AMI as configured in /etc/asterisk/manager_additional.conf
password = "somegoodpassword" #password for Asterisk AMI as configured in /etc/asterisk/manager_additional.conf

debug = 0							#Set default debug mode.

actionindex = 1						#ActionID index required by Asterisk AMI protocol.  Always starts at 1 and increments with each command.

if len(sys.argv) == 1:
   sys.exit("ERROR: No Asterisk Extension Number passed in as argument.  Terminating.")

extension = sys.argv[1]

print("Trying to hangup Extension ",str(extension))

#Connect to Asterisk AMI
tn = telnetlib.Telnet(host,port)

#Wait till asterisk responds
out = tn.read_until('Asterisk Call Manager'.encode('ascii'),2)
if debug: print("RECEIVED:",out)

#Send Login Info
message = "Action: Login\nActionID: " + str(actionindex) + "\nUsername: " + user + "\nSecret: " + password + "\nEvents: off\n\n"
if debug: print("SENT:\n",message)
tn.write(message.encode('ascii'))
actionindex = actionindex + 1

#Wait till AMI responds
out = tn.read_until(b"Message: Authentication accepted",2)
if debug: print("RECEIVED:\n",out)

#Send Status Request
message = "Action: Status\nActionID: " + str(actionindex) + "\n\n"
if debug: print("SENT:\n",message)
tn.write(message.encode('ascii'))
actionindex = actionindex + 1

#Wait till AMI responds
out = tn.read_until(b"Event: StatusComplete",2)
if debug: print("RECEIVED:",out)

#Search for a SIP channel matching the extension we are looking for
matchtext = 'Channel: SIP/' + str(extension) + '-........'
match = re.search(matchtext.encode('ascii'), out)
if match:
   result = match.group()
   channel = result.lstrip('Channel: ')
   print("Extension ",str(extension)," is active. Found Channel ",channel)
   message = "Action: Hangup\nChannel: " + channel + "\nActionID: " + str(actionindex) + "\n\n"
   if debug: print("SENT:\n",message)
   tn.write(message.encode('ascii'))
   actionindex = actionindex + 1
   out = tn.read_until("Hungup",2)
   if debug: print("RECEIVED:",out)
else:
   print("No active Asterisk Channel for Extension ",str(extension)," was found")	

#Logoff
message = "Action: Logoff\nActionID: " + str(actionindex) + "\n\n"
if debug: print("SENT:\n",message)
tn.write(message.encode('ascii'))

#Wait till AMI responds
out = tn.read_until(b"fish",2)
if debug: print("RECEIVED:",out)
print("Done. Terminating.")

The above takes a single command line argument: the extension number you want to hang up. Unfortunately, Asterisk AMI needs the name of a SIP Channel to initiate a hangup. Most of the complexity in this script was associated with parsing AMI’s status report to convert an extension to a SIP channel. The script does nothing if the desired extension has no active SIP Channel open.

Put this in your HA configuration.yaml:

shell_command:
  hangup_doorbell: 'python /path/to/hangup.py 120'
  #the above line will hangup a call on extension 120 if it is active.  Change to desired extension number.

I realize there is a phython script method already built into HA. I couldn’t make it work. The above seemed much easier since none of the script arguments are dynamic. Someone tell me how I should have done this better.

Then add the automation:

- id: 'someuniqueid'
  alias: Cancel Doorbell if Front Door Opens
  trigger:
  - entity_id: binary_sensor.front_door
    from: 'off'
    platform: state
  condition: []
  action:
  - service: shell_command.hangup_doorbell

That’s it. Now my intercom call (or any other SIP call I desire) can be controlled via automation. In the case above, the call drops within two seconds of me opening the door. I believe all of this delay is related to my alarm panel which is how I’m getting the binary sensor for the door. The python hangup script executes in milliseconds based on my testing.

This could pretty easily be used to originate a SIP call also. Or do pretty much anything else programatically to an Asterisk server. So use it for whatever you want. If you make any improvements, then please let me know.

Very neat approach… I just started here with sip and an add-on asterisk and integration is being developed at the moment. But your scenario has never been taken into account :crossed_fingers::+1: thanks for the hint an solution.ill propose to add this directly in the integration .if not I’ll use your script…thumbs up!

Nice, it works! but i think this code was written for python2
Since python2 is not present anymore, i changed 2 lines , the message and channel line… now it can be fired too from HA:



if match:
   result = match.group()
   # channel = result.lstrip('Channel: ')
   channel = result.lstrip(b'Channel: ')
   print("Extension ",str(extension)," is active. Found Channel ",channel)
   message = "Action: Hangup\nChannel: " + str(channel.decode()) + "\nActionID: " + str(actionindex) + "\n\n"
   # message = "Action: Hangup\nChannel: " + channel + "\nActionID: " + str(actionindex) + "\n\n"
   if debug: print("SENT:\n",message)
   tn.write(message.encode('ascii'))
   actionindex = actionindex + 1
   out = tn.read_until(b"Hungup",2)
   if debug: print("RECEIVED:",out)
1 Like

Awesome, thanks. Also, I believe the default SIP driver in Asterisk changed from “SIP” to “PSIP” or something like that. You would have to edit my script as needed based on which driver Asterisk is actually using.

1 Like

yes, pjsip is for later :slight_smile:

Just tested: works perfectly Thanks!

Indeed , for me it’s also instant, no delay

hello all how would that script look like to be used with pjsip?