commit 2021-12-10]15:45:25

cli.go

@@ -1,15 +1,11 @@
 package manager
 
 import (
-	"bufio"
 	"context"
-	"encoding/json"
-	"fmt"
+	"errors"
 	"log"
-	"time"
 
 	"git.godopu.net/lab/etri-smartfarm-poc-controller-serial/puserial"
-	"github.com/jacobsa/go-serial/serial"
 	"github.com/rjeczalik/notify"
 )
 
@@ -21,62 +17,34 @@ var _managerObj *_manager
 func init() {
 	ctx, cancel = context.WithCancel(context.Background())
 	ch_discover = make(chan notify.EventInfo)
+	devWithUUID := map[string]*_device{}
+	devWithIface := map[interface{}]*_device{}
+	chanForSync := map[string]chan map[string]interface{}{}
+
 	_managerObj = &_manager{
-		devices:      map[string]*_device{},
-		SyncListener: nil,
-		RecvListener: nil,
+		devicesWithUUID:  devWithUUID,
+		devicesWithIface: devWithIface,
+		SyncListener:     &SyncHandler{devices: devWithUUID, chanForSync: chanForSync},
+		RecvListener:     &RecvHandler{devices: devWithIface, chanForSync: chanForSync},
 	}
+
 }
 
 func Close() {
 	cancel()
 }
 
