Merge pull request #424 from pixlise/feature/client

Feature/client

Author: pnemere

core/client/client.go

@@ -11,15 +11,59 @@ func main() {
 	// Try to load the config file
 	apiClient, err := client.Authenticate()
 	fmt.Printf("auth err: %v\n", err)
+	if err != nil {
+		return
+	}
+
+	// err = apiClient.UploadImage(&protos.ImageUploadHttpRequest{
+	// 	Name:         "myimage.png",
+	// 	ImageData:    []byte{0x89, 0x50, 0x4e, 0x47, 0xd, 0xa, 0x1a, 0xa, 0x0, 0x0, 0x0, 0xd, 0x49, 0x48, 0x44, 0x52, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0x0, 0x1, 0x8, 0x2, 0x0, 0x0, 0x0, 0x90, 0x77, 0x53, 0xde, 0x0, 0x0, 0x0, 0x1, 0x73, 0x52, 0x47, 0x42, 0x0, 0xae, 0xce, 0x1c, 0xe9, 0x0, 0x0, 0x0, 0xc, 0x49, 0x44, 0x41, 0x54, 0x18, 0x57, 0x63, 0x28, 0x3d, 0xaf, 0xb, 0x0, 0x3, 0x2e, 0x1, 0x72, 0x50, 0x4e, 0xda, 0xdf, 0x0, 0x0, 0x0, 0x0, 0x49, 0x45, 0x4e, 0x44, 0xae, 0x42, 0x60, 0x82},
+	// 	OriginScanId: "069927431",
+	// })
+
+	// return
+
+	// err = apiClient.SaveMapData("my-data", &protos.ClientMap{
+	// 	EntryPMCs:   []int32{7, 8, 10, 12},
+	// 	FloatValues: []float64{79.4, 89.4, 109.4, 129.4},
+	// })
+	// fmt.Printf("SaveMapData: %v\n", err)
+
+	// mapData, err := apiClient.LoadMapData("my-data")
+	// fmt.Printf("LoadMapData: %v|%v\n", err, mapData)
+
+	// return
 
-	// Dev: 500302337 is missing bulk sum?
+	// // Dev: 500302337 is missing bulk sum?
 
 	// spectrum, err := apiClient.GetScanSpectrum("261161477", 15, protos.SpectrumType_SPECTRUM_NORMAL, "A")
 	// fmt.Printf("%v|%v|%v\n", err, len(spectrum.Counts), spectrum)
 
 	// spectrum, err = apiClient.GetScanSpectrum("261161477", 8383824, protos.SpectrumType_SPECTRUM_BULK, "B")
 	// fmt.Printf("%v|%v|%v\n", err, len(spectrum.Counts), spectrum)
 
+	// rangeMap, err := apiClient.GetScanSpectrumRangeAsMap("475070977", 300, 302, "B")
+	// fmt.Printf("%v|%v\n", err, len(rangeMap.EntryPMCs))
+
+	diffMap, err := apiClient.GetDiffractionAsMap("069927431", protos.EnergyCalibrationSource_CAL_BULK_SUM, 100, 120)
+	fmt.Printf("err: %v\n", err)
+	if diffMap != nil {
+		fmt.Printf("PMC, Value\n")
+		for k, v := range diffMap.EntryPMCs {
+			fmt.Printf("All Points,%v,%v\n", v, diffMap.IntValues[k])
+		}
+	}
+
+	// ruffMap, err := apiClient.GetRoughnessAsMap("475070977", protos.EnergyCalibrationSource_CAL_BULK_SUM)
+	// fmt.Printf("err: %v\n", err)
+	// if ruffMap != nil {
+	// 	fmt.Printf("PMC, Value\n")
+	// 	for k, v := range ruffMap.EntryPMCs {
+	// 		fmt.Printf("All Points,%v,%v\n", v, ruffMap.FloatValues[k])
+	// 	}
+	// }
+	return
+
 	xyzs, err := apiClient.GetScanBeamLocations("261161477")
 	fmt.Print(len(xyzs.Locations))
 

core/client/internal/cmdline/main.go

@@ -119,6 +119,13 @@ func getScanSpectrum(scanId string, pmc int32, spectrumType int, detector string
 	})
 }
 
