Add cancellation to api package

README.md

@@ -15,7 +15,7 @@
 - Notification transliteration
 - Call Notifications (ModemManager)
 - Music control
-- Get info from watch (HRM, Battery level, Firmware version)
+- Get info from watch (HRM, Battery level, Firmware version, Motion)
 - Set current time
 - Control socket
 - Firmware upgrades
@@ -29,7 +29,7 @@ This daemon creates a UNIX socket at `/tmp/itd/socket`. It allows you to directl
 The socket accepts JSON requests. For example, sending a notification looks like this:
 
 ```json
-{"type": "notify", "data": {"title": "title1", "body": "body1"}}
+{"type": 5, "data": {"title": "title1", "body": "body1"}}
 ```
 
 It will return a JSON response. A response can have 3 fields: `error`, `msg`, and `value`. Error is a boolean that signals whether an error was returned. If error is true, the msg field will contain the error. Value can contain any data and depends on what the request was.
@@ -83,10 +83,10 @@ This is the `itctl` usage screen:
 Control the itd daemon for InfiniTime smartwatches
 
 Usage:
+  itctl [flags]
   itctl [command]
 
 Available Commands:
-  completion  generate the autocompletion script for the specified shell
   firmware    Manage InfiniTime firmware
   get         Get information from InfiniTime
   help        Help about any command
@@ -94,7 +94,8 @@ Available Commands:
   set         Set information on InfiniTime
 
 Flags:
