2018-12-16 23:33:13 +01:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"flag"
|
|
|
|
"fmt"
|
|
|
|
"os"
|
|
|
|
"os/signal"
|
|
|
|
"strings"
|
|
|
|
"syscall"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
common "miflorad/common"
|
|
|
|
impl "miflorad/common/ble"
|
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
2018-12-16 23:33:13 +01:00
|
|
|
"github.com/go-ble/ble"
|
|
|
|
"github.com/go-ble/ble/examples/lib/dev"
|
|
|
|
"github.com/pkg/errors"
|
|
|
|
)
|
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
const mqttConnectTimeout = 10 * time.Second
|
|
|
|
|
2018-12-23 12:35:33 +01:00
|
|
|
// program version, will be populated on build
|
|
|
|
var version string
|
|
|
|
|
2018-12-16 23:33:13 +01:00
|
|
|
var (
|
2018-12-23 23:16:16 +01:00
|
|
|
scanTimeout = flag.Duration("scantimeout", 10*time.Second, "timeout after that a scan per peripheral will be aborted")
|
|
|
|
readRetries = flag.Int("readretries", 2, "number of times reading will be attempted per peripheral")
|
|
|
|
interval = flag.Duration("interval", 25*time.Second, "metrics collection interval")
|
|
|
|
brokerHost = flag.String("brokerhost", "localhost", "MQTT broker host to send metrics to")
|
|
|
|
brokerUser = flag.String("brokeruser", "", "MQTT broker user used for authentication")
|
|
|
|
brokerPassword = flag.String("brokerpassword", "", "MQTT broker password used for authentication")
|
|
|
|
brokerUseTLS = flag.Bool("brokerusetls", true, "whether TLS should be used for MQTT broker")
|
|
|
|
brokerTopicPrefix = flag.String("brokertopicprefix", "", "MQTT topic prefix for messages")
|
2019-01-02 12:57:26 +01:00
|
|
|
publishFormatFlag = flag.String("publishformat", "graphite", "MQTT message content format")
|
|
|
|
graphitePrefix = flag.String("graphiteprefix", "", "Graphite metrics name prefix")
|
|
|
|
)
|
|
|
|
|
|
|
|
type publishFormat int
|
|
|
|
|
|
|
|
const (
|
|
|
|
graphiteFormat publishFormat = iota
|
|
|
|
influxFormat publishFormat = iota
|
2018-12-16 23:33:13 +01:00
|
|
|
)
|
|
|
|
|
|
|
|
type peripheral struct {
|
|
|
|
id string
|
|
|
|
lastMetaDataFetch time.Time
|
|
|
|
metaData common.VersionBatteryResponse
|
|
|
|
}
|
|
|
|
|
|
|
|
var allPeripherals []*peripheral
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
type mifloraMetric interface {
|
|
|
|
getPeripheralId() string
|
|
|
|
}
|
|
|
|
|
|
|
|
type mifloraDataMetric struct {
|
2018-12-31 19:42:49 +01:00
|
|
|
peripheralId string
|
|
|
|
metaData common.VersionBatteryResponse
|
|
|
|
sensorData common.SensorDataResponse
|
|
|
|
connectTime float64
|
|
|
|
readoutTime float64
|
|
|
|
rssi int
|
|
|
|
}
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
func (m mifloraDataMetric) getPeripheralId() string {
|
|
|
|
return m.peripheralId
|
|
|
|
}
|
|
|
|
|
|
|
|
type mifloraErrorMetric struct {
|
|
|
|
peripheralId string
|
|
|
|
failed int
|
|
|
|
}
|
|
|
|
|
|
|
|
func (m mifloraErrorMetric) getPeripheralId() string {
|
|
|
|
return m.peripheralId
|
|
|
|
}
|
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
type mqttLogger struct {
|
|
|
|
level string
|
|
|
|
}
|
|
|
|
|
|
|
|
func (logger mqttLogger) Println(a ...interface{}) {
|
|
|
|
fmt.Fprintln(os.Stderr, fmt.Sprintf("mqtt %s:", logger.level), a)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (logger mqttLogger) Printf(format string, a ...interface{}) {
|
|
|
|
fmt.Fprintf(os.Stderr, "mqtt %s: "+format, logger.level, a)
|
|
|
|
}
|
2018-12-16 23:33:13 +01:00
|
|
|
|
|
|
|
func checkTooShortInterval() error {
|
|
|
|
numPeripherals := int64(len(flag.Args()))
|
2018-12-22 12:19:43 +01:00
|
|
|
numReadRetries := int64(*readRetries)
|
|
|
|
if (*scanTimeout).Nanoseconds()*numReadRetries*numPeripherals >= (*interval).Nanoseconds() {
|
2018-12-16 23:33:13 +01:00
|
|
|
return errors.New(fmt.Sprintf(
|
|
|
|
"The interval of %s is too short given the scan timeout of %s for %d peripheral(s) with %d retries each! Exiting...\n",
|
|
|
|
*interval, *scanTimeout, numPeripherals, *readRetries))
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2018-12-23 12:35:33 +01:00
|
|
|
func getVersion() string {
|
|
|
|
if version == "" {
|
|
|
|
return "dev"
|
|
|
|
} else {
|
|
|
|
return version
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
func getMQTTOptions() *mqtt.ClientOptions {
|
|
|
|
if *brokerUseTLS {
|
|
|
|
return mqtt.NewClientOptions().
|
|
|
|
AddBroker(fmt.Sprintf("ssl://%s:8883", *brokerHost)).
|
|
|
|
SetUsername(*brokerUser).
|
|
|
|
SetPassword(*brokerPassword)
|
|
|
|
} else {
|
|
|
|
return mqtt.NewClientOptions().
|
|
|
|
AddBroker(fmt.Sprintf("tcp://%s:1883", *brokerHost)).
|
|
|
|
SetUsername(*brokerUser).
|
|
|
|
SetPassword(*brokerPassword)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-31 19:42:49 +01:00
|
|
|
func readData(peripheral *peripheral, client ble.Client) (common.SensorDataResponse, error) {
|
2018-12-22 12:19:43 +01:00
|
|
|
// re-request meta data (for battery level) if last check more than 24 hours ago
|
|
|
|
// Source: https://github.com/open-homeautomation/miflora/blob/ffd95c3e616df8843cc8bff99c9b60765b124092/miflora/miflora_poller.py#L92
|
2019-05-20 18:28:29 +02:00
|
|
|
if time.Since(peripheral.lastMetaDataFetch) >= 1*time.Hour {
|
2018-12-22 12:19:43 +01:00
|
|
|
metaData, err := impl.RequestVersionBattery(client)
|
2018-12-16 23:33:13 +01:00
|
|
|
if err != nil {
|
2018-12-31 19:42:49 +01:00
|
|
|
return common.SensorDataResponse{}, errors.Wrap(err, "can't request version battery")
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
peripheral.metaData = metaData
|
|
|
|
peripheral.lastMetaDataFetch = time.Now()
|
|
|
|
}
|
|
|
|
|
|
|
|
if peripheral.metaData.RequiresModeChangeBeforeRead() {
|
2018-12-22 12:19:43 +01:00
|
|
|
err2 := impl.RequestModeChange(client)
|
2018-12-16 23:33:13 +01:00
|
|
|
if err2 != nil {
|
2018-12-31 19:42:49 +01:00
|
|
|
return common.SensorDataResponse{}, errors.Wrap(err2, "can't request mode change")
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-22 12:19:43 +01:00
|
|
|
sensorData, err3 := impl.RequestSensorData(client)
|
2018-12-16 23:33:13 +01:00
|
|
|
if err3 != nil {
|
2018-12-31 19:42:49 +01:00
|
|
|
return common.SensorDataResponse{}, errors.Wrap(err3, "can't request sensor data")
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
|
2018-12-31 19:42:49 +01:00
|
|
|
return sensorData, nil
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
func connectPeripheral(peripheral *peripheral, send chan mifloraMetric) error {
|
2018-12-16 23:33:13 +01:00
|
|
|
|
2018-12-22 12:19:43 +01:00
|
|
|
// only way to get back the found advertisement, must be buffered!
|
|
|
|
foundAdvertisementChannel := make(chan ble.Advertisement, 1)
|
|
|
|
|
2018-12-16 23:33:13 +01:00
|
|
|
filter := func(adv ble.Advertisement) bool {
|
2018-12-22 12:19:43 +01:00
|
|
|
if strings.ToUpper(adv.Addr().String()) == strings.ToUpper(peripheral.id) {
|
|
|
|
foundAdvertisementChannel <- adv
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
return false
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
timeConnectStart := time.Now()
|
|
|
|
|
|
|
|
ctx := ble.WithSigHandler(context.WithTimeout(context.Background(), *scanTimeout))
|
|
|
|
client, err := ble.Connect(ctx, filter)
|
|
|
|
if err != nil {
|
|
|
|
return errors.Wrapf(err, "can't connect to %s", peripheral.id)
|
|
|
|
}
|
|
|
|
|
|
|
|
timeConnectTook := time.Since(timeConnectStart).Seconds()
|
2018-12-22 12:19:43 +01:00
|
|
|
|
2018-12-16 23:33:13 +01:00
|
|
|
// Source: https://github.com/go-ble/ble/blob/master/examples/basic/explorer/main.go#L53
|
|
|
|
// Make sure we had the chance to print out the message.
|
|
|
|
done := make(chan struct{})
|
|
|
|
// Normally, the connection is disconnected by us after our exploration.
|
|
|
|
// However, it can be asynchronously disconnected by the remote peripheral.
|
|
|
|
// So we wait(detect) the disconnection in the go routine.
|
|
|
|
go func() {
|
|
|
|
<-client.Disconnected()
|
|
|
|
close(done)
|
|
|
|
}()
|
|
|
|
|
|
|
|
timeReadoutStart := time.Now()
|
|
|
|
|
|
|
|
if _, err := client.DiscoverProfile(true); err != nil {
|
|
|
|
return errors.Wrap(err, "can't descover profile")
|
|
|
|
}
|
|
|
|
|
2018-12-31 19:42:49 +01:00
|
|
|
sensorData, err2 := readData(peripheral, client)
|
2018-12-16 23:33:13 +01:00
|
|
|
|
|
|
|
timeReadoutTook := time.Since(timeReadoutStart).Seconds()
|
|
|
|
|
|
|
|
client.CancelConnection()
|
|
|
|
|
|
|
|
<-done
|
|
|
|
|
2018-12-31 19:42:49 +01:00
|
|
|
if err2 != nil {
|
|
|
|
return errors.Wrap(err2, "can't read data")
|
|
|
|
}
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
send <- mifloraDataMetric{
|
2018-12-31 19:42:49 +01:00
|
|
|
peripheralId: common.MifloraGetAlphaNumericID(peripheral.id),
|
|
|
|
sensorData: sensorData,
|
|
|
|
metaData: peripheral.metaData,
|
|
|
|
connectTime: timeConnectTook,
|
|
|
|
readoutTime: timeReadoutTook,
|
|
|
|
rssi: (<-foundAdvertisementChannel).RSSI(),
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
|
2019-04-23 16:15:38 +02:00
|
|
|
func readPeripheral(quit chan struct{}, peripheral *peripheral, send chan mifloraMetric) error {
|
2018-12-16 23:33:13 +01:00
|
|
|
var err error
|
2019-04-23 15:40:28 +02:00
|
|
|
fmt.Fprintf(os.Stderr, "Scanning for %s...", peripheral.id)
|
2019-04-23 16:15:38 +02:00
|
|
|
L:
|
2018-12-16 23:33:13 +01:00
|
|
|
for retry := 0; retry < *readRetries; retry++ {
|
2019-04-23 16:15:38 +02:00
|
|
|
// check for quit signal (non-blocking) and terminate
|
|
|
|
select {
|
|
|
|
case <-quit:
|
|
|
|
break L
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
2019-04-23 15:40:28 +02:00
|
|
|
fmt.Fprintf(os.Stderr, " %d", retry+1)
|
2018-12-23 23:16:16 +01:00
|
|
|
err = connectPeripheral(peripheral, send)
|
2018-12-16 23:33:13 +01:00
|
|
|
// stop retrying once we have a success, last err will be returned (or nil)
|
|
|
|
if err == nil {
|
2019-04-23 15:40:28 +02:00
|
|
|
fmt.Fprintf(os.Stderr, ".")
|
2019-04-23 16:15:38 +02:00
|
|
|
break L
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
}
|
2019-04-23 15:40:28 +02:00
|
|
|
fmt.Fprintf(os.Stderr, "\n")
|
2018-12-16 23:33:13 +01:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
func readAllPeripherals(quit chan struct{}, send chan mifloraMetric) {
|
2018-12-16 23:33:13 +01:00
|
|
|
for _, peripheral := range allPeripherals {
|
2019-04-23 16:15:38 +02:00
|
|
|
err := readPeripheral(quit, peripheral, send)
|
2018-12-16 23:33:13 +01:00
|
|
|
if err != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "Failed to read peripheral %s, err: %s\n", peripheral.id, err)
|
2018-12-31 19:42:49 +01:00
|
|
|
// id := common.MifloraGetAlphaNumericID(peripheral.id)
|
|
|
|
// send <- fmt.Sprintf("%s.miflora.%s.failed 1 %d", *prefix, id, time.Now().Unix())
|
2018-12-16 23:33:13 +01:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func main() {
|
|
|
|
flag.Parse()
|
|
|
|
if len(flag.Args()) < 1 {
|
|
|
|
fmt.Fprintf(os.Stderr, "Usage: %s [options] peripheral-id [peripheral-ids...] \n", os.Args[0])
|
|
|
|
flag.PrintDefaults()
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := checkTooShortInterval(); err != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "%s\n", err)
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
|
2019-01-02 12:57:26 +01:00
|
|
|
var format publishFormat
|
|
|
|
switch *publishFormatFlag {
|
|
|
|
case "graphite":
|
|
|
|
format = graphiteFormat
|
|
|
|
case "influx":
|
|
|
|
format = influxFormat
|
|
|
|
default:
|
|
|
|
fmt.Fprintf(os.Stderr, "Unrecognized publish format %s! Exiting...\n", *publishFormatFlag)
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
|
2018-12-23 12:35:33 +01:00
|
|
|
fmt.Fprintf(os.Stderr, "miflorad version %s\n", getVersion())
|
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
mqtt.WARN = mqttLogger{level: "warning"}
|
|
|
|
mqtt.ERROR = mqttLogger{level: "error"}
|
|
|
|
mqtt.CRITICAL = mqttLogger{level: "critical"}
|
|
|
|
|
|
|
|
mqttClient := mqtt.NewClient(getMQTTOptions())
|
|
|
|
|
|
|
|
if token := mqttClient.Connect(); token.WaitTimeout(mqttConnectTimeout) && token.Error() != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "Failed to connect MQTT, err: %s\n", token.Error())
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
|
|
|
|
fmt.Fprintf(os.Stderr, "Connected to MQTT broker %s\n", *brokerHost)
|
|
|
|
|
2018-12-16 23:33:13 +01:00
|
|
|
device, err := dev.NewDevice("default")
|
|
|
|
if err != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "Failed to open device, err: %s\n", err)
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
ble.SetDefaultDevice(device)
|
|
|
|
|
|
|
|
intervalTicker := time.NewTicker(*interval)
|
|
|
|
quit := make(chan struct{})
|
2019-01-02 12:57:26 +01:00
|
|
|
send := make(chan mifloraMetric, 1)
|
2018-12-31 19:42:49 +01:00
|
|
|
publish := make(chan string, 10)
|
2018-12-16 23:33:13 +01:00
|
|
|
|
|
|
|
go func() {
|
2018-12-23 23:16:16 +01:00
|
|
|
fmt.Fprintf(os.Stderr, "Starting loop with %s interval...\n", *interval)
|
2018-12-16 23:33:13 +01:00
|
|
|
|
2018-12-22 12:19:43 +01:00
|
|
|
// populate all peripherals data structure
|
2018-12-16 23:33:13 +01:00
|
|
|
allPeripherals = make([]*peripheral, len(flag.Args()))
|
|
|
|
for i, peripheralID := range flag.Args() {
|
|
|
|
allPeripherals[i] = &peripheral{
|
|
|
|
id: peripheralID,
|
2018-12-22 12:19:43 +01:00
|
|
|
lastMetaDataFetch: time.Unix(0, 0), // force immediate 1st request
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-22 12:19:43 +01:00
|
|
|
// main loop
|
2018-12-23 23:16:16 +01:00
|
|
|
readAllPeripherals(quit, send)
|
2018-12-16 23:33:13 +01:00
|
|
|
for range intervalTicker.C {
|
2018-12-23 23:16:16 +01:00
|
|
|
readAllPeripherals(quit, send)
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
go func() {
|
2019-01-02 12:57:26 +01:00
|
|
|
switch format {
|
|
|
|
case graphiteFormat:
|
|
|
|
for metric := range send {
|
|
|
|
publishGraphite(metric, publish, *graphitePrefix)
|
|
|
|
}
|
|
|
|
case influxFormat:
|
|
|
|
for metric := range send {
|
|
|
|
publishInflux(metric, publish)
|
|
|
|
}
|
2018-12-31 19:42:49 +01:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
for line := range publish {
|
|
|
|
// fmt.Fprintln(os.Stdout, line)
|
|
|
|
token := mqttClient.Publish(*brokerTopicPrefix, 1, false, line)
|
2018-12-23 23:16:16 +01:00
|
|
|
if token.WaitTimeout(1*time.Second) && token.Error() != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "Failed to publish MQTT, err: %s\n", token.Error())
|
|
|
|
continue
|
|
|
|
}
|
2018-12-16 23:33:13 +01:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
signals := make(chan os.Signal, 1)
|
|
|
|
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
|
|
|
|
|
|
|
|
signal := <-signals
|
|
|
|
fmt.Fprintf(os.Stderr, "Received %s! Stopping...\n", signal)
|
|
|
|
intervalTicker.Stop()
|
|
|
|
close(quit)
|
2019-04-23 16:15:38 +02:00
|
|
|
// wait for last connectPeripheral to finish (worst case)
|
|
|
|
time.Sleep(*scanTimeout)
|
2018-12-16 23:33:13 +01:00
|
|
|
|
2018-12-23 23:16:16 +01:00
|
|
|
mqttClient.Disconnect(1000)
|
2018-12-16 23:33:13 +01:00
|
|
|
|
|
|
|
if err := device.Stop(); err != nil {
|
|
|
|
fmt.Fprintf(os.Stderr, "Failed to close device, err: %s\n", err)
|
|
|
|
os.Exit(1)
|
|
|
|
}
|
|
|
|
}
|