Feature/fleet management service

backend/fleet_management/allocation.py

@@ -0,0 +1,136 @@
+from datetime import datetime
+from typing import Optional
+import random
+from .models import TripRequest, User, Vehicle
+from .trip_request import TRIP_REQUESTS, VEHICLE_POOL, EMPLOYEES
+
+# Mock data for demonstration
+ADMIN_MANAGERS = [
+    User("A001", "Mampho Nthunya", "admin_manager", "M001"),
+    User("A002", "Lerato Moeti", "admin_manager", "M002"),
+]
+
+FLEET_MANAGERS = [
+    User("F001", "Pheko Matela", "fleet_manager", "MFM001"),
+]
+
+DRIVERS = [
+    User("D001", "Teboho Mohlomi", "driver", "MFM001"),
+    User("D002", "Lineo Seeiso", "driver", "MFM001"),
+]
+
+
+def get_admin_manager_for_employee(user_id: str) -> Optional[User]:
+    """
+    Finds the admin manager for the employee's ministry.
+    """
+    employee = next((u for u in EMPLOYEES if u.user_id == user_id), None)
+    if not employee:
+        return None
+    return next((a for a in ADMIN_MANAGERS if a.ministry_id == employee.ministry_id), None)
+
+
+def admin_approve_trip(request_id: str, approve: bool) -> str:
+    """
+    Admin Manager approves or rejects the trip based on trip details.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == request_id), None)
+    if not trip:
+        return f"Trip {request_id} not found."
+
+    admin_manager = get_admin_manager_for_employee(trip.user_id)
+    if not admin_manager:
+        return "No Admin Manager found for employee’s ministry."
+
+    if approve:
+        trip.status = "approved"
+        trip.approved_by_admin = admin_manager.name
+        return f"Trip {trip.request_id} approved by {admin_manager.name}."
+    else:
+        trip.status = "rejected"
+        return f"Trip {trip.request_id} rejected by {admin_manager.name}."
+
+
+def fleet_manager_confirm_trip(request_id: str, vehicle_id: Optional[str] = None, driver_id: Optional[str] = None):
+    """
+    Fleet Manager confirms or modifies the system recommendation.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == request_id), None)
+    if not trip or trip.status != "approved":
+        return f"Trip {request_id} not available for allocation."
+
+    # Choose or modify vehicle
+    vehicle = next((v for v in VEHICLE_POOL if v.vehicle_id == vehicle_id), None) if vehicle_id else VEHICLE_POOL[0]
+    driver = next((d for d in DRIVERS if d.user_id == driver_id), None) if driver_id else random.choice(DRIVERS)
+
+    trip.status = "allocated"
+    trip.assigned_vehicle = vehicle.registration_number if vehicle else None
+    trip.assigned_driver = driver.name if driver else None
+    trip.recommended_by_fleet_manager = FLEET_MANAGERS[0].name
+    trip.security_pin = generate_security_pin(trip.request_id)
+
+    return {
+        "trip_id": trip.request_id,
+        "vehicle": trip.assigned_vehicle,
+        "driver": trip.assigned_driver,
+        "security_pin": trip.security_pin,
+    }
+
+def fleet_manager_confirm_trip(request_id: str, fleet_manager_id: str, vehicle_id: Optional[str] = None, driver_id: Optional[str] = None):
+    """
+    Fleet Manager confirms or modifies the system recommendation.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == request_id), None)
+    if not trip or trip.status != "approved":
+        return f"Trip {request_id} not available for allocation."
+
+    fleet_manager = next((f for f in FLEET_MANAGERS if f.user_id == fleet_manager_id), None)
+    if not fleet_manager:
+        return f"Fleet Manager with ID {fleet_manager_id} not found."
+
+    # Choose vehicle and driver (either from parameters or defaults)
+    vehicle = next((v for v in VEHICLE_POOL if v.vehicle_id == vehicle_id), None) if vehicle_id else VEHICLE_POOL[0]
+    driver = next((d for d in DRIVERS if d.user_id == driver_id), None) if driver_id else random.choice(DRIVERS)
+
+    # Update trip
+    trip.status = "allocated"
+    trip.assigned_vehicle = vehicle.registration_number if vehicle else None
+    trip.assigned_driver = driver.name if driver else None
+    trip.recommended_by_fleet_manager = fleet_manager.name
+    trip.security_pin = generate_security_pin(trip.request_id)
+
+    return {
+        "trip_id": trip.request_id,
+        "vehicle": trip.assigned_vehicle,
+        "driver": trip.assigned_driver,
+        "fleet_manager": fleet_manager.name,
+        "security_pin": trip.security_pin,
+    }
+
+
+def rerecommend_trip(request_id: str, new_vehicle_id: Optional[str] = None, new_driver_id: Optional[str] = None):
+    """
+    Fleet Manager can re-recommend a new vehicle or driver after initial allocation.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == request_id), None)
+    if not trip:
+        return f"Trip {request_id} not found."
+
+    if new_vehicle_id:
+        vehicle = next((v for v in VEHICLE_POOL if v.vehicle_id == new_vehicle_id), None)
+        trip.assigned_vehicle = vehicle.registration_number if vehicle else trip.assigned_vehicle
+
+    if new_driver_id:
+        driver = next((d for d in DRIVERS if d.user_id == new_driver_id), None)
+        trip.assigned_driver = driver.name if driver else trip.assigned_driver
+
+    trip.recommended_by_fleet_manager = FLEET_MANAGERS[0].name
+    return f"Trip {trip.request_id} rerecommended with updates: vehicle={trip.assigned_vehicle}, driver={trip.assigned_driver}."
+
+
+def generate_security_pin(request_id: str) -> str:
+    """
+    Generates a unique 4-digit PIN for trip verification.
+    """
+    random.seed(request_id)
+    return str(random.randint(1000, 9999))

