feat: bootstrap measurement data objects in Redis

- load measurement hierarchy and metadata from PostgreSQL
- generate 3 types Redis hashes
- map measurement type enums to TM, TS, TC, TA, and SP
- validate token uniqueness, mode, type, and JSONB fields
- track initialized hashes and clean up stale measurement keys
- initialize measurement data objects during modelRT startup

constants/data-object.go

@@ -17,3 +17,16 @@ const (
 	// MeasurementModeAutomatic indicates that the measurement runs automatically.
 	MeasurementModeAutomatic int16 = 1
 )
+
+const (
+	// MeasurementTypeTelemetry represents TM (遥测).
+	MeasurementTypeTelemetry int16 = 0
+	// MeasurementTypeTelesignal represents TS (遥信).
+	MeasurementTypeTelesignal int16 = 1
+	// MeasurementTypeTelecommand represents TC (遥控).
+	MeasurementTypeTelecommand int16 = 2
+	// MeasurementTypeTeleadjusting represents TA (遥调).
+	MeasurementTypeTeleadjusting int16 = 3
+	// MeasurementTypeSetpoint represents SP (定值).
+	MeasurementTypeSetpoint int16 = 4
+)

constants/redis.go

@@ -8,4 +8,8 @@ const (
 	// RedisParameterDataObjectKeySet tracks parameter hashes created during
 	// startup so stale parameter data-object keys can be removed safely.
 	RedisParameterDataObjectKeySet = "modelrt:parameter-data-object:keys"
+
+	// RedisMeasurementDataObjectKeySet tracks measurement hashes created during
+	// startup so stale measurement data-object keys can be removed safely.
+	RedisMeasurementDataObjectKeySet = "modelrt:measurement-data-object:keys"
 )

database/query_measurement_initialization.go

@@ -0,0 +1,73 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"modelRT/constants"
+	"modelRT/model"
+	modelsql "modelRT/sql"
+
+	"gorm.io/gorm"
+)
+
+// QueryMeasurementInitializationRecords loads every measurement that can be
+// addressed through the seven-part, four-part, and two-part token forms.
+func QueryMeasurementInitializationRecords(ctx context.Context, db *gorm.DB) ([]model.MeasurementInitializationRecord, error) {
+	if db == nil {
+		return nil, fmt.Errorf("postgres client is nil")
+	}
+
+	var records []model.MeasurementInitializationRecord
+	if err := db.WithContext(ctx).
+		Raw(compactMeasurementSQL(modelsql.MeasurementInitializationRows)).
+		Scan(&records).Error; err != nil {
+		return nil, fmt.Errorf("query measurement initialization records: %w", err)
+	}
+	if err := validateMeasurementInitializationRecords(records); err != nil {
+		return nil, fmt.Errorf("validate measurement initialization records: %w", err)
+	}
+	return records, nil
+}
+
+func validateMeasurementInitializationRecords(records []model.MeasurementInitializationRecord) error {
+	for _, record := range records {
+		if record.MeasurementID <= 0 {
+			return fmt.Errorf("measurement %q has invalid id %d", record.MeasurementTag, record.MeasurementID)
+		}
+		if record.ComponentUUID == "" {
+			return fmt.Errorf("measurement %q has empty component uuid", record.MeasurementTag)
+		}
+		if record.GridTag == "" ||
+			record.ZoneTag == "" ||
+			record.StationTag == "" ||
+			record.ComponentNSPath == "" ||
+			record.ComponentTag == "" ||
+			record.MeasurementTag == "" {
+			return fmt.Errorf("measurement %d contains an empty data-object token segment", record.MeasurementID)
+		}
+		if record.MeasurementMode != constants.MeasurementModeManual &&
+			record.MeasurementMode != constants.MeasurementModeAutomatic {
+			return fmt.Errorf(
+				"measurement %q mode must be %d or %d, got %d",
+				record.MeasurementTag,
+				constants.MeasurementModeManual,
+				constants.MeasurementModeAutomatic,
+				record.MeasurementMode,
+			)
+		}
+		if _, err := model.MeasurementTypeString(record.MeasurementType); err != nil {
+			return fmt.Errorf("measurement %q: %w", record.MeasurementTag, err)
+		}
+		if record.MeasurementDataSource == nil {
+			return fmt.Errorf("measurement %q has null data_source", record.MeasurementTag)
+		}
+		if record.MeasurementEventPlan == nil {
+			return fmt.Errorf("measurement %q has null event_plan", record.MeasurementTag)
+		}
+		if record.MeasurementBinding == nil {
+			return fmt.Errorf("measurement %q has null binding", record.MeasurementTag)
+		}
+	}
+	return nil
+}

