-
Notifications
You must be signed in to change notification settings - Fork 0
/
demand2thingspeak.py
58 lines (45 loc) · 1.94 KB
/
demand2thingspeak.py
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
# Receive demand values from the RabbitMQ queue and send to ThingSpeak.
import pika
from lxml import etree
import requests
from ConfigParser import SafeConfigParser
# Read in app config values.
config = SafeConfigParser()
config.read('monitor-apps-config.ini')
upload_url = config.get('ThingSpeak', 'upload_url')
API_key = config.get('ThingSpeak', 'API_key')
# Set up the connection to the RabbitMQ server.
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))
channel = connection.channel()
# Set up the link to the selected RabbitMQ exchange.
channel.exchange_declare(exchange='vangorp.home.eagle',
type='fanout')
# Grab a queue name and bind it to the exchange.
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='vangorp.home.eagle',
queue=queue_name)
# Set up callback function to process queue messages.
def callback(ch, method, properties, body):
# Parse the message into an XML document.
message = etree.fromstring(body)
# Extract timestamp and remove the 's' at the end.
timestamp = message.attrib['timestamp']
timestamp = timestamp[:-1]
# Extract set of elements from Eagle gateway post.
fragment = message[0]
# Process the InstantaneousDemand fragment.
if fragment.tag == 'InstantaneousDemand':
# Extract demand measurement value.
demand = int(fragment.findtext('Demand'), base=0)
multiplier = int(fragment.findtext('Multiplier'), base=0)
divisor = int(fragment.findtext('Divisor'), base=0)
InstantaneousDemand = (demand * multiplier)/float(divisor)
# Push data to ThingSpeak.
payload = {'key': API_key, 'field1': InstantaneousDemand}
response = requests.post(upload_url, params=payload)
channel.basic_consume(callback,
queue=queue_name,
no_ack=True)
channel.start_consuming()