初步实现联锁驱采Modbus服务

api/projectRunConfig.go

@@ -12,6 +12,7 @@ import (
 	"joylink.club/bj-rtsts-server/middleware"
 	"joylink.club/bj-rtsts-server/service"
 	"joylink.club/bj-rtsts-server/sys_error"
+	"joylink.club/iot/service/proto"
 )
 
 func InitProjectRunConfigRouter(api *gin.RouterGroup, authMiddleware *jwt.GinJWTMiddleware) {
@@ -204,6 +205,7 @@ func parseRunCofigStruct(m interface{}) []*dto.RunConfigDescription {
 		c := &dto.RunConfigDescription{
 			FieldName:    field.Tag.Get("json"),
 			Description:  field.Tag.Get("description"),
+			DefaultValue: field.Tag.Get("default"),
 		}
 		k := field.Type.Kind()
 		switch k {
@@ -217,7 +219,63 @@ func parseRunCofigStruct(m interface{}) []*dto.RunConfigDescription {
 		default:
 			c.Type = k.String()
 		}
+		// slog.Warn("运行配置字段类型", "fieldType", field.Type, "fieldTypeName", field.Type.Name(), "fieldTypeString", field.Type.String())
+		switch field.Type.String() {
+		case "proto.Modbus_Endianness":
+			c.SelectOptions = modbus_EndiannessSelectOptions
+		case "proto.Modbus_Function":
+			c.SelectOptions = modbus_FunctionSelectOptions
+		case "proto.Modbus_WriteStrategy":
+			c.SelectOptions = modbus_WriteStrategySelectOptions
+		case "proto.DataType":
+			c.SelectOptions = dataTypeSelectOptions
+		default:
+		}
 		cs = append(cs, c)
 	}
 	return cs
 }
+
+var modbus_EndiannessSelectOptions = buildModbus_EndiannessSelectOptions()
+
+func buildModbus_EndiannessSelectOptions() []*dto.RunConfigSelectOption {
+	return []*dto.RunConfigSelectOption{
+		{Label: "大端", Value: int32(proto.Modbus_BigEndian)},
+		{Label: "小端", Value: int32(proto.Modbus_LittleEndian)},
+	}
+}
+
+var modbus_FunctionSelectOptions = buildModbus_FunctionSelectOptions()
+
+func buildModbus_FunctionSelectOptions() []*dto.RunConfigSelectOption {
+	return []*dto.RunConfigSelectOption{
+		{Label: "读线圈", Value: int32(proto.Modbus_ReadCoil)},
+		{Label: "读离散输入", Value: int32(proto.Modbus_ReadDiscreteInput)},
+		{Label: "读保持寄存器", Value: int32(proto.Modbus_ReadHoldingRegister)},
+		{Label: "读输入寄存器", Value: int32(proto.Modbus_ReadInputRegister)},
+		{Label: "写单个线圈", Value: int32(proto.Modbus_WriteCoil)},
+		{Label: "写多个线圈", Value: int32(proto.Modbus_WriteCoils)},
+		{Label: "写单个寄存器", Value: int32(proto.Modbus_WriteRegister)},
+		{Label: "写多个寄存器", Value: int32(proto.Modbus_WriteRegisters)},
+		{Label: "读写多个线圈", Value: int32(proto.Modbus_RWCoils)},
+		{Label: "读写多个寄存器", Value: int32(proto.Modbus_RWRegisters)},
+	}
+}
+
+var modbus_WriteStrategySelectOptions = buildModbus_WriteStrategySelectOptions()
+
+func buildModbus_WriteStrategySelectOptions() []*dto.RunConfigSelectOption {
+	return []*dto.RunConfigSelectOption{
+		{Label: "数据更新时写", Value: int32(proto.Modbus_OnUpdate)},
+		{Label: "定时写", Value: int32(proto.Modbus_OnScheduled)},
+	}
+}
+
+var dataTypeSelectOptions = buildDataTypeSelectOptions()
+
+func buildDataTypeSelectOptions() []*dto.RunConfigSelectOption {
+	return []*dto.RunConfigSelectOption{
+		{Label: "采集数据", Value: int32(proto.DataType_CollectTable)},
+		{Label: "驱动数据", Value: int32(proto.DataType_DriveTable)},
+	}
+}

config/bj_local.yml

@@ -49,11 +49,6 @@ logging:
   # 控制台是否输出
   stdout: false
 messaging:
-  # centrifugo:
-  #   tokenSecret: aaa
-  #   apiKey: bbb
-  #   apiEndpoint: /api
-  #   address: 10.60.1.111:8001
   mqtt:
     address: tcp://10.60.1.111:1883
     username: rtsts_service

config/config.go

@@ -7,6 +7,7 @@ import (
 	"os"
 
 	"github.com/spf13/viper"
+	"joylink.club/iot/service/proto"
 )
 
 const (
@@ -42,7 +43,6 @@ type log struct {
 }
 
 type messaging struct {
-	// Centrifugo centrifugo
 	Mqtt mqtt
 }
 
@@ -69,6 +69,26 @@ type ThridPartyConfig struct {
 	RsspAxleCfgs      []RsspAxleConfig        `json:"rsspAxleCfgs" description:"所有联锁集中站计轴RSSP-I配置"`
 	ElectricMachinery ElectricMachineryConfig `json:"electricMachinery" description:"电机配置"`
 	BtmCanet          BtmCanetConfig          `json:"btmCanet" description:"BTM关联的网关设备CANET配置"`
+	CidcModbus        []CidcModbusConfig      `json:"cidcModbus" description:"联锁驱采Modbus接口配置"`
+}
+
+type CidcModbusConfig struct {
+	Open       bool                    `json:"open" description:"是否开启"`
+	Url        string                  `json:"url" description:"接口URL(格式tcp://{ip}:{port})" default:"tcp://127.0.0.1:502"` // 连接地址
+	UnitId     uint32                  `json:"unitId" description:"从机unitId"`                                              // 从机unitId
+	Endianness proto.Modbus_Endianness `json:"endianness" description:"字节序(大端/小端)"`                                        // 16位寄存器字节序
+	Interval   uint32                  `json:"interval" description:"定时请求间隔" default:"1000"`                               // 循环请求间隔(毫秒),0表示不主动请求,只当有变化时请求
+	Timeout    uint32                  `json:"timeout" description:"请求超时时间" default:"1000"`                                // 超时时间（毫秒）
+	Ecs        string                  `json:"ecs" description:"联锁集中站"`                                                    // 所属集中站
+	Mapping    []ModbusDcMapping       `json:"mapping" description:"modbus数据与驱动/采集码表映射配置"`
+}
+type ModbusDcMapping struct {
+	Function      proto.Modbus_Function      `json:"function" description:"Modbus功能"`                                   // 功能
+	Addr          uint32                     `json:"addr" description:"Modbus功能起始地址,位类型的功能为起始位地址,寄存器类型的功能为起始字(2个字节)地址"` // 起始地址,当功能为位功能时,表示起始位地址,当功能为寄存器功能时,表示起始字(2个字节)地址
+	Quantity      uint32                     `json:"quantity" description:"Modbus读取数量,位类型的为位数,寄存器类型的为字(2个字节)数"`         // 数量,当功能为位功能时,表示位数,当功能为寄存器功能时,表示字(2个字节)数
+	WriteStrategy proto.Modbus_WriteStrategy `json:"writeStrategy" description:"写入策略"`                                  // 当功能为写入类功能时(不包含读写类功能),写策略
+	Type          proto.DataType             `json:"type" description:"映射的数据类型"`                                        // 对应数据类型
+	Start         uint32                     `json:"start" description:"映射数据的起始地址,位类型的为起始位地址,寄存器类型的为起始字节地址"`            // 映射起始地址
 }
 type DynamicsConfig struct {
 	Ip                 string `json:"ip" description:"IP配置"`

config/dev.yml

@@ -50,11 +50,6 @@ logging:
   # 控制台是否输出
   stdout: true
 messaging:
-  # centrifugo:
-  #   tokenSecret: aaa
-  #   apiKey: bbb
-  #   apiEndpoint: /api
-  #   address: 192.168.3.233:10000
   mqtt:
     address: tcp://192.168.3.233:1883
     username: rtsts_service

config/local.yml

@@ -50,11 +50,6 @@ logging:
   # 控制台是否输出
   stdout: false
 messaging:
-  centrifugo:
-    tokenSecret: aaa
-    apiKey: bbb
-    apiEndpoint: /api
-    address: 192.168.0.203:10000
   mqtt:
     address: tcp://192.168.0.203:1883
     username: rtsts_service

config/test_local.yml

@@ -49,11 +49,6 @@ logging:
   # 控制台是否输出
   stdout: false
 messaging:
-  centrifugo:
-    tokenSecret: aaa
-    apiKey: bbb
-    apiEndpoint: /api
-    address: 192.168.3.233:10000
   mqtt:
     address: tcp://192.168.3.233:1883
     username: rtsts_service

dto/projectRunConfig.go

@@ -27,8 +27,14 @@ type RunConfigDescription struct {
 	FieldName      string                   `json:"fieldName" form:"fieldName"`
 	Description    string                   `json:"description" form:"description"`
 	Type           string                   `json:"type" form:"type"`
+	DefaultValue   string                   `json:"defaultValue" form:"defaultValue"`
+	SelectOptions  []*RunConfigSelectOption `json:"selectOptions" form:"selectOptions"`
 	ItemTypeFields []*RunConfigDescription  `json:"itemTypeFields" form:"itemTypeFields"`
 }
+type RunConfigSelectOption struct {
+	Label string `json:"label" form:"label"`
+	Value int32  `json:"value" form:"value"`
+}
 
 func ConvertToRunConfigDto(gi *model.ProjectRunConfig) *ProjectRunConfigDto {
 	return &ProjectRunConfigDto{

third_party/cidc_modbus/cidc_modbus.go

@@ -0,0 +1,196 @@
+package cidcmodbus
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"joylink.club/bj-rtsts-server/config"
+	"joylink.club/bj-rtsts-server/ts/simulation/wayside/memory"
+	"joylink.club/iot/service"
+	"joylink.club/iot/service/model"
+	"joylink.club/iot/service/proto"
+	"joylink.club/rtsssimulation/component"
+	"joylink.club/rtsssimulation/entity"
+)
+
+// 联锁驱采Modbus服务
+type CidcModbusService interface {
+	Stop()
+}
+
+var serviceManage = &cidcModbusServiceManage{}
+
+type cidcModbusServiceManage struct {
+	services []*cidcModbusService
+}
+
+func (m *cidcModbusServiceManage) Stop() {
+	for _, s := range m.services {
+		s.Stop()
+	}
+	m.services = nil
+}
+
+func TryStop() {
+	serviceManage.Stop()
+}
+
+func TryStart(vs *memory.VerifySimulation) error {
+	for _, cmc := range vs.GetCidcModbusConfig() {
+		if cmc.Open {
+			cms, err := newCidcModbusService(vs, &cmc)
+			if err != nil {
+				return err
+			}
+			serviceManage.services = append(serviceManage.services, cms)
+		}
+	}
+	return nil
+}
+
+type cidcModbusService struct {
+	vs           *memory.VerifySimulation // 仿真对象
+	modbusConfig *config.CidcModbusConfig // modbus驱采配置
+	ecsUid       string                   // 联锁集中站uid
+	dcData       model.QC                 // 驱采数据
+	mdms         service.IotService       // modbus驱采映射服务
+	cancel       context.CancelFunc
+}
+
+func (s *cidcModbusService) run(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		s.update()
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
+func (s *cidcModbusService) update() {
+	wd := entity.GetWorldData(s.vs.World)
+	qce := wd.FindQcEntityByEcsId(s.ecsUid)
+	if qce == nil {
+		panic(fmt.Sprintf("联锁驱采Modbus服务查询状态失败,仿真不存在集中站联锁: %s", s.ecsUid))
+	}
+	qcs := component.CiQcStateType.Get(qce)
+	if qcs == nil {
+		panic(fmt.Sprintf("联锁驱采Modbus服务查询状态失败,集中站联锁驱采状态不存在: %s", s.ecsUid))
+	}
+	s.dcData.UpdateCollectByBytes(0, qcs.Cbs)
+	qs := s.dcData.GetDrive()
+	qcs.Qbs = qs
+}
+
+// Stop implements CidcModbusService.
+func (s *cidcModbusService) Stop() {
+	s.cancel()
+	s.mdms.Stop()
+}
+
+func newCidcModbusService(vs *memory.VerifySimulation, modbusConfig *config.CidcModbusConfig) (*cidcModbusService, error) {
+	station := vs.Repo.FindStationByStationName(modbusConfig.Ecs)
+	if station == nil {
+		return nil, fmt.Errorf("联锁驱采Modbus服务创建失败,未找到联锁集中站: %s", modbusConfig.Ecs)
+	}
+	ecsId := station.Id()
+	// 获取
+	wd := entity.GetWorldData(vs.World)
+	qce := wd.FindQcEntityByEcsId(ecsId)
+	if qce == nil {
+		return nil, fmt.Errorf("联锁驱采Modbus服务创建失败,仿真不存在集中站联锁: %s", ecsId)
+	}
+	qcs := component.CiQcStateType.Get(qce)
+	err := checkConfigMapping(qcs, modbusConfig)
+	if err != nil {
+		return nil, fmt.Errorf("联锁驱采Modbus服务创建失败,Modbus地址映射配置错误: %s", err)
+	}
+	dc := model.NewDC(make([]byte, len(qcs.Qbs)), make([]byte, len(qcs.Cbs)))
+	mdms, err := service.NewModbusQcService(converToModbusDcConfig(modbusConfig), dc)
+	if err != nil {
+		return nil, fmt.Errorf("联锁驱采Modbus服务创建失败: %s", err)
+	}
+	cms := &cidcModbusService{
+		vs:           vs,
+		modbusConfig: modbusConfig,
+		ecsUid:       ecsId,
+		dcData:       dc,
+		mdms:         mdms,
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go cms.run(ctx)
+	cms.cancel = cancel
+	return cms, nil
+}
+
+func checkConfigMapping(qcs *component.CiQcState, modbusConfig *config.CidcModbusConfig) error {
+	for _, mdm := range modbusConfig.Mapping {
+		if mdm.Type == proto.DataType_CollectTable {
+			switch mdm.Function {
+			case proto.Modbus_ReadCoil, proto.Modbus_ReadDiscreteInput, proto.Modbus_WriteCoil, proto.Modbus_WriteCoils, proto.Modbus_RWCoils:
+				end := mdm.Start + mdm.Quantity
+				if end > uint32(len(qcs.Cbs)*8) {
+					return fmt.Errorf("Modbus地址映射配置错误,采集表地址超出范围: 起始位地址=%d,位数量=%d,实际位长度=%d", mdm.Start, mdm.Quantity, len(qcs.Cbs)*8)
+				}
+			case proto.Modbus_ReadInputRegister, proto.Modbus_ReadHoldingRegister, proto.Modbus_WriteRegister, proto.Modbus_WriteRegisters, proto.Modbus_RWRegisters:
+				end := mdm.Start + mdm.Quantity*2
+				if end > uint32(len(qcs.Cbs)) {
+					return fmt.Errorf("Modbus地址映射配置错误,采集表地址超出范围: 起始字节地址=%d,字数量=%d,实际位长度=%d", mdm.Start, mdm.Quantity, len(qcs.Cbs))
+				}
+			}
+		} else if mdm.Type == proto.DataType_DriveTable {
+			switch mdm.Function {
+			case proto.Modbus_ReadCoil, proto.Modbus_ReadDiscreteInput, proto.Modbus_WriteCoil, proto.Modbus_WriteCoils, proto.Modbus_RWCoils:
+				end := mdm.Start + mdm.Quantity
+				if end > uint32(len(qcs.Qbs)*8) {
+					return fmt.Errorf("Modbus地址映射配置错误,驱动表地址超出范围: 起始位地址=%d,位数量=%d,实际位长度=%d", mdm.Start, mdm.Quantity, len(qcs.Qbs)*8)
+				}
+			case proto.Modbus_ReadInputRegister, proto.Modbus_ReadHoldingRegister, proto.Modbus_WriteRegister, proto.Modbus_WriteRegisters, proto.Modbus_RWRegisters:
+				end := mdm.Start + mdm.Quantity*2
+				if end > uint32(len(qcs.Qbs)) {
+					return fmt.Errorf("Modbus地址映射配置错误,驱动表地址超出范围: 起始字节地址=%d,字数量=%d,实际位长度=%d", mdm.Start, mdm.Quantity, len(qcs.Qbs))
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func getEcsIdByName(vs *memory.VerifySimulation, ecsName string) string {
+	ecsId := ""
+	for _, s := range vs.Repo.StationList() {
+		if ecsName == s.GetCode() {
+			ecsId = s.Id()
+		}
+	}
+	return ecsId
+}
+
+func converToModbusDcConfig(config *config.CidcModbusConfig) *proto.ModbusConfig {
+	return &proto.ModbusConfig{
+		Url:        config.Url,
+		UnitId:     config.UnitId,
+		Endianness: config.Endianness,
+		Interval:   config.Interval,
+		Timeout:    config.Timeout,
+		Mapping:    convertToModbusDcMapping(config.Mapping),
+	}
+}
+
+func convertToModbusDcMapping(modbusDcMapping []config.ModbusDcMapping) []*proto.ModbusDcMapping {
+	res := make([]*proto.ModbusDcMapping, 0)
+	for _, mdm := range modbusDcMapping {
+		res = append(res, &proto.ModbusDcMapping{
+			Function:      mdm.Function,
+			Addr:          mdm.Addr,
+			Quantity:      mdm.Quantity,
+			WriteStrategy: mdm.WriteStrategy,
+			Type:          mdm.Type,
+			Start:         mdm.Start,
+		})
+	}
+	return res
+}

third_party/dynamics/dynamics.go

@@ -172,7 +172,7 @@ func (d *dynamics) Start(manager DynamicsMessageManager) error {
 		panic("启动动力学消息服务错误: DynamicsMessageManager不能为nil")
 	}
 	if d.manager != nil {
-		panic("启动动力学消息服务错误: 存在正在运行的任务")
+		return fmt.Errorf("启动动力学消息服务错误: 存在正在运行的任务")
 	}
 	d.runConfig = manager.GetDynamicsRunConfig()
 	if d.runConfig == nil || d.runConfig.Ip == "" || !d.runConfig.Open {

ts/simulation/wayside/memory/wayside_simulation.go

@@ -309,6 +309,10 @@ func (s *VerifySimulation) GetDynamicsRunConfig() *config.DynamicsConfig {
 	return &s.runConfig.Dynamics
 }
 
+func (s *VerifySimulation) GetCidcModbusConfig() []config.CidcModbusConfig {
+	return s.runConfig.CidcModbus
+}
+
 // 获取动力学运行资源
 func (s *VerifySimulation) GetDynamicsRunRepository() *message.LineBaseInfo {
 	info := &message.LineBaseInfo{}

ts/test_simulation_manage.go

@@ -2,12 +2,14 @@ package ts
 
 import (
 	"fmt"
-	"joylink.club/bj-rtsts-server/third_party/can_btm"
 	"log/slog"
 	"runtime"
 	"strconv"
 	"sync"
 
+	"joylink.club/bj-rtsts-server/third_party/can_btm"
+	cidcmodbus "joylink.club/bj-rtsts-server/third_party/cidc_modbus"
+
 	"joylink.club/bj-rtsts-server/third_party/axle_device"
 	"joylink.club/bj-rtsts-server/third_party/electrical_machinery"
 
@@ -57,6 +59,7 @@ func CreateSimulation(projectId int32, mapIds []int32, runConfig *dto.ProjectRun
 		// 第三方服务处理
 		err = runThirdParty(verifySimulation)
 		if err != nil {
+			verifySimulation.World.Close()
 			return "", err
 		}
 		simulationMap.Store(simulationId, verifySimulation)
@@ -120,6 +123,11 @@ func runThirdParty(s *memory.VerifySimulation) error {
 	electrical_machinery.Default().Start(s)
 	// 车载BTM启动
 	can_btm.Default().Start(s)
+	// 联锁驱采Modbus服务启动
+	err = cidcmodbus.TryStart(s)
+	if err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -139,7 +147,8 @@ func stopThirdParty(s *memory.VerifySimulation) {
 	electrical_machinery.Default().Stop()
 	// 车载BTM停止
 	can_btm.Default().Stop()
-
+	// 联锁驱采Modbus服务停止
+	cidcmodbus.TryStop()
 }
 
 func createSimulationId(projectId int32) string {