telegram/burrow-telegram.py

   1 #!/usr/bin/env python
   2 # A simple daemon for sending Telegram notifications about failures in the Burrow
   3 # (c) 2022 Martin Mareš <mj@ucw.cz>
   4
   5 from aiogram import Bot, Dispatcher, executor, types
   6 import asyncio
   7 import asyncio_mqtt
   8 from configparser import ConfigParser
   9 from datetime import datetime, timedelta
  10 import logging
  11 from logging.handlers import SysLogHandler
  12 import signal
  13 import ssl
  14 import sys
  15
  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(' ')))
  20 USE_SYSLOG = True
  21
  22 if USE_SYSLOG:
  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: '
  26 else:
  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)
  33
  34 bot = Bot(token=API_TOKEN)
  35 dispatcher = Dispatcher(bot)
  36
  37
  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")
  42
  43
  44 @dispatcher.message_handler()
  45 async def echo(message: types.Message):
  46     print(message)
  47     if message.text.startswith("xyzzy"):
  48         await send_msg("Nothing happens.")
  49     # await message.answer(message.text)
  50
  51
  52 BUCKET_CAPACITY = 5
  53 BUCKET_REFRESH = timedelta(seconds=10)
  54 bucket = BUCKET_CAPACITY
  55 bucket_last_fill = datetime.now()
  56
  57
  58 async def send_msg(text):
  59     global bucket, bucket_last_fill
  60     logger.info(f'Sending message: {text}')
  61
  62     if bucket < BUCKET_CAPACITY:
  63         now = datetime.now()
  64         while bucket < BUCKET_CAPACITY and bucket_last_fill + BUCKET_REFRESH <= now:
  65             bucket_last_fill += BUCKET_REFRESH
  66             bucket += 1
  67             logger.debug(f'Bucket refill: {bucket}')
  68
  69     if bucket > 0:
  70         bucket -= 1
  71         logger.debug(f'Bucket drain: {bucket}')
  72
  73         for chat in CHATS:
  74             await bot.send_message(chat, text + '\n')
  75
  76     else:
  77         logger.info('Bucket empty :(')
  78
  79
  80 hyst_state = {}
  81
  82 def hysteresis(key, value, low, high):
  83     if key in hyst_state:
  84         old_state = hyst_state[key]
  85     else:
  86         old_state = 0
  87     if value is None:
  88         new_state = 0
  89     elif old_state <= 0:
  90         if value >= high:
  91             new_state = 1
  92         else:
  93             new_state = -1
  94     else:
  95         if value <= low:
  96             new_state = -1
  97         else:
  98             new_state = 1
  99     hyst_state[key] = new_state
 100     return new_state
 101
 102
 103 last_temp_state = 1
 104 last_boiler_err = 0
 105
 106
 107 async def mqtt_process_msg(topic, val):
 108     global last_temp_state, last_temp_warning
 109     global last_boiler_err
 110     now = datetime.now()
 111
 112     logger.debug(f'MQTT: {topic} -> {val}')
 113
 114     if topic == 'burrow/temp/catarium':
 115         temp, when = map(float, val.split(' '))
 116         if when < now.timestamp() - 3600:
 117             temp_state = 0
 118         else:
 119             temp_state = hysteresis('catarium-temp', temp, 23.2, 23.5)
 120         if temp_state != last_temp_state:
 121             last_temp_state = temp_state
 122             if temp_state == 0:
 123                 await send_msg('Teplotní čidlo v pracovně nefunguje.')
 124             else:
 125                 await send_msg(f'Teplota v pracovně: {temp} °C')
 126
 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}')
 132
 133
 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')
 139
 140     async with asyncio_mqtt.Client(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())
 146
 147
 148 async def mqtt_watcher():
 149     while True:
 150         try:
 151             logger.info("Starting MQTT")
 152             await mqtt_loop()
 153         except asyncio_mqtt.MqttError as error:
 154             logger.error(f"MQTT error: {error}")
 155         await asyncio.sleep(10)
 156
 157
 158 async def fortunes():
 159     await asyncio.sleep(5*60)
 160     while True:
 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())
 165         else:
 166             logger.error(f'fortune failed with return code {proc.returncode}')
 167         await asyncio.sleep(24*60*60)
 168
 169
 170 async def main():
 171     loop = asyncio.get_event_loop()
 172     coros = [
 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()),
 176     ]
 177     for coro in asyncio.as_completed(coros):
 178         done = await coro
 179         done.result()       # The coroutine probably died of an exception, which is raised here.
 180
 181
 182 asyncio.run(main())