Code: Select all
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
import json
import time
import smtplib
import requests
### script properties ###
# edit the data after ":" to fit your case
p = {"RASP_IP": "192.168.1.121", # IP Adress of your Raspberry PI
"RASP_PORT": "8083", # Z-Way standard port is 8083
"LOGIN": "Z-Way-User", # Your Z-Way Account name
"PWD": "Password2000!", # Your Z-Way Account password
"DEVICES": ["DummyDevice_14"], # All devices you want a notify message from: ["Device1", "Device2", ... ]
"LEVEL": "on", # The status that triggers the notification - same for ALL devices!
"MAIL": "my_zway_mail@gmail.com", # the (Google) mail you want to send from
"MAIL_PWD": "my_zway_mail_pw", # the (Google) mails account password
"MAIL_TO": ["send_me@notification.com", "also_send@here.com"], # recipients you want to send the message
"SUBJECT": "Z-Way SmartHome: Smoke detection warning!", # subject text of notify message
"BODY_TEXT": "Warning: {0} triggered at {1} the smoke alarm!", # text of the mail, {0} and {1} must be there, they will be replaced with device name and time!
"MOBILANT_KEY": "xxxxxxxxxxxxxxxxxxxxxxx", # the API key you can find at Mobilant.com
"PHONE_NUMBERS": ["0049162123456789"], # the number(s) ["123","132",...] you want to send a message / None means no sms notification
"MOBILANT_SMS_ROUTE": "directplus"} # which route should be used (see Mobilant.com for details)
##### DON NOT EDIT BEYOND THIS POINT #####
class ZWayScript:
def __init__(self):
# get pid of this script, write a pid file
self.PID = os.getpid()
self.FILE_NAME = os.path.basename(sys.argv[0])
self.PID_FILE = "pid_{0}_{1}".format(self.FILE_NAME, time.strftime("%d.%m.%Y_%H.%M.%S"))
with open(self.PID_FILE, "w") as f:
f.write(str(self.PID))
f.close()
# ZWAY API CONFIG
self.SERVER = "http://{0}:{1}".format(p["RASP_IP"], p["RASP_PORT"])
print(self.SERVER)
self.API = "/ZAutomation/api/v1"
self.SESSION = requests.Session()
# ZWAY CONFIG
self.ZWAY_LOGIN = p["LOGIN"]
self.ZWAY_PWD = p["PWD"]
self.ZWAY_DEVICES = p["DEVICES"]
self.ZWAY_DEVICE_LEVEL = p["LEVEL"]
# MAIL CONFIG
self.MAIL_FROM = p["MAIL"]
self.MAIL_FROM_PWD = p["MAIL_PWD"]
self.MAIL_TO = p["MAIL_TO"]
self.MAIL_TO_SUBJECT = p["SUBJECT"]
self.MAIL_TO_BODY_TEXT = p["BODY_TEXT"]
# SMS CONFIG
self.MOBILANT_KEY = p["MOBILANT_KEY"]
self.PHONE_NUMBERS = p["PHONE_NUMBERS"]
self.MOBILANT_ROUTE = p["MOBILANT_SMS_ROUTE"]
def login(self):
# create session and login, Requests remembers the login data for the session
login_url = self.SERVER+self.API+"/login"
print(login_url)
payload = {"form":True,
"login":self.ZWAY_LOGIN,
"password":self.ZWAY_PWD,
"keepme":False,
"default_ui":1}
headers = {"Accept": "application/json",
"Content-Type": "application/json"}
auth_r = self.SESSION.post(login_url,
headers=headers,
data=json.dumps(payload))
print("Login-Response:", auth_r.status_code)
def run(self):
while True:
lc = time.localtime()
devices_r = self.SESSION.get(self.SERVER+self.API+"/devices")
print("Devices-Response:", devices_r.status_code)
devices = devices_r.json()
# looking for relevant data
for d in devices["data"]["devices"]:
# Status of the device (fire alarm/switch/fake device)
if d["id"] in self.ZWAY_DEVICES:
device_status = d["metrics"]["level"]
device_name = d["metrics"]["title"]
print("Device status:", str(device_status))
if str(device_status) == self.ZWAY_DEVICE_LEVEL:
# email notification
if self.MAIL_FROM:
self.send_mail_notification(device_name)
else:
print("WARNING: No mail to send from - can't send a mail!!!")
# sms notification
if self.PHONE_NUMBERS and len(self.PHONE_NUMBERS) >= 1:
self.send_sms_notification(device_name)
else:
print("WARNING: No phone number - will not send SMS!")
# check the status all 10 sec.
time.sleep(10)
def send_mail_notification(self, name):
print("Try to send a mail from: {0} to {1}!".format(self.MAIL_FROM, self.MAIL_TO))
# while loop trigger for sending mails
SEND_DONE = False
# create header and body
header = "From: {0}\n".format(self.MAIL_FROM)
header += "To: {0}\n".format(self.MAIL_TO)
header += "Subject: {0}\n\n".format(self.MAIL_TO_SUBJECT)
body = self.MAIL_TO_BODY_TEXT.format(name,
time.strftime("%d.%m.%Y %H:%M:%S"))
msg = header+body
# send mail
print("Try to send...")
while not SEND_DONE:
try:
s = smtplib.SMTP(host="smtp.gmail.com", port=587)
s.ehlo()
s.starttls()
s.login(self.MAIL_FROM, self.MAIL_FROM_PWD)
s.sendmail(self.MAIL_FROM, self.MAIL_TO, msg)
s.quit()
SEND_DONE = True
print("Mail was send!")
break
except:
# print all error messages
e = sys.exc_info()[0]
print(e)
def send_sms_notification(self, name):
for number in self.PHONE_NUMBERS:
print(number)
print("Try to send a SMS to number: {0}".format(number))
sms_msg_text = self.MAIL_TO_BODY_TEXT.format(name,
time.strftime("%d.%m.%Y %H:%M:%S"))
sms_msg_text = sms_msg_text.replace(" ", "+")
sms_send_url = "http://gw.mobilant.net/?key={0}&to={1}&message={2}&route={3}".format(self.MOBILANT_KEY,
number,
sms_msg_text,
self.MOBILANT_ROUTE)
if self.MOBILANT_ROUTE == "lowcostplus" or self.MOBILANT_ROUTE == "directplus":
sms_send_url += "&from=RaZberry"
print(sms_send_url)
self.SESSION.get(sms_send_url)
if __name__ == "__main__":
zs = ZWayScript()
zs.login()
zs.run()
How to run the script:http://<RASPBERRY_IP>:8083/ZAutomation/api/v1/devices
You can also edit your crontab to start the script on Raspberry boot, by adding the following line:python3 /path/to/the/file/name.py
@reboot python3 /path/to/the/file/name.py &
Code: Select all
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import sys
import json
import time
import smtplib
import requests
### script properties ###
# edit the data after ":" to fit your case
p = {"RASP_IP": "192.168.1.121", # IP Adress of your Raspberry PI
"RASP_PORT": "8083", # Z-Way standard port is 8083
"LOGIN": "Z-Way-User", # Your Z-Way Account name
"PWD": "Password2000!", # Your Z-Way Account password
"DEVICES": ["DummyDevice_14"], # All devices you want a notify message from: ["Device1", "Device2", ... ]
"LEVEL": "on", # The status that triggers the notification - same for ALL devices!
"MAIL": "my_zway_mail@gmail.com", # the (Google) mail you want to send from
"MAIL_PWD": "my_zway_mail_pw", # the (Google) mails account password
"MAIL_TO": ["send_me@notification.com", "also_send@here.com"], # recipients you want to send the message
"SUBJECT": "Z-Way SmartHome: Smoke detection warning!", # subject text of notify message
"BODY_TEXT": "Warning: {0} triggered at {1} the smoke alarm!", # text of the mail, {0} and {1} must be there, they will be replaced with device name and time!
"MOBILANT_KEY": "xxxxxxxxxxxxxxxxxxxxxxx", # the API key you can find at Mobilant.com
"PHONE_NUMBERS": ["0049162123456789"], # the number(s) ["123","132",...] you want to send a message / None means no sms notification
"MOBILANT_SMS_ROUTE": "directplus"} # which route should be used (see Mobilant.com for details)
##### DON NOT EDIT BEYOND THIS POINT #####
class ZWayScript:
def __init__(self):
# get pid of this script, write a pid file
self.PID = os.getpid()
self.FILE_NAME = os.path.basename(sys.argv[0])
self.PID_FILE = "pid_{0}_{1}".format(self.FILE_NAME, time.strftime("%d.%m.%Y_%H.%M.%S"))
with open(self.PID_FILE, "w") as f:
f.write(str(self.PID))
f.close()
# ZWAY API CONFIG
self.SERVER = "http://{0}:{1}".format(p["RASP_IP"], p["RASP_PORT"])
print(self.SERVER)
self.API = "/ZAutomation/api/v1"
self.SESSION = requests.Session()
# ZWAY CONFIG
self.ZWAY_LOGIN = p["LOGIN"]
self.ZWAY_PWD = p["PWD"]
self.ZWAY_DEVICES = p["DEVICES"]
self.ZWAY_DEVICE_LEVEL = p["LEVEL"]
self.ZWAY_DEVICE_LEVEL_OLD =p["LEVEL"]
# MAIL CONFIG
self.MAIL_FROM = p["MAIL"]
self.MAIL_FROM_PWD = p["MAIL_PWD"]
self.MAIL_TO = p["MAIL_TO"]
self.MAIL_TO_SUBJECT = p["SUBJECT"]
self.MAIL_TO_BODY_TEXT = p["BODY_TEXT"]
# SMS CONFIG
self.MOBILANT_KEY = p["MOBILANT_KEY"]
self.PHONE_NUMBERS = p["PHONE_NUMBERS"]
self.MOBILANT_ROUTE = p["MOBILANT_SMS_ROUTE"]
def login(self):
# create session and login, Requests remembers the login data for the session
login_url = self.SERVER+self.API+"/login"
print(login_url)
payload = {"form":True,
"login":self.ZWAY_LOGIN,
"password":self.ZWAY_PWD,
"keepme":False,
"default_ui":1}
headers = {"Accept": "application/json",
"Content-Type": "application/json"}
auth_r = self.SESSION.post(login_url,
headers=headers,
data=json.dumps(payload))
print("Login-Response:", auth_r.status_code)
def run(self):
while True:
lc = time.localtime()
devices_r = self.SESSION.get(self.SERVER+self.API+"/devices")
print("Devices-Response:", devices_r.status_code)
devices = devices_r.json()
# looking for relevant data
for d in devices["data"]["devices"]:
# Status of the device (fire alarm/switch/fake device)
if d["id"] in self.ZWAY_DEVICES:
device_status = d["metrics"]["level"]
device_name = d["metrics"]["title"]
print("Device status:", str(device_status))
if str(device_status) != self.ZWAY_DEVICE_LEVEL_OLD:
self.ZWAY_DEVICE_LEVEL_OLD = "none"
if str(device_status) == self.ZWAY_DEVICE_LEVEL and str(device_status) != self.ZWAY_DEVICE_LEVEL_OLD:
self.ZWAY_DEVICE_LEVEL_OLD = str(device_status)
# email notification
if self.MAIL_FROM:
self.send_mail_notification(device_name)
else:
print("WARNING: No mail to send from - can't send a mail!!!")
# sms notification
if self.PHONE_NUMBERS and len(self.PHONE_NUMBERS) >= 1:
self.send_sms_notification(device_name)
else:
print("WARNING: No phone number - will not send SMS!")
# check the status all 10 sec.
time.sleep(10)
def send_mail_notification(self, name):
print("Try to send a mail from: {0} to {1}!".format(self.MAIL_FROM, self.MAIL_TO))
# while loop trigger for sending mails
SEND_DONE = False
# create header and body
header = "From: {0}\n".format(self.MAIL_FROM)
header += "To: {0}\n".format(self.MAIL_TO)
header += "Subject: {0}\n\n".format(self.MAIL_TO_SUBJECT)
body = self.MAIL_TO_BODY_TEXT.format(name,
time.strftime("%d.%m.%Y %H:%M:%S"))
msg = header+body
# send mail
print("Try to send...")
while not SEND_DONE:
try:
s = smtplib.SMTP(host="mail.gmx.net", port=587)
s.ehlo()
s.starttls()
s.login(self.MAIL_FROM, self.MAIL_FROM_PWD)
s.sendmail(self.MAIL_FROM, self.MAIL_TO, msg)
s.quit()
SEND_DONE = True
print("Mail was send!")
break
except:
# print all error messages
e = sys.exc_info()[0]
print(e)
def send_sms_notification(self, name):
for number in self.PHONE_NUMBERS:
print(number)
print("Try to send a SMS to number: {0}".format(number))
sms_msg_text = self.MAIL_TO_BODY_TEXT.format(name,
time.strftime("%d.%m.%Y %H:%M:%S"))
sms_msg_text = sms_msg_text.replace(" ", "+")
sms_send_url = "http://gw.mobilant.net/?key=ENVfJAybda3bd739rxJgmAU&to={0}&message={1}&route={2}".format(number,
sms_msg_text,
self.MOBILANT_ROUTE)
if self.MOBILANT_ROUTE == "lowcostplus" or self.MOBILANT_ROUTE == "directplus":
sms_send_url += "&from=RaZberry"
print(sms_send_url)
self.SESSION.get(sms_send_url)
if __name__ == "__main__":
time.sleep(10)
zs = ZWayScript()
zs.login()
zs.run()
Sure this is also a solution doing it. But I was coding in Python and I thought about SMS sending, my first idea was to use an Mail2SMS service, but Mobilant offered also this easy to use http request method!pofs wrote:I'm pretty sure sending sms can be easily done from JS without calling external python script. Look at http.request() in developer manual.
This would allow direct access to devices, and you won't need to check whether ZWay is running.