Test out admin mode commands; it looks like the basic idea works for now at least

dev/dev.htm

@@ -14,9 +14,9 @@ function initCharts() {
   chart = c3.generate({
     bindto: '#humidity-temp',
     data: {
-      x: 'x',
+      x: 'time',
       columns: [
-        ['x', 1, 2, 3, 4],
+        ['time', 1, 2, 3, 4],
         ['temp', 24, 24.5, 24.3, 24.6],
         ['humidity', 83.1, 99.5, 74.3, 84.6],
       ],
@@ -26,6 +26,11 @@ function initCharts() {
       }
     },
     axis: {
+      x: {
+        tick: {
+          format: (x) => new Date(x).toISOString()
+        }
+      },
       y2: {
         show: true
       }
@@ -37,29 +42,208 @@ function initCharts() {
   chart2 = c3.generate({
     bindto: '#volts',
     data: {
-      x: 'x',
+      x: 'time',
       columns: [
-        ['x', 1, 2, 3, 4],
+        ['time', 1, 2, 3, 4],
         ['voltage', 24, 24.5, 24.3, 24.6],
       ],
       axes: {
         voltage: 'y',
       }
     },
+    axis: {
+      x: {
+        tick: {
+          format: (x) => new Date(x).toISOString()
+        }
+      },
+    },
     zoom: {
       enabled: true
     }
   })
 }
+
+var cur_time_millis = 0
+var max_interval_millis = 5*60*1000
+
+var time = []
+var temp = []
+var humd = []
+var volts = []
+
+function status(msg) {
+  document.getElementById('status').textContent = msg
+}
+
+var autoupdate = true
+var chart_updater = null
+
+async function updateCharts() {
+  status("loading data...")
+  const latest = await fetch("/api/latest")
+  const last_time_millis = await latest.json()
+  if (last_time_millis > cur_time_millis) {
+    // we need to collect more data to catch up
+    // time is in milliseconds so we divide by 1000
+    var diff = (last_time_millis - cur_time_millis) / 1000
+    // worst case load only the last two weeks of data
+    diff = Math.min(diff, max_interval_millis/1000)
+    var path = "/api/last/weeks/"
+    var offset = diff
+    if (diff < 60) {
+      path = "/api/last/seconds/"
+      offset = Math.ceil(diff)
+    } else if (diff < 60*60) {
+      path = "/api/last/minutes/"
+      offset = Math.ceil(diff/60)
+    } else if (diff < 24*60*60) {
+      path = "/api/last/hours/"
+      offset = Math.ceil(diff/(60*60))
+    } else if (diff < 7*24*60*60) {
+      path = "/api/last/days/"
+      offset = Math.ceil(diff/(24*60*60))
+    } else {
+      path = "/api/last/weeks/"
+      offset = Math.ceil(diff/(7*24*60*60))
+    }
+    const new_data = await fetch(path + offset)
+    const new_data_json = await new_data.json()
+    // copy the data into the arrays
+    new_data_json.sort((a, b) => a.t - b.t)
+    new_data_json.forEach((v) => {
+      var last_time = time[time.length-1]
+      if (time.length == 0) {
+        last_time = 0
+      }
+      if (v.t > last_time) {
+        time.push(v.t)
+        temp.push(v.temp)
+        humd.push(v.hum)
+        volts.push(v.hv)
+      }
+    })
+    // truncate all the lists as necessary
+    // using the time array to determine the splicing points
+    cur_time_millis = time[time.length - 1]
+    const min_time = cur_time_millis - max_interval_millis
+
+    var slice_idx = 0
+    for (var i = 0; i < time.length; i++) {
+      if (time[i] >= min_time) {
+        slice_idx = i
+        break
+      }
+    }
+    //console.log("slice idx ", slice_idx)
+
+    temp = temp.slice(slice_idx)
+    humd = humd.slice(slice_idx)
+    volts = volts.slice(slice_idx)
+    time = time.slice(slice_idx)
+
+    chart.load({
+      columns: [
+        ["time"].concat(time),
+        ["temp"].concat(temp),
+        ["humidity"].concat(humd),
+      ]
+    })
+    chart2.load({
+      columns: [
+        ["time"].concat(time),
+        ["voltage"].concat(volts),
+      ]
+    })
+    status("charts updated")
+  }
+
+  if (autoupdate) {
+    chart_updater = chartupdater()
+  } else {
+    chart_updater = null
+  }
+}
+
+const sleep = (ms) => new Promise(r => setTimeout(r, ms))
+var chart_update_millis = 2000
+
+async function chartupdater() {
+  if (chart_updater != null) return
+
+  await sleep(chart_update_millis)
+  // wait at least two seconds to avoid wasting a lot of bandwidth
+  const resp = await fetch("/api/update")
+  chart_updater = null
+  updateCharts()
+}
+
+var status_updater = null 
+async function updateStatus() {
+  const status_resp = await fetch("/api/status")
+  const status = await status_resp.json()
+  //console.log(status)
+  const humidifier_state = (status.humidifier ? "on" : "off")
+  //console.log(humidifier_state)
+  document.getElementById("device-status").textContent = "connected: " + status.connected + ", humidifier: " + humidifier_state
+
+  if (autoupdate) {
+    status_updater = waitThenUpdateStatus()
+  } else {
+    status_updater = null
+  }
+}
+
+async function waitThenUpdateStatus() {
+  if (status_updater != null) return
+  await fetch("/api/status_update")
+  status_updater = null
+  updateStatus()
+}
+
+async function testAdminMode() {
+  const msg = JSON.stringify({
+    auth: "password",
+    data: {
+      manual_mode: true
+    }
+  })
+  await fetch("/api/admin", {
+    method: "POST",
+    body: msg
+  })
+}
+
 window.onload = () => {
   initCharts()
-  const xhr = new XmlHttpRequest()
+  updateCharts()
+  updateStatus()
+
+  document.getElementById('update').addEventListener('click', (e) => {
+    updateCharts()
+    updateStatus()
+  })
+  document.getElementById('autoupdate').addEventListener('click', (e) => {
+    autoupdate = document.getElementById('autoupdate').checked
+  })
+  document.getElementById('test-admin').addEventListener('click', (e) => {
+    testAdminMode()
+  })
 }
     </script>
   </head>
   <body>
+    <div id="device-status"></div>
     <div id="humidity-temp"></div>
     <div id="volts"></div>
-    <p>Test!</p>
+    <div>
+      <form>
+        <input type=checkbox id=autoupdate checked>Autoupdate</input>
+        <input type=button id=update value="Update"></input>
+        <input type=button id=test-admin value="Test admin mode"></input>
+        <!-- TODO add decimation and window size options -->
+        <span id=status></span>
+      </form>
+    </div>
   </body>
 </html>

shroom_controller.py

@@ -0,0 +1,73 @@
+import numpy as np
+
+import serial
+import subprocess
+import time
+
+s = serial.Serial("/dev/ttyUSB0", 115200, timeout=10)
+q = queue.Queue()
+
+process = subprocess.Popen(["ssh", "shrooms@threefortiethofonehamster.com", "python", "/home/shrooms/go/src/shroom-server/shroom-pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+# TODO run thread to process data from process's stdout
+
+def reset_serial():
+  s.close()
+  s = serial.Serial("/dev/ttyUSB0", 115200, timeout = 10)
+  time.sleep(10)
+
+class Humidifier:
+  def __init__(self):
+    self.on = False
+    self.history = np.array(30)
+    self.switch_timeout = 0
+
+  def update(self, volts):
+    self.history[1:] = self.history[:-1]
+    self.history[0] = volts
+    avg = np.sum(self.history)/self.history.shape[0]
+    if self.state:
+      if avg < 0.2:
+        self.on = False
+    else:
+      if avg > 2.6:
+        self.on = True
+
+  def toggle(self, s):
+    if time.time() > self.switch_timeout:
+      s.write(b"h")
+      self.switch_timeout = time.time() + 7
+
+humidifier = Humidifier()
+target_lower = 0.85
+target_higher = 0.90
+try:
+  last_sample = 0
+  while True:
+    now = time.time()
+    if now - last_sample < 0.5:
+      s.write(b"s")
+      resp = s.read(120)
+      if len(resp) == 0:
+        reset_serial()
+        time.sleep(5)
+        continue
+      parts = resp.split(b",")
+      humidity = float(parts[0])
+      temp = float(parts[1])
+      volts = float(parts[2])
+      print(humidity, temp, volts)
+
+      humidifier.update(volts)
+      if humidity < target_lower and humidifier.off:
+        humidifier.toggle(s)
+      elif humidity > target_upper and humidifier.on:
+        humidifier.toggle(s)
+      # TODO check on the process
+    else:
+      time.sleep(0.5-(now - last_sample))
+
+finally:
+  # TODO kill ssh connection
+  process.
+  pass

shroom_controller_mock.py

@@ -0,0 +1,170 @@
+import numpy as np
+
+import json
+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)
+process = subprocess.Popen(["/usr/bin/env", "python", "/home/kelvin/src/shroom-server/shroom_pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+def send_update(msg):
+  global process
+  process.stdin.write(bytes(json.dumps(msg) + "\n", "utf8"))
+  process.stdin.flush()
+
+exiting = False
+# run thread to process data from process's stdout
+def stdout_loop():
+  while not exiting:
+    msg = process.stdout.readline()
+    print("got message ", msg)
+stdout_thread = threading.Thread(target=stdout_loop)
+stdout_thread.start()
+
+class MockSerial:
+  def __init__(self):
+    self.humidity = np.zeros(100)
+    self.humidifier_on = False
+    self.humidity[:] = 0.80
+    self.humidity[-1] = 0.20
+    self.humidity[0] = 0.20
+
+  def write(self, msg):
+    if msg == b'h':
+      print("mock hum toggle")
+      self.humidifier_on = not self.humidifier_on
+
+  def read(self, _):
+    t = time.time()
+    temp = 25 + np.sin(0.01*2*np.pi*t) + 0.5*np.sin(0.0001*2*np.pi*t + 7)
+
+    # very janky model of humidity diffusion
+    # fix end conditions
+    for _ in range(20):
+      self.humidity[-1] = 0.2*0.20 + 0.8*self.humidity[-2]
+      self.humidity[0] = 0.20
+      if self.humidifier_on:
+        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])
+    #print(self.humidity)
+
+    humidity = self.humidity[60] + np.random.random()*0.003
+    if self.humidifier_on:
+      hv = 3.3
+    else:
+      hv = 0.0
+    return bytes("{},{},{}\n".format(humidity, temp, hv), "utf8")
+
+s = MockSerial()
+
+def reset_serial():
+  pass
+  time.sleep(10)
+
+class Humidifier:
+  def __init__(self):
+    self.on = False
+    self.history = np.zeros(10)
+    self.switch_timeout = 0
+
+  @property
+  def off(self):
+    return not self.on
+
+  def update(self, volts):
+    self.history[1:] = self.history[:-1]
+    self.history[0] = volts
+    #print(self.history)
+    avg = np.sum(self.history)/self.history.shape[0]
+    if self.on:
+      if avg < 0.2:
+        self.on = False
+        print("send status off")
+        send_update({"status": 0})
+        self.switch_timeout = time.time() + 1
+    else:
+      if avg > 2.6:
+        self.on = True
+        print("send status on")
+        send_update({"status": 1})
+        self.switch_timeout = time.time() + 1
+
+  def toggle(self, s):
+    if time.time() > self.switch_timeout:
+      s.write(b"h")
+      self.switch_timeout = time.time() + 7
+
+humidifier = Humidifier()
+target_lower = 0.85
+target_upper = 0.90
+
+humidifier_history = np.zeros(30)
+first_sample = False
+try:
+  last_sample = 0
+  while True:
+    now = time.time()
+    if now - last_sample < 0.5:
+      time.sleep(0.5 - (now - last_sample))
+      continue
+    last_sample = now
+
+    s.write(b"s")
+    resp = s.read(120)
+    if len(resp) == 0:
+      reset_serial()
+      time.sleep(5)
+      continue
+    parts = resp.split(b",")
+    humidity = float(parts[0])
+    temp = float(parts[1])
+    volts = float(parts[2])
+    print(humidity, temp, volts)
+    if first_sample:
+      humidifier_history[:] = humidity
+      first_sample = False
+    else:
+      humidifier_history[:-1] = humidifier_history[1:]
+      humidifier_history[-1] = humidity
+
+    # compensate for the slow response time
+    slope = (humidifier_history[-1] - humidifier_history[0])/humidifier_history.shape[0]
+    comp_humidity = humidity + 50*slope
+
+    try:
+      humidifier.update(volts)
+      if comp_humidity < target_lower and humidifier.off:
+        humidifier.toggle(s)
+      elif comp_humidity > target_upper and humidifier.on:
+        humidifier.toggle(s)
+
+      update = {
+          "time": int(now*1000),
+          "temp": temp,
+          "hum": humidity,
+          "hv": volts
+      }
+      send_update(update)
+    except Exception as e:
+      print("pipe errored out, restarting: ", e)
+      # restart the process I guess
+      exiting = True
+      process.kill()
+      time.sleep(0.1)
+      stdout_thread.join()
+      exiting = False
+
+      process = subprocess.Popen(["/usr/bin/env", "python", "/home/kelvin/src/shroom-server/shroom_pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+
+      stdout_thread = threading.Thread(target=stdout_loop)
+      stdout_thread.start()
+
+finally:
+  # kill ssh connection
+  exiting = True
+  process.kill()
+  stdout_thread.join()

shroom_internals/sql.go

@@ -22,10 +22,10 @@ func CreateTable(db *sql.DB) error {
 }
 
 type Datapoint struct {
-	Time            uint64
+	Time            uint64  `json:"t"`
-	Temperature     float32
+	Temperature     float32 `json:"temp"`
-	Humidity        float32
+	Humidity        float32 `json:"hum"`
-	HumidifierVolts float32
+	HumidifierVolts float32 `json:"hv"`
 }
 
 func QueryHistory(db *sql.DB, start int64) ([]Datapoint, error) {

shroom_internals/tcp_server.go

@@ -22,7 +22,7 @@ type ShroomData struct {
 	Temperature     float32 `json:"temp"`
 	Humidity        float32 `json:"hum"`
 	HumidifierVolts float32 `json:"hv"`
-	Status          uint32  `json:"status"`
+	Status          int32   `json:"status"`
 }
 
 type ShroomStatus struct {
@@ -30,6 +30,9 @@ type ShroomStatus struct {
 	HumidifierOn   bool
 	NumConnections int
 	Wait           chan struct{}
+	StatusWait     chan struct{}
+
+	Commands chan []byte
 }
 
 func (s *ShroomStatus) Update() {
@@ -39,13 +42,20 @@ func (s *ShroomStatus) Update() {
 	s.Wait = make(chan struct{})
 }
 
+func (s *ShroomStatus) StatusUpdate() {
+	s.Lock()
+	defer s.Unlock()
+	close(s.StatusWait)
+	s.StatusWait = make(chan struct{})
+}
+
 func parseMsg(line []byte, db *sql.DB, status *ShroomStatus) {
 	data := ShroomData{
 		Time:            0,
-		Temperature:     -1,
+		Temperature:     -274,
 		Humidity:        -1,
 		HumidifierVolts: -1,
-		Status:          0,
+		Status:          -1,
 	}
 	err := json.Unmarshal(line, &data)
 	if err != nil {
@@ -54,21 +64,23 @@ func parseMsg(line []byte, db *sql.DB, status *ShroomStatus) {
 		log.Println(line)
 		return
 	}
-	if data.Time > 0 && data.Temperature > 0 && data.Humidity > 0 && data.HumidifierVolts > 0 {
+	//log.Println("received data ", data)
+	if data.Time > 0 && data.Temperature > -275 && data.Humidity > -1 && data.HumidifierVolts > -1 {
 		err = InsertRow(db, &data)
 		if err != nil {
 			log.Println("unable to write to database: ", err)
 		}
 		// we got a data packet
 		status.Update()
-	} else if data.Status > 0 {
+	} 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
 		status.Unlock()
-		status.Update()
+		status.StatusUpdate()
 	} else {
-		log.Println("unknown packet: ", line)
+		log.Println("unknown packet: ", line, string(line))
 	}
 
 }
@@ -84,23 +96,51 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 			conn, err := ln.Accept()
 			if err != nil {
 				log.Println("tcp accept error: ", err)
-				continue
+				return
 			}
+			// not spawning a goroutine here
+			// should limit the number of connections to
+			// one hopefully
+
+			// wrapping in a func() so that I can use defer
+			// to automatically decrement the number of connections
+			func() {
 				status.Lock()
 				status.NumConnections += 1
 				status.Unlock()
+				status.StatusUpdate()
 
 				defer func() {
 					status.Lock()
 					status.NumConnections -= 1
 					status.Unlock()
+					status.StatusUpdate()
 					log.Println("connection disconnected")
 				}()
 
 				log.Println("connection started")
 
+				// write loop; waits for commands and forwards them
+				exiting := make(chan struct{})
 				go func() {
-				// TODO deal with the write side of the connection
+					for {
+						select {
+						case v, ok := <-status.Commands:
+							if !ok {
+								return
+							}
+							v = append(v, []byte("\n")...)
+							_, err := conn.Write(v)
+							if err != nil {
+								log.Println("tcp write err: ", err)
+							}
+						case <-exiting:
+							return
+						}
+					}
+				}()
+				defer func() {
+					close(exiting)
 				}()
 
 				// deal with the read side of the connection
@@ -109,6 +149,7 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 
 				for {
 					num_read, err := conn.Read(left)
+					//log.Println("received: ", string(left[:num_read]))
 					left = left[num_read:]
 					//log.Println("buf ", buf)
 					//log.Println("left ", left)
@@ -140,6 +181,7 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 					copy(buf[:len(unread)], unread)
 					left = buf[len(unread):]
 				}
+			}()
 		}
 	}()
 }

shroom_server.go

@@ -16,6 +16,8 @@ import (
 	"io/fs"
 	"log"
 	"net/http"
+	"strconv"
+	"strings"
 	"time"
 )
 
@@ -27,10 +29,25 @@ type statusJson struct {
 	Humidifier bool `json:"humidifier"`
 }
 
-func dumpData(db *sql.DB, offset int64) func(http.ResponseWriter, *http.Request) {
+type adminMsg struct {
-	return func(w http.ResponseWriter, _req *http.Request) {
+	Auth string                 `json:"auth"`
+	Msg  map[string]interface{} `json:"data"`
+}
+
+func dumpData(db *sql.DB, multiplier int64) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, req *http.Request) {
 		now := time.Now().Unix()
-		t := now + offset
+		path := strings.Split(req.URL.Path, "/")
+		last := path[len(path)-1]
+		count, err := strconv.Atoi(last)
+		if err != nil {
+			w.WriteHeader(400)
+			w.Write([]byte("could not read integer in path: " + err.Error()))
+			return
+		}
+		offset := int64(count) * multiplier
+		t := now*1000 - offset
+		//log.Println("req start ", t)
 		msg, err := s.GetRows(db, t)
 		if err != nil {
 			w.WriteHeader(500)
@@ -57,6 +74,8 @@ func main() {
 
 	status := s.ShroomStatus{
 		Wait:       make(chan struct{}),
+		StatusWait: make(chan struct{}),
+		Commands:   make(chan []byte),
 	}
 	s.InitTcpServer(db, &status)
 
@@ -65,10 +84,11 @@ func main() {
 		log.Fatal("unable to use subdirectory of embedded fs: ", err)
 	}
 
-	dumpWeek := dumpData(db, -7*24*60*60)
+	dumpWeek := dumpData(db, 7*24*60*60*1000)
-	dumpDay := dumpData(db, -24*60*60)
+	dumpDay := dumpData(db, 24*60*60*1000)
-	dumpHour := dumpData(db, -60*60)
+	dumpHour := dumpData(db, 60*60*1000)
-	dumpMinute := dumpData(db, -60)
+	dumpMinute := dumpData(db, 60*1000)
+	dumpSecond := dumpData(db, 1000)
 
 	lastPoint := func(w http.ResponseWriter, _req *http.Request) {
 		msg, err := s.LastTime(db)
@@ -92,7 +112,7 @@ func main() {
 		msg, err := json.Marshal(s)
 		if err != nil {
 			err = fmt.Errorf("unable to marshal json: %w", err)
-			w.WriteHeader(500)
+			w.WriteHeader(400)
 			w.Write([]byte(err.Error()))
 		} else {
 			w.Write(msg)
@@ -100,9 +120,49 @@ func main() {
 	}
 
 	adminHandler := func(w http.ResponseWriter, req *http.Request) {
-		w.WriteHeader(500)
+		if req.Method != "POST" {
-		w.Write([]byte("unimplemented"))
+			w.WriteHeader(405)
-		// TODO
+			w.Write([]byte("method must be POST"))
+			return
+		}
+		msg := make([]byte, 256)
+		l, err := req.Body.Read(msg)
+		if err != nil && l == 0 {
+			err = fmt.Errorf("unable to read body: %w", err)
+			w.WriteHeader(400)
+			w.Write([]byte(err.Error()))
+			return
+		}
+		adminReq := adminMsg{}
+		err = json.Unmarshal(msg[:l], &adminReq)
+		if err != nil {
+			err = fmt.Errorf("unable to unmarshal body as json: %w", err)
+			w.WriteHeader(400)
+			w.Write([]byte(err.Error()))
+			return
+		}
+
+		// TODO switch to embedded secret
+		if adminReq.Auth != "password" {
+			w.WriteHeader(401)
+			w.Write([]byte(err.Error()))
+			return
+		}
+
+		inner_msg, err := json.Marshal(adminReq.Msg)
+		if err != nil {
+			err = fmt.Errorf("unable to marshal inner message: %w", err)
+			w.WriteHeader(400)
+			w.Write([]byte(err.Error()))
+			return
+		}
+		select {
+		case status.Commands <- inner_msg:
+			w.Write([]byte("ok"))
+		default:
+			w.WriteHeader(503)
+			w.Write([]byte("unable to forward request; controller may not be connected"))
+		}
 	}
 
 	updateHandler := func(w http.ResponseWriter, req *http.Request) {
@@ -112,17 +172,26 @@ func main() {
 		}
 		w.Write([]byte("ok"))
 	}
+	statusUpdateHandler := func(w http.ResponseWriter, req *http.Request) {
+		stillopen := true
+		for stillopen {
+			_, stillopen = <-status.StatusWait
+		}
+		w.Write([]byte("ok"))
+	}
 
 	http.Handle("/d/", http.StripPrefix("/d/", http.FileServer(http.Dir("./dev"))))
 	http.Handle("/", http.FileServer(http.FS(contentSub)))
-	http.HandleFunc("/api/last/week", dumpWeek)
+	http.HandleFunc("/api/last/weeks/", dumpWeek)
-	http.HandleFunc("/api/last/day", dumpDay)
+	http.HandleFunc("/api/last/days/", dumpDay)
-	http.HandleFunc("/api/last/hour", dumpHour)
+	http.HandleFunc("/api/last/hours/", dumpHour)
-	http.HandleFunc("/api/last/minute", dumpMinute)
+	http.HandleFunc("/api/last/minutes/", dumpMinute)
+	http.HandleFunc("/api/last/seconds/", dumpSecond)
 	http.HandleFunc("/api/latest", lastPoint)
 	http.HandleFunc("/api/status", getStatus)
 	http.HandleFunc("/api/admin", adminHandler)
 	http.HandleFunc("/api/update", updateHandler)
+	http.HandleFunc("/api/status_update", statusUpdateHandler)
 
 	// TODO periodically clear old entries from the database