hswro-alarm-bot/main.go

package main

import (
	"flag"
	"fmt"
	"html/template"
	"log"
	"net"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"git.sr.ht/~michalr/go-satel"
	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
)

const (
	PersistenceFilename = "hs_wro_last_seen.bin"
)

type TgSender struct {
	bot     *tgbotapi.BotAPI
	s       SatelNameGetter
	logger  *log.Logger
	chatIds []int64
}

func (self TgSender) Send(msg GenericMessage, tpl *template.Template) error {
	if len(self.chatIds) == 0 {
		return nil
	}
	message := msg.Format(tpl, self.s, self.logger)
	for _, chatId := range self.chatIds {
		toSend := tgbotapi.NewMessage(chatId, message)
		toSend.ParseMode = "HTML"
		_, err := self.bot.Send(toSend)
		if err != nil {
			return err
		}
	}
	return nil
}

func sendTgMessage(tgEvents chan GenericMessage, msg []satel.BasicEventElement, chatId int64) {
	tgEvents <- GenericMessage{msg}
}

type RealSleeper struct {
	duration time.Duration
}

func (self RealSleeper) Sleep(ch chan<- interface{}) {
	go func() {
		time.Sleep(self.duration)
		ch <- nil
	}()
}

func getCmdLineParams(logger *log.Logger) (string, []int64, []satel.ChangeType, []int, time.Duration) {
	satelApiAddr := flag.String("satel-addr", "", "Address that should be used to connect to the SATEL device")
	satelApiPort := flag.String("satel-port", "7094", "Port that should be used to connect to the SATEL device")
	chatIdRaw := flag.String("tg-chat-id", "", "Telegram Chat ID where to send updates. Use \",\" to specify multiple IDs.")
	allowedTypesRaw := flag.String("allowed-types", "", "Satel change types that are allowed. All other types will be discarded. By default all are allowed. Use \",\" to specify multiple types.")
	allowedIndexesRaw := flag.String("allowed-indexes", "", "Satel indexes (zones?) that are allowed. All other indexes will be discarded. By default all are allowed. Use \",\" to specify multiple indexes.")
	satelPoolInterval := flag.Duration("pool-interval", 5*time.Second, "How often should the SATEL device be pooled for changes? Default: 5 seconds.")
	flag.Parse()

	if len(*satelApiAddr) == 0 || len(*satelApiPort) == 0 || len(*chatIdRaw) == 0 {
		logger.Fatal("Use --satel-addr=ADDR, --satel-port=PORT and --tg-chat-id=CHAT_ID command line flags to continue.")
	}
	chatIdsStrings := strings.Split(*chatIdRaw, ",")
	var chatIds []int64
	for _, chatIdStr := range chatIdsStrings {
		chatId, err := strconv.ParseInt(chatIdStr, 10, 64)
		if err != nil {
			logger.Fatalf("Tried to use a non-int value for one of tg_chatIds: %s. That's bad.", chatIdStr)
		}
		chatIds = append(chatIds, chatId)
	}
	allowedTypesStrings := strings.Split(*allowedTypesRaw, ",")
	var allowedTypes []satel.ChangeType
	for _, allowedTypeStr := range allowedTypesStrings {
		if len(allowedTypeStr) == 0 {
			continue
		}
		allowedType, err := StringToSatelChangeType(allowedTypeStr)
		if err != nil {
			logger.Fatalf("Error trying to understand an allowed type: %s.", err)
		}
		allowedTypes = append(allowedTypes, allowedType)
	}
	allowedIndexesStrings := strings.Split(*allowedIndexesRaw, ",")
	var allowedIndexes []int
	for _, allowedIndexStr := range allowedIndexesStrings {
		if len(allowedIndexStr) == 0 {
			continue
		}
		allowedIndex, err := strconv.ParseInt(allowedIndexStr, 10, 0)
		if err != nil {
			logger.Fatalf("Tried to use a non-int value for one of allowed indexes: %s. That's bad.", allowedIndexStr)
		}
		allowedIndexes = append(allowedIndexes, int(allowedIndex))
	}

	satelAddr := fmt.Sprintf("%s:%s", *satelApiAddr, *satelApiPort)
	return satelAddr, chatIds, allowedTypes, allowedIndexes, *satelPoolInterval
}

func makeSatel(satelAddr string, poolInterval time.Duration) *satel.Satel {
	satelConn, err := net.Dial("tcp", satelAddr)
	if err != nil {
		panic(err)
	}
	return satel.NewConfig(satelConn, satel.Config{EventsQueueSize: 10, PoolingInterval: poolInterval})
}

func getPersistenceFilePath() string {
	var stateDir = os.Getenv("STATE_DIRECTORY")
	if len(stateDir) != 0 {
		return filepath.Join(stateDir, PersistenceFilename)
	}
	return PersistenceFilename
}

func main() {
	var (
		wg       sync.WaitGroup
		tgEvents = make(chan GenericMessage, 5)
		logger   = log.New(os.Stderr, "Main", log.Lmicroseconds)
		sleeper  = RealSleeper{time.Second * 60}
	)

	satelAddr, chatIds, allowedTypes, allowedIndexes, poolInterval := getCmdLineParams(logger)

	s := makeSatel(satelAddr, poolInterval)
	logger.Printf("Connected to Satel: %s", satelAddr)

	bot, err := tgbotapi.NewBotAPI(os.Getenv("TELEGRAM_APITOKEN"))
	if err != nil {
		panic(err)
	}
	logger.Print("Created Telegram Bot API client")

	tgSender := TgSender{bot, s, log.New(os.Stderr, "TgFormatter", log.Lmicroseconds), chatIds}

	tpl := template.Must(template.New("TelegramMessage").Parse(TelegramMessageTemplate))

	dataStore := MakeDataStore(log.New(os.Stderr, "DataStore", log.Lmicroseconds), getPersistenceFilePath())

	Consume(
		SendToTg(Throttle(NotifyViaHTTP(tgEvents, &wg, log.New(os.Stderr, "HTTPNotify", log.Lmicroseconds)),
			&wg, sleeper, log.New(os.Stderr, "MessageThrottle", log.Lmicroseconds)),
			tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tpl),
	)

	go CloseSatelOnCtrlC(s)

	for e := range FilterByTypeOrIndex(
		FilterByLastSeen(s.Events, &wg, &dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
		&wg, allowedTypes, allowedIndexes) {
		logger.Print("Received change from SATEL: ", e)
		for _, chatId := range chatIds {
			sendTgMessage(tgEvents, e.BasicEvents, chatId)
		}
	}

	close(tgEvents)
	wg.Wait()
}