import pandas as pd import numpy as np from datetime import timedelta class PredictiveFleetSwapAI: def __init__(self, vehicles_df, locations_df, locations_relations_df, routes_df, segments_df): self.vehicles = vehicles_df.copy() self.locations = locations_df.copy() self.locations_relations = locations_relations_df.copy() self.routes = routes_df.copy() self.segments = segments_df.copy() # Convert date columns to datetime objects self.vehicles['leasing_start_date'] = pd.to_datetime(self.vehicles['leasing_start_date']) self.vehicles['leasing_end_date'] = pd.to_datetime(self.vehicles['leasing_end_date']) self.routes['start_datetime'] = pd.to_datetime(self.routes['start_datetime']) self.routes['end_datetime'] = pd.to_datetime(self.routes['end_datetime']) # Merge routes with segments to get start/end locations for each route route_details = self.segments.sort_values(['route_id', 'seq']).groupby('route_id').agg( start_loc_id=('start_loc_id', 'first'), end_loc_id=('end_loc_id', 'last') ).reset_index() self.routes = pd.merge(self.routes, route_details, on='route_id', how='left') def initial_vehicle_placement(self): """ Places vehicles at the most frequent starting locations of the routes. """ print("Placing vehicles at initial locations...") # Find the most common starting locations top_start_locations = self.routes['start_loc_id'].value_counts().nlargest(len(self.vehicles)).index.tolist() # Assign vehicles to these locations num_locations = len(top_start_locations) for i, vehicle_id in enumerate(self.vehicles['Id']): location_id = top_start_locations[i % num_locations] self.vehicles.loc[self.vehicles['Id'] == vehicle_id, 'current_location_id'] = location_id print(f"Initial placement complete. Vehicles distributed among {num_locations} unique locations.") def solve(self): """ Main function to solve the vehicle assignment problem. """ # 1. Initial vehicle placement self.initial_vehicle_placement() # Prepare vehicle status tracking # 'available_at' tracks when a vehicle finishes its current route self.vehicles['available_at'] = pd.to_datetime('1970-01-01') # Sort routes chronologically self.routes = self.routes.sort_values(by='start_datetime').reset_index(drop=True) assignments = [] total_routes = len(self.routes) print(f"Starting route assignments for {total_routes} routes...") # 2. Iterate through each route and assign a vehicle for i, route in self.routes.iterrows(): if (i + 1) % 1000 == 0: print(f"Processing route {i + 1}/{total_routes}...") route_start_time = route['start_datetime'] route_start_loc = route['start_loc_id'] best_vehicle_id = None min_cost = float('inf') best_vehicle_arrival_time = None # Find the best available vehicle for the current route for _, vehicle in self.vehicles.iterrows(): # Check if vehicle is available before the route starts if vehicle['available_at'] <= route_start_time: vehicle_loc = vehicle['current_location_id'] relocation_cost = 0 arrival_time = route_start_time # Assume instant relocation if at the same place # If vehicle needs to move if vehicle_loc != route_start_loc: # Find relocation details (distance and time) relocation = self.locations_relations[ ((self.locations_relations['id_loc_1'] == vehicle_loc) & ( self.locations_relations['id_loc_2'] == route_start_loc)) ] if not relocation.empty: relocation_dist = relocation.iloc[0]['dist'] relocation_time_hours = relocation.iloc[0]['time'] / 60.0 # time is in minutes # Cost: 1000 PLN + 1 PLN/km + 150 PLN/h relocation_cost = 1000 + (relocation_dist * 1) + (relocation_time_hours * 150) # The vehicle must start moving early to arrive on time arrival_time = route_start_time # Check if there is enough time to relocate required_travel_time = timedelta(hours=relocation_time_hours) if vehicle['available_at'] + required_travel_time > route_start_time: continue # Not enough time to get to the start location else: # No direct relation, this case should be handled, for now skip continue # Simple greedy choice: pick the one with the lowest relocation cost if relocation_cost < min_cost: min_cost = relocation_cost best_vehicle_id = vehicle['Id'] best_vehicle_arrival_time = arrival_time # If no vehicle is available at all, find the one that becomes available earliest if best_vehicle_id is None: # This is a fallback to ensure all routes are covered, even if it means a delay # A more complex solver might re-shuffle previous assignments earliest_available_vehicle = self.vehicles.loc[self.vehicles['available_at'].idxmin()] best_vehicle_id = earliest_available_vehicle['Id'] # The route will be delayed vehicle_loc = earliest_available_vehicle['current_location_id'] relocation_time_hours = 0 if vehicle_loc != route_start_loc: relocation = self.locations_relations[ ((self.locations_relations['id_loc_1'] == vehicle_loc) & ( self.locations_relations['id_loc_2'] == route_start_loc)) ] if not relocation.empty: relocation_time_hours = relocation.iloc[0]['time'] / 60.0 # Route starts after the vehicle is free and has travelled best_vehicle_arrival_time = earliest_available_vehicle['available_at'] + timedelta( hours=relocation_time_hours) # Assign the best found vehicle to the route if best_vehicle_id is not None: vehicle_idx = self.vehicles.index[self.vehicles['Id'] == best_vehicle_id][0] # Update vehicle status self.vehicles.loc[vehicle_idx, 'current_odometer_km'] += route['distance_km'] # If there was a relocation, add that distance too if self.vehicles.loc[vehicle_idx, 'current_location_id'] != route['start_loc_id']: relocation = self.locations_relations[ ((self.locations_relations['id_loc_1'] == self.vehicles.loc[ vehicle_idx, 'current_location_id']) & ( self.locations_relations['id_loc_2'] == route['start_loc_id'])) ] if not relocation.empty: self.vehicles.loc[vehicle_idx, 'current_odometer_km'] += relocation.iloc[0]['dist'] self.vehicles.loc[vehicle_idx, 'current_location_id'] = route['end_loc_id'] # The vehicle becomes available after the route ends. If the route was delayed, this is pushed back. route_end_time = max(best_vehicle_arrival_time, route['start_datetime']) + ( route['end_datetime'] - route['start_datetime']) self.vehicles.loc[vehicle_idx, 'available_at'] = route_end_time assignments.append({ 'route_id': route['route_id'], 'vehicle_id': best_vehicle_id, 'start_datetime': route['start_datetime'], 'end_datetime': route['end_datetime'], 'assigned_start_time': best_vehicle_arrival_time, 'assigned_end_time': route_end_time, 'relocation_cost': min_cost if min_cost != float('inf') else 0 }) print("All routes assigned.") return pd.DataFrame(assignments) if __name__ == '__main__': # Load data print("Loading data files...") try: vehicles_df = pd.read_csv('data/vehicles.csv') locations_df = pd.read_csv('data/locations.csv') locations_relations_df = pd.read_csv('data/locations_relations.csv') routes_df = pd.read_csv('data/routes.csv') segments_df = pd.read_csv('data/segments.csv') # Add route_id to routes_df for merging routes_df.rename(columns={'id': 'route_id'}, inplace=True) except FileNotFoundError as e: print(f"Error loading data: {e}. Make sure the data files are in a 'data/' directory.") exit() # Initialize and run the solver solver = PredictiveFleetSwapAI(vehicles_df, locations_df, locations_relations_df, routes_df, segments_df) assignments_df = solver.solve() # Save the output output_path = 'assignments.csv' assignments_df.to_csv(output_path, index=False) print(f"\nAssignment complete. Results saved to {output_path}") print(f"Total relocation cost: {assignments_df['relocation_cost'].sum():,.2f} PLN") # Display first 5 assignments print("\nSample of assignments:") print(assignments_df.head())