Uploading to ThingSpeak

.gitignore

@@ -1,2 +1,5 @@
 .DS_Store
 ._*
+
+# python
+*.pyc

Aggregation/main.py

@@ -1,66 +1,120 @@
 #! /usr/local/bin/python3
-import glob, os
+import os
+import glob
 import serial
 import time
 from collections import defaultdict
 from queue import Queue
+import re
+import json
+from pprint import pprint
+import numpy as np
 from xbee import XBee, ZigBee
+import thingspeak as ts
+
+# Parameters
+api_keys = {37372: "BKSHA6VI5K13A6HL"}
 
 # Constants
-START_CH = '!'
+START_CH = '^'
+END_CH = '\r'
 
 # Globals
 ser = None
 xbee = None
 rxQueue = defaultdict(Queue)
+channels = defaultdict(ts.channel)
 
 # run function, fn, on all items in iterable, it.
-def each(fn, it): [fn(i) for i in it]
+def each(fn, it):
+    for i in it: fn(i)
 
 # print and pass on, use for inline debugging
 def inprint(x): print(x, end=''); return(x)
 
+# rx data and queue into a chr buffer
 def handle_data_rx(data):
     each(rxQueue[data['source_addr_long']].put, map(chr, data['rf_data']))
-    print(data['source_addr_long'], rxQueue[data['source_addr_long']].qsize())
+    print("    from: ", data['source_addr_long'], "raw", data['rf_data'])
 
 def process_data():
     for addr in rxQueue:
         q = rxQueue[addr]
 
         # search for start char
-        while (q.get() is not START_CH) and not q.empty(): pass
+        while not q.empty() and (q.queue[0] is not START_CH): q.get()
+
+        # wait for full message (comes in blocks)
+        if (not q.empty()) and (END_CH in q.queue):
+            q.get() # pop start character
+
+            # compile message
+            ch = ""
+            payload = ""
+            while (ch is not END_CH) and not q.empty():
+                ch = q.get()
+                print(ch, end='')
+                payload += ch # simple and not so inefficient after all
+                # http://stackoverflow.com/questions/19926089/python-equivalent-of-java-stringbuffer
 
-        # compile message
-        ch = ""
-        while (ch is not '\r') and not q.empty():
-            ch = q.get()
-            print(ch, end='')
-        if not q.empty(): print()
+            # add quotes to keys (proper JSON formatting)
+            payload = re.sub('([{,])([^{:\s"]*):',
+                             lambda m: '%s"%s":'%(m.group(1),m.group(2)),
+                             payload)
+            print(payload)
+
+            def group_in_channels(d):
+                print(d[0]["channel"])
+                channels = np.unique(list(map(lambda x: x["channel"], d)))
+                print(channels)
+                grouped = {}
+                for c in channels:
+                    fields = map(lambda x: x["field"],
+                                 filter(lambda x: x["channel"] == c,
+                                        d))
+                    grouped[c] = [map(lambda x: x["data"],
+                                     filter(lambda x: x["field"] == f,
+                                            filter(lambda x: x["channel"] == c,
+                                                   d))).__next__()
+                                  if f in fields else None
+                                  for f in range(1,9)]
+                return grouped
+
+            payload = json.loads(payload)
+            grouped = group_in_channels(payload)
+            print(grouped)
+            for ch in grouped:
+                channels[ch].update(grouped[ch])
 
 def find_port():
     return "/dev/tty.usbserial-DA01I3FX"
 
 def init():
+    global xbee, ser, channels
     port = find_port()
     baud = 9600
     print("Listening on:", port, "at", baud, "baud")
-    os.system("cat " + port) # This cleared the port for me, good luck charm
+    #os.system("cat " + port) # This cleared the port for me, good luck charm
 
     ser = serial.Serial(port, 9600)
     xbee = ZigBee(ser, escaped=True, callback=handle_data_rx)
 
+    for ch in api_keys:
+        channels[ch] = ts.channel(api_keys[ch], None)
+    pprint(channels)
+
 def main():
     while True:
         try:
             process_data()
-            time.sleep(0.5)
+            time.sleep(0.1)
         except KeyboardInterrupt:
-            break
+            print("quit")
+            raise
 
 def end():
     xbee.halt()
-    serial_port.close()
+    ser.close()
 
 if __name__ == "__main__":
     init()

Aggregation/thingspeak.py

@@ -0,0 +1,45 @@
+# modified from https://github.com/bergey/thingspeak/blob/master/thingspeak.py
+import http.client as httplib
+import urllib.parse as urllib
+
+__version__ = '0.1.1'
+
+field_keys = ['field' + str(n) for n in range(1,9)]
+headers = {"Content-type": "application/x-www-form-urlencoded","Accept": "text/plain"}
+
+def non_null_values(**kwargs):
+    return [(k,v) for (k,v) in kwargs.items() if v != None]
+
+def non_null_in_dict(d):
+    return [(k,v) for (k,v) in d if v != None]
+
+class TooManyFields(ValueError):
+    pass
+
+class channel(object):
+    def __init__(self, write_key, cid):
+        """write_key is the Write API Key.
+        cid is the read_key"""
+        self.write_key = write_key
+
+    def update(self, field_vals, lat=None, long=None, elevation=None, status=None):
+        if len(field_vals) > 8:
+            raise TooManyFields('update can only handle 8 field values per channel')
+        # this verbosity, rather than just using kwargs,
+        # so that callers get a prompt error if they supply an arg `update` cannot handle
+        named_args = non_null_values(lat=lat, long=long, elevation=elevation, status=status)
+        params = urllib.urlencode(non_null_in_dict(zip(field_keys, field_vals)) + [('key', self.write_key)] + named_args)
+        print(params)
+        conn = httplib.HTTPConnection("api.thingspeak.com:80")
+        conn.request("POST", "/update", params, headers)
+        response = conn.getresponse()
+        conn.close()
+        return response
+
+    def fetch(self, format):
+        conn = httplib.HTTPConnection("anpi.thingspeak.com:80")
+        path = "/channels/{0}/feed.{1}".format(self.cid, format)
+        params = urllib.urlencode([('key',self.key)])
+        conn.request("GET", path, params, headers)
+        response = conn.getresponse()
+        return response

