Telegram bot для наших bmw G серии часть 2
В первой части https://habr.com/ru/articles/793112/ постарался объяснить зачем мне это все надо и дал пояснение — как я достаю данные с авто. В данной части разберемся как выслать эти данные в телегу чтобы получать сообщения наподобие вот этих:
Для начала надо установить новые библиотеки.
pip install logging
pip install telebot
pip install http.client
pip install schedule
pip install time
pip install math
С помощью Скедулера будем раз в фиксированное время выполнять обращение к базе данных и сравнивать локации, а так-же проверять количество топлива.
Результат будем высылать в телегу через телеграмм бота.
Для этого нам потребуется создать телеграмм бота с получить его TOKEN. Для этого у https://t.me/BotFather запросим создание нового бота с помощью команды /newbot. Далее надо будет ввести имя нового бота и вуаля — бот готов и новый ТОКЕН для доступа сгенерирован.
Для работы с ботом я выбрал telebot, хотя это не единственное решение. Меня устроило, то что telebot быстро завелся без проблем.
Для проверки локации по необходимым заранее заданным точкам интернет предложил мне следующее решение.
def haversine(lat1, lon1, lat2, lon2):
# Радиус Земли в км
R = 6371.0
# Преобразование координат в радианы
lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
# Разница между широтами и долготами
dlat = lat2 - lat1
dlon = lon2 - lon1
# Формула гаверсинуса для вычисления расстояния
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
c = 2 * atan2(sqrt(a), sqrt(1 - a))
# Расстояние между двумя точками
distance = R * c
return distance
def check_proximity(lat1, lon1, lat2, lon2, radius=1.0):
# Проверка, находится ли точка (lat1, lon1) в пределах заданного радиуса от точки (lat2, lon2)
distance = haversine(lat1, lon1, lat2, lon2)
return distance <= radiusБерем две координаты и сравниваем их локации. Если локации в радиусе 1 км, то возвращаем TRUE. Предыдущую локацию авто придется сохранять заранее в глобальной переменной. Но об этом ниже.
Полный код представил ниже, да, это можно было сделать проще и грамотнее, но все работает и меня все устраивает)). Координаты замазал…
def run_periodic_notificationX3_location():
asyncio.run(periodic_notificationX3_location(chat_id))
async def periodic_notificationX3_location(chat_id):
bot = telebot.TeleBot(TOKEN)
bot.send_message(chat_id, "Checking of X3 location️ ")
print("Checking of X3 location")
# Кострома квартира
target_KF_lon = 40.XXXX # Замените на желаемую широту
target_KF_lat = 57.XXXX # Замените на желаемую долготу
# Калинино
target_KAL_lon = 40.XXXX # Замените на желаемую широту
target_KAL_lat = 57.XXXX # Замените на желаемую долготу
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM bmw ORDER BY id DESC LIMIT 1')
row = cursor.fetchone()
conn.close()
print(row)
print(row[1])
print(row[2])
global Flat_location
global Kalinino_location
global previous_lat
global previous_lon
if row:
current_lat = row[1]
bot.send_message(chat_id, current_lat)
current_lon = row[2]
bot.send_message(chat_id, current_lon)
# Проверка, если машина ранее была в квартире
if Flat_location and not check_proximity(current_lat, current_lon, target_KF_lat, target_KF_lon):
print("X3 покинула квартиру\n")
bot.send_message(chat_id, 'X3 покинула квартиру\n')
Flat_location = False
# Проверка, если машина ранее была в Калинино
if Kalinino_location and not check_proximity(current_lat, current_lon, target_KAL_lat, target_KAL_lon):
print("X3 покинула Калинино\n")
bot.send_message(chat_id, 'X3 покинула Калинино\n')
Kalinino_location = False
# Проверка, если машина входит в квартиру
if not Flat_location and check_proximity(current_lat, current_lon, target_KF_lat, target_KF_lon):
print("X3 у квартиры\n")
bot.send_message(chat_id, 'X3 у квартиры\n')
Flat_location = True
# Проверка, если машина входит в Калинино
if not Kalinino_location and check_proximity(current_lat, current_lon, target_KAL_lat, target_KAL_lon):
print("X3 в Калинино\n")
bot.send_message(chat_id, 'X3 в Калинино\n')
Kalinino_location = True
# Обновление предыдущих координат
bot.send_message(chat_id, previous_lat)
bot.send_message(chat_id, previous_lon)
previous_lat = current_lat
previous_lon = current_lon
bot.send_message(chat_id, previous_lat)
bot.send_message(chat_id, previous_lon)
else:
print("Нет данных в базе.")
Топливо в баке можно без проблем запросить вот так:
def run_periodic_notificationX3():
asyncio.run(periodic_notificationX3(chat_id))
async def periodic_notificationX3(chat_id):
bot = telebot.TeleBot(TOKEN)
bot.send_message(chat_id, "Testing ️X3 ")
print("test X3")
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM bmw ORDER BY id DESC LIMIT 1')
row = cursor.fetchone()
conn.close()
print(row)
fuel_value = row[3]
if fuel_value <= 10.0:
bot.send_message(chat_id, f'Остаток топлива: {fuel_value} литров. Необходимо заправить Х3\n')Далее в скедулере укажем периодичность — для опроса координат запускать асинхронный процесс раз в 30 минут, для проверки уровня топлива — раз в 120 минут. Саму базу данных наполняю раз в 30 минут, ибо делать это чаще побоялся, что BMW забанит из-за частых запросов.
schedule.every(120).minutes.do(lambda: asyncio.run(periodic_notificationX3(chat_id)))
schedule.every(30).minutes.do(lambda: asyncio.run(periodic_notificationX3_location(chat_id)))А теперь ниже приведу полный код для обработки телеграмм бота, который у меня успешно работает.
#telegramm bot notification
import logging
import telebot
import http.client
import schedule
import time
import asyncio
import sqlite3
from math import radians, sin, cos, sqrt, atan2
Flat_location = False # Изначально считаем, что машина не в квартире
Kalinino_location = False # Изначально считаем, что машина не в Калинино
# Предыдущие координаты машины
previous_lon = 40.XXX
previous_lat = 57.XXX
def haversine(lat1, lon1, lat2, lon2):
# Радиус Земли в км
R = 6371.0
# Преобразование координат в радианы
lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
# Разница между широтами и долготами
dlat = lat2 - lat1
dlon = lon2 - lon1
# Формула гаверсинуса для вычисления расстояния
a = sin(dlat / 2)**2 + cos(lat1) * cos(lat2) * sin(dlon / 2)**2
c = 2 * atan2(sqrt(a), sqrt(1 - a))
# Расстояние между двумя точками
distance = R * c
return distance
def check_proximity(lat1, lon1, lat2, lon2, radius=1.0):
# Проверка, находится ли точка (lat1, lon1) в пределах заданного радиуса от точки (lat2, lon2)
distance = haversine(lat1, lon1, lat2, lon2)
return distance <= radius
# Замените 'YOUR_BOT_TOKEN' на токен, который вы получили от BotFather
TOKEN = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
bot = telebot.TeleBot(TOKEN)
chat_id = XXXXXXXXXXXXXX
# Функция для запуска asyncio.run()
def run_periodic_notificationX3():
asyncio.run(periodic_notificationX3(chat_id))
async def periodic_notificationX3(chat_id):
bot = telebot.TeleBot(TOKEN)
bot.send_message(chat_id, "Testing ️X3 ")
print("test X3")
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM bmw ORDER BY id DESC LIMIT 1')
row = cursor.fetchone()
conn.close()
print(row)
fuel_value = row[3]
if fuel_value <= 10.0:
bot.send_message(chat_id, f'Остаток топлива: {fuel_value} литров. Необходимо заправить Х3\n')
def run_periodic_notificationX3_location():
asyncio.run(periodic_notificationX3_location(chat_id))
async def periodic_notificationX3_location(chat_id):
bot = telebot.TeleBot(TOKEN)
bot.send_message(chat_id, "Checking of X3 location️ ")
print("Checking of X3 location")
# Кострома квартира
target_KF_lon = 40.XXX # Замените на желаемую широту
target_KF_lat = 57.XXX # Замените на желаемую долготу
# Калинино
target_KAL_lon = 40.XXX # Замените на желаемую широту
target_KAL_lat = 57.XXX # Замените на желаемую долготу
conn = sqlite3.connect('mydatabase.db')
cursor = conn.cursor()
cursor.execute('SELECT * FROM bmw ORDER BY id DESC LIMIT 1')
row = cursor.fetchone()
conn.close()
print(row)
print(row[1])
print(row[2])
global Flat_location
global Kalinino_location
global previous_lat
global previous_lon
if row:
current_lat = row[1]
bot.send_message(chat_id, current_lat)
current_lon = row[2]
bot.send_message(chat_id, current_lon)
# Проверка, если машина ранее была в квартире
if Flat_location and not check_proximity(current_lat, current_lon, target_KF_lat, target_KF_lon):
print("X3 покинула квартиру\n")
bot.send_message(chat_id, 'X3 покинула квартиру\n')
Flat_location = False
# Проверка, если машина ранее была в Калинино
if Kalinino_location and not check_proximity(current_lat, current_lon, target_KAL_lat, target_KAL_lon):
print("X3 покинула Калинино\n")
bot.send_message(chat_id, 'X3 покинула Калинино\n')
Kalinino_location = False
# Проверка, если машина входит в квартиру
if not Flat_location and check_proximity(current_lat, current_lon, target_KF_lat, target_KF_lon):
print("X3 у квартиры\n")
bot.send_message(chat_id, 'X3 у квартиры\n')
Flat_location = True
# Проверка, если машина входит в Калинино
if not Kalinino_location and check_proximity(current_lat, current_lon, target_KAL_lat, target_KAL_lon):
print("X3 в Калинино\n")
bot.send_message(chat_id, 'X3 в Калинино\n')
Kalinino_location = True
# Обновление предыдущих координат
bot.send_message(chat_id, previous_lat)
bot.send_message(chat_id, previous_lon)
previous_lat = current_lat
previous_lon = current_lon
bot.send_message(chat_id, previous_lat)
bot.send_message(chat_id, previous_lon)
else:
print("Нет данных в базе.")
# Расписание для вызова функции раз в полчаса
schedule.every(120).minutes.do(lambda: asyncio.run(periodic_notificationX3(chat_id)))
schedule.every(30).minutes.do(lambda: asyncio.run(periodic_notificationX3_location(chat_id)))
# Функция, которая будет запускаться при старте
def main():
pass
if __name__ == '__main__':
while True:
try:
main()
schedule.run_pending()
except Exception as e:
print(f"Произошла ошибка: {e}")
time.sleep(600)
time.sleep(1)
Координаты закрыты ХХХ, вместо этого надо будет вставить необходимые координаты.
Код выше конечно можно оптимизировать и нужно это сделать. Но все работает и так лень что то переделывать)…
В принципе кода из данной статьи и первой статьи уже достаточно, чтобы все работало, но со штатного компа и запущенным постоянно Python. Мне этого было недостаточно, я хотел иметь полную автономность и перенести код на VPS, выдать часть функционала на WEB, ну или возможно, в будущем попробовать написать свое приложения для телефона… Об этом в следующей части. Продолжение следует…
