Python

GNU/Linux Python3 Mail Probe

A small script I used to quickly send a mail messages (probe).

#!/usr/bin/env python3
# Small mail probe...
from datetime import datetime
import smtplib
from email.message import EmailMessage
from email.mime.text import MIMEText

# Configured values...
senderid = "my-name"
subject = "MailProbe from " + senderid
body = "*** This is a mail probe message to verify the connection ***"
sender = "sender@example.com"
recipients = ["receiver@example.com"]
password = "store-password-somewhere-else"

# Mail sender function...
def send_email(subject, body, sender, recipients, password):
    now = datetime.now()
    msg = MIMEText(body)
    msg['X-Ne7-Probe-Firetime'] = now.strftime("%Y-%m-%dT%H:%M:%S")
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = ', '.join(recipients)
    smtp_server = smtplib.SMTP_SSL('smtp.example.com', 465)
    smtp_server.login(sender, password)
    smtp_server.sendmail(sender, recipients, msg.as_string())
    smtp_server.quit()

send_email(subject, body, sender, recipients, password)

GNU/Linux PYTHON Script Read Values from JSON File

Another Python script to read values from a JSON file and store them in a mysql database.

import sys
import json
import csv
import MySQLdb
import mysql.connector
from datetime import datetime, date, timedelta

# Get yesterdays date...
yesterday = date.today() - timedelta(1)

# Create database connection...
dbconn = MySQLdb.connect(host='localhost', user='zonpan', passwd='maybe-store-pw-somwhere-else', db='databassename')
cursor = dbconn.cursor()

# Open Mysql Database...
with open('/path/to/directory/yesterday.json') as fd:
     json_data = json.load(fd)

# Insert Query...
add_zopbrengst = ("INSERT INTO ZOpbrengst (Tijdstip, Opbrengst) VALUES (%s, %s)")

# Parse JSON File and insert values...
datum = yesterday.strftime('%Y-%m-%d %H:%M')
opbrengst = json_data['data']['production']['yesterday'].replace(',', '.')
data_zo = (datum, opbrengst)
cursor.execute(add_zopbrengst, data_zo)
print(data_zo)

# Cleanup...
dbconn.commit()
cursor.close()
dbconn.close()

GNU/Linux PYTHON3 Script to Read HUE Sensor DATA

A small python3 script I wrote to read sensor data and store in a logg file.

#!/usr/bin/env python3
import sys
import json
import urllib.request
import requests
import parsedatetime as pdt

# Constants...
cal = pdt.Calendar()
hue_url_sensors = "http://path.to.hue.bridge/api/{api-key}/sensors/"
hue_url_lights = "http://path.to.hue.bridge/api/{api-key}/lights/"
air_url_1 = "http://path.to.other.sensor-1:8082/sensors/"
air_url_2 = "http://path.to.other.sensor-3:8082/sensors/"
rel_url = "http://path.to.relay.raspi:8081/relays/"
light_on = json.dumps({'on': True})
light_off = json.dumps({'on': False})


# Switch function...
def switch(key, value):
  action = 'N/A'
  url = hue_url_lights + key + "/state/"
  if value.lower() == 'off':
    result = requests.put(url, data=light_off.encode('utf-8'))
  elif value.lower() == 'on':
    result = requests.put(url, data=light_on.encode('utf-8'))
  print("INFO: ", "Light=", key, "Action=", value, "Result=", result)

# GetLights function
def getlights():
  # Retrieve huedata (lights)...
  output = ""
  web_data_l = urllib.request.urlopen(hue_url_lights)
  json_data_l = web_data_l.read()
  json_enc_l = web_data_l.info().get_content_charset('utf-8')
  json_lights = json.loads(json_data_l.decode(json_enc_l))
  for key in json_lights:
    if json_lights[key]['state']['reachable'] == True and json_lights[key]['state']['on'] == True:
      state = "on"
    else:
      state = "off"
    output = output + " " + str(key).zfill(2) + "\t" + json_lights[key]['name'].ljust(20) + str(key).ljust(2) + "\t" + state + "\n"
  return output

# GetSensors function
def getSensors():
  # Retreive Sensor Data
  output = ""
  web_data_l = urllib.request.urlopen(hue_url_sensors)
  json_data_l = web_data_l.read()
  json_enc_l = web_data_l.info().get_content_charset('utf-8')
  json_sensors = json.loads(json_data_l.decode(json_enc_l))
  for key in json_sensors:
    isValid = True
    if json_sensors[key]['type'] == 'ZLLLightLevel':
      value = json_sensors[key]['state']['lightlevel']
    elif json_sensors[key]['type'] == 'ZLLPresence':
      value = json_sensors[key]['state']['presence']
    elif json_sensors[key]['type'] == 'ZLLTemperature':
      value = json_sensors[key]['state']['temperature'] /100
    else:
      isValid = False
    if isValid == True:
      output = output + " " + str(key).zfill(2) + "\t"  + str(json_sensors[key]['name']).ljust(20) + "\t\t" + str(value) + ""
  return output

# GetSensors function
def getAirinfo(Url, DeviceId):
  # Retreive Sensor Data
  format = '%Y-%m-%dT%H:%M:%S'
  output = ""
  web_data_l = urllib.request.urlopen(Url)
  json_data_l = web_data_l.read()
  json_enc_l = web_data_l.info().get_content_charset('utf-8')
  json_sensors = json.loads(json_data_l.decode(json_enc_l))['sensors']
  for key in json_sensors:
    output = output + str(key) + ": " + str(json_sensors[key]['value']).ljust(5)
    datetxt = str(cal.parseDT(datetimeString=json_sensors[key]['LastModified'][5:-4])[0].strftime(format))
  return datetxt + " PS" + DeviceId + " " + output + "\n"

# GetRelayState function
def getRelayinfo(Url):
  # Retreive Sensor Data
  r1 = False
  r2 = False
  r8 = False
  VENT = "NA"
  output = ""
  web_data_l = urllib.request.urlopen(Url)
  json_data_l = web_data_l.read()
  json_enc_l = web_data_l.info().get_content_charset('utf-8')
  json_sensors = json.loads(json_data_l.decode(json_enc_l))['relays']
  for key in json_sensors:
    if key == '1':
      if json_sensors[key]['state'] == 1:
        r1=True
    elif key == '2':
      if json_sensors[key]['state'] == 1:
        r2=True
    elif key == '8':
      if json_sensors[key]['state'] == 1:
        r8=True
  if r1 == False and r2 == True:
    VENT = "HIGH"
  if r1 == False and r2 == False:
    VENT = "MEDIUM"
  if r1 == True:
    VENT = "LOW"
  output = " RE\tVENT".ljust(20) + "\t\t" + VENT
  return output


# Construct switch command...
with open("/var/log/ax.log", "a") as logfile:
  logfile.write(getAirinfo(air_url_1, "01"))
  logfile.write(getAirinfo(air_url_2, "02"))