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

qq

Browse Source
This commit is contained in:
qsc
2026-07-13 00:44:09 +08:00
parent 34ec65c8b5
commit ef96af7f21
22 changed files with 1827 additions and 587 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/router.go
+38 -40
View File
@@ -57,6 +57,7 @@ func NewRouter(h *Handler) *gin.Engine {
v.POST("/write-points/:id/write", h.writePoint)
v.GET("/write-logs", h.logs)
v.GET("/history/tree", h.tree)
v.POST("/history/archive/cleanup", h.cleanupHistoryArchive)
v.POST("/history/query", h.historyQuery)
v.POST("/history/query-table", h.historyTable)
v.POST("/history/export", h.exportHistory)
@@ -182,8 +183,10 @@ func (h *Handler) points(kind string) gin.HandlerFunc {
if v := c.Query("data_type"); v != "" && p.DataType != v {
continue
}
if v := c.Query("enabled"); v != "" && strconv.FormatBool(p.Enabled) != v {
continue
if kind == "collection" {
if v := c.Query("enabled"); v != "" && strconv.FormatBool(p.Enabled) != v {
continue
}
}
if kind == "write" {
if v := c.Query("write_enabled"); v != "" && strconv.FormatBool(p.WriteEnabled) != v {
@@ -332,9 +335,9 @@ func (h *Handler) exportConfig(kind string) gin.HandlerFunc {
w.Write([]string{p.Name, p.GroupName, p.DeviceName, p.Address, p.DataType, stringValue(p.Unit), strconv.Itoa(p.CollectInterval), strings.ToUpper(strconv.FormatBool(p.StoreHistory)), strconv.Itoa(p.HistoryInterval), strings.ToUpper(strconv.FormatBool(p.Enabled))})
}
} else {
w.Write([]string{"name", "group_name", "device_name", "address", "data_type", "unit", "enabled", "write_enabled", "readback_tolerance"})
w.Write([]string{"name", "group_name", "device_name", "address", "data_type", "unit", "write_enabled"})
for _, p := range items {
w.Write([]string{p.Name, p.GroupName, p.DeviceName, p.Address, p.DataType, stringValue(p.Unit), strings.ToUpper(strconv.FormatBool(p.Enabled)), strings.ToUpper(strconv.FormatBool(p.WriteEnabled)), strconv.FormatFloat(p.ReadbackTolerance, 'g', -1, 64)})
w.Write([]string{p.Name, p.GroupName, p.DeviceName, p.Address, p.DataType, stringValue(p.Unit), strings.ToUpper(strconv.FormatBool(p.WriteEnabled))})
}
}
}
@@ -345,12 +348,6 @@ func stringValue(v *string) string {
}
return *v
}
func floatPtr(v *float64) string {
if v == nil {
return ""
}
return strconv.FormatFloat(*v, 'g', -1, 64)
}
func (h *Handler) importConfig(kind string) gin.HandlerFunc {
return func(c *gin.Context) {
file, e := c.FormFile("file")
@@ -419,9 +416,13 @@ func (h *Handler) importRow(ctx context.Context, kind string, head map[string]in
}
existing, e := h.Platform.FindPointByName(ctx, kind, get("name"))
created := platform.IsNotFound(e)
enabled, be := parseCSVBool(get("enabled"), true)
if be != nil {
return false, be
enabled := true
if kind == "collection" {
var be error
enabled, be = parseCSVBool(get("enabled"), true)
if be != nil {
return false, be
}
}
unit := get("unit")
p := postgres.PointRow{ID: uuid.Nil, Name: get("name"), GroupName: get("group_name"), DeviceID: device.ID, Enabled: enabled, Address: get("address"), DataType: get("data_type"), Unit: &unit}
@@ -434,7 +435,6 @@ func (h *Handler) importRow(ctx context.Context, kind string, head map[string]in
p.HistoryInterval = parseInt(get("history_interval"), 1)
} else {
p.WriteEnabled, _ = parseCSVBool(get("write_enabled"), false)
p.ReadbackTolerance = parseFloat(get("readback_tolerance"), .0001)
}
return created, h.Platform.SavePoint(ctx, kind, &p)
}
@@ -448,16 +448,6 @@ func parseInt(v string, d int) int {
}
return n
}
func parseFloat(v string, d float64) float64 {
if v == "" {
return d
}
n, e := strconv.ParseFloat(v, 64)
if e != nil {
return d
}
return n
}
func parseCSVBool(v string, d bool) (bool, error) {
if v == "" {
return d, nil
@@ -467,19 +457,18 @@ func parseCSVBool(v string, d bool) (bool, error) {
func (h *Handler) savePoint(kind string, update bool) gin.HandlerFunc {
return func(c *gin.Context) {
var p struct {
ID uuid.UUID `json:"-"`
Name string `json:"name"`
GroupName string `json:"group_name"`
DeviceID uuid.UUID `json:"device_id"`
Enabled *bool `json:"enabled"`
WriteEnabled bool `json:"write_enabled"`
Address string `json:"address"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
CollectInterval int `json:"collect_interval"`
StoreHistory *bool `json:"store_history"`
HistoryInterval int `json:"history_interval"`
ReadbackTolerance float64 `json:"readback_tolerance"`
ID uuid.UUID `json:"-"`
Name string `json:"name"`
GroupName string `json:"group_name"`
DeviceID uuid.UUID `json:"device_id"`
Enabled *bool `json:"enabled"`
WriteEnabled bool `json:"write_enabled"`
Address string `json:"address"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
CollectInterval int `json:"collect_interval"`
StoreHistory *bool `json:"store_history"`
HistoryInterval int `json:"history_interval"`
}
if e := c.ShouldBindJSON(&p); e != nil {
response.Error(c, 400, 41001, "参数格式错误", nil)
@@ -508,10 +497,10 @@ func (h *Handler) savePoint(kind string, update bool) gin.HandlerFunc {
if p.HistoryInterval == 0 {
p.HistoryInterval = 1
}
if p.ReadbackTolerance == 0 {
p.ReadbackTolerance = .0001
if kind == "write" {
enabled = false
}
row := postgres.PointRow{ID: id, Name: p.Name, GroupName: p.GroupName, DeviceID: p.DeviceID, Enabled: enabled, WriteEnabled: p.WriteEnabled, Address: p.Address, DataType: p.DataType, Unit: p.Unit, CollectInterval: p.CollectInterval, StoreHistory: storeHistory, HistoryInterval: p.HistoryInterval, ReadbackTolerance: p.ReadbackTolerance}
row := postgres.PointRow{ID: id, Name: p.Name, GroupName: p.GroupName, DeviceID: p.DeviceID, Enabled: enabled, WriteEnabled: p.WriteEnabled, Address: p.Address, DataType: p.DataType, Unit: p.Unit, CollectInterval: p.CollectInterval, StoreHistory: storeHistory, HistoryInterval: p.HistoryInterval, PointKind: kind}
if e := h.Platform.SavePoint(c, kind, &row); e != nil {
code := 41001
if kind == "write" {
@@ -619,6 +608,15 @@ func (h *Handler) tree(c *gin.Context) {
response.OK(c, map[string]any{"tree": tree})
}
func (h *Handler) cleanupHistoryArchive(c *gin.Context) {
count, e := h.History.CleanupDeletedArchives(c)
if e != nil {
serverError(c, e)
return
}
response.OK(c, map[string]int{"deleted_points": count})
}
type historyRequest struct {
PointIDs []uuid.UUID `json:"point_ids"`
StartTime time.Time `json:"start_time"`
internal/engine/collector/engine.go
+1
View File
@@ -119,6 +119,7 @@ func (e *Engine) collect(ctx context.Context, p pg.PointRow) {
value = &v
quality = 0
reason = nil
e.manager.MarkConnected(p.DeviceID)
} else {
reasonText = "read_error"
reason = &reasonText
internal/engine/collector/manager.go
+11 -1
View File
@@ -100,8 +100,18 @@ func (m *Manager) Times(id uuid.UUID) (online, offline *time.Time) {
}
func (m *Manager) MarkDisconnected(id uuid.UUID) {
m.mu.Lock()
if m.statuses[id] != "disconnected" {
m.lastOffline[id] = time.Now()
}
m.statuses[id] = "disconnected"
m.lastOffline[id] = time.Now()
m.mu.Unlock()
}
func (m *Manager) MarkConnected(id uuid.UUID) {
m.mu.Lock()
if m.statuses[id] != "connected" {
m.lastOnline[id] = time.Now()
}
m.statuses[id] = "connected"
m.mu.Unlock()
}
func (m *Manager) setStatus(id uuid.UUID, s string) {
internal/engine/collector/manager_test.go
+50
View File
@@ -0,0 +1,50 @@
package collector
import (
"github.com/google/uuid"
"testing"
"time"
)
func TestManagerStatusRecoversAfterSuccessfulRead(t *testing.T) {
m := NewManager(nil, nil)
id := uuid.New()
m.MarkConnected(id)
online, offline := m.Times(id)
if m.Status(id) != "connected" || online == nil || offline != nil {
t.Fatalf("expected connected status with online timestamp, got status=%q online=%v offline=%v", m.Status(id), online, offline)
}
firstOnline := *online
m.MarkDisconnected(id)
if m.Status(id) != "disconnected" {
t.Fatalf("expected disconnected status, got %q", m.Status(id))
}
_, offline = m.Times(id)
if offline == nil {
t.Fatal("expected offline timestamp after disconnect")
}
time.Sleep(2 * time.Millisecond)
m.MarkConnected(id)
online, offline = m.Times(id)
if m.Status(id) != "connected" || online == nil || offline == nil {
t.Fatalf("expected recovered connected status, got status=%q online=%v offline=%v", m.Status(id), online, offline)
}
if !online.After(firstOnline) {
t.Fatalf("expected reconnect to refresh online timestamp: first=%v current=%v", firstOnline, *online)
}
}
func TestManagerRepeatedConnectedDoesNotRefreshOnlineTimestamp(t *testing.T) {
m := NewManager(nil, nil)
id := uuid.New()
m.MarkConnected(id)
first, _ := m.Times(id)
m.MarkConnected(id)
second, _ := m.Times(id)
if first == nil || second == nil || !first.Equal(*second) {
t.Fatalf("repeated connected mark changed online timestamp: first=%v second=%v", first, second)
}
}
internal/engine/writer/writer.go
+7 -3
View File
@@ -19,9 +19,13 @@ type Result struct {
TS time.Time
}
// REAL values are compared with a fixed protocol precision. The tolerance is
// intentionally not part of the write-point configuration anymore.
const realReadbackPrecision = 0.0001
func (e *Engine) Execute(ctx context.Context, p pg.PointRow, value float64) (Result, error) {
if !p.Enabled || !p.WriteEnabled {
return Result{}, errors.New("写入点未启用或写入开关关闭")
if !p.WriteEnabled {
return Result{}, errors.New("写入点未允许写入")
}
c, err := e.Manager.Connection(ctx, p.DeviceID)
if err != nil {
@@ -39,7 +43,7 @@ func (e *Engine) Execute(ctx context.Context, p pg.PointRow, value float64) (Res
}
ok := actual == value
if dt == protocol.Real {
ok = math.Abs(actual-value) <= p.ReadbackTolerance
ok = math.Abs(actual-value) <= realReadbackPrecision
}
if ok {
return Result{value, actual, time.Now()}, nil
internal/model/model.go
-1
View File
@@ -35,6 +35,5 @@ type Point struct {
HistoryInterval int
HistoryStartedAt *time.Time
WriteEnabled bool
ReadbackTolerance float64
CreatedAt, UpdatedAt time.Time
}
internal/repository/postgres/store.go
+113 -38
View File
@@ -67,7 +67,7 @@ func (s *Store) DeleteDevice(ctx context.Context, id uuid.UUID) error {
if tag.RowsAffected() == 0 {
return pgx.ErrNoRows
}
for _, q := range []string{`UPDATE collection_points SET deleted=TRUE,enabled=FALSE,updated_at=NOW() WHERE device_id=$1 AND deleted=FALSE`, `UPDATE write_points SET deleted=TRUE,enabled=FALSE,write_enabled=FALSE,updated_at=NOW() WHERE device_id=$1 AND deleted=FALSE`} {
for _, q := range []string{`UPDATE collection_points SET deleted=TRUE,enabled=FALSE,updated_at=NOW() WHERE device_id=$1 AND deleted=FALSE`, `UPDATE write_points SET deleted=TRUE,write_enabled=FALSE,updated_at=NOW() WHERE device_id=$1 AND deleted=FALSE`} {
if _, e = tx.Exec(ctx, q, id); e != nil {
return e
}
@@ -76,25 +76,37 @@ func (s *Store) DeleteDevice(ctx context.Context, id uuid.UUID) error {
}
type PointRow struct {
ID uuid.UUID `json:"id"`
Name string `json:"name"`
GroupName string `json:"group_name"`
DeviceID uuid.UUID `json:"device_id"`
DeviceName string `json:"device_name"`
ProtocolType string `json:"protocol_type"`
Address string `json:"address"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
Enabled bool `json:"enabled"`
CollectInterval int `json:"collect_interval,omitempty"`
StoreHistory bool `json:"store_history,omitempty"`
HistoryInterval int `json:"history_interval,omitempty"`
HistoryStartedAt *time.Time `json:"history_started_at,omitempty"`
WriteEnabled bool `json:"write_enabled,omitempty"`
ReadbackTolerance float64 `json:"readback_tolerance,omitempty"`
LatestValue any `json:"latest_value"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID uuid.UUID `json:"id"`
Name string `json:"name"`
GroupName string `json:"group_name"`
DeviceID uuid.UUID `json:"device_id"`
DeviceName string `json:"device_name"`
ProtocolType string `json:"protocol_type"`
Address string `json:"address"`
DataType string `json:"data_type"`
Unit *string `json:"unit"`
Enabled bool `json:"enabled"`
CollectInterval int `json:"collect_interval,omitempty"`
StoreHistory bool `json:"store_history,omitempty"`
HistoryInterval int `json:"history_interval,omitempty"`
HistoryStartedAt *time.Time `json:"history_started_at,omitempty"`
WriteEnabled bool `json:"write_enabled,omitempty"`
ReadbackValue any `json:"readback_value"`
LatestValue any `json:"latest_value"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
PointKind string `json:"-"`
}
func (p PointRow) MarshalJSON() ([]byte, error) {
type alias PointRow
if p.PointKind == "write" {
return json.Marshal(struct {
alias
Enabled any `json:"enabled,omitempty"`
}{alias: alias(p)})
}
return json.Marshal(alias(p))
}
func (s *Store) GetPoint(ctx context.Context, kind string, id uuid.UUID) (PointRow, error) {
@@ -114,29 +126,26 @@ func (s *Store) SavePoint(ctx context.Context, kind string, p *PointRow) error {
if p.ID == uuid.Nil {
return s.DB.QueryRow(ctx, `INSERT INTO collection_points(name,group_name,device_id,enabled,address,data_type,unit,collect_interval,store_history,history_interval) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) RETURNING id,created_at,updated_at`, p.Name, p.GroupName, p.DeviceID, p.Enabled, p.Address, p.DataType, p.Unit, p.CollectInterval, p.StoreHistory, p.HistoryInterval).Scan(&p.ID, &p.CreatedAt, &p.UpdatedAt)
}
tag, e := s.DB.Exec(ctx, `UPDATE collection_points SET name=$2,group_name=$3,device_id=$4,enabled=$5,address=$6,data_type=$7,unit=$8,collect_interval=$9,store_history=$10,history_interval=$11,updated_at=NOW() WHERE id=$1 AND deleted=FALSE AND (history_started_at IS NULL OR (device_id=$4 AND data_type=$7))`, p.ID, p.Name, p.GroupName, p.DeviceID, p.Enabled, p.Address, p.DataType, p.Unit, p.CollectInterval, p.StoreHistory, p.HistoryInterval)
tag, e := s.DB.Exec(ctx, `UPDATE collection_points SET name=$2,group_name=$3,device_id=$4,enabled=$5,address=$6,data_type=$7,unit=$8,collect_interval=$9,store_history=$10,history_interval=$11,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`, p.ID, p.Name, p.GroupName, p.DeviceID, p.Enabled, p.Address, p.DataType, p.Unit, p.CollectInterval, p.StoreHistory, p.HistoryInterval)
if e == nil && tag.RowsAffected() == 0 {
return pgx.ErrNoRows
}
return e
}
if p.ID == uuid.Nil {
return s.DB.QueryRow(ctx, `INSERT INTO write_points(name,group_name,device_id,enabled,write_enabled,address,data_type,unit,readback_tolerance) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) RETURNING id,created_at,updated_at`, p.Name, p.GroupName, p.DeviceID, p.Enabled, p.WriteEnabled, p.Address, p.DataType, p.Unit, p.ReadbackTolerance).Scan(&p.ID, &p.CreatedAt, &p.UpdatedAt)
return s.DB.QueryRow(ctx, `INSERT INTO write_points(name,group_name,device_id,write_enabled,address,data_type,unit) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING id,created_at,updated_at`, p.Name, p.GroupName, p.DeviceID, p.WriteEnabled, p.Address, p.DataType, p.Unit).Scan(&p.ID, &p.CreatedAt, &p.UpdatedAt)
}
tag, e := s.DB.Exec(ctx, `UPDATE write_points SET name=$2,group_name=$3,device_id=$4,enabled=$5,write_enabled=$6,address=$7,data_type=$8,unit=$9,readback_tolerance=$10,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`, p.ID, p.Name, p.GroupName, p.DeviceID, p.Enabled, p.WriteEnabled, p.Address, p.DataType, p.Unit, p.ReadbackTolerance)
tag, e := s.DB.Exec(ctx, `UPDATE write_points SET name=$2,group_name=$3,device_id=$4,write_enabled=$5,address=$6,data_type=$7,unit=$8,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`, p.ID, p.Name, p.GroupName, p.DeviceID, p.WriteEnabled, p.Address, p.DataType, p.Unit)
if e == nil && tag.RowsAffected() == 0 {
return pgx.ErrNoRows
}
return e
}
func (s *Store) DeletePoint(ctx context.Context, kind string, id uuid.UUID) error {
table := "collection_points"
extra := ""
q := `UPDATE collection_points SET deleted=TRUE,enabled=FALSE,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`
if kind == "write" {
table = "write_points"
extra = ",write_enabled=FALSE"
q = `UPDATE write_points SET deleted=TRUE,write_enabled=FALSE,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`
}
q := fmt.Sprintf(`UPDATE %s SET deleted=TRUE,enabled=FALSE%s,updated_at=NOW() WHERE id=$1 AND deleted=FALSE`, table, extra)
tag, e := s.DB.Exec(ctx, q, id)
if e == nil && tag.RowsAffected() == 0 {
return pgx.ErrNoRows
@@ -166,7 +175,10 @@ func (s *Store) FindPointByName(ctx context.Context, kind, name string) (PointRo
return PointRow{}, pgx.ErrNoRows
}
func (s *Store) Groups(ctx context.Context) ([]map[string]any, error) {
rows, e := s.DB.Query(ctx, `SELECT g.name,COUNT(p.id) FROM collection_groups g LEFT JOIN collection_points p ON p.group_name=g.name AND p.deleted=FALSE GROUP BY g.name ORDER BY g.name`)
rows, e := s.DB.Query(ctx, `SELECT g.name,
(SELECT COUNT(*) FROM collection_points cp WHERE cp.group_name=g.name AND cp.deleted=FALSE),
(SELECT COUNT(*) FROM write_points wp WHERE wp.group_name=g.name AND wp.deleted=FALSE)
FROM collection_groups g ORDER BY g.name`)
if e != nil {
return nil, e
}
@@ -174,11 +186,16 @@ func (s *Store) Groups(ctx context.Context) ([]map[string]any, error) {
out := []map[string]any{}
for rows.Next() {
var name string
var count int
if e = rows.Scan(&name, &count); e != nil {
var collectionCount, writeCount int
if e = rows.Scan(&name, &collectionCount, &writeCount); e != nil {
return nil, e
}
out = append(out, map[string]any{"name": name, "count": count})
out = append(out, map[string]any{
"name": name,
"count": collectionCount,
"collection_count": collectionCount,
"write_count": writeCount,
})
}
return out, rows.Err()
}
@@ -198,6 +215,9 @@ func (s *Store) UpdateGroup(ctx context.Context, oldName, newName string) error
if _, e = tx.Exec(ctx, `UPDATE collection_points SET group_name=$2,updated_at=NOW() WHERE group_name=$1 AND deleted=FALSE`, oldName, newName); e != nil {
return e
}
if _, e = tx.Exec(ctx, `UPDATE write_points SET group_name=$2,updated_at=NOW() WHERE group_name=$1 AND deleted=FALSE`, oldName, newName); e != nil {
return e
}
if _, e = tx.Exec(ctx, `DELETE FROM collection_groups WHERE name=$1`, oldName); e != nil {
return e
}
@@ -212,7 +232,13 @@ func (s *Store) DeleteGroup(ctx context.Context, name string) error {
return e
}
defer tx.Rollback(ctx)
if _, e = tx.Exec(ctx, `UPDATE collection_points SET deleted=TRUE,enabled=FALSE,updated_at=NOW() WHERE group_name=$1 AND deleted=FALSE`, name); e != nil {
if _, e = tx.Exec(ctx, `INSERT INTO collection_groups(name) VALUES('default') ON CONFLICT(name) DO NOTHING`); e != nil {
return e
}
if _, e = tx.Exec(ctx, `UPDATE collection_points SET group_name='default',updated_at=NOW() WHERE group_name=$1 AND deleted=FALSE`, name); e != nil {
return e
}
if _, e = tx.Exec(ctx, `UPDATE write_points SET group_name='default',updated_at=NOW() WHERE group_name=$1 AND deleted=FALSE`, name); e != nil {
return e
}
tag, e := tx.Exec(ctx, `DELETE FROM collection_groups WHERE name=$1`, name)
@@ -236,16 +262,19 @@ func (s *Store) SetRetention(ctx context.Context, days int) error {
func (s *Store) ListPoints(ctx context.Context, kind, keyword string, archived bool) ([]PointRow, error) {
table := "collection_points"
cols := `p.collect_interval,p.store_history,p.history_interval,p.history_started_at,FALSE,0`
enabledColumn := "p.enabled"
cols := `p.collect_interval,p.store_history,p.history_interval,p.history_started_at,FALSE,NULL::text`
if kind == "write" {
table = "write_points"
cols = `NULL::double precision,NULL::double precision,0,FALSE,0,NULL::timestamptz,p.write_enabled,p.readback_tolerance`
enabledColumn = "FALSE"
cols = `0::int,FALSE::bool,0,NULL::timestamptz,p.write_enabled,
(SELECT wl.readback_value FROM write_logs wl WHERE wl.point_id=p.id ORDER BY wl.created_at DESC LIMIT 1)`
}
deleted := "p.deleted=FALSE AND d.deleted=FALSE"
if archived {
deleted = "TRUE"
}
q := fmt.Sprintf(`SELECT p.id,p.name,p.group_name,p.device_id,d.name,d.protocol_type,p.address,p.data_type,p.unit,p.enabled,%s,p.created_at,p.updated_at FROM %s p JOIN devices d ON d.id=p.device_id WHERE %s AND ($1='' OR p.name ILIKE '%%'||$1||'%%' OR p.group_name ILIKE '%%'||$1||'%%') ORDER BY p.group_name,p.name LIMIT 1000`, cols, table, deleted)
q := fmt.Sprintf(`SELECT p.id,p.name,p.group_name,p.device_id,d.name,d.protocol_type,p.address,p.data_type,p.unit,%s,%s,p.created_at,p.updated_at FROM %s p JOIN devices d ON d.id=p.device_id WHERE %s AND ($1='' OR p.name ILIKE '%%'||$1||'%%' OR p.group_name ILIKE '%%'||$1||'%%') ORDER BY p.group_name,p.name LIMIT 1000`, enabledColumn, cols, table, deleted)
rows, e := s.DB.Query(ctx, q, keyword)
if e != nil {
return nil, e
@@ -253,15 +282,61 @@ func (s *Store) ListPoints(ctx context.Context, kind, keyword string, archived b
defer rows.Close()
var out []PointRow
for rows.Next() {
var p PointRow
if e = rows.Scan(&p.ID, &p.Name, &p.GroupName, &p.DeviceID, &p.DeviceName, &p.ProtocolType, &p.Address, &p.DataType, &p.Unit, &p.Enabled, &p.CollectInterval, &p.StoreHistory, &p.HistoryInterval, &p.HistoryStartedAt, &p.WriteEnabled, &p.ReadbackTolerance, &p.CreatedAt, &p.UpdatedAt); e != nil {
p := PointRow{PointKind: kind}
var rawReadback *string
if e = rows.Scan(&p.ID, &p.Name, &p.GroupName, &p.DeviceID, &p.DeviceName, &p.ProtocolType, &p.Address, &p.DataType, &p.Unit, &p.Enabled, &p.CollectInterval, &p.StoreHistory, &p.HistoryInterval, &p.HistoryStartedAt, &p.WriteEnabled, &rawReadback, &p.CreatedAt, &p.UpdatedAt); e != nil {
return nil, e
}
p.ReadbackValue = parseOptional(p.DataType, rawReadback)
out = append(out, p)
}
return out, rows.Err()
}
// ListDeletedCollectionPointIDs returns collection points whose realtime
// metadata no longer exists. The history cleanup flow uses this list to avoid
// touching disabled points that may be re-enabled later.
func (s *Store) ListDeletedCollectionPointIDs(ctx context.Context) ([]uuid.UUID, error) {
rows, e := s.DB.Query(ctx, `SELECT p.id
FROM collection_points p
LEFT JOIN devices d ON d.id=p.device_id
WHERE p.deleted=TRUE OR d.deleted=TRUE
ORDER BY p.id`)
if e != nil {
return nil, e
}
defer rows.Close()
var out []uuid.UUID
for rows.Next() {
var id uuid.UUID
if e = rows.Scan(&id); e != nil {
return nil, e
}
out = append(out, id)
}
return out, rows.Err()
}
// ListDeviceActivity returns the current realtime availability of every
// device, including logically deleted rows used by the history tree.
func (s *Store) ListDeviceActivity(ctx context.Context) (map[uuid.UUID]bool, error) {
rows, e := s.DB.Query(ctx, `SELECT id,enabled,deleted FROM devices`)
if e != nil {
return nil, e
}
defer rows.Close()
out := make(map[uuid.UUID]bool)
for rows.Next() {
var id uuid.UUID
var enabled, deleted bool
if e = rows.Scan(&id, &enabled, &deleted); e != nil {
return nil, e
}
out[id] = enabled && !deleted
}
return out, rows.Err()
}
type WriteLog struct {
ID uuid.UUID `json:"id"`
PointID uuid.UUID `json:"point_id"`
internal/repository/tdengine/store.go
+22
View File
@@ -76,6 +76,24 @@ func (s *Store) HasData(ctx context.Context, id uuid.UUID) bool {
var n int
return s.DB.QueryRowContext(ctx, q).Scan(&n) == nil && n > 0
}
// DropTables removes the per-point child tables for the supplied archived
// points. The caller is responsible for selecting only points that have no
// realtime metadata; TableName validates every derived identifier.
func (s *Store) DropTables(ctx context.Context, ids []uuid.UUID) (int, error) {
removed := 0
for _, id := range ids {
if !s.HasData(ctx, id) {
continue
}
q := fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", s.Database, TableName(id))
if _, e := s.DB.ExecContext(ctx, q); e != nil {
return removed, e
}
removed++
}
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")))
@@ -91,6 +109,10 @@ 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)
if quality == 0 {
x.Quality = "good"
} else {
internal/service/platform/history.go
+27 -2
View File
@@ -36,9 +36,21 @@ func (h *History) Tree(ctx context.Context) ([]map[string]any, error) {
if e != nil {
return nil, e
}
deletedIDs, e := h.PG.ListDeletedCollectionPointIDs(ctx)
if e != nil {
return nil, e
}
deviceActive, e := h.PG.ListDeviceActivity(ctx)
if e != nil {
return nil, e
}
deleted := make(map[uuid.UUID]bool, len(deletedIDs))
for _, id := range deletedIDs {
deleted[id] = true
}
groups := map[string][]map[string]any{}
for _, p := range points {
active := p.Enabled && p.StoreHistory
active := p.Enabled && p.StoreHistory && deviceActive[p.DeviceID]
has := h.TD.HasData(ctx, p.ID)
if !active && !has {
continue
@@ -51,7 +63,7 @@ func (h *History) Tree(ctx context.Context) ([]map[string]any, error) {
if h.Collector != nil && life == "active" {
latest = h.Collector.Latest(p.ID)
}
groups[p.GroupName] = append(groups[p.GroupName], map[string]any{"id": p.ID, "name": p.Name, "type": "collection", "data_type": p.DataType, "unit": p.Unit, "history_interval": p.HistoryInterval, "device_id": p.DeviceID, "device_name": p.DeviceName, "group_name": p.GroupName, "lifecycle_status": life, "has_history_data": has, "latest_value": latest})
groups[p.GroupName] = append(groups[p.GroupName], map[string]any{"id": p.ID, "name": p.Name, "type": "collection", "data_type": p.DataType, "unit": p.Unit, "history_interval": p.HistoryInterval, "device_id": p.DeviceID, "device_name": p.DeviceName, "group_name": p.GroupName, "lifecycle_status": life, "has_history_data": has, "can_cleanup": deleted[p.ID], "latest_value": latest})
}
names := make([]string, 0, len(groups))
for n := range groups {
@@ -66,6 +78,16 @@ func (h *History) Tree(ctx context.Context) ([]map[string]any, error) {
tree = append(tree, map[string]any{"id": "internal-data", "name": "内部数据", "type": "reserved", "children": []map[string]any{{"id": "placeholder", "name": "暂无数据", "type": "placeholder", "disabled": true}}})
return tree, nil
}
// CleanupDeletedArchives drops TDengine history tables only for collection
// points that have been logically deleted (or belong to a deleted device).
func (h *History) CleanupDeletedArchives(ctx context.Context) (int, error) {
ids, e := h.PG.ListDeletedCollectionPointIDs(ctx)
if e != nil {
return 0, e
}
return h.TD.DropTables(ctx, ids)
}
func (h *History) Query(ctx context.Context, ids []uuid.UUID, start, end time.Time, max int) ([]Series, error) {
meta, e := h.PG.ListPoints(ctx, "collection", "", true)
if e != nil {
@@ -159,6 +181,9 @@ type TableResult struct {
}
func (h *History) QueryTable(ctx context.Context, ids []uuid.UUID, start, end time.Time, minutes int) (TableResult, error) {
shanghai := time.FixedZone("Asia/Shanghai", 8*60*60)
start = start.In(shanghai)
end = end.In(shanghai)
step := time.Duration(minutes) * time.Minute
times := []time.Time{}
for t := start; !t.After(end); t = t.Add(step) {
internal/service/platform/service.go
-2
View File
@@ -190,8 +190,6 @@ func (s *Service) SavePoint(ctx context.Context, kind string, p *pg.PointRow) er
if p.HistoryInterval < 1 || p.HistoryInterval > 1440 {
return errors.New("history_interval必须为1~1440")
}
} else if p.ReadbackTolerance < 0 || p.ReadbackTolerance > 1000000 {
return errors.New("readback_tolerance超出范围")
}
return s.Store.SavePoint(ctx, kind, p)
}