Compare commits

...

8 Commits

6 changed files with 617 additions and 79 deletions

View File

@ -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>

73
shroom_controller.py Normal file
View File

@ -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

170
shroom_controller_mock.py Normal file
View File

@ -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()

View File

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

View File

@ -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,62 +96,92 @@ func InitTcpServer(db *sql.DB, status *ShroomStatus) {
conn, err := ln.Accept()
if err != nil {
log.Println("tcp accept error: ", err)
continue
return
}
status.Lock()
status.NumConnections += 1
status.Unlock()
// not spawning a goroutine here
// should limit the number of connections to
// one hopefully
defer func() {
// wrapping in a func() so that I can use defer
// to automatically decrement the number of connections
func() {
status.Lock()
status.NumConnections -= 1
status.NumConnections += 1
status.Unlock()
log.Println("connection disconnected")
}()
status.StatusUpdate()
log.Println("connection started")
defer func() {
status.Lock()
status.NumConnections -= 1
status.Unlock()
status.StatusUpdate()
log.Println("connection disconnected")
}()
go func() {
// TODO deal with the write side of the connection
}()
log.Println("connection started")
// deal with the read side of the connection
buf := make([]byte, 128)
left := buf
for {
num_read, err := conn.Read(left)
left = left[num_read:]
//log.Println("buf ", buf)
//log.Println("left ", left)
if err != nil {
log.Println("tcp read error: ", err)
_ = conn.Close()
log.Println("disconnected from client")
break
}
// parse the message to see if it's finished
unread := buf[:len(buf)-len(left)]
for newlinePos(unread) != -1 {
end := newlinePos(unread)
line := unread[:end]
unread = unread[end+1:]
//log.Println("line ", line)
//log.Println("unread ", unread)
// skip empty lines
if len(line) == 0 {
continue
// write loop; waits for commands and forwards them
exiting := make(chan struct{})
go func() {
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)
}()
parseMsg(line, db, status)
// deal with the read side of the connection
buf := make([]byte, 128)
left := buf
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)
if err != nil {
log.Println("tcp read error: ", err)
_ = conn.Close()
log.Println("disconnected from client")
break
}
// parse the message to see if it's finished
unread := buf[:len(buf)-len(left)]
for newlinePos(unread) != -1 {
end := newlinePos(unread)
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)
}
// shift the remaining data back to the start of the buffer
copy(buf[:len(unread)], unread)
left = buf[len(unread):]
}
// shift the remaining data back to the start of the buffer
copy(buf[:len(unread)], unread)
left = buf[len(unread):]
}
}()
}
}()
}

View File

@ -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) {
return func(w http.ResponseWriter, _req *http.Request) {
type adminMsg struct {
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)
@ -56,7 +73,9 @@ func main() {
}
status := s.ShroomStatus{
Wait: make(chan struct{}),
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)
dumpDay := dumpData(db, -24*60*60)
dumpHour := dumpData(db, -60*60)
dumpMinute := dumpData(db, -60)
dumpWeek := dumpData(db, 7*24*60*60*1000)
dumpDay := dumpData(db, 24*60*60*1000)
dumpHour := dumpData(db, 60*60*1000)
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)
w.Write([]byte("unimplemented"))
// TODO
if req.Method != "POST" {
w.WriteHeader(405)
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/day", dumpDay)
http.HandleFunc("/api/last/hour", dumpHour)
http.HandleFunc("/api/last/minute", dumpMinute)
http.HandleFunc("/api/last/weeks/", dumpWeek)
http.HandleFunc("/api/last/days/", dumpDay)
http.HandleFunc("/api/last/hours/", dumpHour)
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