2 # A simple daemon for sending Telegram notifications about failures in the Burrow
3 # (c) 2022 Martin Mareš <mj@ucw.cz>
5 from aiogram import Bot, Dispatcher, executor, types
8 from configparser import ConfigParser
9 from datetime import datetime, timedelta
11 from logging.handlers import SysLogHandler
16 config = ConfigParser()
17 config.read('/usr/local/etc/burrow-telegram')
18 API_TOKEN = config['telegram']['api_token']
19 CHATS = list(map(int, config['telegram']['chats'].split(' ')))
23 formatter = logging.Formatter(fmt="%(message)s") # systemd will handle the rest
24 log_handler = SysLogHandler('/dev/log', facility=SysLogHandler.LOG_LOCAL1)
25 log_handler.ident = 'burrow-telegram: '
27 formatter = logging.Formatter(fmt="%(asctime)s %(name)s.%(levelname)s: %(message)s", datefmt='%Y-%m-%d %H:%M:%S')
28 log_handler = logging.StreamHandler(stream=sys.stdout)
29 log_handler.setFormatter(formatter)
30 logger = logging.getLogger()
31 logger.setLevel(logging.INFO)
32 logger.addHandler(log_handler)
34 bot = Bot(token=API_TOKEN)
35 dispatcher = Dispatcher(bot)
38 @dispatcher.message_handler(commands=['start', 'help'])
39 async def send_welcome(message: types.Message):
40 logger.info(f'Start from {message.chat}')
41 await message.reply("Brum!\nI'm BurrowBot!\n")
44 @dispatcher.message_handler()
45 async def echo(message: types.Message):
47 if message.text.startswith("xyzzy"):
48 await send_msg("Nothing happens.")
49 # await message.answer(message.text)
53 BUCKET_REFRESH = timedelta(seconds=10)
54 bucket = BUCKET_CAPACITY
55 bucket_last_fill = datetime.now()
58 async def send_msg(text):
59 global bucket, bucket_last_fill
60 logger.info(f'Sending message: {text}')
62 if bucket < BUCKET_CAPACITY:
64 while bucket < BUCKET_CAPACITY and bucket_last_fill + BUCKET_REFRESH <= now:
65 bucket_last_fill += BUCKET_REFRESH
67 logger.debug(f'Bucket refill: {bucket}')
71 logger.debug(f'Bucket drain: {bucket}')
74 await bot.send_message(chat, text + '\n')
77 logger.info('Bucket empty :(')
82 def hysteresis(key, value, low, high):
84 old_state = hyst_state[key]
99 hyst_state[key] = new_state
107 async def mqtt_process_msg(topic, val):
108 global last_temp_state, last_temp_warning
109 global last_boiler_err
112 logger.debug(f'MQTT: {topic} -> {val}')
114 if topic == 'burrow/temp/catarium':
115 temp, when = map(float, val.split(' '))
116 if when < now.timestamp() - 3600:
119 temp_state = hysteresis('catarium-temp', temp, 22, 23)
120 if temp_state != last_temp_state:
121 last_temp_state = temp_state
123 await send_msg('Teplotní čidlo v pracovně nefunguje.')
125 await send_msg(f'Teplota v pracovně: {temp} °C')
127 if topic == 'burrow/heating/error':
128 err = int(val.split(' ')[0])
129 if err != last_boiler_err:
130 last_boiler_err = err
131 await send_msg(f'Chyba kotle: {err}')
134 async def mqtt_loop():
135 sctx = ssl.SSLContext(ssl.PROTOCOL_TLS)
136 sctx.verify_mode = ssl.CERT_REQUIRED
137 sctx.load_cert_chain('/etc/burrow-mqtt/client.crt', '/etc/burrow-mqtt/client.key')
138 sctx.load_verify_locations(cafile='/etc/burrow-mqtt/ca.crt')
140 async with asyncio_mqtt.Client(client_id='telegram', hostname="burrow-mqtt", port=8883, tls_context=sctx) as mqtt:
141 async with mqtt.unfiltered_messages() as messages:
142 await mqtt.subscribe("burrow/heating/#")
143 await mqtt.subscribe("burrow/temp/#")
144 async for msg in messages:
145 await mqtt_process_msg(msg.topic, msg.payload.decode())
148 async def mqtt_watcher():
151 logger.info("Starting MQTT")
153 except asyncio_mqtt.MqttError as error:
154 logger.error(f"MQTT error: {error}")
155 await asyncio.sleep(10)
158 async def fortunes():
159 await asyncio.sleep(5*60)
161 proc = await asyncio.create_subprocess_exec('/usr/games/fortune', stdout=asyncio.subprocess.PIPE)
162 out, err = await proc.communicate()
163 if proc.returncode == 0:
164 await send_msg(out.decode())
166 logger.error(f'fortune failed with return code {proc.returncode}')
167 await asyncio.sleep(24*60*60)
171 loop = asyncio.get_event_loop()
173 loop.create_task(mqtt_watcher()),
174 loop.create_task(dispatcher.start_polling(timeout=60, relax=0.01, fast=True, allowed_updates=None)),
175 loop.create_task(fortunes()),
177 for coro in asyncio.as_completed(coros):
179 done.result() # The coroutine probably died of an exception, which is raised here.