Add code to remove old data and switch the port to 8085

trying to read

.gitignore

@@ -1,2 +1,3 @@
 shroom_server
 shrooms.db
+auth_secret

shroom_controller.py

@@ -1,73 +1,290 @@
 import numpy as np
 
+import json
+import os
 import serial
 import subprocess
+import threading
 import time
 
-s = serial.Serial("/dev/ttyUSB0", 115200, timeout=10)
-q = queue.Queue()
+SERIAL_PATH = "/dev/ttyUSB0"
+SERIAL_BAUD = 115200
 
-process = subprocess.Popen(["ssh", "shrooms@threefortiethofonehamster.com", "python", "/home/shrooms/go/src/shroom-server/shroom-pipe.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+SAMPLE_PERIOD = 0.2
+DECIMATION_RATE = 5
 
-# TODO run thread to process data from process's stdout
+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
+  process.stdin.write(bytes(json.dumps(msg) + "\n", "utf8"))
+  process.stdin.flush()
+
+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.10*(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")
+
+if is_mock:
+  s = MockSerial()
+else:
+  s = serial.Serial(SERIAL_PATH, SERIAL_BAUD, timeout = 10)
 
 def reset_serial():
-  s.close()
-  s = serial.Serial("/dev/ttyUSB0", 115200, timeout = 10)
+  if not is_mock:
+    s.close()
+    s = serial.Serial(SERIAL_PATH, SERIAL_BAUD, timeout = 10)
   time.sleep(10)
 
 class Humidifier:
   def __init__(self):
-    self.on = False
-    self.history = np.array(30)
+    self.off_threshold = 0.2
+    self.on_threshold = 2.6
+    self.toggle_cooldown = 7
+
+    self._on = False
+    self.history = np.zeros(10)
     self.switch_timeout = 0
 
+  @property
+  def on(self):
+    return self._on
+
+  @on.setter
+  def on(self, nv):
+    old_on = self._on
+    self._on = nv
+    if nv:
+      print("send hum on")
+    else:
+      print("send hum off")
+    send_update({"status": {"humidifier": nv}})
+
+  @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.state:
-      if avg < 0.2:
+    if self.on:
+      if avg < self.off_threshold:
         self.on = False
+        self.switch_timeout = time.time() + 1
     else:
-      if avg > 2.6:
+      if avg > self.on_threshold:
         self.on = True
+        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
+      self.switch_timeout = time.time() + self.toggle_cooldown
+
+class Controller:
+  def __init__(self):
+    self.target_lower = 0.85
+    self.target_upper = 0.90
+    self.feedforward_coeff = 50
+
+    self._manual_mode = False
+    self.manual_on = False
+    self.manual_timeout = 0
+    self.manual_duration = 40
+
+    self.humidifier_history = np.zeros(30)
+    self.first_sample = False
+
+  @property
+  def manual_mode(self):
+    return self._manual_mode
+
+  @manual_mode.setter
+  def manual_mode(self, on):
+    self._manual_mode = on
+    send_update({"status": {"manual_mode": on}})
+
+  def update(self, humidifier, humidity):
+    if self.first_sample:
+      self.humidifier_history[:] = humidity
+      self.first_sample = False
+    else:
+      self.humidifier_history[:-1] = self.humidifier_history[1:]
+      self.humidifier_history[-1] = humidity
+
+    # compensate for the slow response time by adding a little feed forward
+    # using the slope of the humidifier data
+    slope = (self.humidifier_history[-1] - self.humidifier_history[0])/self.humidifier_history.shape[0]
+    comp_humidity = humidity + self.feedforward_coeff*slope
+
+    if self.manual_mode and time.time() > self.manual_timeout:
+      self.manual_mode = False
+
+    if self.manual_mode:
+      if humidifier.off and self.manual_on:
+        humidifier.toggle(s)
+      elif humidifier.on and not self.manual_on:
+        humidifier.toggle(s)
+    else:
+      if comp_humidity < self.target_lower and humidifier.off:
+        humidifier.toggle(s)
+      elif comp_humidity > self.target_upper and humidifier.on:
+        humidifier.toggle(s)
 
 humidifier = Humidifier()
-target_lower = 0.85
-target_higher = 0.90
+controller = Controller()
+
+exiting = False
+# run thread to process data from process's stdout
+def stdout_loop():
+  global process, controller, humidifier
+  while not exiting:
+    msg = process.stdout.readline()
+    if len(msg) == 0:
+      continue
+    print("got message ", msg)
+    try:
+      msg_js = json.loads(msg)
+      if "query_params" in msg_js:
+        if msg_js["query_params"]:
+          send_update({"params": {
+            "target_lower": controller.target_lower,
+            "target_upper": controller.target_upper,
+            "feedforward_coeff": controller.feedforward_coeff,
+            "manual_timeout": controller.manual_timeout,
+            "manual_duration_s": controller.manual_duration,
+            "manual_mode": 1.0 if controller.manual_mode else 0.0,
+            "manual_hum_on": 1.0 if controller.manual_on else 0.0,
+
+            "off_threshold_volts": humidifier.off_threshold,
+            "on_threshold_volts": humidifier.on_threshold,
+            "toggle_cooldown": humidifier.toggle_cooldown
+            }
+          })
+      elif "set_params" in msg_js:
+        if type(msg_js["set_params"]) is dict:
+          set_params = msg_js["set_params"]
+          if "name" in set_params and "value" in set_params:
+            name, value = set_params["name"], set_params["value"]
+            if type(value) is float:
+              if name == "target_lower":
+                controller.target_lower = value
+              elif name == "target_upper":
+                controller.target_upper = value
+              elif name == "feedforward_coeff":
+                controller.feedforward_coeff = value
+              elif name == "manual_timeout":
+                controller.manual_timeout = value
+              elif name == "manual_duration_s":
+                controller.manual_duration = value
+              elif name == "off_threshold_volts":
+                humidifier.off_threshold = value
+              elif name == "on_threshold_volts":
+                humidifier.on_threshold = value
+              elif name == "toggle_cooldown":
+                humidifier.toggle_cooldown = value
+      elif "manual_mode" in msg_js:
+        controller.manual_mode = msg_js["manual_mode"]
+        if controller.manual_mode:
+          controller.manual_timeout = time.time() + controller.manual_duration
+      elif "manual_mode_on" in msg_js:
+        controller.manual_on = msg_js["manual_mode_on"]
+    except json.JSONDecodeError as e:
+      print("received bad json ", msg)
+stdout_thread = threading.Thread(target=stdout_loop)
+stdout_thread.start()
+
+
+frame_num = 0
+last_sample = 0
 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)
+    if now - last_sample < SAMPLE_PERIOD:
+      time.sleep(SAMPLE_PERIOD - (now - last_sample) + 0.001)
+      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])
+
+    try:
       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))