+//export getScanSpectrumRangeAsMap
+func getScanSpectrumRangeAsMap(scanId string, channelStart int32, channelEnd int32, detector string) *C.char {
+	return processRequest("getScanSpectrumRangeAsMap", func() (proto.Message, error) {
+		return apiClient.GetScanSpectrumRangeAsMap(scanId, channelStart, channelEnd, detector)
+	})
+}
+
 //export listScans
 func listScans(scanId string) *C.char {
 	return processRequest("listScans", func() (proto.Message, error) { return apiClient.ListScans(scanId) })
@@ -139,9 +146,9 @@ func getScanEntryDataColumns(scanId string) *C.char {
 	return processRequest("getScanEntryDataColumns", func() (proto.Message, error) { return apiClient.GetScanEntryDataColumns(scanId) })
 }
 
-//export getScanEntryDataColumn
-func getScanEntryDataColumn(scanId string, columnName string) *C.char {
-	return processRequest("getScanEntryDataColumn", func() (proto.Message, error) { return apiClient.GetScanEntryDataColumn(scanId, columnName) })
+//export getScanEntryDataColumnAsMap
+func getScanEntryDataColumnAsMap(scanId string, columnName string) *C.char {
+	return processRequest("getScanEntryDataColumnAsMap", func() (proto.Message, error) { return apiClient.GetScanEntryDataColumnAsMap(scanId, columnName) })
 }
 
 //export listScanQuants
@@ -216,14 +223,28 @@ func getDiffractionPeaks(scanId string, calibrationSource int) *C.char {
 	})
 }
 
+//export getDiffractionAsMap
+func getDiffractionAsMap(scanId string, calibrationSource int, channelStart int32, channelEnd int32) *C.char {
+	return processRequest("getDiffractionAsMap", func() (proto.Message, error) {
+		return apiClient.GetDiffractionAsMap(scanId, protos.EnergyCalibrationSource(calibrationSource), channelStart, channelEnd)
+	})
+}
+
+//export getRoughnessAsMap
+func getRoughnessAsMap(scanId string, calibrationSource int) *C.char {
+	return processRequest("getRoughnessAsMap", func() (proto.Message, error) {
+		return apiClient.GetRoughnessAsMap(scanId, protos.EnergyCalibrationSource(calibrationSource))
+	})
+}
+
 //export getQuantColumns
 func getQuantColumns(quantId string) *C.char {
 	return processRequest("getQuantColumns", func() (proto.Message, error) { return apiClient.GetQuantColumns(quantId) })
 }
 
