MARTe-Integrated-Components/Common/Client/go/wshub/export.go

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