Recently, I created this simple circuit (and corresponding code) which makes an emergency phone call alert to an Arduino via an Asterisk PBX. This project was for someone who needed to be alerted (silently) if someone made an emergency phone call on their Asterisk PBX. By the way, the free and open source Asterisk PBX is a fantastic phone system which can be easily adapted (via custom scripting) to do amazing things!
For this project, the Asterisk dialing plan for the emergency number is set to invoke an AGI script (which in this case was written in python). The AGI python script reads the SIP configuration and creates an Asterisk call file for each SIP peer which puts them all in a conference room. The scenario for this is something like a condominium where there might not be a dedicated person to answer the emergency call, instead everyone in the building (so to speak) is called. The python script also accesses the WEB page of the server running on an Arduino which will rapidly flash an “emergency” light.
I created this simple relay circuit with an S108T02 solid state relay to control the light (CFL). Here is the Fritzing diagram:
You can download the Fritzing code for the circuit here and modify it for your needs. Here is what the circuit looks like when it is built:
Note that the board is about double the size needed so I can add a second relay if future needs require. 🙄
The Arduino code uses the UIPEthernet library to basically create a WEB server. When it is 1st accessed the light is turned on. Access it again and it is turned off. It is a ‘toggle’ so it can be tested from a browser client. The circuit is attached to an arduino via the pinToToggle value (and also +5V and GND). Additionally the Arduino is connected to an ENC28J60 module for LAN/Internet access. See my previous posts here and here for connecting an Arduino Leonardo to an ENC28J60.
Here is the Arduino code:
//From earl@microcontrollerelectronics.com
#include <SPI.h>
#include <UIPEthernet.h>
byte mac[] = { 0x54, 0x34, 0x41, 0x30, 0x30, 0x31 };
// Change this to match your network
IPAddress ip(192, 168, 0, 50);
EthernetServer server(80);
long interval = 500;
unsigned long currentTime = 0;
unsigned long previousTime = 0;
int ledState = LOW;
// Pin the relay is connected to
#define pinToToggle 8
void setup() {
pinMode(pinToToggle,OUTPUT);
Serial.begin(115200);
Serial.println("ARDUINO WEB SERVER ");
Serial.println();
Ethernet.begin(mac, ip);
server.begin();
Serial.print("IP Address: ");
Serial.println(Ethernet.localIP());
}
void loop() {
EthernetClient client = server.available();
if (client) {
Serial.println("-> New Connection");
boolean currentLineIsBlank = true;
while (client.connected()) {
if (client.available()) {
char c = client.read();
if (c == '\n' && currentLineIsBlank) {
client.println("<!DOCTYPE html>");
client.println("<html xmlns='http://www.w3.org/1999/xhtml'>");
client.println("<head>\n<meta charset='UTF-8'>");
client.println("<title>Arduino WEB Server</title>");
client.println("</head>\n<body>");
client.println("<H2>Arduino Leonardo</H2>");
client.println("<pre>");
if (ledState == LOW) {
ledState = HIGH;
client.println("Light On!<br>");
digitalWrite(pinToToggle,HIGH);
}
else {
ledState = LOW;
client.println("Light Off!<br>");
digitalWrite(pinToToggle,LOW);
}
client.println("</pre>");
client.print("</body>\n</html>");
break;
}
if (c == '\n') currentLineIsBlank = true;
else if (c != '\r') {
currentLineIsBlank = false;
}
}
}
delay(10);
client.stop();
Serial.println(" Disconnected\n");
}
if (ledState == HIGH) {
currentTime = millis();
if(currentTime - previousTime > interval) {
previousTime = currentTime;
digitalWrite(pinToToggle, !digitalRead(pinToToggle));
}
}
}
The python AGI script:
#!/bin/env python
#
# Copyright (C) 2006-2016 earl@micpc.com
#
import sys, string, os, re, urllib2
SENDMAIL = "/usr/sbin/sendmail"
class AGI:
"""
Class AGI facilitates writing AGI scripts in Python.
Exported functions:
Write(message):
Writes message to Asterisk Console.
Cmd(command):
Send command to Asterisk, read result.
The result is a two element tuple:
[0] A text string giving the entire result returned by Asterisk
[1] The "result=" integer extracted from the result line
If we get an unhappy response from Asterisk or if the result
returned to the right of the equals sign is not an integer we
issue an error message and terminate the script.
Exported Variables:
env
Dictionary containing the various environment startup items, as
passed to us by Asterisk.
"""
def Write(self,data):
"""
Write unbuffered line output to STDERR.
Ensures data is flushed out.
"""
sys.stderr.write(str(data) + "\n")
sys.stderr.flush()
def Error(self,data):
"""
Write unbuffered line output to STDERR.
Ensures data is flushed out.
"""
sys.stderr.write(str(data) + "\n")
sys.stderr.flush()
def Cmd(self,command):
"""
Send an AGI command to Asterisk; read back the response.
The result is a two element tuple:
[0] A text string giving the entire result returned by Asterisk
[1] The "result=" integer extracted from the result line
If we get an unhappy response from Asterisk or if the result
returned to the right of the equals sign is not an integer we
issue an error message and terminate the script.
"""
try:
sys.stdout.write(str(command) + "\n")
sys.stdout.flush()
Response = sys.stdin.readline()
if Response[:11] <> '200 result=':
#we didn't get a happy response for AGI
raise "AgiError"
#accumulate integer portion
J = 11
while J < len(Response) and (Response[J] in '0123456789'):
J += 1
res = Response[11:J]
try:
ResInt = int(res)
except ValueError:
#there is no integer immediately to the right of the equal sign
raise "AgiError"
except "AgiError":
self.Error('Unexpected response to AGI command.')
self.Error('Command: %s'%command)
self.Error('Response: %s'%Response)
self.Error('Script terminated.')
sys.exit()
return (Response,ResInt)
def __init__(self):
"""
Read until blank line to get the AGI environment.
The lines read are used to build the dictionary 'env'.
"""
self.env = {}
while 1:
line = string.strip(sys.stdin.readline())
if line == '':
break
key,data = string.split(line,':')
key = string.strip(key)
data = string.strip(data)
if key <> '':
self.env[key] = data
# change this to match your Arduino WEB address
try:
urllib2.urlopen("http://192.168.0.50").read()
except:
pass
A = AGI()
sys.stderr.write(str(A.env) + "\n")
sys.stderr.flush()
cid = A.env['agi_callerid']
cidname = A.env['agi_calleridname']
p1 = re.compile('^\s*\[(.*?)\]')
p2 = re.compile('^\s*(\w+)\s*=>\s*(.+)\s*;?.*$')
p3 = re.compile('^\s*(\w+)\s*=\s*(.+)\s*;?.*$')
in_file = open("/etc/asterisk/sip.conf","r")
for in_line in in_file.readlines():
in_line = string.strip(in_line[:-1])
m1 = p1.match(in_line)
if m1:
s = p1.split(in_line)
if (s[1] == 'general'): continue
if (s[1] == 'authentication'): continue
user = s[1]
callfile = "/var/spool/asterisk/outgoing/" + user
ufile = open(callfile,'w')
rec = "Channel: SIP/"
rec += user + "\n"
rec += "MaxRetries: 5\nExtension: 700\nContext: wholegang\n"
rec += "Priority: 1\n"
rec += "Callerid: EMERGENCY " + cid + " : " + cidname + "\n"
rec += "RetryTime:30\nWaitTime: 60\n"
ufile.write(rec)
ufile.close()
continue
in_file.close()
body = "\n***Emergency Call***\n"
for k,v in A.env.iteritems():
body += k + " : " + v + "\n"
body += "\n"
body += "Phone Call from Number: " + A.env['agi_callerid']
body += " Id: " + A.env['agi_calleridname']
body += "\n"
p = os.popen("%s -t" % SENDMAIL, "w")
p.write("To: my_email_address@mydomain.com\n")
p.write("Subject: Emergency Call\n")
p.write("\n")
p.write(body)
status = p.close()
sys.exit()
If you try this out make sure to change the IP addresses in both scripts to match that of your Arduino. Also note that the python AGI script emails someone that the call was made. Check out the bottom part of the AGI script and change the email address to the correct one. (Or comment out that part if you don’t want it to email anyone.)
Recent Comments