ammeter/platform/platform.go

package platform

import (
	"encoding/json"
	"reflect"
	"time"

	"github.com/yuguorong/go/log"

	bus "github.com/ammeter/Bus"
	"github.com/ammeter/cloudserver"
	"github.com/ammeter/config"
	drviers "github.com/ammeter/drivers"
	"github.com/ammeter/util"
)

const (
	DEF_REQ_AMGW_URL = "/platform/dev/get-4G-gateway-list"
	REQ_AMMETER_URL  = "/platform/dev/get-ammeter-list"
	REQ_AIR_URL      = "/platform/dev/get-sair-list"
	DEF_FTP_PORT     = 10010
)

type GatewayConfig struct {
	bus.QMqtt
	Tenants string `json:"userName"`
}

type PaPlatform struct {
	gwUrl       string
	jsGwList    string
	jsDevList   map[string]string
	GatewayList map[string]*Gateway
	modelList   map[string]string
}

func (p *PaPlatform) LoadSyncGWConfig() *[]GatewayConfig {
	if p.gwUrl == "" {
		p.gwUrl = DEF_REQ_AMGW_URL
	}
	mqttList := []GatewayConfig{}
	prof := config.GetSysConfig().GetProfile("remote_config", cloudserver.GetCloudConfig())
	cs := cloudserver.GetCloudServer(prof)
	if err := cs.GetClientData(p.gwUrl, &mqttList, nil); err == nil {
		for i := 0; i < len(mqttList); {
			if mqttList[i].QMqtt.User == "" || mqttList[i].QMqtt.MqttUrl == "" || mqttList[i].QMqtt.Uid == "" || mqttList[i].QMqtt.Name == "" {
				mqttList = append(mqttList[:i], mqttList[i+1:]...)
			} else {
				i++
			}
		}
		jmq, _ := json.Marshal(&mqttList)
		if string(jmq) != p.jsGwList {
			config.GetDB().Save(&mqttList)
			p.jsGwList = string(jmq)
			//need check different
		}
	} else {
		config.GetDB().Table(&mqttList)
	}
	return &mqttList
}

func (p *PaPlatform) LoadSyncAllDevices() {
	for name, url := range p.modelList {
		drv := drviers.Install(name, nil)
		model := drv.GetModel()
		if url != "" && model != nil {
			prof := config.GetSysConfig().GetProfile("remote_config", cloudserver.GetCloudConfig())
			cs := cloudserver.GetCloudServer(prof)
			param := map[string]string{
				"gwDevId": "",
			}
			bupdate := true
			err := cs.GetClientData(url, model, &param)
			if err == nil {
				jdev, _ := json.Marshal(model)
				if js, has := p.jsDevList[name]; has {
					bupdate = (js != string(jdev))
				}
				p.jsDevList[name] = string(jdev)
				if bupdate {
					config.GetDB().Delete(drv.GetModel()) //clear first
					config.GetDB().CreateTbl(model)
					config.GetDB().Save(model)
				}
			}
		}
	}
}

func (p *PaPlatform) LoadGatewayProfile() {
	tscur := time.Now().UnixMilli()
	p.LoadSyncAllDevices()
	gwMqttList := p.LoadSyncGWConfig()
	for _, mq := range *gwMqttList {
		gw, has := p.GatewayList[mq.Uid]
		if !has {
			gw = InitGateway(&mq.QMqtt, mq.Tenants, p.modelList)
			p.GatewayList[gw.Uid] = gw
		}
		gw.InstallDrivers(mq.Uid)
		gw.tsUpdate = tscur
	}

	for name, gw := range p.GatewayList {
		if gw.tsUpdate != tscur {
			gw.Remove(gw.Uid)
			delete(p.GatewayList, name)
		}
	}
}

func (p *PaPlatform) ListDevices() interface{} {

	devmap := make(map[string]interface{})
	devmap["Version"] = config.GetSysConfig().GetValue("GatwayVersion", "0.0.0.1")

	for _, gw := range p.GatewayList {
		mgw := make(map[string]interface{})
		for _, drv := range gw.drvList {
			for n, dev := range drv.GetDeviceList() {
				mgw[n] = dev.ListDevices()
			}
		}
		//mgw["ID"] = gw.Uid
		mgw["Tenants"] = gw.Name
		mgw["URL"] = gw.Mqttcfg.MqttUrl
		devmap[gw.Mqttcfg.Name] = mgw
	}
	return devmap
}

func (p *PaPlatform) Init() {
	p.GatewayList = make(map[string]*Gateway)
	p.jsDevList = make(map[string]string)
	p.LoadModles()
	p.LoadGatewayProfile()
}

func (p *PaPlatform) SetModel(sname string, surl string) {
	if p.modelList == nil {
		p.modelList = make(map[string]string)
	}
	p.modelList[sname] = surl
}

func (p *PaPlatform) SaveModel() {
	config.GetSysConfig().SetProfile("model_list", &p.modelList)
}

func (p *PaPlatform) LoadModles() {
	p.modelList = make(map[string]string)
	list := config.GetSysConfig().GetProfile("model_list", &p.modelList)
	if list != nil {
		p.modelList = *list.(*map[string]string)
	}
}

func (p *PaPlatform) SetGatewayUrl(url string) {
	p.gwUrl = url
}

func StartServer() {
	p := PaPlatform{}
	p.Init()
	msgAPI := util.GetMsgHandler("API")

	ticksAutoFresh := time.NewTicker(time.Duration(time.Minute * 60))

	for loop := true; loop; {
		select {
		case msg := <-msgAPI:
			if reflect.TypeOf(msg).Kind() == reflect.String {
				switch msg.(string) {
				case "UpdateDevice":
					p.LoadGatewayProfile()
				case "ListDevices":
					chn := <-msgAPI
					chn.(chan interface{}) <- p.ListDevices()
				case "exit":
					loop = false
				}
			}
		case <-ticksAutoFresh.C:
			log.Info("Auto refresh devices")
			p.LoadGatewayProfile()
		}
	}
	log.Info("exit now")

}

func init() {
	config.GetDB().CreateTbl(&GatewayConfig{})
}