+      controller.update(humidifier, humidity)
+
+      if frame_num == 0:
+        print(humidity, temp, volts)
+        update = {
+            "data": {
+              "time": int(now*1000),
+              "temp": temp,
+              "hum": humidity,
+              "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
+      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:
-  # TODO kill ssh connection
-  process.
-  pass
+  # kill ssh connection
+  exiting = True
+  process.kill()
+  stdout_thread.join()

shroom_controller_mock.py

@@ -1,170 +0,0 @@
-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

@@ -66,8 +66,32 @@ func LatestTime(db *sql.DB) (int64, error) {
 	return t, nil
 }
 
-func InsertRow(db *sql.DB, s *ShroomData) error {
+func OldestTime(db *sql.DB) (int64, error) {
+	query := "SELECT MIN(time) FROM shrooms"
+	rows, err := db.Query(query)
+	if err != nil {
+		return -1, fmt.Errorf("sql error: %w", err)
+	}
+	defer rows.Close()
+	if !rows.Next() {
+		// i guess the database is empty
+		return 0, nil
+	}
+	t := int64(0)
+	err = rows.Scan(&t)
+	if err != nil {
+		return -1, fmt.Errorf("sql scan error: %w", err)
+	}
+	return t, nil
+}
+
+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
+}
+
+func ClearOldRows(db *sql.DB, min_time int64) error {
+	_, err := db.Exec("DELETE FROM shrooms WHERE t < ?", min_time)
 	return err
 }

shroom_internals/tcp_server.go

@@ -3,9 +3,11 @@ package shroom_internals
 import (
 	"database/sql"
 	"encoding/json"
+	"fmt"
 	"log"
 	"net"
 	"sync"
+	"time"
 )
 
 func newlinePos(s []byte) int {
@@ -17,47 +19,131 @@ func newlinePos(s []byte) int {
 	return -1
 }
 
-type ShroomData struct {
-	Time            uint64  `json:"time"`
-	Temperature     float32 `json:"temp"`
-	Humidity        float32 `json:"hum"`
-	HumidifierVolts float32 `json:"hv"`
-	Status          int32   `json:"status"`
+type StatusJson struct {
+	HumOn      *bool `json:"humidifier"`
+	ManualMode *bool `json:"manual_mode"`
 }
 
-type ShroomStatus struct {
+type DataJson struct {
+	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"`
+	Params map[string]float32 `json:"params"`
+}
+
+type ShroomState struct {
 	sync.RWMutex
 	HumidifierOn   bool
+	ManualMode     bool
 	NumConnections int
-	Wait           chan struct{}
-	StatusWait     chan struct{}
+
+	Params     map[string]float32
+	Wait       chan struct{}
+	StatusWait chan struct{}
+	ParamsWait chan struct{}
 
 	Commands chan []byte
 }
 
-func (s *ShroomStatus) Update() {
+func NewShroomState() ShroomState {
+	return ShroomState{
+		Params:     make(map[string]float32),
+		Wait:       make(chan struct{}),
+		StatusWait: make(chan struct{}),
+		ParamsWait: make(chan struct{}),
+		Commands:   make(chan []byte),
+	}
+}
+
+func (s *ShroomState) Update() {
 	s.Lock()
 	defer s.Unlock()
 	close(s.Wait)
 	s.Wait = make(chan struct{})
 }
 
-func (s *ShroomStatus) StatusUpdate() {
+func (s *ShroomState) 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:     -274,
-		Humidity:        -1,
-		HumidifierVolts: -1,
-		Status:          -1,
+func (s *ShroomState) ParamsUpdate() {
+	s.Lock()
+	defer s.Unlock()
+	close(s.ParamsWait)
+	s.ParamsWait = make(chan struct{})
+}
+
+func WaitForClose(c chan struct{}) {
+	stillopen := true
+	for stillopen {
+		_, stillopen = <-c
 	}
-	err := json.Unmarshal(line, &data)
+}
+
+func WaitForCloseWithTimeout(c chan struct{}, t time.Duration) bool {
+	stillopen := true
+	timeout := make(chan struct{})
+	go func() {
+		time.Sleep(t)
+		timeout <- struct{}{}
+	}()
+
+	for stillopen {
+		select {
+		case _, stillopen = <-c:
+			if stillopen {
+				continue
+			}
+		case <-timeout:
+			return false
+		}
+	}
+	return true
+}
+
+type queryParams struct {
+	QueryParams bool `json:"query_params"`
+}
+
+func QueryParams(state *ShroomState) ([]byte, error) {
+	msg, err := json.Marshal(queryParams{
+		QueryParams: true,
+	})
+	if err != nil {
+		err = fmt.Errorf("unable to marshal query request: %w", err)
+		return nil, err
+	}
+	select {
+	case state.Commands <- msg:
+		if !WaitForCloseWithTimeout(state.ParamsWait, 2*time.Second) {
+			return nil, fmt.Errorf("request timed out")
+		}
+		state.RLock()
+		resp, err := json.Marshal(state.Params)
+		state.RUnlock()
+		if err != nil {
+			err = fmt.Errorf("unable to unmarshal params as json: %w", err)
+			return nil, err
+		}
+		return resp, nil
+	default:
+		return nil, fmt.Errorf("unable to forward request; controller may not be connected")
+	}
+
+}
+
+func parseMsg(line []byte, db *sql.DB, state *ShroomState) {
+	packet := ShroomPacket{}
+	err := json.Unmarshal(line, &packet)
 	if err != nil {
 		log.Println("unable to parse tcp line: ", err)
 		log.Println(string(line))
@@ -65,27 +151,38 @@ 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 {
-		err = InsertRow(db, &data)
-		if err != nil {
-			log.Println("unable to write to database: ", err)
+	if packet.Data != nil {
+		if packet.Data.Time != nil && packet.Data.Temperature != nil && packet.Data.Humidity != nil && packet.Data.HumidifierVolts != nil {
+			err = InsertRow(db, packet.Data)
+			if err != nil {
+				log.Println("unable to write to database: ", err)
+			}
+			// we got a data packet
+			state.Update()
 		}
-		// we got a data packet
-		status.Update()
-	} else if data.Status != -1 {
+	} else if packet.Status != nil {
 		//log.Println("received status ", data.Status)
-		status.Lock()
-		// TODO change to have more detailed data
-		status.HumidifierOn = (data.Status & 1) == 1
-		status.Unlock()
-		status.StatusUpdate()
+		state.Lock()
+		if packet.Status.HumOn != nil {
+			state.HumidifierOn = *packet.Status.HumOn
+		}
+		if packet.Status.ManualMode != nil {
+			state.ManualMode = *packet.Status.ManualMode
+		}
+		state.Unlock()
+		state.StatusUpdate()
+	} else if packet.Params != nil {
+		state.Lock()
+		state.Params = packet.Params
+		state.Unlock()
+		state.ParamsUpdate()
 	} else {
 		log.Println("unknown packet: ", line, string(line))
 	}
 
 }
 
-func InitTcpServer(db *sql.DB, status *ShroomStatus) {
+func InitTcpServer(db *sql.DB, state *ShroomState) {
 	// start TCP server for the pipe from the raspberry pi
 	ln, err := net.Listen("tcp", ":9876")
 	if err != nil {
@@ -105,16 +202,16 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 			// 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()
+				state.Lock()
+				state.NumConnections += 1
+				state.Unlock()
+				state.StatusUpdate()
 
 				defer func() {
-					status.Lock()
-					status.NumConnections -= 1
-					status.Unlock()
-					status.StatusUpdate()
+					state.Lock()
+					state.NumConnections -= 1
+					state.Unlock()
+					state.StatusUpdate()
 					log.Println("connection disconnected")
 				}()
 
@@ -125,7 +222,7 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 				go func() {
 					for {
 						select {
-						case v, ok := <-status.Commands:
+						case v, ok := <-state.Commands:
 							if !ok {
 								return
 							}
@@ -144,15 +241,17 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 				}()
 
 				// deal with the read side of the connection
-				buf := make([]byte, 128)
+				buf := make([]byte, 1024)
 				left := buf
 
 				for {
+					if len(left) == 0 {
+						log.Println("overflow detected, truncating data")
+						left = buf
+					}
+
 					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)
 
 					if err != nil {
 						log.Println("tcp read error: ", err)
@@ -168,14 +267,12 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
 						line := unread[:end]
 						unread = unread[end+1:]
 
-						//log.Println("line ", line)
-						//log.Println("unread ", unread)
 						// skip empty lines
 						if len(line) == 0 {
 							continue
 						}
 
-						parseMsg(line, db, status)
+						parseMsg(line, db, state)
 					}
 					// shift the remaining data back to the start of the buffer
 					copy(buf[:len(unread)], unread)

shroom_pipe.py

@@ -13,7 +13,9 @@ async def connect_stdin_stdout():
 
 async def pipe_loop(reader, writer):
   while True:
+    #res = await reader.readline()
     res = await reader.read(1024)
+    #res = await reader.readchunk()
     if not res:
       break
     writer.write(res)

shroom_server.go

@@ -21,12 +21,16 @@ import (
 	"time"
 )
 
+//go:embed auth_secret
+var auth_secret string
+
 //go:embed static/*
 var content embed.FS
 
 type statusJson struct {
 	Connected  bool `json:"connected"`
 	Humidifier bool `json:"humidifier"`
+	ManualMode bool `json:"manual_mode"`
 }
 
 type adminMsg struct {
@@ -34,6 +38,8 @@ type adminMsg struct {
 	Msg  map[string]interface{} `json:"data"`
 }
 
+// returns a function that multiplies the number at the very last segment of the url
+// and returns the data that was collected in the last n*multiplier milliseconds
 func dumpData(db *sql.DB, multiplier int64) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, req *http.Request) {
 		now := time.Now().Unix()
@@ -72,12 +78,8 @@ func main() {
 		log.Fatal("unable to create table ", err)
 	}
 
-	status := s.ShroomStatus{
-		Wait:       make(chan struct{}),
-		StatusWait: make(chan struct{}),
-		Commands:   make(chan []byte),
-	}
-	s.InitTcpServer(db, &status)
+	state := s.NewShroomState()
+	s.InitTcpServer(db, &state)
 
 	contentSub, err := fs.Sub(content, "static")
 	if err != nil {
@@ -101,13 +103,15 @@ func main() {
 	}
 
 	getStatus := func(w http.ResponseWriter, _req *http.Request) {
-		status.RLock()
-		num_connections := status.NumConnections
-		humidifier := status.HumidifierOn
-		status.RUnlock()
+		state.RLock()
+		num_connections := state.NumConnections
+		humidifier := state.HumidifierOn
+		manual_mode := state.ManualMode
+		state.RUnlock()
 		s := statusJson{
 			Connected:  num_connections > 0,
 			Humidifier: humidifier,
+			ManualMode: manual_mode,
 		}
 		msg, err := json.Marshal(s)
 		if err != nil {
@@ -142,10 +146,10 @@ func main() {
 			return
 		}
 
-		// TODO switch to embedded secret
-		if adminReq.Auth != "password" {
+		// switch to embedded secret
+		if adminReq.Auth != auth_secret {
 			w.WriteHeader(401)
-			w.Write([]byte(err.Error()))
+			w.Write([]byte("invalid secret"))
 			return
 		}
 
@@ -157,7 +161,7 @@ func main() {
 			return
 		}
 		select {
-		case status.Commands <- inner_msg:
+		case state.Commands <- inner_msg:
 			w.Write([]byte("ok"))
 		default:
 			w.WriteHeader(503)
@@ -165,18 +169,22 @@ func main() {
 		}
 	}
 
-	updateHandler := func(w http.ResponseWriter, req *http.Request) {
-		stillopen := true
-		for stillopen {
-			_, stillopen = <-status.Wait
+	paramsHandler := func(w http.ResponseWriter, req *http.Request) {
+		msg, err := s.QueryParams(&state)
+		if err != nil {
+			w.WriteHeader(500)
+			w.Write([]byte(err.Error()))
+			return
 		}
+		w.Write(msg)
+	}
+
+	updateHandler := func(w http.ResponseWriter, req *http.Request) {
+		s.WaitForClose(state.Wait)
 		w.Write([]byte("ok"))
 	}
 	statusUpdateHandler := func(w http.ResponseWriter, req *http.Request) {
-		stillopen := true
-		for stillopen {
-			_, stillopen = <-status.StatusWait
-		}
+		s.WaitForClose(state.StatusWait)
 		w.Write([]byte("ok"))
 	}
 
@@ -190,12 +198,32 @@ func main() {
 	http.HandleFunc("/api/latest", lastPoint)
 	http.HandleFunc("/api/status", getStatus)
 	http.HandleFunc("/api/admin", adminHandler)
+	http.HandleFunc("/api/params", paramsHandler)
 	http.HandleFunc("/api/update", updateHandler)
 	http.HandleFunc("/api/status_update", statusUpdateHandler)
 
-	// TODO periodically clear old entries from the database
+	// periodically clear old entries from the database
+	go func() {
+		// TODO maybe make this exit gracefully
+		for {
+			t, err := s.OldestTime(db)
+			if err != nil {
+				log.Println("unable to get oldest time: ", err)
+			}
+			now := time.Now().Unix()
+			diff := now*1000 - t
+			log.Println("oldest time", t, " current time", now, "diff", diff)
+			if diff > 2*7*24*60*60*1000 {
+				err = s.ClearOldRows(db, now-8*24*60*60*1000)
+				if err != nil {
+					log.Println("unable to delete rows: ", err)
+				}
+			}
+			time.Sleep(24 * time.Hour)
+		}
+	}()
 
-	err = http.ListenAndServe("localhost:8080", nil)
+	err = http.ListenAndServe("localhost:8085", nil)
 	if err != nil {
 		log.Fatal("unable to start server: ", err)
 	}

static/admin.html

@@ -0,0 +1,195 @@
+<!doctype html>
+<html>
+  <head>
+    <script>
+function status(msg) {
+  document.getElementById('status').textContent = msg
+}
+async function queryParams() {
+  const resp = await fetch('/api/params')
+  //console.log(json)
+  //console.log(Object.keys(json))
+  if (!resp.ok) {
+    var err_msg = null
+    if (resp.body != null) {
+      err_msg = await resp.text()
+    }
+    if (err_msg != null) {
+      status("query failed: " + resp.status + " " + err_msg)
+    } else {
+      status("query failed: " + resp.status)
+    }
+  }
+  const json = await resp.json()
+  status('parameter query successful!')
+  const paramList = document.getElementById('param-list')
+  while (paramList.firstChild) {
+    paramList.removeChild(paramList.lastChild)
+  }
+  Object.keys(json).forEach((name) => {
+    const elem = document.createElement('div')
+    elem.textContent = name + " = " + json[name]
+    paramList.appendChild(elem)
+  })
+}
+
+async function setParam(auth, name, value) {
+  const msg = JSON.stringify({
+    auth: auth,
+    data: {
+      set_params: {
+        name: name,
+        value: value
+      }
+    }
+  })
+  const resp = await fetch('/api/admin', {
+    method: 'POST',
+    body: msg
+  })
+  if (!resp.ok) {
+    var err_msg = null
+    if (resp.body != null) {
+      err_msg = await resp.text()
+    }
+    if (err_msg != null) {
+      status("set failed: " + resp.status + " " + err_msg)
+    } else {
+      status("set failed: " + resp.status)
+    }
+  } else {
+    status("parameter set successful!")
+    await queryParams()
+  }
+}
+
+var manual_mode = false
+var manual_loop_running = false
+const sleep = (ms) => new Promise(r => setTimeout(r, ms))
+async function manualModeLoop(auth) {
+  if (manual_loop_running) return
+  manual_loop_running = true
+  try {
+    while (manual_mode) {
+      await sleep(30*1000)
+      if (manual_mode) await manualMode(auth, manual_mode)
+    }
+  } finally {
+    manual_loop_running = false
+  }
+}
+async function manualMode(auth, on) {
+  const msg = JSON.stringify({
+    auth: auth,
+    data: {
+      manual_mode: on
+    }
+  })
+  const resp = await fetch('/api/admin', {
+    method: 'POST',
+    body: msg
+  })
+  if (!resp.ok) {
+    var err_msg = null
+    if (resp.body != null) {
+      err_msg = await resp.text()
+    }
+    if (err_msg != null) {
+      status("manual mode set failed: " + resp.status + " " + err_msg)
+    } else {
+      status("manual mode set failed: " + resp.status)
+    }
+  } else {
+    status("manual mode set successful!")
+    manual_mode = on
+    if (manual_mode) {
+      manualModeLoop(auth)
+    }
+    //await queryParams()
+  }
+}
+
+async function manualHumidifier(auth, on) {
+  const msg = JSON.stringify({
+    auth: auth,
+    data: {
+      manual_mode_on: on
+    }
+  })
+  const resp = await fetch('/api/admin', {
+    method: 'POST',
+    body: msg
+  })
+  if (!resp.ok) {
+    var err_msg = null
+    if (resp.body != null) {
+      err_msg = await resp.text()
+    }
+    if (err_msg != null) {
+      status("manual hum set failed: " + resp.status + " " + err_msg)
+    } else {
+      status("manual hum set failed: " + resp.status)
+    }
+  } else {
+    status("manual hum set successful!")
+  }
+}
+
+window.onload = () => {
+  document.getElementById('query-params').addEventListener('click', (e) => {
+    queryParams()
+  })
+
+  document.getElementById('set-param').addEventListener('click', (e) => {
+    const auth = document.getElementById('password').value
+    const name = document.getElementById('name').value
+    const value = parseFloat(document.getElementById('value').value)
+    if (isNaN(value)) {
+      status('invalid value set')
+      return
+    }
+    setParam(auth, name, value)
+  })
+
+  document.getElementById('manual-mode').addEventListener('click', (e) => {
+    const auth = document.getElementById('password').value
+    const on = document.getElementById('manual-mode').checked
+    manualMode(auth, on)
+  })
+  document.getElementById('manual-on').addEventListener('click', (e) => {
+    const auth = document.getElementById('password').value
+    const on = document.getElementById('manual-on').checked
+    manualHumidifier(auth, on)
+  })
+}
+    </script>
+  </head>
+  <body>
+    <fieldset>
+      <legend>Status</legend>
+      <div id=status></div>
+    </fieldset>
+    <form>
+      <fieldset>
+        <legend>Parameter list</legend>
+        <div id=param-list></div>
+        <input id=query-params type=button value="Query params"></input><br />
+      </fieldset>
+      <label for=password>Auth password: </label>
+      <input id=password type=password></input><br />
+      <fieldset>
+        <legend>Set parameter</legend>
+        <label for=name>Name: </label>
+        <input id=name type=text></input><br />
+        <label for=value>Value: </label>
+        <input id=value type=text></input><br />
+        <input id=set-param type=button value="Set parameter"></input><br />
+      </fieldset>
+      <fieldset>
+        <legend>Manual mode</legend>
+        <input id=manual-mode type=checkbox></input><label for=manual-mode>manual mode</label><br />
+        <input id=manual-on type=checkbox></input><label for=manual-on>humidifier on</label><br />
+      </fieldset>
+    </form>
+  </body>
+</html>

static/index.htm

@@ -1,14 +0,0 @@
-<!doctype html>
-<html>
-  <head>
-    <!-- Load c3.css -->
-    <link href="c3.css" rel="stylesheet" type=text/css>
-
-    <!-- Load d3.js and c3.js -->
-    <script src="d3.v7.min.js" charset="utf-8"></script>
-    <script src="c3.min.js"></script>
-  </head>
-  <body>
-    <div id="chart"></div>
-  </body>
-</html>

static/index.html

@@ -0,0 +1,60 @@
+<!doctype html>
+<html>
+  <head>
+    <title>mushroom humidifier datalog</title>
+    <!-- Load c3.css -->
+    <link href="/c3.css" rel="stylesheet" type=text/css>
+
+    <!-- Load d3.js and c3.js -->
+    <script src="/d3.v5.min.js" charset="utf-8"></script>
+    <script src="/c3.min.js"></script>
+    <script src="index.js"></script>
+  </head>
+  <body>
+    <form>
+        <input type=checkbox id=autoupdate checked>Autoupdate</input>
+        <input type=button id=update value="Update"></input>
+        <span id="device-status"></span>
+        <span id=status></span>
+    </form>
+
+    <div id="humidity-temp"></div>
+    <div id="volts"></div>
+    <div>
+      <form>
+        <fieldset>
+          <legend>Data decimation rate</legend>
+          <input type=radio name=decim value="1" checked>1x</input>
+          <input type=radio name=decim value="2">2x</input>
+          <input type=radio name=decim value="4">4x</input>
+          <input type=radio name=decim value="10">10x</input>
+          <input type=radio name=decim value="20">20x</input>
+          <input type=radio name=decim value="100">100x</input>
+        </fieldset>
+        <fieldset>
+          <legend>History length</legend>
+          <input type=radio name=duration value="5" checked>5 minutes</input>
+          <input type=radio name=duration value="15">15 minutes</input>
+          <input type=radio name=duration value="60">1 hour</input>
+          <input type=radio name=duration value="1440">1 day</input>
+          <input type=radio name=duration value="10080">1 week</input>
+        </fieldset>
+        <fieldset>
+          <legend>Update rate</legend>
+          <input type=radio name=update value="100">0.1 second</input>
+          <input type=radio name=update value="1000">1 second</input>
+          <input type=radio name=update value="2000" checked>2 seconds</input>
+          <input type=radio name=update value="5000">5 seconds</input>
+          <input type=radio name=update value="10000">10 seconds</input>
+          <input type=radio name=update value="20000">20 seconds</input>
+        </fieldset>
+
+
+        <!--
+        <input type=button id=test-admin value="Test admin mode"></input>
+        -->
+        <!-- TODO add decimation and window size options -->
+      </form>
+    </div>
+  </body>
+</html>

static/index.js

@@ -1,13 +1,3 @@
-<!doctype html>
-<html>
-  <head>
-    <!-- Load c3.css -->
-    <link href="/c3.css" rel="stylesheet" type=text/css>
-
-    <!-- Load d3.js and c3.js -->
-    <script src="/d3.v5.min.js" charset="utf-8"></script>
-    <script src="/c3.min.js"></script>
-    <script>
 var chart = null
 var chart2 = null
 function initCharts() {
@@ -64,9 +54,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 = []
@@ -142,26 +133,29 @@ async function updateCharts() {
     volts = volts.slice(slice_idx)
     time = time.slice(slice_idx)
 
+    const temp_d = temp.filter((_, idx) => (idx % decimation_rate) == 0)
+    const humd_d = humd.filter((_, idx) => (idx % decimation_rate) == 0)
+    const volts_d = volts.filter((_, idx) => (idx % decimation_rate) == 0)
+    const time_d = time.filter((_, idx) => (idx % decimation_rate) == 0)
+
     chart.load({
       columns: [
-        ["time"].concat(time),
-        ["temp"].concat(temp),
-        ["humidity"].concat(humd),
+        ["time"].concat(time_d),
+        ["temp"].concat(temp_d),
+        ["humidity"].concat(humd_d),
       ]
     })
     chart2.load({
       columns: [
-        ["time"].concat(time),
-        ["voltage"].concat(volts),
+        ["time"].concat(time_d),
+        ["voltage"].concat(volts_d),
       ]
     })
     status("charts updated")
   }
 
   if (autoupdate) {
-    chart_updater = chartupdater()
-  } else {
-    chart_updater = null
+    chartupdater()
   }
 }
 
@@ -169,43 +163,59 @@ const sleep = (ms) => new Promise(r => setTimeout(r, ms))
 var chart_update_millis = 2000
 
 async function chartupdater() {
-  if (chart_updater != null) return
+  if (chart_updater) return
+  chart_updater = true
 
-  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()
+  try {
+    while (autoupdate) {
+      await sleep(chart_update_millis)
+      // wait at least two seconds to avoid wasting a lot of bandwidth
+      const resp = await fetch("/api/update")
+      updateCharts()
+    }
+  } finally {
+    chart_updater = false
+  }
 }
 
-var status_updater = null 
+var status_updater = false
 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")
+  const manual_mode_state = (status.manual_mode ? "on" : "off")
   //console.log(humidifier_state)
-  document.getElementById("device-status").textContent = "connected: " + status.connected + ", humidifier: " + humidifier_state
+  document.getElementById("device-status").textContent = "connected: " +
+    status.connected + ", manual mode: " + manual_mode_state + ", humidifier: "
+    + humidifier_state
 
   if (autoupdate) {
-    status_updater = waitThenUpdateStatus()
-  } else {
-    status_updater = null
+    waitThenUpdateStatus()
   }
 }
 
 async function waitThenUpdateStatus() {
-  if (status_updater != null) return
-  await fetch("/api/status_update")
-  status_updater = null
-  updateStatus()
+  if (status_updater) return
+  status_updater = true
+  try {
+    while (autoupdate) {
+      await fetch("/api/status_update")
+      updateStatus()
+    }
+  } finally {
+    status_updater = false
+  }
 }
 
 async function testAdminMode() {
   const msg = JSON.stringify({
     auth: "password",
     data: {
-      manual_mode: true
+      set_params: {
+        name: "target_lower",
+        value: 0.87
+      }
     }
   })
   await fetch("/api/admin", {
@@ -226,24 +236,42 @@ window.onload = () => {
   document.getElementById('autoupdate').addEventListener('click', (e) => {
     autoupdate = document.getElementById('autoupdate').checked
   })
+  document.getElementsByName('decim').forEach((elem) => {
+    elem.addEventListener('click', (e) => {
+      if (elem.checked) {
+        decimation_rate = parseInt(elem.value)
+      }
+    })
+  })
+  document.getElementsByName('update').forEach((elem) => {
+    elem.addEventListener('click', (e) => {
+      if (elem.checked) {
+        chart_update_millis = parseInt(elem.value)
+      }
+    })
+  })
+
+  document.getElementsByName('duration').forEach((elem) => {
+    elem.addEventListener('click', (e) => {
+      const old_millis = max_interval_millis
+      if (elem.checked) {
+        max_interval_millis = parseInt(elem.value)*1000*60
+        if (max_interval_millis != old_millis) {
+          // reset the chart data to force a reload on the next update
+          cur_time_millis = 0
+          time = []
+          temp = []
+          humd = []
+          volts = []
+        }
+      }
+    })
+  })
+
+  /*
   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>
-    <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>
+