-//export getQuantColumn
-func getQuantColumn(quantId string, columnName string, detector string) *C.char {
-	return processRequest("getQuantColumn", func() (proto.Message, error) { return apiClient.GetQuantColumn(quantId, columnName, detector) })
+//export getQuantColumnAsMap
+func getQuantColumnAsMap(quantId string, columnName string, detector string) *C.char {
+	return processRequest("getQuantColumnAsMap", func() (proto.Message, error) { return apiClient.GetQuantColumnAsMap(quantId, columnName, detector) })
 }
 
 //export createROI
@@ -238,5 +259,66 @@ func createROI(roiBuff string, isMist bool) *C.char {
 	return processRequest("createROI", func() (proto.Message, error) { return apiClient.CreateROI(roiItem, isMist) })
 }
 
+//export saveMapData
+func saveMapData(key string, dataBuff string) *C.char {
+	// Here we can read the data string as a protobuf message and create the right structure
+	mapItem := &protos.ClientMap{}
+	err := protojson.Unmarshal([]byte(dataBuff), mapItem)
+	if err != nil {
+		return C.CString(fmt.Sprintf("saveMapData: Failed to decode data: %v", err))
+	}
+
+	if apiClient == nil {
+		return C.CString("Not authenticated")
+	}
+
+	err = apiClient.SaveMapData(key, mapItem)
+	if err != nil {
+		return C.CString(fmt.Sprintf("saveMapData error: %v", err))
+	}
+
+	return emptyCString
+}
+
+//export loadMapData
+func loadMapData(key string) *C.char {
+	return processRequest("loadMapData", func() (proto.Message, error) { return apiClient.LoadMapData(key) })
+}
+
+//export uploadImage
+func uploadImage(imageUpload string) *C.char {
+	// Here we can read the data string as a protobuf message and create the right structure
+	upload := &protos.ImageUploadHttpRequest{}
+	err := protojson.Unmarshal([]byte(imageUpload), upload)
+	if err != nil {
+		return C.CString(fmt.Sprintf("uploadImage: Failed to decode imageUpload: %v", err))
+	}
+
+	if apiClient == nil {
+		return C.CString("Not authenticated")
+	}
+
+	err = apiClient.UploadImage(upload)
+	if err != nil {
+		return C.CString(fmt.Sprintf("uploadImage error: %v", err))
+	}
+
+	return emptyCString
+}
+
+//export deleteImage
+func deleteImage(imageName string) *C.char {
+	if apiClient == nil {
+		return C.CString("Not authenticated")
+	}
+
+	err := apiClient.DeleteImage(imageName)
+	if err != nil {
+		return C.CString(fmt.Sprintf("deleteImage error: %v", err))
+	}
+
+	return emptyCString
+}
+
 func main() {
 }

core/client/lib/main.go

@@ -31,18 +31,29 @@ type Auth0Info struct {
 }
 
 type SocketConn struct {
-	JWT       string
-	UserId    string
-	send      chan []byte
-	recv      chan []byte
-	recvList  [][]byte // msgs received in past
-	interrupt chan os.Signal
-	done      chan struct{}
-	reqCount  uint32
+	Host         string
+	HostProtocol string
+	JWT          string
+	UserId       string
+	send         chan []byte
+	recv         chan []byte
+	recvList     [][]byte // msgs received in past
+	interrupt    chan os.Signal
+	done         chan struct{}
+	reqCount     uint32
 }
 
 const maxResponsesBuffered = 100
 
+func (s *SocketConn) GetHost(path string) (*url.URL, error) {
+	if len(s.HostProtocol) <= 0 || len(s.Host) <= 0 {
+		// Host is empty, stop here
+		return nil, fmt.Errorf("Host not set")
+	}
+
+	return &url.URL{Scheme: s.HostProtocol, Host: s.Host, Path: path}, nil
+}
+
 // Inspired by: https://tradermade.com/tutorials/golang-websocket-client
 func (s *SocketConn) Connect(connectParams ConnectInfo, auth0Params Auth0Info) error {
 	token, err := s.getWSConnectToken(connectParams, auth0Params)
@@ -239,6 +250,10 @@ func (s *SocketConn) getWSConnectToken(connectParams ConnectInfo, auth0Params Au
 		return "", err
 	}
 
+	// Remember this host for later
+	s.Host = hostUrl
+	s.HostProtocol = protocol
+
 	return respBody.ConnToken, nil
 }
 

core/client/socketconn.go

@@ -0,0 +1,55 @@
+package utils
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/pixlise/core/v4/core/timestamper"
+)
+
+type RateLimiter struct {
+	requestTimestamps []int64
+	timestamper       timestamper.ITimeStamper
+	softLimitInWindow int
+	hardLimitInWindow int
+	timeWindowSec     int64
+	softLimitSleepSec int
+}
+
+func MakeRateLimiter(timestamper timestamper.ITimeStamper, softLimit int, hardLimit int, timeWindowSec int64, softLimitSleepSec int) *RateLimiter {
+	return &RateLimiter{
+		requestTimestamps: []int64{},
+		timestamper:       timestamper,
+		softLimitInWindow: softLimit,
+		hardLimitInWindow: hardLimit,
+		timeWindowSec:     timeWindowSec,
+		softLimitSleepSec: softLimitSleepSec,
+	}
+}
+
+func (r *RateLimiter) CheckRateLimit() {
+	now := r.timestamper.GetTimeNowSec()
+
+	oldest := now - r.timeWindowSec
+
+	// Clear too old
+	validTimestamps := []int64{}
+	for _, ts := range r.requestTimestamps {
+		if ts >= oldest {
+			validTimestamps = append(validTimestamps, ts)
+		}
+	}
+
+	// Add ours
+	r.requestTimestamps = append(validTimestamps, now)
+
+	// Check if we need to limit
+	if len(r.requestTimestamps) > r.hardLimitInWindow {
+		panic("Message hard rate limit exceeded")
+	}
+
+	if len(r.requestTimestamps) > r.softLimitInWindow {
+		fmt.Printf("Rate limiting %v sec...\n", r.softLimitSleepSec)
+		time.Sleep(time.Duration(r.softLimitSleepSec) * time.Second)
+	}
+}

core/utils/rateLimit.go

@@ -0,0 +1,56 @@
+package utils
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/pixlise/core/v4/core/timestamper"
+)
+
+func Example_rateLimiter() {
+	defer func() {
+		if r := recover(); r != nil {
+			fmt.Println("Hard limit")
+		}
+	}()
+
+	ts := timestamper.MockTimeNowStamper{
+		QueuedTimeStamps: []int64{1234567000, 1234567000, 1234567001, 1234567003, 1234567009, 12345670010, 12345670012, 12345670013, 1234567017, 1234567025, 1234567030, 1234567030, 1234567031, 1234567032},
+	}
+	r := MakeRateLimiter(&ts, 3, 5, 10, 5)
+
+	runLoop(r, len(ts.QueuedTimeStamps))
+
+	// Output:
+	// No limit
+	// No limit
+	// No limit
+	// Rate limiting 5 sec...
+	// Soft limit
+	// Rate limiting 5 sec...
+	// Soft limit
+	// No limit
+	// No limit
+	// No limit
+	// Rate limiting 5 sec...
+	// Soft limit
+	// Rate limiting 5 sec...
+	// Soft limit
+	// Rate limiting 5 sec...
+	// Soft limit
+	// Hard limit
+}
+
+func runLoop(r *RateLimiter, iterations int) {
+	for c := 0; c < iterations; c++ {
+		startMs := time.Now().UnixMilli()
+		r.CheckRateLimit()
+		endMs := time.Now().UnixMilli()
+
+		if endMs-startMs > 4000 {
+			fmt.Println("Soft limit")
+		} else {
+			fmt.Println("No limit")
+		}
+	}
+}