diff --git a/constants/data-object.go b/constants/data-object.go index f57158e..8e624aa 100644 --- a/constants/data-object.go +++ b/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 +) diff --git a/constants/redis.go b/constants/redis.go index 6a3adb0..3d4d347 100644 --- a/constants/redis.go +++ b/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" ) diff --git a/database/query_measurement_initialization.go b/database/query_measurement_initialization.go new file mode 100644 index 0000000..393c128 --- /dev/null +++ b/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 +} diff --git a/database/query_measurement_initialization_test.go b/database/query_measurement_initialization_test.go new file mode 100644 index 0000000..132bd8e --- /dev/null +++ b/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{}, + } +} diff --git a/handler/data_object_attribute_query.go b/handler/data_object_attribute_query.go index d4592e9..77c2faf 100644 --- a/handler/data_object_attribute_query.go +++ b/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 diff --git a/handler/data_object_attribute_query_test.go b/handler/data_object_attribute_query_test.go index 5ff9efc..f887def 100644 --- a/handler/data_object_attribute_query_test.go +++ b/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, diff --git a/main.go b/main.go index 95754b7..32affa1 100644 --- a/main.go +++ b/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) diff --git a/model/measurement_attribute.go b/model/measurement_attribute.go index 51c0cc0..dda45a6 100644 --- a/model/measurement_attribute.go +++ b/model/measurement_attribute.go @@ -2,50 +2,27 @@ package model import ( "fmt" - "strings" "modelRT/constants" - "modelRT/orm" ) -var allowedMeasurementTypes = map[string]struct{}{ - "TM": {}, - "TS": {}, - "TC": {}, - "TA": {}, - "SP": {}, -} - -// 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) +// MeasurementTypeString converts measurement.type from PostgreSQL to the +// electric-element type stored in the Redis data-object hash. +func MeasurementTypeString(measurementType int16) (string, error) { + switch measurementType { + case constants.MeasurementTypeTelemetry: + 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) } - 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) { diff --git a/model/measurement_attribute_test.go b/model/measurement_attribute_test.go index 8ca8e38..80e8199 100644 --- a/model/measurement_attribute_test.go +++ b/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) { - for _, measurementType := range []string{"TM", "TS", "TC", "TA", "SP"} { - t.Run(measurementType, func(t *testing.T) { - actual, err := MeasurementTypeFromDataSource(orm.JSONMap{ - "type": float64(1), - "io_address": map[string]any{ - "channel": strings.ToLower(measurementType) + "1_test", - }, - }) +func TestMeasurementTypeString(t *testing.T) { + tests := []struct { + value int16 + expected string + }{ + {value: 0, expected: "TM"}, + {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) { - tests := []orm.JSONMap{ - {"type": float64(2), "io_address": map[string]any{"channel": "tm1"}}, - {"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) +func TestMeasurementTypeStringRejectsInvalidValues(t *testing.T) { + for _, value := range []int16{-1, 5, 100} { + _, err := MeasurementTypeString(value) require.Error(t, err) + assert.Contains(t, err.Error(), "unsupported measurement type") } } diff --git a/model/measurement_data_object_init.go b/model/measurement_data_object_init.go new file mode 100644 index 0000000..04a987c --- /dev/null +++ b/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 +} diff --git a/model/measurement_data_object_init_test.go b/model/measurement_data_object_init_test.go new file mode 100644 index 0000000..549ef74 --- /dev/null +++ b/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), + }, + }, + } +} diff --git a/sql/data_object_measurement_initialization.go b/sql/data_object_measurement_initialization.go new file mode 100644 index 0000000..3157d34 --- /dev/null +++ b/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 <> ''` +)