"""Fleet Agent - Manages vehicles, capacity, and real-time tracking.""" import uuid from datetime import datetime, timedelta from typing import Dict, List, Any, Optional from dataclasses import dataclass from enum import Enum from core.agent import SpecializedAgent from core.types import AgentTask, MessageType, Vehicle from core.logger import logger class VehicleStatus(str, Enum): AVAILABLE = "available" ASSIGNED = "assigned" IN_TRANSIT = "in_transit" AT_HUB = "at_hub" MAINTENANCE = "maintenance" OFFLINE = "offline" @dataclass class VehicleAssignment: assignment_id: str vehicle_id: str route_id: str order_ids: List[str] assigned_at: datetime estimated_return: datetime current_location: Dict[str, float] status: str = "active" class FleetAgent(SpecializedAgent): """Fleet Agent - Manages vehicle fleet and logistics operations.""" def __init__(self): super().__init__( agent_id="FLEET_AGENT", domain="fleet_management", description="Manages vehicle fleet, capacity, and real-time tracking" ) self._vehicles = self._init_fleet() self._assignments: Dict[str, VehicleAssignment] = {} self._maintenance_schedule: Dict[str, datetime] = {} self._hub_capacity: Dict[str, Dict] = {} # BUG FIX: hub_capacity init was dead code inside _init_fleet (after return). # Moved here so it actually runs after _vehicles is populated. for vehicle in self._vehicles.values(): hub = vehicle["hub"] if hub not in self._hub_capacity: self._hub_capacity[hub] = {"available": 0, "total": 0} self._hub_capacity[hub]["total"] += 1 if vehicle["status"] == "available": self._hub_capacity[hub]["available"] += 1 def _init_fleet(self) -> Dict[str, Dict]: return { # Delhi "DL-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "DL-HUB-01", "location": {"lat": 28.6139, "lng": 77.2090}}, "DL-V-002": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "DL-HUB-01", "location": {"lat": 28.6139, "lng": 77.2090}}, "DL-V-003": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "DL-HUB-02", "location": {"lat": 28.5355, "lng": 77.2100}}, "DL-B-001": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "DL-HUB-01", "location": {"lat": 28.6139, "lng": 77.2090}}, "DL-B-002": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "DL-HUB-02", "location": {"lat": 28.5355, "lng": 77.2100}}, # Mumbai "MU-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "MU-HUB-01", "location": {"lat": 19.0760, "lng": 72.8777}}, "MU-V-002": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "MU-HUB-01", "location": {"lat": 19.0760, "lng": 72.8777}}, "MU-V-003": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "MU-HUB-02", "location": {"lat": 19.1650, "lng": 72.8500}}, "MU-B-001": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "MU-HUB-02", "location": {"lat": 19.1650, "lng": 72.8500}}, "MU-T-001": {"type": "truck", "capacity_kg": 2000, "capacity_vol": 25, "status": "available", "hub": "MU-HUB-01", "location": {"lat": 19.0760, "lng": 72.8777}}, # Bangalore "BL-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "BL-HUB-01", "location": {"lat": 12.9716, "lng": 77.5946}}, "BL-B-001": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "BL-HUB-01", "location": {"lat": 12.9716, "lng": 77.5946}}, "BL-B-002": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "BL-HUB-01", "location": {"lat": 12.9716, "lng": 77.5946}}, # Hyderabad "HY-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "HY-HUB-01", "location": {"lat": 17.3850, "lng": 78.4867}}, "HY-T-001": {"type": "truck", "capacity_kg": 2000, "capacity_vol": 25, "status": "available", "hub": "HY-HUB-01", "location": {"lat": 17.3850, "lng": 78.4867}}, # Pune "PU-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "PU-HUB-01", "location": {"lat": 18.5204, "lng": 73.8567}}, "PU-B-001": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "PU-HUB-01", "location": {"lat": 18.5204, "lng": 73.8567}}, # Kolkata "KL-V-001": {"type": "van", "capacity_kg": 500, "capacity_vol": 8, "status": "available", "hub": "KL-HUB-01", "location": {"lat": 22.5726, "lng": 88.3639}}, "KL-B-001": {"type": "bike", "capacity_kg": 15, "capacity_vol": 0.5, "status": "available", "hub": "KL-HUB-01", "location": {"lat": 22.5726, "lng": 88.3639}}, } async def handle_task(self, task: AgentTask) -> Dict[str, Any]: handlers = { "assign_vehicle": self._assign_vehicle, "release_vehicle": self._release_vehicle, "get_availability": self._get_availability, "track_vehicle": self._track_vehicle, "update_location": self._update_location, "schedule_maintenance": self._schedule_maintenance, "get_fleet_status": self._get_fleet_status, "optimize_allocation": self._optimize_allocation, } handler = handlers.get(task.task_type, self._unknown_task) return await handler(task) async def _assign_vehicle(self, task: AgentTask) -> Dict[str, Any]: route_id = task.data.get("route_id") route_type = task.data.get("route_type", "last_mile") from_hub = task.data.get("from_hub") order_weight = task.data.get("order_weight", 0) logger.info(f"Fleet Agent: Assigning vehicle for route {route_id}") suitable_vehicles = self._find_suitable_vehicles(from_hub, order_weight, route_type) if not suitable_vehicles: suitable_vehicles = self._find_nearest_available_vehicles(from_hub, order_weight) if not suitable_vehicles: logger.warning(f"No suitable vehicles available for route {route_id}") return {"status": "no_vehicle_available", "route_id": route_id, "message": "No suitable vehicles available"} vehicle_id = suitable_vehicles[0] vehicle = self._vehicles[vehicle_id] vehicle["status"] = "assigned" if vehicle["hub"] in self._hub_capacity: self._hub_capacity[vehicle["hub"]]["available"] -= 1 assignment_id = f"ASN-{uuid.uuid4().hex[:8].upper()}" assignment = VehicleAssignment( assignment_id=assignment_id, vehicle_id=vehicle_id, route_id=route_id, order_ids=[task.data.get("order_id", "")], assigned_at=datetime.now(), estimated_return=datetime.now() + timedelta(hours=2), current_location=vehicle["location"], status="active" ) self._assignments[assignment_id] = assignment logger.info(f"Vehicle {vehicle_id} ({vehicle['type']}) assigned to route {route_id}") await self.send_message( recipient="DISPATCH_AGENT", message_type=MessageType.VEHICLE_ASSIGNED, payload={"route_id": route_id, "vehicle_id": vehicle_id, "assignment_id": assignment_id}, ) return { "status": "assigned", "assignment_id": assignment_id, "vehicle_id": vehicle_id, "vehicle_type": vehicle["type"], "capacity_kg": vehicle["capacity_kg"], "capacity_vol": vehicle["capacity_vol"], "current_location": vehicle["location"], } async def _release_vehicle(self, task: AgentTask) -> Dict[str, Any]: vehicle_id = task.data.get("vehicle_id") if vehicle_id not in self._vehicles: return {"status": "error", "message": f"Vehicle {vehicle_id} not found"} vehicle = self._vehicles[vehicle_id] vehicle["status"] = "available" if vehicle["hub"] in self._hub_capacity: self._hub_capacity[vehicle["hub"]]["available"] += 1 logger.info(f"Vehicle {vehicle_id} released") return {"status": "released", "vehicle_id": vehicle_id, "hub": vehicle["hub"]} async def _get_availability(self, task: AgentTask) -> Dict[str, Any]: hub_id = task.data.get("hub_id") vehicle_type = task.data.get("vehicle_type") available = [] for vid, vehicle in self._vehicles.items(): if vehicle["hub"] == hub_id and vehicle["status"] == "available": if vehicle_type is None or vehicle["type"] == vehicle_type: available.append({ "vehicle_id": vid, "type": vehicle["type"], "capacity_kg": vehicle["capacity_kg"], "capacity_vol": vehicle["capacity_vol"], }) return {"hub_id": hub_id, "available_vehicles": available, "total_available": len(available)} async def _track_vehicle(self, task: AgentTask) -> Dict[str, Any]: vehicle_id = task.data.get("vehicle_id") if vehicle_id not in self._vehicles: return {"status": "error", "message": f"Vehicle {vehicle_id} not found"} vehicle = self._vehicles[vehicle_id] active_assignment = next( (a for a in self._assignments.values() if a.vehicle_id == vehicle_id and a.status == "active"), None, ) return { "vehicle_id": vehicle_id, "type": vehicle["type"], "status": vehicle["status"], "location": vehicle["location"], "hub": vehicle["hub"], "current_assignment": { "assignment_id": active_assignment.assignment_id, "route_id": active_assignment.route_id, "estimated_return": active_assignment.estimated_return.isoformat(), } if active_assignment else None, } async def _update_location(self, task: AgentTask) -> Dict[str, Any]: vehicle_id = task.data.get("vehicle_id") new_location = task.data.get("location") if vehicle_id not in self._vehicles: return {"status": "error", "message": f"Vehicle {vehicle_id} not found"} self._vehicles[vehicle_id]["location"] = new_location for assignment in self._assignments.values(): if assignment.vehicle_id == vehicle_id and assignment.status == "active": assignment.current_location = new_location break return {"status": "updated", "vehicle_id": vehicle_id, "new_location": new_location} async def _schedule_maintenance(self, task: AgentTask) -> Dict[str, Any]: vehicle_id = task.data.get("vehicle_id") maintenance_date = task.data.get("date") if vehicle_id not in self._vehicles: return {"status": "error", "message": f"Vehicle {vehicle_id} not found"} if maintenance_date: self._maintenance_schedule[vehicle_id] = datetime.fromisoformat(maintenance_date) else: self._maintenance_schedule[vehicle_id] = datetime.now() + timedelta(days=3) self._vehicles[vehicle_id]["status"] = "maintenance" return { "status": "scheduled", "vehicle_id": vehicle_id, "maintenance_date": self._maintenance_schedule[vehicle_id].isoformat(), } async def _get_fleet_status(self, task: AgentTask) -> Dict[str, Any]: status_summary = { "total_vehicles": len(self._vehicles), "by_type": {}, "by_status": {}, "by_hub": {}, "vehicles": [], } for vehicle_id, vehicle in self._vehicles.items(): vtype = vehicle["type"] vstatus = vehicle["status"] hub = vehicle["hub"] status_summary["by_type"][vtype] = status_summary["by_type"].get(vtype, 0) + 1 status_summary["by_status"][vstatus] = status_summary["by_status"].get(vstatus, 0) + 1 if hub not in status_summary["by_hub"]: status_summary["by_hub"][hub] = {"total": 0, "available": 0} status_summary["by_hub"][hub]["total"] += 1 if vstatus == "available": status_summary["by_hub"][hub]["available"] += 1 status_summary["vehicles"].append({ "vehicle_id": vehicle_id, "type": vtype, "status": vstatus, "hub": hub, "location": vehicle["location"], }) return status_summary async def _optimize_allocation(self, task: AgentTask) -> Dict[str, Any]: logger.info("Fleet Agent: Optimizing vehicle allocation") recommendations = [] for hub_id, capacity in self._hub_capacity.items(): utilization = (capacity["total"] - capacity["available"]) / capacity["total"] if capacity["total"] > 0 else 0 if utilization < 0.3: recommendations.append({ "action": "redistribute", "from_hub": hub_id, "reason": f"Low utilization ({utilization*100:.0f}%)", "vehicles_to_move": 1, }) elif utilization > 0.9: recommendations.append({ "action": "request_reinforcement", "to_hub": hub_id, "reason": f"High utilization ({utilization*100:.0f}%)", "vehicles_needed": 2, }) return {"status": "optimized", "recommendations": recommendations} async def _unknown_task(self, task: AgentTask) -> Dict[str, Any]: return {"status": "error", "message": f"Unknown task: {task.task_type}"} def _find_suitable_vehicles(self, hub_id: str, weight_kg: float, route_type: str) -> List[str]: suitable = [] for vehicle_id, vehicle in self._vehicles.items(): if vehicle["hub"] != hub_id: continue if vehicle["status"] != "available": continue if vehicle["capacity_kg"] < weight_kg: continue if route_type == "last_mile" and vehicle["type"] in ["van", "truck"]: continue suitable.append(vehicle_id) return suitable def _find_nearest_available_vehicles(self, hub_id: str, weight_kg: float) -> List[str]: available = [ vid for vid, v in self._vehicles.items() if v["status"] == "available" and v["capacity_kg"] >= weight_kg ] return available[:1] async def think(self, context: str, options: List[str] = None) -> str: return f"[FLEET_AGENT reasoning]: {context}"