Compare commits
8 Commits
25d3788478
...
3a7bb60360
Author | SHA1 | Date |
---|---|---|
|
3a7bb60360 | |
|
f1723ff5bd | |
|
9692132e75 | |
|
008cde9d5e | |
|
ce21b6fa85 | |
|
00868284bd | |
|
18ffe01347 | |
|
0ca4da4b6c |
196
dev/dev.htm
196
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>
|
||||
|
|
|
@ -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
|
|
@ -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()
|
|
@ -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) {
|
||||
|
|
|
@ -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):]
|
||||
}
|
||||
}()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
|
101
shroom_server.go
101
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) {
|
||||
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
|
||||
|
||||
|
|
Loading…
Reference in New Issue