Refactor the controller-server format a bit; also add in some protection to prevent overrun issues

dev/dev.htm

@@ -64,9 +64,10 @@ function initCharts() {
   })
 }
 
-var cur_time_millis = 0
 var max_interval_millis = 5*60*1000
+var decimation_rate = 1
 
+var cur_time_millis = 0
 var time = []
 var temp = []
 var humd = []

shroom_controller.py

@@ -1,13 +1,23 @@
 import numpy as np
 
 import json
+import os
 import serial
 import subprocess
 import threading
 import time
 
-#process = subprocess.Popen(["ssh", "shrooms@localhost", "/usr/bin/env", "python", "/home/shrooms/go/src/shroom-server/shroom-pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+SERIAL_PATH = "/dev/ttyUSB0"
-process = subprocess.Popen(["/usr/bin/env", "python", "/home/kelvin/src/shroom-server/shroom_pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+SERIAL_BAUD = 115200
+
+SAMPLE_PERIOD = 0.2
+DECIMATION_RATE = 10
+
+is_mock = os.environ['MOCK']
+if is_mock:
+  process = subprocess.Popen(["/usr/bin/env", "python", "/home/kelvin/src/shroom-server/shroom_pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+else:
+  process = subprocess.Popen(["ssh", "shrooms@threefortiethofonehamster.com", "/usr/bin/env", "python", "/home/shrooms/go/src/shroom-server/shroom-pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
 
 def send_update(msg):
   global process
@@ -46,10 +56,10 @@ class MockSerial:
       self.humidity[-1] = 0.2*0.20 + 0.8*self.humidity[-2]
       self.humidity[0] = 0.20
       if self.humidifier_on:
-        self.humidity[20] = 1.5
+        self.humidity[20] = 2
       # use the gradient to determine the change in humidity
       avg = 0.5*(self.humidity[:-2] + self.humidity[2:])
-      self.humidity[1:-1] += 0.2*(avg - self.humidity[1:-1])
+      self.humidity[1:-1] += 0.10*(avg - self.humidity[1:-1])
     #print(self.humidity)
 
     humidity = self.humidity[60] + np.random.random()*0.003
@@ -59,10 +69,15 @@ class MockSerial:
       hv = 0.0
     return bytes("{},{},{}\n".format(humidity, temp, hv), "utf8")
 
-s = MockSerial()
+if is_mock:
+  s = MockSerial()
+else:
+  s = serial.Serial(SERIAL_PATH, SERIAL_BAUD, timeout = 10)
 
 def reset_serial():
-  pass
+  if not is_mock:
+    s.close()
+    s = serial.Serial(SERIAL_PATH, SERIAL_BAUD, timeout = 10)
   time.sleep(10)
 
 class Humidifier:
@@ -84,13 +99,13 @@ class Humidifier:
       if avg < 0.2:
         self.on = False
         print("send status off")
-        send_update({"status": 0})
+        send_update({"status": {"humidifier": False}})
         self.switch_timeout = time.time() + 1
     else:
       if avg > 2.6:
         self.on = True
         print("send status on")
-        send_update({"status": 1})
+        send_update({"status": {"humidifier": True}})
         self.switch_timeout = time.time() + 1
 
   def toggle(self, s):
@@ -105,12 +120,13 @@ feedforward_coeff = 50
 
 humidifier_history = np.zeros(30)
 first_sample = False
+frame_num = 0
 try:
   last_sample = 0
   while True:
     now = time.time()
-    if now - last_sample < 0.5:
+    if now - last_sample < SAMPLE_PERIOD:
-      time.sleep(0.5 - (now - last_sample))
+      time.sleep(SAMPLE_PERIOD - (now - last_sample) + 0.001)
       continue
     last_sample = now
 
@@ -144,13 +160,17 @@ try:
       elif comp_humidity > target_upper and humidifier.on:
         humidifier.toggle(s)
 
-      update = {
+      if frame_num == 0:
-          "time": int(now*1000),
+        update = {
-          "temp": temp,
+            "data": {
-          "hum": humidity,
+              "time": int(now*1000),
-          "hv": volts
+              "temp": temp,
-      }
+              "hum": humidity,
-      send_update(update)
+              "hv": volts
+            }
+        }
+        send_update(update)
+      frame_num = (frame_num + 1) % DECIMATION_RATE
     except Exception as e:
       print("pipe errored out, restarting: ", e)
       # restart the process I guess

shroom_internals/sql.go

@@ -66,8 +66,8 @@ func LatestTime(db *sql.DB) (int64, error) {
 	return t, nil
 }
 
-func InsertRow(db *sql.DB, s *ShroomData) error {
+func InsertRow(db *sql.DB, s *DataJson) error {
 	_, err := db.Exec("INSERT INTO shrooms (time, temperature, humidity, humidifier_volts) VALUES (?, ?, ?, ?)",
-		s.Time, s.Temperature, s.Humidity, s.HumidifierVolts)
+		*s.Time, *s.Temperature, *s.Humidity, *s.HumidifierVolts)
 	return err
 }

shroom_internals/tcp_server.go

@@ -17,12 +17,20 @@ func newlinePos(s []byte) int {
 	return -1
 }
 
-type ShroomData struct {
+type StatusJson struct {
-	Time            uint64  `json:"time"`
+	HumOn *bool `json:"humidifier"`
-	Temperature     float32 `json:"temp"`
+}
-	Humidity        float32 `json:"hum"`
+
-	HumidifierVolts float32 `json:"hv"`
+type DataJson struct {
-	Status          int32   `json:"status"`
+	Time            *uint64  `json:"time"`
+	Temperature     *float32 `json:"temp"`
+	Humidity        *float32 `json:"hum"`
+	HumidifierVolts *float32 `json:"hv"`
+}
+
+type ShroomPacket struct {
+	Data   *DataJson   `json:"data"`
+	Status *StatusJson `json:"status"`
 }
 
 type ShroomStatus struct {
@@ -50,14 +58,8 @@ func (s *ShroomStatus) StatusUpdate() {
 }
 
 func parseMsg(line []byte, db *sql.DB, status *ShroomStatus) {
-	data := ShroomData{
+	packet := ShroomPacket{}
-		Time:            0,
+	err := json.Unmarshal(line, &packet)
-		Temperature:     -274,
-		Humidity:        -1,
-		HumidifierVolts: -1,
-		Status:          -1,
-	}
-	err := json.Unmarshal(line, &data)
 	if err != nil {
 		log.Println("unable to parse tcp line: ", err)
 		log.Println(string(line))
@@ -65,18 +67,22 @@ func parseMsg(line []byte, db *sql.DB, status *ShroomStatus) {
 		return
 	}
 	//log.Println("received data ", data)
-	if data.Time > 0 && data.Temperature > -275 && data.Humidity > -1 && data.HumidifierVolts > -1 {
+	if packet.Data != nil {
-		err = InsertRow(db, &data)
+		if packet.Data.Time != nil && packet.Data.Temperature != nil && packet.Data.Humidity != nil && packet.Data.HumidifierVolts != nil {
-		if err != nil {
+			err = InsertRow(db, packet.Data)
-			log.Println("unable to write to database: ", err)
+			if err != nil {
+				log.Println("unable to write to database: ", err)
+			}
+			// we got a data packet
+			status.Update()
 		}
-		// we got a data packet
+	} else if packet.Status != nil {
-		status.Update()
-	} else if data.Status != -1 {
 		//log.Println("received status ", data.Status)
 		status.Lock()
 		// TODO change to have more detailed data
-		status.HumidifierOn = (data.Status & 1) == 1
+		if packet.Status.HumOn != nil {
+			status.HumidifierOn = *packet.Status.HumOn
+		}
 		status.Unlock()
 		status.StatusUpdate()
 	} else {
@@ -153,6 +159,10 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 					left = left[num_read:]
 					//log.Println("buf ", buf)
 					//log.Println("left ", left)
+					if len(left) == 0 {
+						log.Println("overflow detected, truncating data")
+						left = buf
+					}
 
 					if err != nil {
 						log.Println("tcp read error: ", err)