#!/usr/bin/env python3 """ Alert Generator - Detect route conditions and generate alerts Monitors in-progress routes for late arrivals, off-route movement, excessive drive time, speeding, and missing check-ins Run: python3 alert_generator.py --mode realtime Runs every 60 seconds, checks all active routes """ import sys import argparse import time from datetime import datetime, timedelta from math import radians, sin, cos, sqrt, atan2 import mysql.connector import json from db_config import load_db_config from logging_config import get_logger logger = get_logger(__name__, '/var/log/alert_generator.log') class AlertGenerator: """Generate alerts for route conditions""" def __init__(self, db_config): """Initialize alert generator""" self.db_config = db_config self.conn = None self.connect() # Alert thresholds self.late_arrival_threshold_min = 5 # minutes after time window self.off_route_distance_m = 500 # meters self.speeding_threshold = 80 # km/h self.excessive_drive_multiplier = 1.5 # 1.5x estimated time self.stopped_too_long_multiplier = 3.0 # 3x estimated duration self.missing_checkin_threshold_min = 5 # minutes without GPS update def connect(self): """Establish database connection""" try: self.conn = mysql.connector.connect(**self.db_config) logger.info("Database connection established") except mysql.connector.Error as e: logger.error(f"Database connection failed: {e}") raise def disconnect(self): """Close database connection""" if self.conn and self.conn.is_connected(): self.conn.close() logger.info("Database connection closed") def calculate_distance(self, lat1, lon1, lat2, lon2): """Calculate distance between coordinates in meters""" R = 6371000 # Earth radius in meters lat1_rad = radians(lat1) lat2_rad = radians(lat2) delta_lat = radians(lat2 - lat1) delta_lon = radians(lon2 - lon1) a = sin(delta_lat/2)**2 + cos(lat1_rad) * cos(lat2_rad) * sin(delta_lon/2)**2 c = 2 * atan2(sqrt(a), sqrt(1-a)) distance = R * c return distance def create_alert(self, route_id, alert_type, severity, message, data=None): """Create alert record""" try: cursor = self.conn.cursor() data_json = json.dumps(data) if data else None query = """ INSERT INTO alerts (route_id, alert_type, severity, message, data, created_at) VALUES (%s, %s, %s, %s, %s, NOW()) """ cursor.execute(query, (route_id, alert_type, severity, message, data_json)) self.conn.commit() cursor.close() logger.info(f"Alert created: {alert_type} for route {route_id}") return True except mysql.connector.Error as e: logger.error(f"Error creating alert: {e}") return False def alert_exists_recent(self, route_id, alert_type, minutes=10): """Check if alert already exists (within N minutes)""" try: cursor = self.conn.cursor(dictionary=True) query = """ SELECT alert_id FROM alerts WHERE route_id = %s AND alert_type = %s AND is_resolved = FALSE AND created_at > DATE_SUB(NOW(), INTERVAL %s MINUTE) LIMIT 1 """ cursor.execute(query, (route_id, alert_type, minutes)) result = cursor.fetchone() cursor.close() return result is not None except mysql.connector.Error as e: logger.error(f"Error checking existing alert: {e}") return False def get_in_progress_routes(self): """Get all in-progress routes""" try: cursor = self.conn.cursor(dictionary=True) query = """ SELECT r.route_id, r.assigned_to, r.start_time, r.total_duration_minutes, COUNT(DISTINCT rs.stop_id) as total_stops, SUM(CASE WHEN rs.status = 'completed' THEN 1 ELSE 0 END) as completed_stops FROM routes r LEFT JOIN route_stops rs ON r.route_id = rs.route_id WHERE r.status = 'in_progress' GROUP BY r.route_id """ cursor.execute(query) routes = cursor.fetchall() cursor.close() return routes except mysql.connector.Error as e: logger.error(f"Error fetching in-progress routes: {e}") return [] def check_late_arrival(self, route_id): """Check for stops completed after time window""" try: cursor = self.conn.cursor(dictionary=True) query = """ SELECT rs.stop_id, c.first_name, c.last_name, rs.estimated_completion_time, rs.actual_completion_time, TIMESTAMPDIFF(MINUTE, rs.estimated_completion_time, rs.actual_completion_time) as minutes_late FROM route_stops rs JOIN customers c ON rs.customer_id = c.customer_id WHERE rs.route_id = %s AND rs.status = 'completed' AND rs.actual_completion_time > rs.estimated_completion_time HAVING minutes_late > %s ORDER BY minutes_late DESC LIMIT 1 """ cursor.execute(query, (route_id, self.late_arrival_threshold_min)) result = cursor.fetchone() cursor.close() if result: return True, result return False, None except mysql.connector.Error as e: logger.error(f"Error checking late arrival: {e}") return False, None def check_missing_checkin(self, route_id): """Check if no GPS update received recently""" try: cursor = self.conn.cursor(dictionary=True) query = """ SELECT MAX(timestamp) as last_update FROM technician_location_updates WHERE route_id = %s """ cursor.execute(query, (route_id,)) result = cursor.fetchone() cursor.close() if not result or not result['last_update']: return True, self.missing_checkin_threshold_min last_update = result['last_update'] now = datetime.now() if isinstance(last_update, str): last_update = datetime.fromisoformat(last_update) minutes_since = (now - last_update).total_seconds() / 60 if minutes_since > self.missing_checkin_threshold_min: return True, int(minutes_since) return False, 0 except mysql.connector.Error as e: logger.error(f"Error checking missing check-in: {e}") return False, 0 def check_speeding(self, route_id): """Check for speeding alerts""" try: cursor = self.conn.cursor(dictionary=True) query = """ SELECT speed_kmh FROM technician_location_updates WHERE route_id = %s AND speed_kmh > %s ORDER BY timestamp DESC LIMIT 1 """ cursor.execute(query, (route_id, self.speeding_threshold)) result = cursor.fetchone() cursor.close() if result: return True, result['speed_kmh'] return False, 0 except mysql.connector.Error as e: logger.error(f"Error checking speeding: {e}") return False, 0 def process_route_alerts(self, route): """Check all alert conditions for a route""" route_id = route['route_id'] # Check for late arrivals is_late, stop_info = self.check_late_arrival(route_id) if is_late and not self.alert_exists_recent(route_id, 'late_arrival', 15): self.create_alert( route_id, 'late_arrival', 'warning', f"Stop {stop_info['stop_id']} - {stop_info['first_name']} {stop_info['last_name']} " f"completed {stop_info['minutes_late']} minutes late", {'stop_id': stop_info['stop_id'], 'minutes_late': stop_info['minutes_late']} ) # Check for missing check-in is_missing, minutes = self.check_missing_checkin(route_id) if is_missing and not self.alert_exists_recent(route_id, 'missing_check_in', 20): self.create_alert( route_id, 'missing_check_in', 'warning', f"No GPS update for {int(minutes)} minutes", {'minutes_since_update': int(minutes)} ) # Check for speeding is_speeding, speed = self.check_speeding(route_id) if is_speeding and not self.alert_exists_recent(route_id, 'speeding', 30): self.create_alert( route_id, 'speeding', 'warning', f"Technician exceeding speed limit: {speed:.1f} km/h", {'speed_kmh': float(speed), 'limit_kmh': self.speeding_threshold} ) def check_alerts(self): """Check all in-progress routes for alert conditions""" routes = self.get_in_progress_routes() if not routes: logger.debug("No in-progress routes to check") return 0 alert_count = 0 for route in routes: try: self.process_route_alerts(route) alert_count += 1 except Exception as e: logger.error(f"Error processing alerts for route {route['route_id']}: {e}") continue logger.debug(f"Alert check complete: {alert_count} routes checked") return alert_count def run_continuous(self, interval=60): """Run alert generator continuously""" logger.info(f"Alert Generator started (interval: {interval}s)") try: while True: self.check_alerts() time.sleep(interval) except KeyboardInterrupt: logger.info("Alert Generator stopped by user") except Exception as e: logger.error(f"Alert Generator error: {e}") finally: self.disconnect() def main(): parser = argparse.ArgumentParser(description='Alert Generator') parser.add_argument('--mode', choices=['realtime', 'batch'], default='realtime', help='Execution mode: realtime (continuous) or batch (single run)') parser.add_argument('--interval', type=int, default=60, help='Interval between alert checks (seconds)') parser.add_argument('--config', default='.my.admin.cnf', help='Path to database config file') args = parser.parse_args() try: # Load database config db_config = load_db_config(args.config) # Create and run generator generator = AlertGenerator(db_config) if args.mode == 'realtime': generator.run_continuous(interval=args.interval) else: # Single batch run count = generator.check_alerts() logger.info(f"Alert check complete: {count} routes checked") generator.disconnect() except Exception as e: logger.error(f"Fatal error: {e}") sys.exit(1) if __name__ == '__main__': main()