fixed and improved ui
This commit is contained in:
Martino Ferrari
@@ -0,0 +1,119 @@
|
||||
package wshub
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/parquet-go/parquet-go"
|
||||
)
|
||||
|
||||
// ExportSample is one row of the binary export: a single stored sample, in
|
||||
// long ("tidy") form, keyed by source and signal with its own timestamp.
|
||||
//
|
||||
// Keeping each signal's samples as its own rows — rather than resampling onto a
|
||||
// shared time grid — is what makes the export hole-free: per-signal streams of
|
||||
// different lengths export exactly as stored, nothing is fabricated, and
|
||||
// nothing is dropped.
|
||||
type ExportSample struct {
|
||||
Source string `parquet:"source"`
|
||||
Signal string `parquet:"signal"`
|
||||
Time float64 `parquet:"time"`
|
||||
Value float64 `parquet:"value"`
|
||||
}
|
||||
|
||||
// exportChunkRows bounds each batched write and, via MaxRowsPerRowGroup, the
|
||||
// size of each parquet row group: memory stays bounded however large the
|
||||
// export is, because a finished row group is flushed to the HTTP stream.
|
||||
const exportChunkRows = 65536
|
||||
|
||||
// exportWriteBuffer is the parquet writer's output buffer: larger than the
|
||||
// 32KiB default means fewer writes on the HTTP stream for a multi-GB export.
|
||||
const exportWriteBuffer = 1 << 20
|
||||
|
||||
// HandleExport serves GET /api/export?t0=..&t1=..[&signals=a,b] as a Parquet
|
||||
// file containing every stored sample of the named signals in [t0, t1].
|
||||
//
|
||||
// Unlike /api/zoom there is no decimation: the file holds the full contents of
|
||||
// the rings. At rates above the ring budget those contents are min/max buckets
|
||||
// (the finest resolution the hub retains); at lower rates they are verbatim.
|
||||
func (h *Hub) HandleExport(w http.ResponseWriter, r *http.Request) {
|
||||
q := r.URL.Query()
|
||||
t0, err0 := strconv.ParseFloat(q.Get("t0"), 64)
|
||||
t1, err1 := strconv.ParseFloat(q.Get("t1"), 64)
|
||||
if err0 != nil || err1 != nil || t1 <= t0 {
|
||||
http.Error(w, "invalid t0/t1", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
var keys []string
|
||||
if s := strings.TrimSpace(q.Get("signals")); s != "" {
|
||||
keys = strings.Split(s, ",")
|
||||
for i := range keys {
|
||||
keys[i] = strings.TrimSpace(keys[i])
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot the rings we will read. A signal removed mid-export must not
|
||||
// silently drop rows from the file.
|
||||
h.ringsMu.RLock()
|
||||
refs := make(map[string]*sigRing)
|
||||
if keys == nil {
|
||||
for k, rb := range h.rings {
|
||||
refs[k] = rb
|
||||
}
|
||||
} else {
|
||||
for _, k := range keys {
|
||||
if rb, ok := h.rings[k]; ok {
|
||||
refs[k] = rb
|
||||
}
|
||||
}
|
||||
}
|
||||
h.ringsMu.RUnlock()
|
||||
if len(refs) == 0 {
|
||||
http.Error(w, "no signals", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
// Deterministic column order.
|
||||
names := make([]string, 0, len(refs))
|
||||
for k := range refs {
|
||||
names = append(names, k)
|
||||
}
|
||||
sort.Strings(names)
|
||||
|
||||
w.Header().Set("Content-Type", "application/vnd.apache.parquet")
|
||||
w.Header().Set("Content-Disposition",
|
||||
fmt.Sprintf("attachment; filename=\"signals_%d.parquet\"", time.Now().Unix()))
|
||||
|
||||
writer := parquet.NewGenericWriter[ExportSample](w,
|
||||
parquet.MaxRowsPerRowGroup(exportChunkRows),
|
||||
parquet.WriteBufferSize(exportWriteBuffer),
|
||||
)
|
||||
batch := make([]ExportSample, 0, exportChunkRows)
|
||||
for _, key := range names {
|
||||
st, sv := refs[key].slice(t0, t1)
|
||||
colon := strings.IndexByte(key, ':')
|
||||
source, signal := key, key
|
||||
if colon >= 0 {
|
||||
source = key[:colon]
|
||||
signal = key[colon+1:]
|
||||
}
|
||||
for i := range st {
|
||||
batch = append(batch, ExportSample{Source: source, Signal: signal, Time: st[i], Value: sv[i]})
|
||||
if len(batch) >= exportChunkRows {
|
||||
if _, err := writer.Write(batch); err != nil {
|
||||
// Client went away or the stream broke; stop writing.
|
||||
return
|
||||
}
|
||||
batch = batch[:0]
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(batch) > 0 {
|
||||
_, _ = writer.Write(batch)
|
||||
}
|
||||
_ = writer.Close()
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package wshub
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/parquet-go/parquet-go"
|
||||
)
|
||||
|
||||
func TestHandleExportParquetFullResolution(t *testing.T) {
|
||||
h := NewHub()
|
||||
// Two signals with different lengths and offset time bases: the export must
|
||||
// keep every sample of each, on its own timestamps (no holes, no
|
||||
// resampling, no decimation).
|
||||
sig1 := newSigRing(10000)
|
||||
sig2 := newSigRing(10000)
|
||||
t1, v1 := make([]float64, 1000), make([]float64, 1000)
|
||||
for i := range t1 {
|
||||
t1[i] = float64(i) * 0.001
|
||||
v1[i] = float64(i) * 2
|
||||
}
|
||||
sig1.write(t1, v1)
|
||||
t2, v2 := make([]float64, 500), make([]float64, 500)
|
||||
for i := range t2 {
|
||||
t2[i] = 0.1 + float64(i)*0.002
|
||||
v2[i] = -float64(i)
|
||||
}
|
||||
sig2.write(t2, v2)
|
||||
h.rings["s1:Ch1"] = sig1
|
||||
h.rings["s1:Ch2"] = sig2
|
||||
|
||||
req := httptest.NewRequest("GET", "/api/export?t0=0&t1=2&signals=s1:Ch1,s1:Ch2", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
h.HandleExport(rec, req)
|
||||
if rec.Code != 200 {
|
||||
t.Fatalf("status = %d, want 200 (body: %s)", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
reader := parquet.NewGenericReader[ExportSample](bytes.NewReader(rec.Body.Bytes()))
|
||||
defer reader.Close()
|
||||
|
||||
var got []ExportSample
|
||||
buf := make([]ExportSample, 1000)
|
||||
for {
|
||||
n, err := reader.Read(buf)
|
||||
got = append(got, buf[:n]...)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(got) != 1500 {
|
||||
t.Fatalf("rows = %d, want 1500 (every sample of both signals)", len(got))
|
||||
}
|
||||
ch1 := filterExportSamples(got, "s1", "Ch1")
|
||||
ch2 := filterExportSamples(got, "s1", "Ch2")
|
||||
if len(ch1) != 1000 || len(ch2) != 500 {
|
||||
t.Fatalf("ch1=%d ch2=%d rows, want 1000/500 (no holes, no resampling)", len(ch1), len(ch2))
|
||||
}
|
||||
if ch1[0].Time != 0 || ch1[0].Value != 0 || ch1[999].Time != 0.999 || ch1[999].Value != 1998 {
|
||||
t.Fatalf("ch1 endpoints wrong: first=%+v last=%+v", ch1[0], ch1[999])
|
||||
}
|
||||
if ch2[0].Time != 0.1 || ch2[499].Time != 0.1+499*0.002 || ch2[499].Value != -499 {
|
||||
t.Fatalf("ch2 endpoints wrong: first=%+v last=%+v", ch2[0], ch2[499])
|
||||
}
|
||||
}
|
||||
|
||||
func filterExportSamples(rows []ExportSample, source, signal string) []ExportSample {
|
||||
out := make([]ExportSample, 0, len(rows))
|
||||
for _, r := range rows {
|
||||
if r.Source == source && r.Signal == signal {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestHandleExportParquetBadRange(t *testing.T) {
|
||||
h := NewHub()
|
||||
h.rings["s1:Ch1"] = newSigRing(10)
|
||||
req := httptest.NewRequest("GET", "/api/export?t0=2&t1=1", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
h.HandleExport(rec, req)
|
||||
if rec.Code != 400 {
|
||||
t.Fatalf("status = %d, want 400 for inverted range", rec.Code)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user