-  -h, --help   help for itctl
+  -h, --help                 help for itctl
+  -s, --socket-path string   Path to itd socket
 
 Use "itctl [command] --help" for more information about a command.
 ```

api/client.go

@@ -0,0 +1,117 @@
+package api
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"net"
+
+	"github.com/mitchellh/mapstructure"
+	"go.arsenm.dev/infinitime"
+	"go.arsenm.dev/itd/internal/types"
+)
+
+// Default socket address
+const DefaultAddr = "/tmp/itd/socket"
+
+// Client is the socket API client
+type Client struct {
+	conn          net.Conn
+	respCh        chan types.Response
+	heartRateCh   chan uint8
+	battLevelCh   chan uint8
+	stepCountCh   chan uint32
+	motionCh      chan infinitime.MotionValues
+	dfuProgressCh chan DFUProgress
+}
+
+// New creates a new client and sets it up
+func New(addr string) (*Client, error) {
+	conn, err := net.Dial("unix", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	out := &Client{
+		conn:   conn,
+		respCh: make(chan types.Response, 5),
+	}
+
+	go func() {
+		scanner := bufio.NewScanner(conn)
+		for scanner.Scan() {
+			var res types.Response
+			err = json.Unmarshal(scanner.Bytes(), &res)
+			if err != nil {
+				continue
+			}
+			out.handleResp(res)
+		}
+	}()
+	return out, err
+}
+
+func (c *Client) Close() error {
+	err := c.conn.Close()
+	if err != nil {
+		return err
+	}
+	close(c.respCh)
+	return nil
+}
+
+// request sends a request to itd and waits for and returns the response
+func (c *Client) request(req types.Request) (types.Response, error) {
+	// Encode request into connection
+	err := json.NewEncoder(c.conn).Encode(req)
+	if err != nil {
+		return types.Response{}, err
+	}
+
+	res := <-c.respCh
+
+	if res.Error {
+		return res, errors.New(res.Message)
+	}
+
+	return res, nil
+}
+
+// requestNoRes sends a request to itd and does not wait for the response
+func (c *Client) requestNoRes(req types.Request) error {
+	// Encode request into connection
+	err := json.NewEncoder(c.conn).Encode(req)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// handleResp handles the received response as needed
+func (c *Client) handleResp(res types.Response) error {
+	switch res.Type {
+	case types.ResTypeWatchHeartRate:
+		c.heartRateCh <- uint8(res.Value.(float64))
+	case types.ResTypeWatchBattLevel:
+		c.battLevelCh <- uint8(res.Value.(float64))
+	case types.ResTypeWatchStepCount:
+		c.stepCountCh <- uint32(res.Value.(float64))
+	case types.ResTypeWatchMotion:
+		out := infinitime.MotionValues{}
+		err := mapstructure.Decode(res.Value, &out)
+		if err != nil {
+			return err
+		}
+		c.motionCh <- out
+	case types.ResTypeDFUProgress:
+		out := DFUProgress{}
+		err := mapstructure.Decode(res.Value, &out)
+		if err != nil {
+			return err
+		}
+		c.dfuProgressCh <- out
+	default:
+		c.respCh <- res
+	}
+	return nil
+}

api/info.go

@@ -0,0 +1,158 @@
+package api
+
+import (
+	"reflect"
+
+	"github.com/mitchellh/mapstructure"
+	"go.arsenm.dev/infinitime"
+	"go.arsenm.dev/itd/internal/types"
+)
+
+// Address gets the bluetooth address of the connected device
+func (c *Client) Address() (string, error) {
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeBtAddress,
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return res.Value.(string), nil
+}
+
+// Version gets the firmware version of the connected device
+func (c *Client) Version() (string, error) {
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeFwVersion,
+	})
+	if err != nil {
+		return "", err
+	}
+
+	return res.Value.(string), nil
+}
+
+// BatteryLevel gets the battery level of the connected device
+func (c *Client) BatteryLevel() (uint8, error) {
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeBattLevel,
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return uint8(res.Value.(float64)), nil
+}
+
+// WatchBatteryLevel returns a channel which will contain
+// new battery level values as they update. Do not use after
+// calling cancellation function
+func (c *Client) WatchBatteryLevel() (<-chan uint8, func(), error) {
+	c.battLevelCh = make(chan uint8, 2)
+	err := c.requestNoRes(types.Request{
+		Type: types.ReqTypeBattLevel,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	cancel := c.cancelFn(types.ReqTypeCancelBattLevel, c.battLevelCh)
+	return c.battLevelCh, cancel, nil
+}
+
+// HeartRate gets the heart rate from the connected device
+func (c *Client) HeartRate() (uint8, error) {
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeHeartRate,
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return uint8(res.Value.(float64)), nil
+}
+
+// WatchHeartRate returns a channel which will contain
+// new heart rate values as they update. Do not use after
+// calling cancellation function
+func (c *Client) WatchHeartRate() (<-chan uint8, func(), error) {
+	c.heartRateCh = make(chan uint8, 2)
+	err := c.requestNoRes(types.Request{
+		Type: types.ReqTypeWatchHeartRate,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	cancel := c.cancelFn(types.ReqTypeCancelHeartRate, c.heartRateCh)
+	return c.heartRateCh, cancel, nil
+}
+
+// cancelFn generates a cancellation function for the given
+// request type and channel
+func (c *Client) cancelFn(reqType int, ch interface{}) func() {
+	return func() {
+		reflectCh := reflect.ValueOf(ch)
+		reflectCh.Close()
+		reflectCh.Set(reflect.Zero(reflectCh.Type()))
+		c.requestNoRes(types.Request{
+			Type: reqType,
+		})
+	}
+}
+
+// StepCount gets the step count from the connected device
+func (c *Client) StepCount() (uint32, error) {
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeStepCount,
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return uint32(res.Value.(float64)), nil
+}
+
+// WatchStepCount returns a channel which will contain
+// new step count values as they update. Do not use after
+// calling cancellation function
+func (c *Client) WatchStepCount() (<-chan uint32, func(), error) {
+	c.stepCountCh = make(chan uint32, 2)
+	err := c.requestNoRes(types.Request{
+		Type: types.ReqTypeWatchStepCount,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	cancel := c.cancelFn(types.ReqTypeCancelStepCount, c.stepCountCh)
+	return c.stepCountCh, cancel, nil
+}
+
+// Motion gets the motion values from the connected device
+func (c *Client) Motion() (infinitime.MotionValues, error) {
+	out := infinitime.MotionValues{}
+	res, err := c.request(types.Request{
+		Type: types.ReqTypeMotion,
+	})
+	if err != nil {
+		return out, err
+	}
+	err = mapstructure.Decode(res.Value, &out)
+	if err != nil {
+		return out, err
+	}
+	return out, nil
+}
+
+// WatchMotion returns a channel which will contain
+// new motion values as they update. Do not use after
+// calling cancellation function
+func (c *Client) WatchMotion() (<-chan infinitime.MotionValues, func(), error) {
+	c.motionCh = make(chan infinitime.MotionValues, 2)
+	err := c.requestNoRes(types.Request{
+		Type: types.ReqTypeWatchMotion,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	cancel := c.cancelFn(types.ReqTypeCancelMotion, c.motionCh)
+	return c.motionCh, cancel, nil
+}

api/time.go

@@ -0,0 +1,33 @@
+package api
+
+import (
+	"time"
+
+	"go.arsenm.dev/itd/internal/types"
+)
+
+// SetTime sets the given time on the connected device
+func (c *Client) SetTime(t time.Time) error {
+	_, err := c.request(types.Request{
+		Type: types.ReqTypeSetTime,
+		Data: t.Format(time.RFC3339),
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// SetTimeNow sets the time on the connected device to
+// the current time. This is more accurate than
+// SetTime(time.Now()) due to RFC3339 formatting
+func (c *Client) SetTimeNow() error {
+	_, err := c.request(types.Request{
+		Type: types.ReqTypeSetTime,
+		Data: "now",
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}

api/upgrade.go

@@ -0,0 +1,37 @@
+package api
+
+import (
+	"encoding/json"
+
+	"go.arsenm.dev/itd/internal/types"
+)
+
+// DFUProgress stores the progress of a DFU upfate
+type DFUProgress types.DFUProgress
+
+// UpgradeType indicates the type of upgrade to be performed
+type UpgradeType uint8
+
+// Type of DFU upgrade
+const (
+	UpgradeTypeArchive UpgradeType = iota
+	UpgradeTypeFiles
+)
+
+// FirmwareUpgrade initiates a DFU update and returns the progress channel
+func (c *Client) FirmwareUpgrade(upgType UpgradeType, files ...string) (<-chan DFUProgress, error) {
+	err := json.NewEncoder(c.conn).Encode(types.Request{
+		Type: types.ReqTypeFwUpgrade,
+		Data: types.ReqDataFwUpgrade{
+			Type:  int(upgType),
+			Files: files,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	c.dfuProgressCh = make(chan DFUProgress, 5)
+
+	return c.dfuProgressCh, nil
+}

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/srwiley/oksvg v0.0.0-20210519022825-9fc0c575d5fe // indirect
 	github.com/srwiley/rasterx v0.0.0-20210519020934-456a8d69b780 // indirect
 	github.com/yuin/goldmark v1.4.1 // indirect
-	go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b
+	go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72
 	golang.org/x/image v0.0.0-20210628002857-a66eb6448b8d // indirect
 	golang.org/x/net v0.0.0-20211011170408-caeb26a5c8c0 // indirect
 	golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac // indirect

go.sum

@@ -366,8 +366,8 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1
 github.com/yuin/goldmark v1.3.8/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 github.com/yuin/goldmark v1.4.1 h1:/vn0k+RBvwlxEmP5E7SZMqNxPhfMVFEJiykr15/0XKM=
 github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
-go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b h1:2VitKPwSYSWXmL5BH88nfTPLSIYPCt4yubpEJHhcQBc=
-go.arsenm.dev/infinitime v0.0.0-20211022195951-45baea10486b/go.mod h1:gaepaueUz4J5FfxuV19B4w5pi+V3mD0LTef50ryxr/Q=
+go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72 h1:e8kOuL6Jj8ZjJzkGwJ3xqpGG9EhUzfvZk9AlSsm3X1U=
+go.arsenm.dev/infinitime v0.0.0-20211023042633-53aa6f8a0c72/go.mod h1:gaepaueUz4J5FfxuV19B4w5pi+V3mD0LTef50ryxr/Q=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0/go.mod h1:IJHfcCEKxYu1Os13ZdwCwIUTUVGYTSAM3YSwc9/Ac1g=
 go.etcd.io/etcd/client/v2 v2.305.0/go.mod h1:h9puh54ZTgAKtEbut2oe9P4L/oqKCVB6xsXlzd7alYQ=

internal/types/types.go

@@ -9,11 +9,35 @@ const (
 	ReqTypeNotify
 	ReqTypeSetTime
 	ReqTypeWatchHeartRate
+	ReqTypeCancelHeartRate
 	ReqTypeWatchBattLevel
+	ReqTypeCancelBattLevel
 	ReqTypeMotion
 	ReqTypeWatchMotion
+	ReqTypeCancelMotion
 	ReqTypeStepCount
 	ReqTypeWatchStepCount
+	ReqTypeCancelStepCount
+)
+
+const (
+	ResTypeHeartRate = iota
+	ResTypeBattLevel
+	ResTypeFwVersion
+	ResTypeDFUProgress
+	ResTypeBtAddress
+	ResTypeNotify
+	ResTypeSetTime
+	ResTypeWatchHeartRate
+	ResTypeCancelHeartRate
+	ResTypeWatchBattLevel
+	ResTypeCancelBattLevel
+	ResTypeMotion
+	ResTypeWatchMotion
+	ResTypeCancelMotion
+	ResTypeStepCount
+	ResTypeWatchStepCount
+	ResTypeCancelStepCount
 )
 
 const (
@@ -27,6 +51,7 @@ type ReqDataFwUpgrade struct {
 }
 
 type Response struct {
+	Type    int         `json:"type"`
 	Value   interface{} `json:"value,omitempty"`
 	Message string      `json:"msg,omitempty"`
 	Error   bool        `json:"error"`

socket.go

@@ -81,6 +81,11 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 		return
 	}
 
+	heartRateDone := make(chan struct{})
+	battLevelDone := make(chan struct{})
+	stepCountDone := make(chan struct{})
+	motionDone := make(chan struct{})
+
 	// Create new scanner on connection
 	scanner := bufio.NewScanner(conn)
 	for scanner.Scan() {
@@ -102,21 +107,36 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 			}
 			// Encode heart rate to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeHeartRate,
 				Value: heartRate,
 			})
 		case types.ReqTypeWatchHeartRate:
-			heartRateCh, err := dev.WatchHeartRate()
+			heartRateCh, cancel, err := dev.WatchHeartRate()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every heart rate value
 				for heartRate := range heartRateCh {
-					json.NewEncoder(conn).Encode(types.Response{
-						Value: heartRate,
-					})
+					select {
+					case <-heartRateDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
+						json.NewEncoder(conn).Encode(types.Response{
+							Type:  types.ResTypeWatchHeartRate,
+							Value: heartRate,
+						})
+					}
 				}
 			}()
+		case types.ReqTypeCancelHeartRate:
+			// Stop heart rate notifications
+			heartRateDone <- struct{}{}
+			json.NewEncoder(conn).Encode(types.Response{})
 		case types.ReqTypeBattLevel:
 			// Get battery level from watch
 			battLevel, err := dev.BatteryLevel()
@@ -126,21 +146,36 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 			}
 			// Encode battery level to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeBattLevel,
 				Value: battLevel,
 			})
 		case types.ReqTypeWatchBattLevel:
-			battLevelCh, err := dev.WatchBatteryLevel()
+			battLevelCh, cancel, err := dev.WatchBatteryLevel()
 			if err != nil {
-				connErr(conn, err, "Error getting heart rate channel")
+				connErr(conn, err, "Error getting battery level channel")
 				break
 			}
 			go func() {
+				// For every battery level value
 				for battLevel := range battLevelCh {
-					json.NewEncoder(conn).Encode(types.Response{
-						Value: battLevel,
-					})
+					select {
+					case <-battLevelDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
+						json.NewEncoder(conn).Encode(types.Response{
+							Type:  types.ResTypeWatchBattLevel,
+							Value: battLevel,
+						})
+					}
 				}
 			}()
+		case types.ReqTypeCancelBattLevel:
+			// Stop battery level notifications
+			battLevelDone <- struct{}{}
+			json.NewEncoder(conn).Encode(types.Response{})
 		case types.ReqTypeMotion:
 			// Get battery level from watch
 			motionVals, err := dev.Motion()
@@ -150,21 +185,36 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 			}
 			// Encode battery level to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeMotion,
 				Value: motionVals,
 			})
 		case types.ReqTypeWatchMotion:
-			motionValCh, _, err := dev.WatchMotion()
+			motionValCh, cancel, err := dev.WatchMotion()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every motion event
 				for motionVals := range motionValCh {
-					json.NewEncoder(conn).Encode(types.Response{
-						Value: motionVals,
-					})
+					select {
+					case <-motionDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
+						json.NewEncoder(conn).Encode(types.Response{
+							Type:  types.ResTypeWatchMotion,
+							Value: motionVals,
+						})
+					}
 				}
 			}()
+		case types.ReqTypeCancelMotion:
+			// Stop motion notifications
+			motionDone <- struct{}{}
+			json.NewEncoder(conn).Encode(types.Response{})
 		case types.ReqTypeStepCount:
 			// Get battery level from watch
 			stepCount, err := dev.StepCount()
@@ -174,35 +224,52 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 			}
 			// Encode battery level to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeStepCount,
 				Value: stepCount,
 			})
 		case types.ReqTypeWatchStepCount:
-			stepCountCh, _, err := dev.WatchStepCount()
+			stepCountCh, cancel, err := dev.WatchStepCount()
 			if err != nil {
 				connErr(conn, err, "Error getting heart rate channel")
 				break
 			}
 			go func() {
+				// For every step count value
 				for stepCount := range stepCountCh {
-					json.NewEncoder(conn).Encode(types.Response{
-						Value: stepCount,
-					})
+					select {
+					case <-stepCountDone:
+						// Stop notifications if done signal received
+						cancel()
+						return
+					default:
+						// Encode response to connection if no done signal received
+						json.NewEncoder(conn).Encode(types.Response{
+							Type:  types.ResTypeWatchStepCount,
+							Value: stepCount,
+						})
+					}
 				}
 			}()
+		case types.ReqTypeCancelStepCount:
+			// Stop step count notifications
+			stepCountDone <- struct{}{}
+			json.NewEncoder(conn).Encode(types.Response{})
 		case types.ReqTypeFwVersion:
 			// Get firmware version from watch
 			version, err := dev.Version()
 			if err != nil {
-				connErr(conn, err, "Error getting battery level")
+				connErr(conn, err, "Error getting firmware version")
 				break
 			}
 			// Encode version to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeFwVersion,
 				Value: version,
 			})
 		case types.ReqTypeBtAddress:
 			// Encode bluetooth address to connection
 			json.NewEncoder(conn).Encode(types.Response{
+				Type:  types.ResTypeBtAddress,
 				Value: dev.Address(),
 			})
 		case types.ReqTypeNotify:
@@ -229,7 +296,7 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				break
 			}
 			// Encode empty types.Response to connection
-			json.NewEncoder(conn).Encode(types.Response{})
+			json.NewEncoder(conn).Encode(types.Response{Type: types.ResTypeNotify})
 		case types.ReqTypeSetTime:
 			// If no data, return error
 			if req.Data == nil {
@@ -261,7 +328,7 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				break
 			}
 			// Encode empty types.Response to connection
-			json.NewEncoder(conn).Encode(types.Response{})
+			json.NewEncoder(conn).Encode(types.Response{Type: types.ResTypeSetTime})
 		case types.ReqTypeFwUpgrade:
 			// If no data, return error
 			if req.Data == nil {
@@ -326,6 +393,7 @@ func handleConnection(conn net.Conn, dev *infinitime.Device) {
 				for event := range progress {
 					// Encode event on connection
 					json.NewEncoder(conn).Encode(types.Response{
+						Type:  types.ResTypeDFUProgress,
 						Value: event,
 					})
 				}