#!/usr/bin/env python3
from __future__ import annotations
import __py_include
import logging
import locale
import queue
import time
import threading
import paho.mqtt.client as mqtt
from homekit.telegram import bot
from homekit.mqtt import Mqtt
from homekit.config import config
from homekit.util import chunks
from syncleo import (
Kettle,
PowerType,
DeviceListener,
IncomingMessageListener,
ConnectionStatusListener,
ConnectionStatus
)
import syncleo.protocol as kettle_proto
from typing import Optional, Tuple, List, Union
from collections import namedtuple
from functools import partial
from datetime import datetime
from abc import abstractmethod
from telegram.error import TelegramError
from telegram import (
ReplyKeyboardMarkup,
InlineKeyboardMarkup,
InlineKeyboardButton,
Message
)
from telegram.ext import (
CallbackQueryHandler,
MessageHandler,
CommandHandler
)
logger = logging.getLogger(__name__)
config.load_app('polaris_kettle_bot')
primary_choices = (70, 80, 90, 100)
all_choices = range(
config['kettle']['temp_min'],
config['kettle']['temp_max']+1,
config['kettle']['temp_step'])
bot.initialize()
bot.lang.ru(
start_message="Выберите команду на клавиатуре:",
invalid_command="Неизвестная команда",
unexpected_callback_data="Ошибка: неверные данные",
disable="❌ Выключить",
server_error="Ошибка сервера",
back="🔙 Назад",
smth_went_wrong="😱 Что-то пошло не так",
# /status
status_not_connected="😟 Связь с чайником не установлена",
status_on="🟢 Чайник включён (до %d °C)",
status_off="🔴 Чайник выключен",
status_current_temp="Сейчас: %d °C",
status_update_time="Обновлено %s",
status_update_time_fmt="%d %b в %H:%M:%S",
# /temp
select_temperature="Выберите температуру:",
# enable/disable
enabling="💤 Чайник включается...",
disabling="💤 Чайник выключается...",
enabled="🟢 Чайник включён.",
enabled_target="%s Цель: %d °C",
enabled_reached="✅ Готово! Чайник вскипел, температура %d °C.",
disabled="✅ Чайник выключен.",
please_wait="⏳ Ожидайте..."
)
bot.lang.en(
start_message="Select command on the keyboard:",
invalid_command="Unknown command",
unexpected_callback_data="Unexpected callback data",
disable="❌ Turn OFF",
server_error="Server error",
back="🔙 Back",
smth_went_wrong="😱 Something went wrong",
# /status
status_not_connected="😟 No connection",
status_on="🟢 Turned ON! Target: %d °C",
status_off="🔴 Turned OFF",
status_current_temp="Now: %d °C",
status_update_time="Updated on %s",
status_update_time_fmt="%b %d, %Y at %H:%M:%S",
# /temp
select_temperature="Select a temperature:",
# enable/disable
enabling="💤 Turning on...",
disabling="💤 Turning off...",
enabled="🟢 The kettle is turned ON.",
enabled_target="%s Target: %d °C",
enabled_reached="✅ Done! The kettle has boiled, the temperature is %d °C.",
disabled="✅ The kettle is turned OFF.",
please_wait="⏳ Please wait..."
)
kc: Optional[KettleController] = None
RenderedContent = Tuple[str, Optional[Union[InlineKeyboardMarkup, ReplyKeyboardMarkup]]]
tasks_lock = threading.Lock()
def run_tasks(tasks: queue.SimpleQueue, done: callable):
def next_task(r: Optional[kettle_proto.MessageResponse]):
if r is not None:
try:
assert r is not False, 'server error'
except AssertionError as exc:
logger.exception(exc)
tasks_lock.release()
return done(False)
if not tasks.empty():
task = tasks.get()
args = task[1:]
args.append(next_task)
f = getattr(kc.kettle, task[0])
f(*args)
else:
tasks_lock.release()
return done(True)
tasks_lock.acquire()
next_task(None)
def temperature_emoji(temp: int) -> str:
if temp > 90:
return '🔥'
elif temp >= 40:
return '♨️'
elif temp >= 35:
return '🌡'
else:
return '❄️'
class KettleInfoListener:
@abstractmethod
def info_updated(self, field: str):
pass
# class that holds data coming from the kettle over mqtt
class KettleInfo:
update_time: int
_mode: Optional[PowerType]
_temperature: Optional[int]
_target_temperature: Optional[int]
_update_listener: KettleInfoListener
def __init__(self, update_listener: KettleInfoListener):
self.update_time = 0
self._mode = None
self._temperature = None
self._target_temperature = None
self._update_listener = update_listener
def _update(self, field: str):
self.update_time = int(time.time())
if self._update_listener:
self._update_listener.info_updated(field)
@property
def temperature(self) -> int:
return self._temperature
@temperature.setter
def temperature(self, value: int):
self._temperature = value
self._update('temperature')
@property
def mode(self) -> PowerType:
return self._mode
@mode.setter
def mode(self, value: PowerType):
self._mode = value
self._update('mode')
@property
def target_temperature(self) -> int:
return self._target_temperature
@target_temperature.setter
def target_temperature(self, value: int):
self._target_temperature = value
self._update('target_temperature')
class KettleController(threading.Thread,
Mqtt,
DeviceListener,
IncomingMessageListener,
KettleInfoListener,
ConnectionStatusListener):
kettle: Kettle
info: KettleInfo
_logger: logging.Logger
_stopped: bool
_restart_server_at: int
_lock: threading.Lock
_info_lock: threading.Lock
_accumulated_updates: dict
_info_flushed_time: float
_mqtt_root_topic: str
_muts: List[MessageUpdatingTarget]
def __init__(self):
# basic setup
Mqtt.__init__(self, clean_session=False)
threading.Thread.__init__(self)
self._logger = logging.getLogger(self.__class__.__name__)
self.kettle = Kettle(mac=config['kettle']['mac'],
device_token=config['kettle']['token'],
read_timeout=config['kettle']['read_timeout'])
self.kettle_reconnect()
# info
self.info = KettleInfo(update_listener=self)
self._accumulated_updates = {}
self._info_flushed_time = 0
# mqtt
self._mqtt_root_topic = '/polaris/6/'+config['kettle']['token']+'/#'
self.connect_and_loop(loop_forever=False)
# thread loop related
self._stopped = False
# self._lock = threading.Lock()
self._info_lock = threading.Lock()
self._restart_server_at = 0
# bot
self._muts = []
self._muts_lock = threading.Lock()
self.start()
def kettle_reconnect(self):
self.kettle.discover(wait=False, listener=self)
def stop_all(self):
self.kettle.stop_all()
self._stopped = True
def add_updating_message(self, mut: MessageUpdatingTarget):
with self._muts_lock:
for m in self._muts:
if m.user_id == m.user_id and m.user_did_turn_on() or m.user_did_turn_on() != mut.user_did_turn_on():
m.delete()
self._muts.append(mut)
# ---------------------
# threading.Thread impl
def run(self):
while not self._stopped:
updates = []
deletions = []
forget = []
with self._muts_lock and self._info_lock:
if self._muts and self._accumulated_updates and (self._info_flushed_time == 0 or time.time() - self._info_flushed_time >= 1):
deletions = []
for mut in self._muts:
upd = mut.update(
mode=self.info.mode,
current_temp=self.info.temperature,
target_temp=self.info.target_temperature)
if upd.finished or upd.delete:
forget.append(mut)
if upd.delete:
deletions.append((mut, upd))
elif upd.changed:
updates.append((mut, upd))
self._info_flushed_time = time.time()
self._accumulated_updates = {}
# edit messages
for mut, upd in updates:
self._logger.debug(f'loop: got update: {upd}')
try:
do_edit = True
if upd.finished:
# try to delete the old message and send a new one, to notify user more effectively
try:
bot.delete_message(upd.user_id, upd.message_id)
do_edit = False
except TelegramError as exc:
self._logger.error(f'loop: failed to delete old message (in order to send a new one)')
self._logger.exception(exc)
if do_edit:
bot.edit_message_text(upd.user_id, upd.message_id,
text=upd.html,
reply_markup=upd.markup)
else:
bot.notify_user(upd.user_id, upd.html, reply_markup=upd.markup)
except TelegramError as exc:
if "Message can't be edited" in exc.message:
self._logger.warning("message can't be edited, adding it to forget list")
forget.append(upd)
self._logger.error(f'loop: edit_message_text failed for update: {upd}')
self._logger.exception(exc)
# delete messages
for mut, upd in deletions:
self._logger.debug(f'loop: got deletion: {upd}')
try:
bot.delete_message(upd.user_id, upd.message_id)
except TelegramError as exc:
self._logger.error(f'loop: delete_message failed for update: {upd}')
self._logger.exception(exc)
# delete muts, if needed
if forget:
with self._muts_lock:
for mut in forget:
self._logger.debug(f'loop: removing mut {mut}')
self._muts.remove(mut)
time.sleep(0.5)
# -------------------
# DeviceListener impl
def device_updated(self):
self._logger.info(f'device updated: {self.kettle.device.si}')
self.kettle.start_server_if_needed(incoming_message_listener=self,
connection_status_listener=self)
# -----------------------
# KettleInfoListener impl
def info_updated(self, field: str):
with self._info_lock:
newval = getattr(self.info, field)
self._logger.debug(f'info_updated: updated {field}, new value is {newval}')
self._accumulated_updates[field] = newval
# ----------------------------
# IncomingMessageListener impl
def incoming_message(self, message: kettle_proto.Message) -> Optional[kettle_proto.Message]:
self._logger.info(f'incoming message: {message}')
if isinstance(message, kettle_proto.ModeMessage):
self.info.mode = message.pt
elif isinstance(message, kettle_proto.CurrentTemperatureMessage):
self.info.temperature = message.current_temperature
elif isinstance(message, kettle_proto.TargetTemperatureMessage):
self.info.target_temperature = message.temperature
return kettle_proto.AckMessage()
# -----------------------------
# ConnectionStatusListener impl
def connection_status_updated(self, status: ConnectionStatus):
self._logger.info(f'connection status updated: {status}')
if status == ConnectionStatus.DISCONNECTED:
self.kettle.stop_all()
self.kettle_reconnect()
# -------------
# MQTTBase impl
def on_connect(self, client: mqtt.Client, userdata, flags, rc):
super().on_connect(client, userdata, flags, rc)
client.subscribe(self._mqtt_root_topic, qos=1)
self._logger.info(f'subscribed to {self._mqtt_root_topic}')
def on_message(self, client: mqtt.Client, userdata, msg):
try:
topic = msg.topic[len(self._mqtt_root_topic)-2:]
pld = msg.payload.decode()
self._logger.debug(f'mqtt: on message: topic={topic} pld={pld}')
if topic == 'state/sensor/temperature':
self.info.temperature = int(float(pld))
elif topic == 'state/mode':
self.info.mode = PowerType(int(pld))
elif topic == 'state/temperature':
self.info.target_temperature = int(float(pld))
except Exception as e:
self._logger.exception(str(e))
class Renderer:
@classmethod
def index(cls, ctx: bot.Context) -> RenderedContent:
html = f'{ctx.lang("settings")}\n\n'
html += ctx.lang('select_place')
return html, None
@classmethod
def status(cls, ctx: bot.Context,
connected: bool,
mode: PowerType,
current_temp: int,
target_temp: int,
update_time: int) -> RenderedContent:
if not connected:
return cls.not_connected(ctx)
else:
# power status
if mode != PowerType.OFF:
html = ctx.lang('status_on', target_temp)
else:
html = ctx.lang('status_off')
# current temperature
html += '\n'
html += ctx.lang('status_current_temp', current_temp)
# updated on
html += '\n'
html += cls.updated(ctx, update_time)
return html, None
@classmethod
def temp(cls, ctx: bot.Context, choices) -> RenderedContent:
buttons = []
for chunk in chunks(choices, 5):
buttons.append([f'{temperature_emoji(n)} {n}' for n in chunk])
buttons.append([ctx.lang('back')])
return ctx.lang('select_temperature'), ReplyKeyboardMarkup(buttons)
@classmethod
def turned_on(cls, ctx: bot.Context,
target_temp: int,
current_temp: int,
mode: PowerType,
update_time: Optional[int] = None,
reached=False,
no_keyboard=False) -> RenderedContent:
if mode == PowerType.OFF and not reached:
html = ctx.lang('enabling')
else:
if not reached:
html = ctx.lang('enabled')
# target temperature
html += '\n'
html += ctx.lang('enabled_target', temperature_emoji(target_temp), target_temp)
# current temperature
html += '\n'
html += temperature_emoji(current_temp) + ' '
html += ctx.lang('status_current_temp', current_temp)
else:
html = ctx.lang('enabled_reached', current_temp)
# updated on
if not reached and update_time is not None:
html += '\n'
html += cls.updated(ctx, update_time)
return html, None if no_keyboard else cls.wait_buttons(ctx)
@classmethod
def turned_off(cls, ctx: bot.Context,
mode: PowerType,
update_time: Optional[int] = None,
reached=False,
no_keyboard=False) -> RenderedContent:
if mode != PowerType.OFF:
html = ctx.lang('disabling')
else:
html = ctx.lang('disabled')
# updated on
if not reached and update_time is not None:
html += '\n'
html += cls.updated(ctx, update_time)
return html, None if no_keyboard else cls.wait_buttons(ctx)
@classmethod
def not_connected(cls, ctx: bot.Context) -> RenderedContent:
return ctx.lang('status_not_connected'), None
@classmethod
def smth_went_wrong(cls, ctx: bot.Context) -> RenderedContent:
html = ctx.lang('smth_went_wrong')
return html, None
@classmethod
def updated(cls, ctx: bot.Context, update_time: int):
locale_bak = locale.getlocale(locale.LC_TIME)
locale.setlocale(locale.LC_TIME, 'ru_RU.UTF-8' if ctx.user_lang == 'ru' else 'en_US.UTF-8')
dt = datetime.fromtimestamp(update_time)
html = ctx.lang('status_update_time', dt.strftime(ctx.lang('status_update_time_fmt')))
locale.setlocale(locale.LC_TIME, locale_bak)
return html
@classmethod
def wait_buttons(cls, ctx: bot.Context):
return InlineKeyboardMarkup([
[
InlineKeyboardButton(ctx.lang('please_wait'), callback_data='wait')
]
])
MUTUpdate = namedtuple('MUTUpdate', 'message_id, user_id, finished, changed, delete, html, markup')
class MessageUpdatingTarget:
ctx: bot.Context
message: Message
user_target_temp: Optional[int]
user_enabled_power_mode: PowerType
initial_power_mode: PowerType
need_to_delete: bool
rendered_content: Optional[RenderedContent]
def __init__(self,
ctx: bot.Context,
message: Message,
user_enabled_power_mode: PowerType,
initial_power_mode: PowerType,
user_target_temp: Optional[int] = None):
self.ctx = ctx
self.message = message
self.initial_power_mode = initial_power_mode
self.user_enabled_power_mode = user_enabled_power_mode
self.ignore_pm = initial_power_mode is PowerType.OFF and self.user_did_turn_on()
self.user_target_temp = user_target_temp
self.need_to_delete = False
self.rendered_content = None
self.last_reported_temp = None
def set_rendered_content(self, content: RenderedContent):
self.rendered_content = content
def rendered_content_changed(self, content: RenderedContent) -> bool:
return content != self.rendered_content
def update(self,
mode: PowerType,
current_temp: int,
target_temp: int) -> MUTUpdate:
# determine whether status updating is finished
finished = False
reached = False
if self.ignore_pm:
if mode != PowerType.OFF:
self.ignore_pm = False
elif mode == PowerType.OFF:
reached = True
if self.user_did_turn_on():
# when target is 100 degrees, this kettle sometimes turns off at 91, sometimes at 95, sometimes at 98.
# it's totally unpredictable, so in this case, we keep updating the message until it reaches at least 97
# degrees, or if temperature started dropping.
if self.user_target_temp < 100 \
or current_temp >= self.user_target_temp - 3 \
or current_temp < self.last_reported_temp:
finished = True
else:
finished = True
self.last_reported_temp = current_temp
# render message
if self.user_did_turn_on():
rc = Renderer.turned_on(self.ctx,
target_temp=target_temp,
current_temp=current_temp,
mode=mode,
reached=reached,
no_keyboard=finished)
else:
rc = Renderer.turned_off(self.ctx,
mode=mode,
reached=reached,
no_keyboard=finished)
changed = self.rendered_content_changed(rc)
update = MUTUpdate(message_id=self.message.message_id,
user_id=self.ctx.user_id,
finished=finished,
changed=changed,
delete=self.need_to_delete,
html=rc[0],
markup=rc[1])
if changed:
self.set_rendered_content(rc)
return update
def user_did_turn_on(self) -> bool:
return self.user_enabled_power_mode in (PowerType.ON, PowerType.CUSTOM)
def delete(self):
self.need_to_delete = True
@property
def user_id(self) -> int:
return self.ctx.user_id
@bot.handler(command='status')
def status(ctx: bot.Context) -> None:
text, markup = Renderer.status(ctx,
connected=kc.kettle.is_connected(),
mode=kc.info.mode,
current_temp=kc.info.temperature,
target_temp=kc.info.target_temperature,
update_time=kc.info.update_time)
ctx.reply(text, markup=markup)
@bot.handler(command='temp')
def temp(ctx: bot.Context) -> None:
text, markup = Renderer.temp(
ctx, choices=all_choices)
ctx.reply(text, markup=markup)
def enable(temp: int, ctx: bot.Context) -> None:
if not kc.kettle.is_connected():
text, markup = Renderer.not_connected(ctx)
ctx.reply(text, markup=markup)
return
tasks = queue.SimpleQueue()
if temp == 100:
power_mode = PowerType.ON
else:
power_mode = PowerType.CUSTOM
tasks.put(['set_target_temperature', temp])
tasks.put(['set_power', power_mode])
def done(ok: bool):
if not ok:
html, markup = Renderer.smth_went_wrong(ctx)
else:
html, markup = Renderer.turned_on(ctx,
target_temp=temp,
current_temp=kc.info.temperature,
mode=kc.info.mode)
message = ctx.reply(html, markup=markup)
logger.debug(f'ctx.reply returned message: {message}')
if ok:
mut = MessageUpdatingTarget(ctx, message,
initial_power_mode=kc.info.mode,
user_enabled_power_mode=power_mode,
user_target_temp=temp)
mut.set_rendered_content((html, markup))
kc.add_updating_message(mut)
run_tasks(tasks, done)
@bot.handler(message='disable')
def disable(ctx: bot.Context):
if not kc.kettle.is_connected():
text, markup = Renderer.not_connected(ctx)
ctx.reply(text, markup=markup)
return
def done(ok: bool):
mode = kc.info.mode
if not ok:
html, markup = Renderer.smth_went_wrong(ctx)
else:
kw = {}
if mode == PowerType.OFF:
kw['reached'] = True
kw['no_keyboard'] = True
html, markup = Renderer.turned_off(ctx, mode=mode, **kw)
message = ctx.reply(html, markup=markup)
logger.debug(f'ctx.reply returned message: {message}')
if ok and mode != PowerType.OFF:
mut = MessageUpdatingTarget(ctx, message,
initial_power_mode=mode,
user_enabled_power_mode=PowerType.OFF)
mut.set_rendered_content((html, markup))
kc.add_updating_message(mut)
tasks = queue.SimpleQueue()
tasks.put(['set_power', PowerType.OFF])
run_tasks(tasks, done)
@bot.handler(message='back')
def back(ctx: bot.Context):
bot.start(ctx)
@bot.defaultreplymarkup
def defaultmarkup(ctx: Optional[bot.Context]) -> Optional[ReplyKeyboardMarkup]:
buttons = [
[f'{temperature_emoji(n)} {n}' for n in primary_choices],
[ctx.lang('disable')]
]
return ReplyKeyboardMarkup(buttons, one_time_keyboard=False)
if __name__ == '__main__':
for temp in primary_choices:
bot.handler(text=f'{temperature_emoji(temp)} {temp}')(partial(enable, temp))
for temp in all_choices:
bot.handler(text=f'{temperature_emoji(temp)} {temp}')(partial(enable, temp))
kc = KettleController()
bot.run()
# bot library handles signals, so when sigterm or something like that happens, we should stop all other threads here
kc.stop_all()