As we advance the Smart Commuter implementation, we’ll integrate traffic data and smart notifications to refine the user experience. For previous step see first part.
3. Telegram Bot Handler
# telegram_bot.py
import os
from typing import Optional
from telegram import Update, Bot
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
filters,
ContextTypes
)
import asyncio
class TelegramCommuteBot:
"""Telegram bot for commute notifications and commands"""
def __init__(self, token: str, chat_id: str, config):
self.token = token
self.chat_id = chat_id
self.config = config
self.bot = Bot(token=token)
self.application = None
async def send_notification(self, title: str, message: str, parse_mode: str = 'Markdown'):
"""Send notification message"""
try:
full_message = f"*{title}*\n\n{message}"
await self.bot.send_message(
chat_id=self.chat_id,
text=full_message,
parse_mode=parse_mode
)
print(f"✅ Telegram notification sent: {title}")
return True
except Exception as e:
print(f"❌ Telegram error: {e}")
return False
async def send_departure_alert(self,
travel_time: str,
route_summary: str,
traffic_status: str,
waypoints: list):
"""Send main departure notification"""
message = f"🏠 *Time to Head Home!*\n\n"
message += f"⏱️ *Travel time:* {travel_time}\n"
message += f"🚦 *Traffic:* {traffic_status}\n"
message += f"🗺️ *Best route:* {route_summary}\n"
if waypoints:
message += f"\n📍 *Your stops:*\n"
for i, stop in enumerate(waypoints, 1):
message += f" {i}. {stop['address']}\n"
message += f"\n_Have a safe trip home!_ 🚗"
await self.send_notification("Commute Alert", message)
async def send_early_warning(self,
travel_time: str,
minutes_early: int,
traffic_status: str):
"""Send early warning for heavy traffic"""
message = f"⚠️ *Heavy Traffic Detected*\n\n"
message += f"Consider leaving *{minutes_early} minutes early*\n\n"
message += f"⏱️ *Current travel time:* {travel_time}\n"
message += f"🚦 {traffic_status}\n\n"
message += f"_I'll notify you again when it's time to leave._"
await self.send_notification("Traffic Warning", message)
def setup_commands(self):
"""Setup bot command handlers"""
self.application = Application.builder().token(self.token).build()
# Command handlers
self.application.add_handler(CommandHandler("start", self.cmd_start))
self.application.add_handler(CommandHandler("status", self.cmd_status))
self.application.add_handler(CommandHandler("add", self.cmd_add_stop))
self.application.add_handler(CommandHandler("remove", self.cmd_remove_stop))
self.application.add_handler(CommandHandler("stops", self.cmd_list_stops))
self.application.add_handler(CommandHandler("clear", self.cmd_clear_stops))
self.application.add_handler(CommandHandler("check", self.cmd_check_now))
# Message handler for natural language
self.application.add_handler(
MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message)
)
async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /start command"""
message = (
"🏠 *Welcome to Your Commute Assistant!*\n\n"
"I'll help you get home on time by monitoring traffic "
"and notifying you when to leave.\n\n"
"*Available Commands:*\n"
"• `/status` - Current commute info\n"
"• `/add ` - Add stop on your way\n"
"• `/remove ` - Remove a stop\n"
"• `/stops` - List all stops\n"
"• `/clear` - Clear all stops\n"
"• `/check` - Check traffic now\n\n"
"*Natural Language:*\n"
"You can also say:\n"
"• 'Add gym'\n"
"• 'Stop at grocery store'\n"
"• 'Clear all stops'\n\n"
f"I'll check traffic daily at *{self.config.check_time}* "
f"and notify you when it's time to leave!"
)
await update.message.reply_text(message, parse_mode='Markdown')
async def cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /status command"""
message = f"📊 *Current Configuration*\n\n"
message += f"🏢 *Work:* {self.config.work_address}\n"
message += f"🏠 *Home:* {self.config.home_address}\n"
message += f"⏰ *Check time:* {self.config.check_time}\n"
message += f"🎯 *Target arrival:* {self.config.desired_arrival_time}\n"
message += f"⏱️ *Buffer:* {self.config.buffer_minutes} min\n\n"
if self.config.waypoints:
message += f"📍 *Active stops:*\n"
for i, stop in enumerate(self.config.waypoints, 1):
message += f" {i}. {stop}\n"
else:
message += f"📍 *No stops configured*\n"
await update.message.reply_text(message, parse_mode='Markdown')
async def cmd_add_stop(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /add command"""
if not context.args:
await update.message.reply_text(
"Please specify a location.\n"
"Example: `/add Whole Foods Market`",
parse_mode='Markdown'
)
return
stop = ' '.join(context.args)
if self.config.add_waypoint(stop):
await update.message.reply_text(
f"✅ Added stop: *{stop}*\n\n"
f"You now have {len(self.config.waypoints)} stop(s).",
parse_mode='Markdown'
)
else:
await update.message.reply_text(
f"⚠️ Stop *{stop}* already exists!",
parse_mode='Markdown'
)
async def cmd_remove_stop(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /remove command"""
if not context.args:
await update.message.reply_text(
"Please specify which stop to remove.\n"
"Use `/stops` to see your stops.",
parse_mode='Markdown'
)
return
stop = ' '.join(context.args)
if self.config.remove_waypoint(stop):
await update.message.reply_text(
f"✅ Removed stop: *{stop}*",
parse_mode='Markdown'
)
else:
await update.message.reply_text(
f"⚠️ Stop *{stop}* not found!",
parse_mode='Markdown'
)
async def cmd_list_stops(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /stops command"""
if not self.config.waypoints:
await update.message.reply_text("You have no stops configured.")
return
message = f"📍 *Your stops ({len(self.config.waypoints)}):*\n\n"
for i, stop in enumerate(self.config.waypoints, 1):
message += f"{i}. {stop}\n"
await update.message.reply_text(message, parse_mode='Markdown')
async def cmd_clear_stops(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /clear command"""
count = len(self.config.waypoints)
self.config.clear_waypoints()
await update.message.reply_text(
f"✅ Cleared {count} stop(s).",
parse_mode='Markdown'
)
async def cmd_check_now(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle /check command - trigger immediate traffic check"""
await update.message.reply_text(
"🔍 Checking traffic now...",
parse_mode='Markdown'
)
# This will be called by the main app
# Store the update context for callback
context.user_data['check_requested'] = True
async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle natural language messages"""
text = update.message.text.lower()
# Pattern matching for natural commands
if any(word in text for word in ['add', 'stop at', 'stop by']):
# Extract location (simple approach)
for keyword in ['add', 'stop at', 'stop by']:
if keyword in text:
location = text.split(keyword, 1)[1].strip()
if location:
self.config.add_waypoint(location)
await update.message.reply_text(
f"✅ Added: *{location}*",
parse_mode='Markdown'
)
return
elif 'clear' in text and 'stop' in text:
count = len(self.config.waypoints)
self.config.clear_waypoints()
await update.message.reply_text(
f"✅ Cleared {count} stop(s)."
)
return
elif 'status' in text or 'info' in text:
await self.cmd_status(update, context)
return
# Default response
await update.message.reply_text(
"I'm not sure what you mean. Try:\n"
"• `/status` - See your settings\n"
"• `/add ` - Add a stop\n"
"• `/help` - See all commands"
)
def run_bot(self):
"""Start the bot (blocking)"""
self.setup_commands()
print("🤖 Telegram bot started")
self.application.run_polling()
Key features:
- Interactive Telegram commands (/add, /remove, /status, etc.)
- Natural language processing for casual commands
- Real-time stop management
- Beautiful formatted messages with emojis
4. Main Application
# smart_commute.py
import os
import schedule
import time
import asyncio
from datetime import datetime, timedelta
from config import CommuteConfig
from traffic_monitor import TrafficMonitor
from telegram_bot import TelegramCommuteBot
class SmartCommuteAssistant:
"""Main application orchestrating the commute assistant"""
def __init__(self):
# Load configuration
self.config = CommuteConfig.from_env()
# Initialize components
self.traffic_monitor = TrafficMonitor(
os.getenv('GOOGLE_MAPS_API_KEY')
)
self.telegram_bot = TelegramCommuteBot(
token=os.getenv('TELEGRAM_BOT_TOKEN'),
chat_id=os.getenv('TELEGRAM_CHAT_ID'),
config=self.config
)
# State tracking
self.last_notification_time = None
self.notification_sent_today = False
def check_traffic_and_notify(self):
"""Main logic: Check traffic and send notification if needed"""
current_time = datetime.now()
print(f"\n🕐 Checking traffic at {current_time.strftime('%H:%M:%S')}")
# Get route with all waypoints
route = self.config.get_full_route()
origin = route[0]
destination = route[-1]
waypoints = route[1:-1] if len(route) > 2 else None
print(f"📍 Route: {origin}")
if waypoints:
for wp in waypoints:
print(f" ↓ via {wp}")
print(f" → {destination}")
# Get current traffic data
traffic_data = self.traffic_monitor.get_route_with_traffic(
origin=origin,
destination=destination,
waypoints=waypoints
)
if not traffic_data:
print("❌ Could not retrieve traffic data")
return
# Calculate when to leave
departure_time, travel_minutes = self.traffic_monitor.calculate_departure_time(
traffic_data,
self.config.desired_arrival_time,
self.config.buffer_minutes
)
# Analyze traffic
traffic_status = self.traffic_monitor.analyze_traffic(
traffic_data,
self.config.heavy_traffic_threshold
)
# Format travel time
travel_time_str = self._format_duration(
traffic_data['total_duration_traffic']
)
# Calculate time until departure
time_until_departure = (departure_time - current_time).total_seconds() / 60
# Print status
print(f"⏱️ Travel time: {travel_time_str}")
print(f"🚦 Traffic: {traffic_status}")
print(f"🏁 Recommended departure: {departure_time.strftime('%H:%M')}")
print(f"⏰ Minutes until departure: {int(time_until_departure)}")
# Send notifications based on timing
if 0 <= time_until_departure <= 5 and not self.notification_sent_today:
# Time to leave!
asyncio.run(self._send_departure_notification(
travel_time_str,
traffic_data,
traffic_status
))
self.notification_sent_today = True
self.last_notification_time = current_time
elif (traffic_data['traffic_ratio'] >= self.config.heavy_traffic_threshold
and 5 < time_until_departure <= 20
and not self.notification_sent_today):
# Heavy traffic - send early warning
asyncio.run(self._send_early_warning(
travel_time_str,
int(time_until_departure),
traffic_status
))
# Don't set notification_sent_today yet, still need to send main alert
elif time_until_departure > 30:
print("⏳ Too early to leave, continuing to monitor...")
elif time_until_departure < 0:
print("⚠️ You should have already left!")
if not self.notification_sent_today:
asyncio.run(self._send_late_warning(travel_time_str))
self.notification_sent_today = True
async def _send_departure_notification(self,
travel_time: str,
traffic_data: dict,
traffic_status: str):
"""Send the main departure notification"""
await self.telegram_bot.send_departure_alert(
travel_time=travel_time,
route_summary=traffic_data['summary'],
traffic_status=traffic_status,
waypoints=traffic_data.get('waypoints', [])
)
async def _send_early_warning(self,
travel_time: str,
minutes_early: int,
traffic_status: str):
"""Send early warning for heavy traffic"""
await self.telegram_bot.send_early_warning(
travel_time,
minutes_early,
traffic_status
)
async def _send_late_warning(self, travel_time: str):
"""Send warning if user is late"""
message = f"⚠️ *You're Running Late!*\n\n"
message += f"⏱️ *Travel time:* {travel_time}\n\n"
message += f"_Leave now to minimize delay!_"
await self.telegram_bot.send_notification("Late Alert", message)
def _format_duration(self, seconds: int) -> str:
"""Format duration in seconds to readable string"""
minutes = int(seconds / 60)
if minutes < 60:
return f"{minutes} min"
else:
hours = minutes // 60
mins = minutes % 60
return f"{hours}h {mins}min"
def _check_if_in_monitoring_window(self):
"""Check if we're within the monitoring window"""
now = datetime.now()
# Reset daily flag at midnight
if now.hour == 0 and now.minute < 5:
self.notification_sent_today = False
# Parse check time and arrival time
check_hour, check_minute = map(int, self.config.check_time.split(':'))
arrival_hour, arrival_minute = map(int, self.config.desired_arrival_time.split(':'))
check_time = now.replace(hour=check_hour, minute=check_minute, second=0)
arrival_time = now.replace(hour=arrival_hour, minute=arrival_minute, second=0)
# Monitor from check_time until 30 minutes after arrival_time
end_time = arrival_time + timedelta(minutes=30)
if check_time <= now <= end_time:
self.check_traffic_and_notify()
def start(self):
"""Start the commute assistant"""
print("\n" + "="*60)
print("🚀 Smart Commute Assistant Started")
print("="*60)
print(f"\n📍 Work: {self.config.work_address}")
print(f"📍 Home: {self.config.home_address}")
print(f"⏰ Check time: {self.config.check_time}")
print(f"🎯 Target arrival: {self.config.desired_arrival_time}")
print(f"📊 Monitoring for traffic conditions...")
# Schedule checks
check_time = self.config.check_time
# Main scheduled check
schedule.every().day.at(check_time).do(self.check_traffic_and_notify)
# Continuous monitoring during window (every 5 minutes)
schedule.every(5).minutes.do(self._check_if_in_monitoring_window)
# Run once immediately for testing
print(f"\n🔍 Running initial check...")
self.check_traffic_and_notify()
# Start the scheduling loop
print(f"\n✅ System is running. Press Ctrl+C to stop.")
print("="*60 + "\n")
try:
while True:
schedule.run_pending()
time.sleep(30) # Check every 30 seconds
except KeyboardInterrupt:
print("\n\n👋 Smart Commute Assistant stopped")
def main():
"""Entry point"""
assistant = SmartCommuteAssistant()
assistant.start()
if __name__ == "__main__":
main()
Key features:
- Intelligent scheduling with monitoring windows
- Daily notification reset at midnight
- Multiple notification types (departure, early warning, late alert)
- Automatic waypoint handling
- Continuous monitoring during commute window
Usage Examples
Basic Setup
- Configure your .env file:
GOOGLE_MAPS_API_KEY=AIzaSyB...
TELEGRAM_BOT_TOKEN=123456789:ABC...
TELEGRAM_CHAT_ID=987654321
WORK_ADDRESS=1600 Amphitheatre Parkway, Mountain View, CA
HOME_ADDRESS=123 Main Street, Mountain View, CA
CHECK_TIME=17:00
DESIRED_ARRIVAL_TIME=18:00
BUFFER_MINUTES=10
- Run the system:
python smart_commute.py
Interactive Commands
While the system is running, chat with your bot on Telegram:
Check current status:
You: /status
Bot: 📊 Current Configuration
🏢 Work: 1600 Amphitheatre Parkway...
🏠 Home: 123 Main Street...
⏰ Check time: 17:00
🎯 Target arrival: 18:00
⏱️ Buffer: 10 min
📍 No stops configured
Add a stop:
You: /add Whole Foods Market
Bot: ✅ Added stop: Whole Foods Market
You now have 1 stop(s).
Or use natural language:
You: Stop at the gym on my way home
Bot: ✅ Added: the gym on my way home
List your stops:
You: /stops
Bot: 📍 Your stops (2):
1. Whole Foods Market
2. the gym on my way home
Remove a stop:
You: /remove gym
Bot: ✅ Removed stop: the gym on my way home
Clear all stops:
You: /clear
Bot: ✅ Cleared 1 stop(s).
Notification Examples
Normal departure notification:
🏠 Time to Head Home!
⏱️ Travel time: 32 min
🚦 Traffic: 🟢 Light traffic
🗺️ Best route: I-280 S
📍 Your stops:
1. Whole Foods Market
Have a safe trip home! 🚗
Heavy traffic warning:
⚠️ Heavy Traffic Detected
Consider leaving 15 minutes early
⏱️ Current travel time: 45 min
🚦 🔴 Heavy traffic
I'll notify you again when it's time to leave.
Top comments (0)