backend/fleet_management/fleet_management_api.py

@@ -0,0 +1,88 @@
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from datetime import datetime
+from fleet_management.trip_request import create_trip_request, recommend_vehicle
+from fleet_management.allocation import admin_approve_trip, fleet_manager_confirm_trip, rerecommend_trip
+from fleet_management.monitoring import authenticate_trip, end_trip
+
+app = FastAPI(title="GovRide AI Fleet Management API")
+
+# ----------------------------
+# Request Models
+# ----------------------------
+
+class TripRequestIn(BaseModel):
+    user_id: str
+    pickup_location: str
+    destination: str
+    trip_date: datetime
+    purpose: str
+
+class AdminApprovalIn(BaseModel):
+    approve: bool
+
+class FleetAllocationIn(BaseModel):
+    fleet_manager_id: str
+    vehicle_id: str | None = None
+    driver_id: str | None = None
+
+class FleetReRecommendationIn(BaseModel):
+    new_vehicle_id: str | None = None
+    new_driver_id: str | None = None
+
+class TripAuthIn(BaseModel):
+    employee_pin: str
+    driver_pin: str
+
+
+# ----------------------------
+# Endpoints
+# ----------------------------
+
+@app.post("/trip-request/")
+def submit_trip_request(trip_data: TripRequestIn):
+    trip = create_trip_request(
+        user_id=trip_data.user_id,
+        pickup=trip_data.pickup_location,
+        destination=trip_data.destination,
+        trip_date=trip_data.trip_date,
+        purpose=trip_data.purpose,
+    )
+    recommend_vehicle(trip)
+    return {"message": f"Trip {trip.request_id} created successfully", "trip_id": trip.request_id}
+
+
+@app.patch("/trip-request/{request_id}/approve")
+def approve_trip(request_id: str, data: AdminApprovalIn):
+    result = admin_approve_trip(request_id, approve=data.approve)
+    if "not found" in result:
+        raise HTTPException(status_code=404, detail=result)
+    return {"message": result}
+
+
+@app.patch("/trip-request/{request_id}/allocate")
+def confirm_trip_allocation(request_id: str, data: FleetAllocationIn):
+    result = fleet_manager_confirm_trip(request_id, data.fleet_manager_id, data.vehicle_id, data.driver_id)
+    if isinstance(result, str):
+        raise HTTPException(status_code=404, detail=result)
+    return result
+
+
+@app.patch("/trip-request/{request_id}/rerecommend")
+def rerecommend_allocation(request_id: str, data: FleetReRecommendationIn):
+    result = rerecommend_trip(request_id, data.new_vehicle_id, data.new_driver_id)
+    return {"message": result}
+
+
+@app.patch("/trip-request/{request_id}/start")
+def start_trip(request_id: str, data: TripAuthIn):
+    result = authenticate_trip(request_id, data.employee_pin, data.driver_pin)
+    if "failed" in result:
+        raise HTTPException(status_code=400, detail=result)
+    return {"message": result}
+
+
+@app.patch("/trip-request/{request_id}/complete")
+def complete_trip(request_id: str):
+    result = end_trip(request_id)
+    return {"message": result}