database/query_measurement_initialization_test.go

@@ -0,0 +1,131 @@
+package database
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"modelRT/model"
+	modelsql "modelRT/sql"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/postgres"
+	"gorm.io/gorm"
+)
+
+func TestMeasurementInitializationSQLJoinsRequiredHierarchy(t *testing.T) {
+	statement := compactMeasurementSQL(modelsql.MeasurementInitializationRows)
+
+	assert.Contains(t, statement, "component.station_id = station.id")
+	assert.Contains(t, statement, "measurement.component_uuid = component.global_uuid")
+	assert.Contains(t, statement, "bay.bay_uuid = measurement.bay_uuid")
+	assert.Contains(t, statement, "measurement.type AS measurement_type")
+	assert.Contains(t, statement, "measurement.tag <> ''")
+	assert.NotContains(t, strings.ToLower(statement), "dev_")
+}
+
+func TestQueryMeasurementInitializationRecords(t *testing.T) {
+	sqlDB, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sqlDB.Close() })
+
+	db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{})
+	require.NoError(t, err)
+
+	mock.ExpectQuery(`(?s)SELECT.*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*INNER JOIN public\.measurement.*INNER JOIN public\.bay`).
+		WillReturnRows(measurementInitializationRows().
+			AddRow(
+				"grid000",
+				"zone000",
+				"station000",
+				"component-uuid",
+				"nspath",
+				"component",
+				int64(10),
+				"IA_rms",
+				"A相保护电流有效值",
+				int16(0),
+				int16(1),
+				1,
+				`{"type":1,"io_address":{"channel":"TM1"}}`,
+				`{}`,
+				`{"ct":{"ratio":1250}}`,
+			))
+
+	records, err := QueryMeasurementInitializationRecords(context.Background(), db)
+	require.NoError(t, err)
+	require.Len(t, records, 1)
+	assert.Equal(t, int64(10), records[0].MeasurementID)
+	assert.Equal(t, "IA_rms", records[0].MeasurementTag)
+	assert.Equal(t, int16(0), records[0].MeasurementType)
+	assert.Equal(t, float64(1), records[0].MeasurementDataSource["type"])
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestValidateMeasurementInitializationRecords(t *testing.T) {
+	record := validMeasurementInitializationRecord()
+
+	invalidMode := record
+	invalidMode.MeasurementMode = 3
+	err := validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{invalidMode})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "mode must be 0 or 1")
+
+	emptySegment := record
+	emptySegment.ComponentNSPath = ""
+	err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{emptySegment})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "empty data-object token segment")
+
+	nullDataSource := record
+	nullDataSource.MeasurementDataSource = nil
+	err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{nullDataSource})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "null data_source")
+
+	invalidType := record
+	invalidType.MeasurementType = -1
+	err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{invalidType})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unsupported measurement type -1")
+}
+
+func measurementInitializationRows() *sqlmock.Rows {
+	return sqlmock.NewRows([]string{
+		"grid_tag",
+		"zone_tag",
+		"station_tag",
+		"component_uuid",
+		"component_nspath",
+		"component_tag",
+		"measurement_id",
+		"measurement_tag",
+		"measurement_name",
+		"measurement_type",
+		"measurement_mode",
+		"measurement_size",
+		"measurement_data_source",
+		"measurement_event_plan",
+		"measurement_binding",
+	})
+}
+
+func validMeasurementInitializationRecord() model.MeasurementInitializationRecord {
+	return model.MeasurementInitializationRecord{
+		GridTag:               "grid",
+		ZoneTag:               "zone",
+		StationTag:            "station",
+		ComponentUUID:         "component-uuid",
+		ComponentNSPath:       "nspath",
+		ComponentTag:          "component",
+		MeasurementID:         1,
+		MeasurementTag:        "measurement",
+		MeasurementType:       0,
+		MeasurementMode:       1,
+		MeasurementDataSource: map[string]any{},
+		MeasurementEventPlan:  map[string]any{},
+		MeasurementBinding:    map[string]any{},
+	}
+}

