qsc20001102/AquaControlAI
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

OK

Browse Source
This commit is contained in:
qsc20001102
2026-07-13 11:13:00 +08:00
parent ef96af7f21
commit dac7457c60
9 changed files with 297 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/router.go
+4 -6
View File
@@ -711,19 +711,17 @@ func (h *Handler) exportHistory(c *gin.Context) {
n += "_" + col.PointID.String()[:8]
}
used[n] = true
head = append(head, n, n+"_质量")
head = append(head, n)
}
w.Write(head)
for i, t := range data.TimeColumn {
row := []string{t.Format("2006-01-02 15:04:05")}
for _, col := range data.Columns {
x := col.Data[i]
if x.Quality == "good" && x.Value != nil {
row = append(row, strconv.FormatFloat(*x.Value, 'f', -1, 64), "good")
} else if x.Quality == "bad" {
row = append(row, "—", "bad")
if x.Value != nil {
row = append(row, strconv.FormatFloat(*x.Value, 'f', -1, 64))
} else {
row = append(row, "—", "—")
row = append(row, "")
}
}
w.Write(row)
internal/repository/tdengine/store.go
+10 -7
View File
@@ -28,11 +28,18 @@ func (s *Store) Insert(ctx context.Context, p pg.PointRow, d model.Device, value
if reason != nil {
reasonLiteral = sqlString(*reason)
}
q := fmt.Sprintf("INSERT INTO `%s`.`%s` USING `%s`.`collection_data` TAGS(%s,%s,%s) VALUES(%s,%s,%d,%s,%s,%s)", s.Database, TableName(p.ID), s.Database, sqlString(d.ID.String()), sqlString(d.Name), sqlString(p.DataType), sqlString(ts.Format("2006-01-02 15:04:05.000")), valueLiteral, quality, reasonLiteral, sqlString(p.ID.String()), sqlString(p.Name))
q := fmt.Sprintf("INSERT INTO `%s`.`%s` USING `%s`.`collection_data` TAGS(%s,%s,%s) VALUES(%s,%s,%d,%s,%s,%s)", s.Database, TableName(p.ID), s.Database, sqlString(d.ID.String()), sqlString(d.Name), sqlString(p.DataType), sqlString(tdTimeLiteral(ts)), valueLiteral, quality, reasonLiteral, sqlString(p.ID.String()), sqlString(p.Name))
_, e := s.DB.ExecContext(ctx, q)
return e
}
func sqlString(v string) string { return "'" + strings.ReplaceAll(v, "'", "''") + "'" }
var shanghai = time.FixedZone("Asia/Shanghai", 8*60*60)
func tdTimeLiteral(t time.Time) string {
return t.In(shanghai).Format("2006-01-02T15:04:05.000-07:00")
}
func (s *Store) SetRetention(ctx context.Context, days int) error {
if days < 1 || days > 730 {
return fmt.Errorf("保留天数超出范围")
@@ -95,8 +102,7 @@ func (s *Store) DropTables(ctx context.Context, ids []uuid.UUID) (int, error) {
return removed, nil
}
func (s *Store) Query(ctx context.Context, id uuid.UUID, start, end time.Time) ([]Sample, error) {
shanghai := time.FixedZone("Asia/Shanghai", 8*60*60)
q := fmt.Sprintf("SELECT ts,`value`,quality,quality_reason FROM `%s`.`%s` WHERE ts >= %s AND ts <= %s ORDER BY ts", s.Database, TableName(id), sqlString(start.In(shanghai).Format("2006-01-02 15:04:05.000")), sqlString(end.In(shanghai).Format("2006-01-02 15:04:05.000")))
q := fmt.Sprintf("SELECT ts,`value`,quality,quality_reason FROM `%s`.`%s` WHERE ts >= %s AND ts <= %s ORDER BY ts", s.Database, TableName(id), sqlString(tdTimeLiteral(start)), sqlString(tdTimeLiteral(end)))
rows, e := s.DB.QueryContext(ctx, q)
if e != nil {
return nil, e
@@ -109,10 +115,7 @@ func (s *Store) Query(ctx context.Context, id uuid.UUID, start, end time.Time) (
if e = rows.Scan(&x.TS, &x.Value, &quality, &x.QualityReason); e != nil {
return nil, e
}
// The REST driver returns TDengine's local wall-clock timestamp with a
// UTC location. Rebuild it in Asia/Shanghai instead of converting the
// instant, otherwise an eight-hour offset would be applied twice.
x.TS = time.Date(x.TS.Year(), x.TS.Month(), x.TS.Day(), x.TS.Hour(), x.TS.Minute(), x.TS.Second(), x.TS.Nanosecond(), shanghai)
x.TS = x.TS.In(shanghai)
if quality == 0 {
x.Quality = "good"
} else {
internal/repository/tdengine/store_test.go
+8
View File
@@ -3,6 +3,7 @@ package tdengine
import (
"github.com/google/uuid"
"testing"
"time"
)
func TestTableName(t *testing.T) {
@@ -16,3 +17,10 @@ func TestSQLStringEscapesQuote(t *testing.T) {
t.Fatalf("unexpected literal: %s", got)
}
}
func TestTDTimeLiteralUsesShanghaiOffset(t *testing.T) {
ts := time.Date(2026, 7, 13, 2, 10, 10, 123000000, time.UTC)
if got := tdTimeLiteral(ts); got != "2026-07-13T10:10:10.123+08:00" {
t.Fatalf("unexpected TDengine timestamp literal: %s", got)
}
}
internal/service/platform/history.go
+33 -10
View File
@@ -21,14 +21,15 @@ type History struct {
}
}
type Series struct {
PointID uuid.UUID `json:"point_id"`
PointName string `json:"point_name"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
Sampled bool `json:"sampled"`
RawCount int `json:"raw_count"`
SampleCount int `json:"sample_count"`
Data []td.Sample `json:"data"`
PointID uuid.UUID `json:"point_id"`
PointName string `json:"point_name"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
HistoryInterval int `json:"history_interval"`
Sampled bool `json:"sampled"`
RawCount int `json:"raw_count"`
SampleCount int `json:"sample_count"`
Data []td.Sample `json:"data"`
}
func (h *History) Tree(ctx context.Context) ([]map[string]any, error) {
@@ -108,13 +109,35 @@ func (h *History) Query(ctx context.Context, ids []uuid.UUID, start, end time.Ti
return nil, e
}
raw := len(data)
if raw > max {
data = withHistoryGaps(data, time.Duration(p.HistoryInterval)*time.Minute)
sampled := len(data) > max
if sampled {
data = minMax(data, max)
}
out = append(out, Series{id, p.Name, p.DataType, p.Unit, raw > len(data), raw, len(data), data})
out = append(out, Series{id, p.Name, p.DataType, p.Unit, p.HistoryInterval, sampled, raw, len(data), data})
}
return out, nil
}
func withHistoryGaps(data []td.Sample, interval time.Duration) []td.Sample {
if len(data) < 2 || interval <= 0 {
return data
}
threshold := interval * 3
out := make([]td.Sample, 0, len(data))
out = append(out, data[0])
for i := 1; i < len(data); i++ {
prev := data[i-1]
current := data[i]
gap := current.TS.Sub(prev.TS)
if gap > threshold {
out = append(out, td.Sample{TS: prev.TS.Add(gap / 2), Quality: "none"})
}
out = append(out, current)
}
return out
}
func minMax(data []td.Sample, max int) []td.Sample {
if len(data) <= max {
return data
internal/service/platform/history_test.go
+47
View File
@@ -0,0 +1,47 @@
package platform
import (
td "aquacontrolai/internal/repository/tdengine"
"testing"
"time"
)
func sampleAt(ts time.Time, value float64) td.Sample {
return td.Sample{TS: ts, Value: &value, Quality: "good"}
}
func TestWithHistoryGapsInsertsNullBreakpointForLongOutage(t *testing.T) {
start := time.Date(2026, 7, 13, 1, 59, 0, 0, time.UTC)
data := []td.Sample{
sampleAt(start, 1),
sampleAt(start.Add(time.Minute), 2),
sampleAt(start.Add(6*time.Hour), 3),
}
got := withHistoryGaps(data, time.Minute)
if len(got) != 4 {
t.Fatalf("expected gap breakpoint, got %d rows", len(got))
}
if got[2].Value != nil || got[2].Quality != "none" {
t.Fatalf("expected null gap marker, got value=%v quality=%q", got[2].Value, got[2].Quality)
}
if !got[2].TS.After(got[1].TS) || !got[2].TS.Before(got[3].TS) {
t.Fatalf("gap marker should be inside outage: marker=%v before=%v after=%v", got[2].TS, got[1].TS, got[3].TS)
}
}
func TestWithHistoryGapsAllowsExpectedIntervalJitter(t *testing.T) {
start := time.Date(2026, 7, 13, 1, 59, 0, 0, time.UTC)
data := []td.Sample{
sampleAt(start, 1),
sampleAt(start.Add(2*time.Minute), 2),
sampleAt(start.Add(3*time.Minute), 3),
}
got := withHistoryGaps(data, time.Minute)
if len(got) != len(data) {
t.Fatalf("expected no gap breakpoint for short jitter, got %d rows", len(got))
}
}