backend/fleet_management/models.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Optional
+
+@dataclass
+class Ministry:
+    """
+    Represents a government ministry that uses the system.
+    """
+    ministry_id: str
+    name: str
+    budget_code: str
+
+
+@dataclass
+class User:
+    """
+    Represents a system user (employee, driver, admin, etc.)
+    """
+    user_id: str
+    name: str
+    role: str  # ('employee', 'admin_manager', 'fleet_manager', 'driver', 'admin')
+    ministry_id: str
+
+
+@dataclass
+class Vehicle:
+    """
+    Represents a vehicle in the fleet.
+    """
+    vehicle_id: str
+    registration_number: str
+    vehicle_type: str
+    fuel_type: str
+    current_status: str  # ('available', 'in_use', 'maintenance')
+    acquisition_date: datetime
+
+
+@dataclass
+class TripRequest:
+    """
+    Represents a vehicle trip request made by a government employee.
+    """
+    request_id: str
+    user_id: str
+    pickup_location: str
+    destination: str
+    trip_date: datetime
+    purpose: str
+    passengers: int = 1
+    status: str = "pending"
+    assigned_driver: Optional[str] = None
+    assigned_vehicle: Optional[str] = None
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    approved_by_admin: Optional[str] = None
+    recommended_by_fleet_manager: Optional[str] = None
+    security_pin: Optional[str] = None

backend/fleet_management/monitoring.py

@@ -0,0 +1,36 @@
+#This code implements trip execution and authentication functionality for starting and ending trips.
+
+from datetime import datetime
+from .models import TripRequest
+from .allocation import generate_security_pin
+from .trip_request import TRIP_REQUESTS
+
+def authenticate_trip(trip_id: str, employee_pin: str, driver_input_pin: str) -> str:
+    """
+    Trip can only start if employee and driver pins match.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == trip_id), None)
+    if not trip:
+        return f"Trip {trip_id} not found."
+
+    correct_pin = generate_security_pin(trip_id)
+
+    if employee_pin == correct_pin and driver_input_pin == correct_pin:
+        trip.status = "ongoing"
+        trip.start_time = datetime.now()
+        return f"Trip {trip_id} has started at {trip.start_time}."
+    else:
+        return "Authentication failed. Trip cannot start."
+
+
+def end_trip(trip_id: str) -> str:
+    """
+    Mark trip as completed and record end time.
+    """
+    trip = next((t for t in TRIP_REQUESTS if t.request_id == trip_id), None)
+    if not trip or trip.status != "ongoing":
+        return f"Trip {trip_id} is not ongoing."
+
+    trip.status = "completed"
+    trip.end_time = datetime.now()
+    return f"Trip {trip_id} completed at {trip.end_time}."

backend/fleet_management/trip_request.py

@@ -0,0 +1,45 @@
+#This is the Trip Request Processor which collects trip requests from employees and gathers required information
+from dataclasses import dataclass
+from datetime import datetime
+from typing import List, Optional
+from .models import TripRequest, Vehicle, User
+
+VEHICLE_POOL = [
+    Vehicle("V001", "ABC-123", "sedan", "petrol", "available", datetime(2020, 5, 10)),
+    Vehicle("V002", "XYZ-456", "4x4", "diesel", "available", datetime(2021, 7, 20)),
+]
+
+EMPLOYEES = [
+    User("U001", "Thabo Mohapi", "employee", "M001"),
+    User("U002", "Molupi Lekhanya", "employee", "M002"),
+]
+
+TRIP_REQUESTS: List[TripRequest] = []
+
+def create_trip_request(user_id: str, pickup: str, destination: str, trip_date: datetime, purpose: str, passengers: int = 1) -> TripRequest:
+    user = next((u for u in EMPLOYEES if u.user_id == user_id), None)
+    if not user:
+        raise ValueError(f"User {user_id} not found")
+
+    request_id = f"TR{len(TRIP_REQUESTS)+1:03d}"
+    trip = TripRequest(
+        request_id=request_id,
+        user_id=user_id,
+        pickup_location=pickup,
+        destination=destination,
+        trip_date=trip_date,
+        purpose=purpose,
+        passengers=passengers,
+        status="pending"
+    )
+    TRIP_REQUESTS.append(trip)
+    print(f"Trip request {trip.request_id} created successfully for {user.name}")
+    return trip
+
+def recommend_vehicle(trip: TripRequest) -> Optional[Vehicle]:
+    for vehicle in VEHICLE_POOL:
+        if vehicle.current_status == "available":
+            print(f"Recommended vehicle {vehicle.registration_number} ({vehicle.vehicle_type}) for trip {trip.request_id}")
+            return vehicle
+    print(f"No vehicles available for trip {trip.request_id}")
+    return None