handler/data_object_attribute_query.go

@@ -294,7 +294,7 @@ func buildMeasurementAttributeValue(
 	case "meta":
 		return "MEASUREMENT", nil
 	case "type":
-		return model.MeasurementTypeFromDataSource(measurement.DataSource)
+		return model.MeasurementTypeString(measurement.Type)
 	case "name":
 		// The resolved measurement and component prove that token4.token7 exists.
 		return component.NSPath + "." + measurement.Tag, nil

handler/data_object_attribute_query_test.go

@@ -190,6 +190,7 @@ func TestBuildMeasurementAttributeValue(t *testing.T) {
 	measurement := &orm.Measurement{
 		Tag:        "IA_rms",
 		Name:       "A相电流",
+		Type:       0,
 		Mode:       1,
 		Size:       10,
 		DataSource: dataSource,

main.go

@@ -259,6 +259,18 @@ func main() {
 			panic(err)
 		}
 
+		measurementRecords, err := database.QueryMeasurementInitializationRecords(ctx, tx)
+		if err != nil {
+			logger.Error(ctx, "load measurement data objects from postgres failed", "error", err)
+			panic(err)
+		}
+
+		err = model.InitializeMeasurementDataObjects(ctx, measurementRecords)
+		if err != nil {
+			logger.Error(ctx, "initialize measurement data objects failed", "error", err)
+			panic(err)
+		}
+
 		allMeasurement, err := database.GetAllMeasurements(ctx, tx)
 		if err != nil {
 			logger.Error(ctx, "load topologic info from postgres failed", "error", err)

model/measurement_attribute.go

@@ -2,50 +2,27 @@ package model
 
 import (
 	"fmt"
-	"strings"
 
 	"modelRT/constants"
-	"modelRT/orm"
 )
 
-var allowedMeasurementTypes = map[string]struct{}{
+// MeasurementTypeString converts measurement.type from PostgreSQL to the
-	"TM": {},
+// electric-element type stored in the Redis data-object hash.
-	"TS": {},
+func MeasurementTypeString(measurementType int16) (string, error) {
-	"TC": {},
+	switch measurementType {
-	"TA": {},
+	case constants.MeasurementTypeTelemetry:
-	"SP": {},
+		return "TM", nil
+	case constants.MeasurementTypeTelesignal:
+		return "TS", nil
+	case constants.MeasurementTypeTelecommand:
+		return "TC", nil
+	case constants.MeasurementTypeTeleadjusting:
+		return "TA", nil
+	case constants.MeasurementTypeSetpoint:
+		return "SP", nil
+	default:
+		return "", fmt.Errorf("unsupported measurement type %d", measurementType)
 	}
-
-// MeasurementTypeFromDataSource returns the two-character measurement type
-// encoded in a CL3611 channel. Only TM, TS, TC, TA, and SP are valid.
-func MeasurementTypeFromDataSource(dataSource orm.JSONMap) (string, error) {
-	dataSourceType, err := integerJSONValue(dataSource["type"])
-	if err != nil {
-		return "", fmt.Errorf("invalid measurement data_source type: %w", err)
-	}
-	if dataSourceType != constants.DataSourceTypeCL3611 {
-		return "", fmt.Errorf("measurement type requires data_source type %d, got %d", constants.DataSourceTypeCL3611, dataSourceType)
-	}
-
-	ioAddress, ok := dataSource["io_address"].(map[string]any)
-	if !ok {
-		if value, jsonMapOK := dataSource["io_address"].(orm.JSONMap); jsonMapOK {
-			ioAddress = map[string]any(value)
-		} else {
-			return "", fmt.Errorf("measurement data_source io_address is not an object")
-		}
-	}
-
-	channel, ok := ioAddress["channel"].(string)
-	if !ok || len(channel) < 2 {
-		return "", fmt.Errorf("measurement data_source channel must contain at least two characters")
-	}
-
-	measurementType := strings.ToUpper(channel[:2])
-	if _, ok := allowedMeasurementTypes[measurementType]; !ok {
-		return "", fmt.Errorf("unsupported measurement type %q", measurementType)
-	}
-	return measurementType, nil
 }
 
 func integerJSONValue(value any) (int, error) {

model/measurement_attribute_test.go

@@ -1,40 +1,37 @@
 package model
 
 import (
-	"strings"
 	"testing"
 
-	"modelRT/orm"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func TestMeasurementTypeFromDataSource(t *testing.T) {
+func TestMeasurementTypeString(t *testing.T) {
-	for _, measurementType := range []string{"TM", "TS", "TC", "TA", "SP"} {
+	tests := []struct {
-		t.Run(measurementType, func(t *testing.T) {
+		value    int16
-			actual, err := MeasurementTypeFromDataSource(orm.JSONMap{
+		expected string
-				"type": float64(1),
+	}{
-				"io_address": map[string]any{
+		{value: 0, expected: "TM"},
-					"channel": strings.ToLower(measurementType) + "1_test",
+		{value: 1, expected: "TS"},
-				},
+		{value: 2, expected: "TC"},
-			})
+		{value: 3, expected: "TA"},
+		{value: 4, expected: "SP"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.expected, func(t *testing.T) {
+			actual, err := MeasurementTypeString(test.value)
 			require.NoError(t, err)
-			assert.Equal(t, measurementType, actual)
+			assert.Equal(t, test.expected, actual)
 		})
 	}
 }
 
-func TestMeasurementTypeFromDataSourceRejectsInvalidValues(t *testing.T) {
+func TestMeasurementTypeStringRejectsInvalidValues(t *testing.T) {
-	tests := []orm.JSONMap{
+	for _, value := range []int16{-1, 5, 100} {
-		{"type": float64(2), "io_address": map[string]any{"channel": "tm1"}},
+		_, err := MeasurementTypeString(value)
-		{"type": float64(1), "io_address": map[string]any{"channel": "xx1"}},
-		{"type": float64(1), "io_address": map[string]any{"channel": "t"}},
-		{"type": "1", "io_address": map[string]any{"channel": "tm1"}},
-	}
-
-	for _, dataSource := range tests {
-		_, err := MeasurementTypeFromDataSource(dataSource)
 		require.Error(t, err)
+		assert.Contains(t, err.Error(), "unsupported measurement type")
 	}
 }

model/measurement_data_object_init.go

@@ -0,0 +1,238 @@
+package model
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"modelRT/constants"
+	"modelRT/diagram"
+	"modelRT/logger"
+	"modelRT/orm"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const measurementDataObjectPipelineSize = 500
+
+type measurementDataObjectHash struct {
+	Key    string
+	Fields map[string]any
+}
+
+// MeasurementInitializationRecord contains a measurement and the hierarchy
+// needed to create all supported Redis data-object token aliases.
+type MeasurementInitializationRecord struct {
+	GridTag               string      `gorm:"column:grid_tag"`
+	ZoneTag               string      `gorm:"column:zone_tag"`
+	StationTag            string      `gorm:"column:station_tag"`
+	ComponentUUID         string      `gorm:"column:component_uuid"`
+	ComponentNSPath       string      `gorm:"column:component_nspath"`
+	ComponentTag          string      `gorm:"column:component_tag"`
+	MeasurementID         int64       `gorm:"column:measurement_id"`
+	MeasurementTag        string      `gorm:"column:measurement_tag"`
+	MeasurementName       string      `gorm:"column:measurement_name"`
+	MeasurementType       int16       `gorm:"column:measurement_type"`
+	MeasurementMode       int16       `gorm:"column:measurement_mode"`
+	MeasurementSize       int         `gorm:"column:measurement_size"`
+	MeasurementDataSource orm.JSONMap `gorm:"column:measurement_data_source"`
+	MeasurementEventPlan  orm.JSONMap `gorm:"column:measurement_event_plan"`
+	MeasurementBinding    orm.JSONMap `gorm:"column:measurement_binding"`
+}
+
+// InitializeMeasurementDataObjects creates the seven-part, four-part, and
+// token4.token7 Redis hashes for measurements loaded from PostgreSQL.
+func InitializeMeasurementDataObjects(ctx context.Context, records []MeasurementInitializationRecord) error {
+	hashes, err := buildMeasurementDataObjectHashes(records)
+	if err != nil {
+		return fmt.Errorf("build measurement data-object hashes: %w", err)
+	}
+	if err := storeMeasurementDataObjectHashes(ctx, diagram.GetRedisClientInstance(), hashes); err != nil {
+		return fmt.Errorf("store measurement data-object hashes in redis: %w", err)
+	}
+
+	logger.Info(ctx, "initialize measurement data objects completed",
+		"postgres_record_count", len(records),
+		"redis_hash_count", len(hashes),
+	)
+	return nil
+}
+
+func buildMeasurementDataObjectHashes(records []MeasurementInitializationRecord) ([]measurementDataObjectHash, error) {
+	hashes := make([]measurementDataObjectHash, 0, len(records)*3)
+	seenKeys := make(map[string]string, len(records)*3)
+	for _, record := range records {
+		if record.MeasurementMode != constants.MeasurementModeManual &&
+			record.MeasurementMode != constants.MeasurementModeAutomatic {
+			return nil, fmt.Errorf(
+				"measurement %q mode must be %d or %d, got %d",
+				record.MeasurementTag,
+				constants.MeasurementModeManual,
+				constants.MeasurementModeAutomatic,
+				record.MeasurementMode,
+			)
+		}
+		if record.MeasurementDataSource == nil ||
+			record.MeasurementEventPlan == nil ||
+			record.MeasurementBinding == nil {
+			return nil, fmt.Errorf("measurement %q contains a null JSONB field", record.MeasurementTag)
+		}
+
+		fullToken := strings.Join([]string{
+			record.GridTag,
+			record.ZoneTag,
+			record.StationTag,
+			record.ComponentNSPath,
+			record.ComponentTag,
+			"bay",
+			record.MeasurementTag,
+		}, ".")
+		fourPartToken := strings.Join([]string{
+			record.ComponentNSPath,
+			record.ComponentTag,
+			"bay",
+			record.MeasurementTag,
+		}, ".")
+		twoPartToken := record.ComponentNSPath + "." + record.MeasurementTag
+
+		for _, token := range []string{fullToken, fourPartToken, twoPartToken} {
+			if err := validateInitializedMeasurementToken(token); err != nil {
+				return nil, err
+			}
+		}
+
+		measurementType, err := MeasurementTypeString(record.MeasurementType)
+		if err != nil {
+			return nil, fmt.Errorf("derive type for measurement %q: %w", fullToken, err)
+		}
+		dataSource, err := measurementInitializationJSON(record.MeasurementDataSource)
+		if err != nil {
+			return nil, fmt.Errorf("encode data_source for measurement %q: %w", fullToken, err)
+		}
+		eventPlan, err := measurementInitializationJSON(record.MeasurementEventPlan)
+		if err != nil {
+			return nil, fmt.Errorf("encode event_plan for measurement %q: %w", fullToken, err)
+		}
+		binding, err := measurementInitializationJSON(record.MeasurementBinding)
+		if err != nil {
+			return nil, fmt.Errorf("encode binding for measurement %q: %w", fullToken, err)
+		}
+
+		fields := map[string]any{
+			"mode":        record.MeasurementMode,
+			"meta":        "MEASUREMENT",
+			"type":        measurementType,
+			"name":        twoPartToken,
+			"description": record.MeasurementName,
+			"id":          fullToken,
+			"size":        record.MeasurementSize,
+			"data_source": dataSource,
+			"event_plan":  eventPlan,
+			"binding":     binding,
+		}
+		owner := fmt.Sprintf("%d/%s", record.MeasurementID, record.ComponentUUID)
+		for _, token := range []string{fullToken, fourPartToken, twoPartToken} {
+			if err := appendMeasurementDataObjectHash(&hashes, seenKeys, token, owner, fields); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return hashes, nil
+}
+
+func validateInitializedMeasurementToken(token string) error {
+	dataObjectType, err := ClassifyDataObjectToken(token)
+	if err != nil {
+		return fmt.Errorf("generated invalid measurement token %q: %w", token, err)
+	}
+	if dataObjectType != constants.DataObjectTypeMeasurement {
+		return fmt.Errorf("generated token %q is not a measurement", token)
+	}
+	return nil
+}
+
+func appendMeasurementDataObjectHash(
+	hashes *[]measurementDataObjectHash,
+	seenKeys map[string]string,
+	key string,
+	owner string,
+	fields map[string]any,
+) error {
+	if existingOwner, exists := seenKeys[key]; exists {
+		return fmt.Errorf(
+			"ambiguous measurement token %q is produced by %q and %q",
+			key,
+			existingOwner,
+			owner,
+		)
+	}
+	seenKeys[key] = owner
+	*hashes = append(*hashes, measurementDataObjectHash{Key: key, Fields: fields})
+	return nil
+}
+
+func measurementInitializationJSON(value orm.JSONMap) (string, error) {
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return "", err
+	}
+	return string(encoded), nil
+}
+
+func storeMeasurementDataObjectHashes(
+	ctx context.Context,
+	rdb *redis.Client,
+	hashes []measurementDataObjectHash,
+) error {
+	if rdb == nil {
+		return fmt.Errorf("redis client is nil")
+	}
+
+	oldKeys, err := rdb.SMembers(ctx, constants.RedisMeasurementDataObjectKeySet).Result()
+	if err != nil {
+		return fmt.Errorf("query previously initialized measurement keys: %w", err)
+	}
+	currentKeys := make(map[string]struct{}, len(hashes))
+	for start := 0; start < len(hashes); start += measurementDataObjectPipelineSize {
+		end := min(start+measurementDataObjectPipelineSize, len(hashes))
+		pipeline := rdb.TxPipeline()
+		keyMembers := make([]any, 0, end-start)
+		for _, hash := range hashes[start:end] {
+			pipeline.Del(ctx, hash.Key)
+			pipeline.HSet(ctx, hash.Key, hash.Fields)
+			keyMembers = append(keyMembers, hash.Key)
+			currentKeys[hash.Key] = struct{}{}
+		}
+		if len(keyMembers) > 0 {
+			pipeline.SAdd(ctx, constants.RedisMeasurementDataObjectKeySet, keyMembers...)
+		}
+		if _, err := pipeline.Exec(ctx); err != nil {
+			return fmt.Errorf("write measurement data-object hash batch starting at %d: %w", start, err)
+		}
+	}
+
+	staleKeys := make([]string, 0)
+	for _, key := range oldKeys {
+		if _, exists := currentKeys[key]; !exists {
+			staleKeys = append(staleKeys, key)
+		}
+	}
+	cleanupPipeline := rdb.TxPipeline()
+	for start := 0; start < len(staleKeys); start += measurementDataObjectPipelineSize {
+		end := min(start+measurementDataObjectPipelineSize, len(staleKeys))
+		cleanupPipeline.Del(ctx, staleKeys[start:end]...)
+		members := make([]any, 0, end-start)
+		for _, key := range staleKeys[start:end] {
+			members = append(members, key)
+		}
+		cleanupPipeline.SRem(ctx, constants.RedisMeasurementDataObjectKeySet, members...)
+	}
+	if len(hashes) == 0 {
+		cleanupPipeline.Del(ctx, constants.RedisMeasurementDataObjectKeySet)
+	}
+	if _, err := cleanupPipeline.Exec(ctx); err != nil {
+		return fmt.Errorf("remove stale measurement data-object hashes: %w", err)
+	}
+	return nil
+}

model/measurement_data_object_init_test.go

@@ -0,0 +1,101 @@
+package model
+
+import (
+	"testing"
+
+	"modelRT/orm"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildMeasurementDataObjectHashesCreatesAllTokenForms(t *testing.T) {
+	record := measurementInitializationRecordForTest()
+
+	hashes, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{record})
+	require.NoError(t, err)
+	require.Len(t, hashes, 3)
+
+	fullToken := "grid000.zone000.station000.220kV_xuefulu1.CTA.bay.IA_rms"
+	fourPartToken := "220kV_xuefulu1.CTA.bay.IA_rms"
+	twoPartToken := "220kV_xuefulu1.IA_rms"
+	assert.Equal(t, fullToken, hashes[0].Key)
+	assert.Equal(t, fourPartToken, hashes[1].Key)
+	assert.Equal(t, twoPartToken, hashes[2].Key)
+
+	fields := hashes[0].Fields
+	assert.NotContains(t, fields, "value")
+	assert.Equal(t, int16(1), fields["mode"])
+	assert.Equal(t, "MEASUREMENT", fields["meta"])
+	assert.Equal(t, "TM", fields["type"])
+	assert.Equal(t, twoPartToken, fields["name"])
+	assert.Equal(t, "A相保护电流有效值", fields["description"])
+	assert.Equal(t, fullToken, fields["id"])
+	assert.Equal(t, 1, fields["size"])
+	assert.Equal(t, `{"io_address":{"channel":"TM1","device":"CTA","dtype":1,"option":"rms","station":"001"},"type":1}`, fields["data_source"])
+	assert.Equal(t, `{}`, fields["event_plan"])
+	assert.Equal(t, `{"ct":{"index":0,"polarity":1,"ratio":1250}}`, fields["binding"])
+	assert.Equal(t, fields, hashes[1].Fields)
+	assert.Equal(t, fields, hashes[2].Fields)
+}
+
+func TestBuildMeasurementDataObjectHashesRejectsAmbiguousShortToken(t *testing.T) {
+	first := measurementInitializationRecordForTest()
+	second := first
+	second.GridTag = "grid001"
+	second.ZoneTag = "zone001"
+	second.StationTag = "station001"
+	second.ComponentUUID = "component-uuid-2"
+	second.ComponentTag = "CTB"
+	second.MeasurementID = 2
+
+	_, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{first, second})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "ambiguous measurement token")
+	assert.Contains(t, err.Error(), "220kV_xuefulu1.IA_rms")
+}
+
+func TestBuildMeasurementDataObjectHashesRejectsUnsupportedMeasurementType(t *testing.T) {
+	record := measurementInitializationRecordForTest()
+	record.MeasurementType = 5
+
+	_, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{record})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "derive type")
+	assert.Contains(t, err.Error(), "unsupported measurement type 5")
+}
+
+func measurementInitializationRecordForTest() MeasurementInitializationRecord {
+	return MeasurementInitializationRecord{
+		GridTag:         "grid000",
+		ZoneTag:         "zone000",
+		StationTag:      "station000",
+		ComponentUUID:   "component-uuid-1",
+		ComponentNSPath: "220kV_xuefulu1",
+		ComponentTag:    "CTA",
+		MeasurementID:   1,
+		MeasurementTag:  "IA_rms",
+		MeasurementName: "A相保护电流有效值",
+		MeasurementType: 0,
+		MeasurementMode: 1,
+		MeasurementSize: 1,
+		MeasurementDataSource: orm.JSONMap{
+			"type": float64(1),
+			"io_address": map[string]any{
+				"station": "001",
+				"device":  "CTA",
+				"channel": "TM1",
+				"dtype":   float64(1),
+				"option":  "rms",
+			},
+		},
+		MeasurementEventPlan: orm.JSONMap{},
+		MeasurementBinding: orm.JSONMap{
+			"ct": map[string]any{
+				"index":    float64(0),
+				"ratio":    float64(1250),
+				"polarity": float64(1),
+			},
+		},
+	}
+}

sql/data_object_measurement_initialization.go

@@ -0,0 +1,38 @@
+// Package sql defines reusable database SQL statements
+package sql
+
+const (
+	// MeasurementInitializationRows joins measurements to the hierarchy used by
+	// all supported data-object token forms. The bay join guarantees that
+	// token6=bay refers to an existing bay record.
+	MeasurementInitializationRows = `SELECT
+			grid.tagname AS grid_tag,
+			zone.tagname AS zone_tag,
+			station.tagname AS station_tag,
+			component.global_uuid::text AS component_uuid,
+			component.nspath AS component_nspath,
+			component.tag AS component_tag,
+			measurement.id AS measurement_id,
+			measurement.tag AS measurement_tag,
+			measurement.name AS measurement_name,
+			measurement.type AS measurement_type,
+			measurement.mode AS measurement_mode,
+			measurement.size AS measurement_size,
+			measurement.data_source AS measurement_data_source,
+			measurement.event_plan AS measurement_event_plan,
+			measurement.binding AS measurement_binding
+		FROM public.grid AS grid
+		INNER JOIN public.zone AS zone ON zone.grid_id = grid.id
+		INNER JOIN public.station AS station ON station.zone_id = zone.id
+		INNER JOIN public.component AS component ON component.station_id = station.id
+		INNER JOIN public.measurement AS measurement
+			ON measurement.component_uuid = component.global_uuid
+		INNER JOIN public.bay AS bay
+			ON bay.bay_uuid = measurement.bay_uuid
+		WHERE grid.tagname <> ''
+			AND zone.tagname <> ''
+			AND station.tagname <> ''
+			AND component.nspath <> ''
+			AND component.tag <> ''
+			AND measurement.tag <> ''`
+)