package main import ( "flag" "fmt" "html/template" "log" "net" "os" "path/filepath" "strconv" "strings" "sync" "sync/atomic" "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" retryDelaySec = 25 ) 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} stopRequested = atomic.Bool{} ) stopRequested.Store(false) 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, &stopRequested) for stopRequested.Load() == false { 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) tgEvents <- GenericMessage{e.BasicEvents} } if stopRequested.Load() == false { logger.Printf("Waiting %d seconds before trying again", retryDelaySec) time.Sleep(retryDelaySec * time.Second) } } close(tgEvents) wg.Wait() }