GenericDataLogger/GenericDataLogger.ino

@@ -2,7 +2,12 @@
  * Generic Data Logger example
  * A starting framework to make easy-to-manage dataloggers.
  *
- * Uses sparkfun weather shield.
+ * This exmaple uses sparkfun weather shield, but is simple to extend to other sensors.
+ * It will send data over Serial, using various formats (see the dump functions).
+ *
+ * Troubleshooting:
+ * - Check your paramters and constants. e.g., did you add or remove sensors or dumps?
+ *   Is N_SENSORS and N_DUMPS still the correct size?
  *
  * Created by Ashley Gillman, 09/05/15
 */
@@ -16,18 +21,29 @@
 #include "MPL3115A2.h" // Pressure sensor
 #include "HTU21D.h" // Humidity sensor
 
+#define DEBUG 1
+
 // pins
 const byte STAT1 = 7;
 const byte STAT2 = 8;
 const byte LIGHT = A1;
 const byte REFERENCE_3V3 = A3;
 
 // parameters
-const unsigned long UPDATE = 15500; // update interval, 15.5s (ThingSpeak preferred)
+const unsigned long UPDATE = 5000;//15500; // update interval, 15.5s (ThingSpeak preferred)
 const short N_SENSORS = 4;
+const short N_DUMPS = 2;
+
+// Sensors
+const ReadingFnPointer sensors[] =
+  {*getHumidityStr, *getTemperatureStr, *getPressureStr, *getLightStr};
+const unsigned int channels[] =
+  {37372,           37372,              37372,           37372       };
+const int fields[] =
+  {1,               2,                  3,               4           };
 
 // other Constants
-const char READING_WIDTH = 10; // xxxxxx.xxx
+const char READING_WIDTH = 12; // xxxxxx.xxx
 const char READING_PRECISION = 3;
 
 // global vars
@@ -48,14 +64,14 @@ void setup() {
   setupSensors();
 
   // Configure DataLogger
-  dataLogger = DataLogger(N_SENSORS, READING_WIDTH, 1);
+  dataLogger = DataLogger(N_SENSORS, READING_WIDTH, N_DUMPS);
   // Configure data logger inputs
-  dataLogger.addReading(*getHumidityStr);
-  dataLogger.addReading(*getTemperatureStr);
-  dataLogger.addReading(*getPressureStr);
-  dataLogger.addReading(*getLightStr);
+  for (int i=0; i<N_SENSORS; ++i) {
+    dataLogger.addReading(sensors[i]);
+  }
   // configure data logger outputs
   dataLogger.addDump(*dumpToSerial);
+  dataLogger.addDump(*dumpToSerialJSON);
 
   Serial.println(" done.");
 }
@@ -138,15 +154,42 @@ void dumpToSerial(char* data, short height, short width) {
   Serial.println();
 }
 
+// Writes to Serial for ThingSpeak uploading
+// Start char: '^', end char: '\n', body is in JSON format.
+void dumpToSerialJSON(char* data, short height, short width) {
+  Serial.print("^[");
+  for (int i=0; i<height; ++i) {
+    //debugStats();
+    delay(5);
+    Serial.print(F("{channel: "));
+    Serial.print(channels[i]);
+    Serial.print(F(",field: "));
+    Serial.print(fields[i]);
+    Serial.print(F(",data: "));
+    for (int j=0; j<width; ++j) {
+      if (data[i*width+j] != ' ') // optional: omit whitespace
+        Serial.print(data[i*width+j]);
+    }
+    if (i<height-1)
+      Serial.print("},");
+    else
+      Serial.print('}');
+  }
+  Serial.println(']');
+}
+
+
 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
 // Debugging Code
 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
 
 void debugStats() {
-  Serial.print("FM: "); Serial.println(freeMemory());
-  check_mem();
-  Serial.print("HP: "); Serial.println((unsigned int) heapptr);
-  Serial.print("SP: "); Serial.println((unsigned int) stackptr);
+  if (DEBUG) {
+    Serial.print("FM: "); Serial.println(freeMemory());
+    check_mem();
+    Serial.print("HP: "); Serial.println((unsigned int) heapptr);
+    Serial.print("SP: "); Serial.println((unsigned int) stackptr);
+  }
 }
 
 

README.md

@@ -56,10 +56,12 @@ There are no direct dependencies for the library, however the example file uses
 - [MPL3115A2](https://github.com/sparkfun/MPL3115A2_Breakout): Sparufun weather shield pressure sensor.
 
 ### Python
-> pip install xbee # may require sudo, might be pip3 not pip
->
-> pip install pyserial # may require sudo, might be pip3 not pip
+If you don't already have Python 3:
+> sudo apt-get install python3 pip3
+
+This may require you to use sudo, the command is likely pip3 but may be pip.
+> pip3 install numpy, pyserial, xbee
 
 # Useful References
-http://blog.hcilab.org/gadgeteer/2012/12/xbee-zigbee-howto/
+http://blog.hcilab.org/gadgeteer/2012/12/xbee-zigbee-howto/  
 http://serdmanczyk.github.io/XBeeAPI-PythonArduino-Tutorial/