-func register(iface string) {
-	fmt.Println("Handle] ", iface)
-
-	// change permission
-	err := ChangePermission(iface)
-	if err != nil {
-		panic(err)
+func Sync(key string, param map[string]interface{}) {
+	_managerObj.onSync(key, param)
 }
 
-	// Set up options.
-	options := serial.OpenOptions{
-		PortName:        iface,
-		BaudRate:        9600,
-		DataBits:        8,
-		StopBits:        1,
-		MinimumReadSize: 16,
+func SetDevicePropsToSync(uuid string, propsToSync []string) error {
+	device, ok := _managerObj.devicesWithUUID[uuid]
+	if !ok {
+		return errors.New("device not found")
 	}
-
-	// Open the port.
-	port, err := serial.Open(options)
-	if err != nil {
-		log.Fatalf("serial.Open: %v", err)
-	}
-
-	// Make sure to close it later.
-	defer port.Close()
-	reader := bufio.NewReader(port)
-	dev := &_device{}
-
-	for {
-		b, _, _ := reader.ReadLine()
-
-		err := json.Unmarshal(b, &dev)
-
-		if err != nil {
-			continue
-		}
-
-		fmt.Println(dev)
-		port.Write([]byte{200})
-		time.Sleep(time.Second)
-		break
-	}
-
-	// var data string
+	device.propsToSync = propsToSync
+	return nil
 }
 
 func Run() error {
@@ -87,7 +55,7 @@ func Run() error {
 		}
 	} else {
 		// _managerObj.onAdd(iface)
-		register(iface)
+		_managerObj.onAdded(iface)
 	}
 
 	go puserial.WatchNewDevice(ctx, ch_discover)
@@ -98,6 +66,6 @@ func Run() error {
 			log.Println("manager exit")
 			return nil
 		}
-		register(e.Path())
+		_managerObj.onAdded(e.Path())
 	}
 }

communication.go

@@ -3,12 +3,13 @@ package manager
 import (
 	"bufio"
 	"encoding/json"
-	"fmt"
 	"io"
 	"log"
 )
 
-func recv(reader bufio.Reader, input chan<- Event) {
+func recv(port io.Reader, r Receiver) {
+	reader := bufio.NewReader(port)
+
 	for {
 		b, _, err := reader.ReadLine()
 		if err != nil {
@@ -19,20 +20,14 @@ func recv(reader bufio.Reader, input chan<- Event) {
 		}
 		recvObj := map[string]interface{}{}
 		err = json.Unmarshal(b, &recvObj)
-		// err = readJsonFromSerial(recvObj, decoder)
 
 		if err == nil {
-			fmt.Println("line : ", string(b))
-
+			r.onRecv(port, recvObj)
 		}
 		// data = string(b)
 	}
 }
 
-func send(sender io.Writer, output <-chan Event) {
-
-}
-
 // for {
 // 	fmt.Print("> ")
 // 	cmd, _, _ := cmdReader.ReadLine()

examples/main.go

@@ -1,8 +1,11 @@
 package main
 
 import (
+	"bufio"
 	"fmt"
-	"time"
+	"os"
+	"strconv"
+	"strings"
 
 	manager "git.godopu.net/lab/etri-smartfarm-poc-controller-serial"
 )
@@ -10,13 +13,28 @@ import (
 func main() {
 	go manager.Run()
 
+	for {
 		fmt.Print("> ")
-	var cmd string
-	fmt.Scanln(&cmd)
-
+		reader := bufio.NewReader(os.Stdin)
+		b, _, _ := reader.ReadLine()
+		cmd := string(b)
+		fmt.Println("cmd: ", cmd)
 		if cmd == "exit" {
-		manager.Close()
+			return
+		} else if cmd == "setup" {
+			manager.SetDevicePropsToSync("DEVICE-A-UUID", []string{"fan", "servo"})
+			continue
 		}
 
-	time.Sleep(time.Second)
+		tkns := strings.Split(cmd, " ")
+		key := tkns[0]
+		value, err := strconv.Atoi(tkns[2])
+		if err != nil {
+			continue
+		}
+		param := map[string]interface{}{
+			tkns[1]: value,
+		}
+		manager.Sync(key, param)
+	}
 }

interface.go

@@ -1,8 +1,17 @@
 package manager
 
 type Event interface {
-	ToJson()
+	Params() map[string]interface{}
+	Key() interface{}
 }
 type EventHandler interface {
 	Handle(e Event)
 }
+
+type Receiver interface {
+	onRecv(key interface{}, params map[string]interface{})
+}
+
+type Syncronizer interface {
+	onSync(key interface{}, params map[string]interface{})
+}

listener.go

@@ -1,13 +1,108 @@
 package manager
 
-type SyncHandler struct{}
-
-func (_ *SyncHandler) Handle() {
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+)
 
+type SyncHandler struct {
+	devices     map[string]*_device
+	chanForSync map[string]chan map[string]interface{}
 }
 
-type RecvHandler struct{}
-
-func (_ *RecvHandler) Handle() {
-
+func compareMap(src map[string]interface{}, dst map[string]interface{}, props []string) bool {
+	for _, prop := range props {
+		srcV, ok := src[prop].(int)
+		if !ok {
+			srcV = int(src[prop].(float64))
+		}
+
+		dstV, ok := dst[prop].(int)
+		if !ok {
+			dstV = int(dst[prop].(float64))
+		}
+		if srcV != dstV {
+			fmt.Println("diff ", prop, "] ", srcV, " vs ", dstV)
+			return false
+		}
+	}
+	return true
+}
+
+func (sh *SyncHandler) Handle(e Event) {
+
+	device, ok := sh.devices[e.Key().(string)]
+	// fmt.Println("sync] ", device.IfaceName)
+	// fmt.Println(sh.devices)
+	if !ok {
+		return
+	}
+
+	params := e.Params()
+	if params == nil {
+		return
+	}
+
+	go func() {
+		encoder := json.NewEncoder(device.Iface)
+		origin := map[string]interface{}{}
+		origin["code"] = 200
+		props := device.propsToSync
+		// props := []string{"fan", "light", "servo"}
+		for _, pname := range props {
+			prop, ok := params[pname]
+			if !ok {
+				origin[pname] = device.states[pname]
+			} else {
+				origin[pname] = prop
+			}
+		}
+
+		err := encoder.Encode(origin)
+		if err != nil {
+			return
+		}
+
+		sh.chanForSync[device.IfaceName] = make(chan map[string]interface{})
+		for state := range sh.chanForSync[device.IfaceName] {
+			if compareMap(origin, state, props) {
+				log.Println("Same!!")
+				close(sh.chanForSync[device.IfaceName])
+				delete(sh.chanForSync, device.IfaceName)
+				return
+			}
+			log.Println("wrong: ", state)
+			log.Println("resend: ", origin)
+			err := encoder.Encode(origin)
+			if err != nil {
+				return
+			}
+		}
+	}()
+}
+
+type RecvHandler struct {
+	devices     map[interface{}]*_device
+	chanForSync map[string]chan map[string]interface{}
+}
+
+func (rh *RecvHandler) Handle(e Event) {
+	device, ok := rh.devices[e.Key()]
+	if !ok {
+		return
+	}
+
+	param := e.Params()
+	code, _ := param["code"].(float64)
+	if int(code) != 100 {
+		device.states = param
+	}
+
+	channel, ok := rh.chanForSync[device.IfaceName]
+	if ok {
+		channel <- device.states
+	}
+
+	// fmt.Println("recv] ", device.states)
 }

manager.go

@@ -3,18 +3,25 @@ package manager
 import (
 	"bufio"
 	"encoding/json"
+	"fmt"
 	"log"
 
 	"github.com/jacobsa/go-serial/serial"
 )
 
 type _manager struct {
-	devices      map[string]*_device
+	devicesWithUUID  map[string]*_device
+	devicesWithIface map[interface{}]*_device
 	SyncListener     EventHandler
 	RecvListener     EventHandler
 }
 
-func (m *_manager) onAdd(iface string) {
+func (m *_manager) onRegistered(dev *_device) {
+	go recv(dev.Iface, m)
+	log.Println("> End onRegistered")
+}
+
+func (m *_manager) onAdded(iface string) {
 	err := ChangePermission(iface)
 	if err != nil {
 		panic(err)
@@ -33,43 +40,55 @@ func (m *_manager) onAdd(iface string) {
 	go func() {
 		port, err := serial.Open(options)
 		if err != nil {
-			log.Fatalf("serial.Open: %v", err)
-		}
-
-		reader := bufio.NewReader(port)
-
-		for {
-			port.Write([]byte("Introduce\n"))
-
-			b, _, err := reader.ReadLine()
-			if err != nil {
+			log.Printf("serial.Open: %v", err)
 			return
 		}
 
-			var obj map[string]interface{}
-			err = json.Unmarshal(b, &obj)
+		reader := bufio.NewReader(port)
+		encoder := json.NewEncoder(port)
+
+		sndMsg := map[string]interface{}{}
+		sndMsg["code"] = 100
+
+		for {
+			b, _, _ := reader.ReadLine()
+			rcvMsg := map[string]interface{}{}
+			err := json.Unmarshal(b, &rcvMsg)
+
 			if err != nil {
 				continue
 			}
 
-			uuid, ok := obj["uuid"]
-			if ok {
-				m.devices[iface] = &_device{
-					UUID:      uuid.(string),
+			code, ok := rcvMsg["code"].(float64)
+			if ok && code == 100 {
+				newDevice := &_device{
+					UUID:      rcvMsg["uuid"].(string),
 					IfaceName: iface,
 					Iface:     port,
+					states:    map[string]interface{}{},
 				}
-				break
+				m.devicesWithUUID[rcvMsg["uuid"].(string)] = newDevice
+				m.devicesWithIface[port] = newDevice
+
+				m.onRegistered(newDevice)
+				fmt.Println("onAdded sub-routine is died")
+				return
 			}
+			encoder.Encode(sndMsg)
 		}
 	}()
 
+	fmt.Println("onAdded main-routine is died")
 }
 
-func (m *_manager) onSync(e Event) {
-
+func (m *_manager) onSync(key interface{}, params map[string]interface{}) {
+	if m.SyncListener != nil {
+		m.SyncListener.Handle(&EventStruct{key: key, params: params})
+	}
 }
 
-func (m *_manager) onRecv(e Event) {
-
+func (m *_manager) onRecv(key interface{}, params map[string]interface{}) {
+	if m.RecvListener != nil {
+		m.RecvListener.Handle(&EventStruct{key: key, params: params})
+	}
 }

model.go

@@ -6,9 +6,12 @@ type _device struct {
 	UUID        string
 	IfaceName   string
 	Iface       io.ReadWriter
-	states    []map[string]interface{}
+	states      map[string]interface{}
+	propsToSync []string
 }
 
+
+
 type RecvEvent struct {
 	Params map[string]interface{}
 }

structure.go

@@ -0,0 +1,15 @@
+package manager
+
+type EventStruct struct {
+	Event
+	params map[string]interface{}
+	key    interface{}
+}
+
+func (es *EventStruct) Params() map[string]interface{} {
+	return es.params
+}
+
+func (es *EventStruct) Key() interface{} {
+	return es.key
+}