package main import ( "bytes" "context" "encoding/binary" "encoding/hex" "encoding/json" "fmt" "math" "strconv" "strings" "time" "github.com/AFASystems/presence/internal/pkg/config" "github.com/AFASystems/presence/internal/pkg/kafkaclient" "github.com/AFASystems/presence/internal/pkg/model" "github.com/AFASystems/presence/internal/pkg/mqttclient" "github.com/segmentio/kafka-go" ) func main() { // Load global context to init beacons and latest list appCtx := model.AppContext{ Beacons: model.BeaconsList{ Beacons: make(map[string]model.Beacon), }, Settings: model.Settings{ Settings: model.SettingsVal{ Location_confidence: 4, Last_seen_threshold: 15, Beacon_metrics_size: 30, HA_send_interval: 5, HA_send_changes_only: false, }, }, BeaconEvents: model.BeaconEventList{ Beacons: make(map[string]model.BeaconEvent), }, BeaconsLookup: make(map[string]struct{}), } cfg := config.Load() // Kafka reader for Raw MQTT beacons rawReader := kafkaclient.KafkaReader(cfg.KafkaURL, "rawbeacons", "gid-raw") defer rawReader.Close() // Kafka reader for API server updates apiReader := kafkaclient.KafkaReader(cfg.KafkaURL, "apibeacons", "gid-api") defer apiReader.Close() alertWriter := kafkaclient.KafkaWriter(cfg.KafkaURL, "alertbeacons") defer alertWriter.Close() fmt.Println("Decoder initialized, subscribed to Kafka topics") // // Kafka reader for latest list updates // latestReader := kafkaclient.KafkaReader(cfg.KafkaURL, "latestbeacons", "gid-latest") // defer latestReader.Close() // // Kafka reader for settings updates // settingsReader := kafkaclient.KafkaReader(cfg.KafkaURL, "settings", "gid-settings") // defer settingsReader.Close() // declare channel for collecting Kafka messages chRaw := make(chan model.Incoming_json, 2000) chApi := make(chan model.ApiUpdate, 2000) // chLatest := make(chan model.Incoming_json, 2000) // chSettings := make(chan model.SettingsVal, 10) go kafkaclient.Consume(rawReader, chRaw) go kafkaclient.Consume(apiReader, chApi) // go kafkaclient.Consume(latestReader, chLatest) // go kafkaclient.Consume(settingsReader, chSettings) for { select { case msg := <-chRaw: processIncoming(msg, &appCtx, alertWriter) case msg := <-chApi: switch msg.Method { case "POST": id := msg.Beacon.Beacon_id appCtx.BeaconsLookup[id] = struct{}{} case "DELETE": fmt.Println("Incoming delete message") } // case msg := <-chLatest: // fmt.Println("latest msg: ", msg) // case msg := <-chSettings: // appCtx.Settings.Lock.Lock() // appCtx.Settings.Settings = msg // fmt.Println("settings channel: ", msg) // appCtx.Settings.Lock.Unlock() } } } func processIncoming(incoming model.Incoming_json, ctx *model.AppContext, writer *kafka.Writer) { id := mqttclient.GetBeaconID(incoming) _, ok := ctx.BeaconsLookup[id] if !ok { return } err := decodeBeacon(incoming, ctx, writer) if err != nil { fmt.Println("error in decoding") return } } func decodeBeacon(incoming model.Incoming_json, ctx *model.AppContext, writer *kafka.Writer) error { beacon := strings.TrimSpace(incoming.Data) id := mqttclient.GetBeaconID(incoming) if beacon == "" { return nil // How to return error?, do I even need to return error } b, err := hex.DecodeString(beacon) if err != nil { return err } // check for flag byte, if first AD structure is flag bytes, remove it if len(b) > 1 && b[1] == 0x01 { l := int(b[0]) // length of AD structure if 1+l <= len(b) { b = b[1+l:] } } adStructureIndeces := ParseADFast(b) event := model.BeaconEvent{} for _, r := range adStructureIndeces { ad := b[r[0]:r[1]] if checkIngics(ad) { event = parseIngicsState(ad) event.Id = id event.Name = id break } else if checkEddystoneTLM(ad) { event = parseEddystoneState(ad) event.Id = id event.Name = id break } else if checkMinewB7(ad) { fmt.Println("Minew B7 vendor format") break } } if event.Id != "" { prevEvent, ok := ctx.BeaconEvents.Beacons[id] ctx.BeaconEvents.Beacons[id] = event if ok && bytes.Equal(prevEvent.Hash(), event.Hash()) { return nil } eMsg, err := json.Marshal(event) if err != nil { return err } err = writer.WriteMessages(context.Background(), kafka.Message{ Value: eMsg, }) if err != nil { return err } fmt.Println("Message sent") } return nil } func checkIngics(ad []byte) bool { if len(ad) >= 6 && ad[1] == 0xFF && ad[2] == 0x59 && ad[3] == 0x00 && ad[4] == 0x80 && ad[5] == 0xBC { return true } return false } func parseIngicsState(ad []byte) model.BeaconEvent { return model.BeaconEvent{ Battery: uint32(binary.LittleEndian.Uint16(ad[6:8])), Event: int(ad[8]), Type: "Ingics", } } func checkEddystoneTLM(ad []byte) bool { if len(ad) >= 4 && ad[1] == 0x16 && ad[2] == 0xAA && ad[3] == 0xFE && ad[4] == 0x20 { return true } return false } func parseEddystoneState(ad []byte) model.BeaconEvent { return model.BeaconEvent{ Battery: uint32(binary.BigEndian.Uint16(ad[6:8])), Type: "Eddystone", } } // I dont think this is always true, but for testing is ok func checkMinewB7(ad []byte) bool { if len(ad) >= 4 && ad[1] == 0x16 && ad[2] == 0xE1 && ad[3] == 0xFF { return true } return false } func ParseADFast(b []byte) [][2]int { var res [][2]int i := 0 for i < len(b) { l := int(b[i]) if l == 0 || i+1+l > len(b) { break } res = append(res, [2]int{i, i + 1 + l}) i += 1 + l } return res } func getBeaconDistance(incoming model.Incoming_json) float64 { rssi := incoming.RSSI power := incoming.TX_power distance := 100.0 ratio := float64(rssi) * (1.0 / float64(twos_comp(power))) if ratio < 1.0 { distance = math.Pow(ratio, 10) } else { distance = (0.89976)*math.Pow(ratio, 7.7095) + 0.111 } return distance } func updateBeacon(beacon *model.Beacon, incoming model.Incoming_json) { now := time.Now().Unix() beacon.Incoming_JSON = incoming beacon.Last_seen = now beacon.Beacon_type = incoming.Beacon_type beacon.HB_ButtonCounter = incoming.HB_ButtonCounter beacon.HB_Battery = incoming.HB_Battery beacon.HB_RandomNonce = incoming.HB_RandomNonce beacon.HB_ButtonMode = incoming.HB_ButtonMode if beacon.Beacon_metrics == nil { beacon.Beacon_metrics = make([]model.BeaconMetric, 10) } metric := model.BeaconMetric{} metric.Distance = getBeaconDistance(incoming) metric.Timestamp = now metric.Rssi = int64(incoming.RSSI) metric.Location = incoming.Hostname beacon.Beacon_metrics = append(beacon.Beacon_metrics, metric) } func twos_comp(inp string) int64 { i, _ := strconv.ParseInt("0x"+inp, 0, 64) return i - 256 }