Matterbridge is optional and can appear many times

Matterbridge sender
2024-12-29 22:28:09 +01:00 · 2024-12-27 12:31:40 +01:00
5 changed files with 46 additions and 107 deletions
--- a/README.md
+++ b/README.md
@ -7,74 +7,24 @@ Warning: this is a proof of concept, don't rely on it
 It was very basically tested, and while it seems to work, approach it without any expectations.
 In other words - treat it as a toy, not as a tool that will save your life or valuables, because it probably won't.

-## Running
+## Usage
+

 ```
-$ TELEGRAM_APITOKEN=YOUR_API_TOKEN ./alarm-bot
+$ TELEGRAM_APITOKEN=YOUR_API_TOKEN ./alarm_bot --satel_addr=127.0.0.1 --satel_port=31337 --tg_chat_id=YOUR_CHAT_ID_FROM_BOTFATHER
 ```

-Remember that `hswro-alarm-bot.yml` should be present in the current directory.
-
-## Configuration via YAML file
-
-Configuration should be stored in `hswro-alarm-bot.yml` in current directory.
-
-```yaml
-satel-addr: "192.168.13.37"
-tg-chat-ids:
- - 1234
- - 5678
- - 9876
-allowed-types:
- - "zone-isolate"
- - "zone-alarm"
-allowed-indexes:
- - 5678
- - 1337
-pool-interval: 5m
-arm-callback-urls:
- - "http://192.168.1.10/hello"
- - "http://example.com/api"
-disarm-callback-urls:
- - "http://192.168.1.10/bye"
- - "http://example.com/api2"
-alarm-callback-urls:
- - "http://192.168.1.10/ohno"
- - "http://example.com/api3"
-matterbridge:
- - uri: "http://localhost:4242/api/message"
-   token: "test_token_1"
-   gateway: "test_gateway_1"
-   username: "test_username_1"
-```
-
- - `pool-interval` sets how often will the Satel device be asked for changes
- - `satel-addr` sets the IP address of the Satel device
- - `tg-chat-ids` sets which telegram groups will receive notifications about alarm state changes
-
-### Matterbridge integration
-
-As Telegram is kinda questionable it's beneficial to get the notifications somewhere else as well. This bot can utilize Matterbridge's API to send notifications to many different chat platforms. Check the [Matterbridge API documentation](https://github.com/42wim/matterbridge/wiki/Api) to learn how to configure your existing MAtterbridge installation.
-
-Tip: You can still have independent Telegram and Matterbridge notifications - simply set up an additional gateway in Matterbridge that pushes the messages to your non-telegram chat groups but omits the Telegram group to avoid duplicated messages. Those other chat groups can be the same as those where your regular messages are bridged - Matterbridge will handle that just fine without the need for duplicated bots etc.
-
-To configure that set the `matterbridge` part of the config file with the following parameters:
- - `uri` - URI to your Matterbridge's API
- - `token` - Token that you've set up for yout Matterbridge API. Situation where this token is not configured is not taken into account.
- - `gateway` - Set it to the same value that is set in the `name` parameter to your `[[gateway]]` in `matterbridge.toml`
- - `username` - The username from which the alarm bot messages will appear to API.
-
 ### Notification via HTTP callbacks

-Set the values in following parts of the config file:
+Set the following environment variables:

- `arm-callback-urls` - for an URL that will be POST when partition **0** is armed
- `disarm-callback-urls` - for an URL that will be POST when partition **0** is unarmed
- `alarm-callback-urls` - for an URL that will be POST when **any** partition alarm is activated
+- `NOTIFY_URL_ARM` - for an URL that will be POST when partition **0** is armed
+- `NOTIFY_URL_DISARM` - for an URL that will be POST when partition **0** is unarmed
+- `ALARM_URL_ARM` - for an URL that will be POST when **any** partition alarm is activated

 ### Filtering events by change type

-It's possible to filter events by change type. Use the `allowed-types` part of the config file to do that. If that parameter is not provided, then all change types are allowed, otherwise only the provided ones will be used for notifications.
+It's possible to filter events by change type. Use the `--allowed-types=TYPE1,TYPE2,...` command line parameter to do that. If that parameter is not provided, then all change types are allowed, otherwise only the provided ones will be used for notifications.

 Supported types are:
 - `zone-violation`
@ -119,7 +69,7 @@ Supported types are:

 ### Filtering events by index (which meaning depends on the change type)

-Use the `allowed-indexes` part of config file to set the list of allowed indexes. If that parameter is not provided, then all indexes are allowed; otherwise the notification is sent only for provided indexes.
+Use the `--allowed-indexes=1,2,3,...` command line parameter to set the list of allowed indexes (of course provide your own list instead of `1,2,3,...`). If that parameter is not provided, then all indexes are allowed; otherwise the notification is sent for all indexes.

 ## example systemd unit

@ -131,8 +81,10 @@ After=network.target

 [Service]
 Type=simple
-ExecStart=/path/to/alarm_bot
+ExecStart=/path/to/alarm_bot --satel-addr=192.168.13.37 --satel-port=7094 --tg-chat-id=1234,4567,9876
 Environment=TELEGRAM_APITOKEN=YOUR_API_TOKEN
+Environment=NOTIFY_URL_DISARM="http://localhost/disarmed"
+Environment=NOTIFY_URL_ARM="http://localhost/armed"
 DynamicUser=True
 RuntimeDirectory=hswro-alarm-bot
 StateDirectory=hswro-alarm-bot
--- a/debug_utils.go
+++ b/debug_utils.go
@ -21,10 +21,9 @@ func dumpMemoryProfile(log *log.Logger) {
 }

 func WriteMemoryProfilePeriodically(wg *sync.WaitGroup, log *log.Logger, close <-chan interface{}) {
-	wg.Add(1)
 	go func() {
+		wg.Add(1)
 		defer wg.Done()
-
 		memoryProfileTicker := time.NewTicker(24 * time.Hour)
 		defer memoryProfileTicker.Stop()
 		select {
--- a/filters.go
+++ b/filters.go
@ -26,20 +26,19 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes

 	if (len(allowedTypes) == 0) && (len(allowedIndexes) == 0) {
 		// no allowed types == all types are allowed
-		wg.Add(1)
 		go func() {
+			wg.Add(1)
 			defer wg.Done()
-			defer close(returnChan)

 			for e := range ev {
 				returnChan <- e
 			}
+			close(returnChan)
 		}()
 	} else {
-		wg.Add(1)
 		go func() {
+			wg.Add(1)
 			defer wg.Done()
-			defer close(returnChan)

 			for e := range ev {
 				retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
@ -52,6 +51,7 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes
 					returnChan <- retEv
 				}
 			}
+			close(returnChan)
 		}()
 	}

@ -61,10 +61,9 @@ func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes
 func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *DataStore, logger *log.Logger) <-chan satel.Event {
 	returnChan := make(chan satel.Event)

-	wg.Add(1)
 	go func() {
+		wg.Add(1)
 		defer wg.Done()
-		defer close(returnChan)

 		for e := range ev {
 			retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
@ -82,6 +81,7 @@ func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *Data
 			}
 		}
 		logger.Print("Satel disconnected.")
+		close(returnChan)
 	}()

 	return returnChan
@ -111,11 +111,8 @@ func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sle
 	returnChan := make(chan GenericMessage)
 	timeoutEvents := make(chan interface{})

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-		defer close(returnChan)
-
+		wg.Add(1)
 		var currentEvent *GenericMessage = nil
 	loop:
 		for {
@ -140,6 +137,8 @@ func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sle
 		if currentEvent != nil {
 			returnChan <- *currentEvent
 		}
+		close(returnChan)
+		wg.Done()
 	}()

 	return returnChan
--- a/filters_test.go
+++ b/filters_test.go
@ -16,13 +16,12 @@ func TestSatelEventTypeFiltering(t *testing.T) {
 	receivedEvents := make([]satel.Event, 0)
 	wg := sync.WaitGroup{}

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -45,13 +44,12 @@ func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testin
 	receivedEvents := make([]satel.Event, 0)
 	wg := sync.WaitGroup{}

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	for index, ct := range SUPPORTED_CHANGE_TYPES {
@ -72,13 +70,12 @@ func TestSatelIndexFiltering(t *testing.T) {
 	receivedEvents := make([]satel.Event, 0)
 	wg := sync.WaitGroup{}

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{1, 3}) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -102,13 +99,12 @@ func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T)
 	wg := sync.WaitGroup{}
 	myReasonableMaxIndex := 100 // I wanted to use math.MaxInt at first, but it's kind of a waste of time here

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	for i := 0; i < myReasonableMaxIndex; i++ {
@ -136,13 +132,12 @@ func TestSatelLastSeenFiltering(t *testing.T) {
 	fakeLog := log.New(io.Discard, "", log.Ltime)
 	ds := MakeDataStore(fakeLog, tempFileName)

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -173,13 +168,12 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
 	fakeLog := log.New(io.Discard, "", log.Ltime)
 	ds := MakeDataStore(fakeLog, tempFileName)

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
@ -200,13 +194,12 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
 	testEvents = make(chan satel.Event)
 	receivedEvents = make([]satel.Event, 0)
 	ds = MakeDataStore(fakeLog, tempFileName)
-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	receivedEvents = make([]satel.Event, 0)
@ -250,13 +243,12 @@ func TestThrottle(t *testing.T) {
 		tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
 	)

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-
+		wg.Add(1)
 		for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1}}
@ -303,12 +295,12 @@ func TestThrottle_ManyMessagesInOneEvent(t *testing.T) {
 		tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
 	)

-	wg.Add(1)
 	go func() {
-		defer wg.Done()
+		wg.Add(1)
 		for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
 			receivedEvents = append(receivedEvents, e)
 		}
+		wg.Done()
 	}()

 	testEvents <- makeMassiveEvent(tplMessageTest1, 100)
--- a/sender_worker.go
+++ b/sender_worker.go
@ -34,11 +34,9 @@ func Consume(events <-chan GenericMessage) {
 func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage {
 	returnEvents := make(chan GenericMessage)

-	wg.Add(1)
 	go func() {
+		wg.Add(1)
 		defer wg.Done()
-		defer close(returnEvents)
-
 		for e := range events {
 			returnEvents <- e
 			err := s.Send(e, tpl)
@ -47,6 +45,7 @@ func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger
 				panic(err)
 			}
 		}
+		close(returnEvents)
 	}()

 	return returnEvents
@ -78,11 +77,9 @@ func notifyAllHttp(urls []string, logger *log.Logger, wg *sync.WaitGroup) {
 func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage {
 	returnEvents := make(chan GenericMessage)

-	wg.Add(1)
 	go func() {
+		wg.Add(1)
 		defer wg.Done()
-		defer close(returnEvents)
-
 		for e := range events {
 			returnEvents <- e
 		inner_arm:
@ -107,6 +104,7 @@ func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.Wait
 			}

 		}
+		close(returnEvents)
 	}()

 	return returnEvents
@ -121,11 +119,9 @@ type MatterbridgeMessage struct {
 func SendToMatterbridge(events <-chan GenericMessage, s SatelNameGetter, config AppConfig, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage {
 	returnEvents := make(chan GenericMessage)

-	wg.Add(1)
 	go func() {
+		wg.Add(1)
 		defer wg.Done()
-		defer close(returnEvents)
-
 		for e := range events {
 			returnEvents <- e
 			for _, matterbridgeConfig := range config.Matterbridge {
@ -147,6 +143,7 @@ func SendToMatterbridge(events <-chan GenericMessage, s SatelNameGetter, config
 				logger.Print("Notified via Matterbridge with result ", res.StatusCode)
 			}
 		}
+		close(returnEvents)
 	}()

 	return returnEvents
Author	SHA1	Message	Date
Michał Rudowicz	0a7c15691e	Matterbridge is optional and can appear many times	2024-12-29 22:28:09 +01:00
Michał Rudowicz	d7dc07ecb3	Matterbridge sender	2024-12-27 12:31:40 +01:00