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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
//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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
#!/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