import torch
import [Link] as nn
import [Link] as F
import pandas as pd
import numpy as np
from collections import defaultdict, deque
import heapq
import itertools
from typing import List, Tuple, Dict, Any
import random
import [Link] as plt
import [Link] as animation
import networkx as nx
from [Link] import FancyBboxPatch, Circle
import [Link] as mpatches
from [Link] import FFMpegWriter, PillowWriter
import time
import math
# Set device
device = [Link]('cuda' if [Link].is_available() else 'cpu')
print(f"Using device: {device}")
class AnimatedPyTorchRouteOptimizer:
def __init__(self):
[Link] = defaultdict(list)
[Link] = {}
[Link] = None
self.traffic_probabilities = {}
self.transport_modes = ['car', 'bike', 'walk']
# Animation properties
self.animation_data = []
self.current_step = 0
[Link] = None
[Link] = None
[Link] = None
self.G = None
# Convert to PyTorch tensors for GPU acceleration
self.base_speeds = [Link]([50.0, 20.0, 5.0], device=device) # car, bike, walk
self.traffic_multipliers = [Link]([1.0, 0.7, 0.4], device=device) # low, medium, high
# HMM transition probabilities as tensors
self.transition_matrix = [Link]([
[0.7, 0.25, 0.05], # from low
[0.3, 0.5, 0.2], # from medium
[0.1, 0.3, 0.6] # from high
], device=device)
self.state_to_idx = {'low': 0, 'medium': 1, 'high': 2}
self.idx_to_state = {0: 'low', 1: 'medium', 2: 'high'}
# Monte Carlo parameters
self.mc_samples = 100
self.current_traffic_state = None
def add_edge(self, start: int, end: int, distance: float):
"""Add edge to graph with distance"""
[Link][start].append((end, distance))
[Link][end].append((start, distance))
[Link][(start, end)] = distance
[Link][(end, start)] = distance
def load_dataset_from_csv(self, csv_file_path: str):
"""Load dataset from CSV file with traffic information and perform feature engineering"""
try:
[Link] = pd.read_csv(csv_file_path)
print(f"Successfully loaded {len([Link])} rows from {csv_file_path}")
[Link] = self.feature_engineering([Link])
print("Feature engineering completed successfully")
except FileNotFoundError:
print(f"Error: Could not find file '{csv_file_path}'")
print("Please make sure the CSV file exists in the current directory")
raise
except Exception as e:
print(f"Error loading CSV file: {str(e)}")
raise
def feature_engineering(self, df: [Link]) -> [Link]:
"""Perform feature engineering on the dataset"""
df = [Link]()
# Convert time to hour
df['time'] = pd.to_datetime(df['time'], format='%H:%M:%S').[Link]
# Create day of week
days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
df['day'] = [days[i % 7] for i in range(len(df))]
# Rush hour classification
def classify_rush_hour(hour):
if 7 <= hour <= 9:
return 'morning_rush'
elif 17 <= hour <= 19:
return 'evening_rush'
elif 22 <= hour <= 6:
return 'night'
else:
return 'normal'
df['rush_hour'] = df['time'].apply(classify_rush_hour)
# Traffic state derivation
def derive_traffic_state_vectorized(df):
base_score = [Link](len(df), device=device)
vehicle_counts = [Link](df['vehicle_count'].values, device=device,
dtype=torch.float32)
vehicle_score = [Link](vehicle_counts / 200, 0, 1) * 40
base_score += vehicle_score
rush_multipliers = {
'morning_rush': 25, 'evening_rush': 25,
'normal': 10, 'night': 5
}
rush_scores = [Link]([rush_multipliers.get(rush, 10) for rush in df['rush_hour']],
device=device, dtype=torch.float32)
base_score += rush_scores
weather_scores = {
'No Rain': 0, 'Slow Rain': 10, 'Normal Rain': 15,
'Heavy Rain': 25, 'Snow': 30
}
weather_score_tensor = [Link]([weather_scores.get(w, 0) for w in df['weather']],
device=device, dtype=torch.float32)
base_score += weather_score_tensor
road_scores = {
'Better': -5, 'Good': 0, 'Poor': 15, 'Bad': 20
}
road_score_tensor = [Link]([road_scores.get(r, 0) for r in df['road_condition']],
device=device, dtype=torch.float32)
base_score += road_score_tensor
event_scores = [Link]([15 if event == 'Yes' else 0 for event in df['special_event']],
device=device, dtype=torch.float32)
base_score += event_scores
traffic_states = []
base_score_cpu = base_score.cpu().numpy()
for score in base_score_cpu:
if score <= 30:
traffic_states.append('low')
elif score <= 60:
traffic_states.append('medium')
else:
traffic_states.append('high')
return traffic_states
df['traffic_state'] = derive_traffic_state_vectorized(df)
return df
def calculate_bayes_probabilities_pytorch(self, day: str, time: int, vehicle_count: int,
road_cond: str, weather: str, special_event: str,
start_node: int, end_node: int) -> Dict[str, float]:
"""Calculate traffic state probabilities using Bayes theorem with PyTorch"""
edge_data = [Link][
([Link]['start_node'] == start_node) &
([Link]['end_node'] == end_node)
]
if len(edge_data) == 0:
return {'low': 0.4, 'medium': 0.4, 'high': 0.2}
traffic_states = ['low', 'medium', 'high']
probabilities = [Link](3, device=device)
# Determine rush hour
if 7 <= time <= 9:
rush_hour = 'morning_rush'
elif 17 <= time <= 19:
rush_hour = 'evening_rush'
elif 22 <= time <= 6:
rush_hour = 'night'
else:
rush_hour = 'normal'
for i, state in enumerate(traffic_states):
state_data = edge_data[edge_data['traffic_state'] == state]
if len(state_data) == 0:
probabilities[i] = 0.1
continue
likelihood = [Link](1.0, device=device)
# Calculate various likelihoods
day_match = len(state_data[state_data['day'] == day]) / len(state_data)
likelihood *= max(day_match, 0.1)
rush_match = len(state_data[state_data['rush_hour'] == rush_hour]) / len(state_data)
likelihood *= max(rush_match, 0.1)
time_window = 2
time_match = len(state_data[
(state_data['time'] >= time - time_window) &
(state_data['time'] <= time + time_window)
]) / len(state_data)
likelihood *= max(time_match, 0.1)
vehicle_window_pct = 0.2
vehicle_min = vehicle_count * (1 - vehicle_window_pct)
vehicle_max = vehicle_count * (1 + vehicle_window_pct)
vehicle_match = len(state_data[
(state_data['vehicle_count'] >= vehicle_min) &
(state_data['vehicle_count'] <= vehicle_max)
]) / len(state_data)
likelihood *= max(vehicle_match, 0.1)
road_match = len(state_data[state_data['road_condition'] == road_cond]) / len(state_data)
likelihood *= max(road_match, 0.1)
weather_match = len(state_data[state_data['weather'] == weather]) / len(state_data)
likelihood *= max(weather_match, 0.1)
event_match = len(state_data[state_data['special_event'] == special_event]) / len(state_data)
likelihood *= max(event_match, 0.1)
prior = len(state_data) / len(edge_data)
probabilities[i] = likelihood * prior
probabilities = [Link](probabilities, dim=0)
result = {}
for i, state in enumerate(traffic_states):
result[state] = probabilities[i].item()
return result
def predict_next_traffic_state_hmm(self, current_state: str) -> str:
"""Predict next traffic state using HMM"""
if current_state not in self.state_to_idx:
current_state = 'medium' # Default state
current_idx = self.state_to_idx[current_state]
transition_probs = self.transition_matrix[current_idx]
# Sample from the transition probabilities
next_idx = [Link](transition_probs, 1).item()
return self.idx_to_state[next_idx]
def monte_carlo_path_evaluation(self, paths: List[List[int]], day: str, time: int,
vehicle_count: int, road_cond: str, weather: str,
special_event: str, current_traffic_state: str) -> List[float]:
"""Evaluate paths using Monte Carlo simulation"""
path_scores = []
for path in paths:
total_cost = 0
num_samples = self.mc_samples
for _ in range(num_samples):
path_cost = 0
current_state = current_traffic_state
for i in range(len(path) - 1):
start_node = path[i]
end_node = path[i + 1]
# Get base distance
base_distance = [Link]((start_node, end_node), 10)
# Get traffic probabilities
traffic_probs = self.calculate_bayes_probabilities_pytorch(
day, time, vehicle_count, road_cond, weather,
special_event, start_node, end_node
)
# Normalize probabilities before sampling
states = list(traffic_probs.keys())
probs = [Link](list(traffic_probs.values()))
probs /= [Link]() if [Link]() > 0 else 1 # Normalize
# Sample traffic state
sampled_state = [Link](states, p=probs)
# Calculate cost based on traffic state
if sampled_state == 'low':
multiplier = 1.0
elif sampled_state == 'medium':
multiplier = 1.5
else: # high
multiplier = 2.5
edge_cost = base_distance * multiplier
path_cost += edge_cost
# Update traffic state using HMM
current_state = self.predict_next_traffic_state_hmm(current_state)
total_cost += path_cost
average_cost = total_cost / num_samples
path_scores.append(average_cost)
return path_scores
def find_top_k_paths(self, start: int, end: int, k: int = 3, max_depth: int = 6) -> List[List[int]]:
"""Find top k shortest paths using modified DFS"""
if start == end:
return [[start]]
all_paths = []
def dfs(current_path, visited, depth):
if depth > max_depth:
return
current_node = current_path[-1]
if current_node == end:
all_paths.append(current_path.copy())
return
# Get neighbors sorted by distance
neighbors = [(neighbor, dist) for neighbor, dist in [Link][current_node]]
[Link](key=lambda x: x[1])
for neighbor, distance in neighbors:
if neighbor not in visited and len(all_paths) < k * 3: # Get more paths to filter
current_path.append(neighbor)
[Link](neighbor)
dfs(current_path, visited, depth + 1)
current_path.pop()
[Link](neighbor)
visited = {start}
dfs([start], visited, 0)
# Sort paths by length and return top k
all_paths.sort(key=len)
return all_paths[:k]
def get_best_next_node(self, current_node: int, end_node: int, day: str, time: int,
vehicle_count: int, road_cond: str, weather: str,
special_event: str, visited: set) -> Tuple[int, List[List[int]], Dict]:
"""Get the best next node using Bayes, HMM, and Monte Carlo methods"""
# Find top 3 paths from current node
remaining_paths = self.find_top_k_paths(current_node, end_node, k=3)
# Filter paths that don't go through visited nodes (except current)
valid_paths = []
for path in remaining_paths:
if len(path) > 1 and path[1] not in visited:
valid_paths.append(path)
if not valid_paths:
# Fallback: find any unvisited neighbor
for neighbor, _ in [Link][current_node]:
if neighbor not in visited:
valid_paths = [[current_node, neighbor]]
break
if not valid_paths:
return None, [], {}
# Calculate traffic probabilities for edges from current node
edge_traffic = {}
for neighbor, distance in [Link][current_node]:
if neighbor not in visited:
probs = self.calculate_bayes_probabilities_pytorch(
day, time, vehicle_count, road_cond, weather,
special_event, current_node, neighbor
)
edge_traffic[(current_node, neighbor)] = probs
# Use Monte Carlo to evaluate paths
if self.current_traffic_state is None:
self.current_traffic_state = 'medium' # Default initial state
path_scores = self.monte_carlo_path_evaluation(
valid_paths, day, time, vehicle_count, road_cond, weather,
special_event, self.current_traffic_state
)
# Choose the path with minimum expected cost
best_path_idx = [Link](path_scores)
best_path = valid_paths[best_path_idx]
best_next_node = best_path[1] if len(best_path) > 1 else None
# Update current traffic state using HMM
self.current_traffic_state = self.predict_next_traffic_state_hmm(self.current_traffic_state)
return best_next_node, valid_paths, edge_traffic
def setup_animation_graph(self):
"""Setup the graph for animation"""
self.G = [Link]()
# Add nodes and edges
for node in [Link]():
self.G.add_node(node)
for node, neighbors in [Link]():
for neighbor, distance in neighbors:
if not self.G.has_edge(node, neighbor):
self.G.add_edge(node, neighbor, weight=distance)
# Calculate positions
[Link] = nx.spring_layout(self.G, seed=42, k=3, iterations=50)
def animate_pathfinding(self, start: int, end: int, day: str, time: int,
vehicle_count: int, road_cond: str, weather: str,
special_event: str, save_animation: bool = True,
video_format: str = 'mp4'):
"""Animate the pathfinding process step by step"""
print("Setting up animation...")
self.setup_animation_graph()
# Initialize animation data
self.animation_data = []
self.current_traffic_state = 'medium' # Initial state
# Step-by-step pathfinding simulation
print("Simulating step-by-step pathfinding process...")
current_node = start
visited = {start}
path_so_far = [start]
step_number = 0
# Initial step
initial_step = {
'step_number': step_number,
'current_node': current_node,
'visited': [Link](),
'path_so_far': path_so_far.copy(),
'edge_traffic': {},
'possible_paths': [],
'step_type': 'start',
'traffic_state': self.current_traffic_state,
'analysis': 'Starting pathfinding journey...'
}
self.animation_data.append(initial_step)
while current_node != end:
step_number += 1
# Get best next node and analysis data
next_node, possible_paths, edge_traffic = self.get_best_next_node(
current_node, end, day, time, vehicle_count,
road_cond, weather, special_event, visited
)
if next_node is None:
# No valid next node found
break
# Create analysis text
analysis_text = f"Analyzing {len(possible_paths)} possible paths from node
{current_node}.\n"
analysis_text += f"Current traffic state: {self.current_traffic_state}\n"
analysis_text += f"Using Bayes theorem to predict traffic probabilities.\n"
analysis_text += f"Monte Carlo simulation with {self.mc_samples} samples.\n"
analysis_text += f"Best next node: {next_node}"
# Store step data
step_data = {
'step_number': step_number,
'current_node': current_node,
'visited': [Link](),
'path_so_far': path_so_far.copy(),
'edge_traffic': edge_traffic.copy(),
'possible_paths': possible_paths.copy(),
'step_type': 'analyzing',
'traffic_state': self.current_traffic_state,
'analysis': analysis_text,
'next_node': next_node
}
self.animation_data.append(step_data)
# Move to next node
current_node = next_node
[Link](current_node)
path_so_far.append(current_node)
# Movement step
step_number += 1
move_step = {
'step_number': step_number,
'current_node': current_node,
'visited': [Link](),
'path_so_far': path_so_far.copy(),
'edge_traffic': edge_traffic.copy(),
'possible_paths': possible_paths.copy(),
'step_type': 'moving',
'traffic_state': self.current_traffic_state,
'analysis': f"Moving to node {current_node}. Path so far: {' -> '.join(map(str,
path_so_far))}",
'next_node': current_node
}
self.animation_data.append(move_step)
# Prevent infinite loops
if step_number > 20:
break
# Final step
final_step = {
'step_number': step_number + 1,
'current_node': current_node,
'visited': [Link](),
'path_so_far': path_so_far.copy(),
'edge_traffic': {},
'possible_paths': [],
'step_type': 'completed',
'traffic_state': self.current_traffic_state,
'analysis': f"Destination reached! Final path: {' -> '.join(map(str, path_so_far))}",
'next_node': None
}
self.animation_data.append(final_step)
# Create animation
print("Creating animation...")
[Link], [Link] = [Link](figsize=(16, 12))
# Animation function
def animate_step(frame):
[Link]()
if frame >= len(self.animation_data):
frame = len(self.animation_data) - 1
step_data = self.animation_data[frame]
current_node = step_data['current_node']
visited = step_data['visited']
path_so_far = step_data['path_so_far']
edge_traffic = step_data['edge_traffic']
possible_paths = step_data['possible_paths']
step_type = step_data['step_type']
traffic_state = step_data['traffic_state']
analysis = step_data['analysis']
next_node = step_data.get('next_node')
# Draw all edges with default color
for edge in [Link]():
node1, node2 = edge
nx.draw_networkx_edges(self.G, [Link], [(node1, node2)],
edge_color='lightgray', width=1,
alpha=0.3, ax=[Link])
# Draw edges with traffic information
for edge_key, traffic_probs in edge_traffic.items():
node1, node2 = edge_key
dominant_state = max(traffic_probs, key=traffic_probs.get)
dominant_prob = traffic_probs[dominant_state]
if dominant_state == 'low':
edge_color = 'green'
edge_alpha = 0.7
elif dominant_state == 'medium':
edge_color = 'orange'
edge_alpha = 0.8
else: # high
edge_color = 'red'
edge_alpha = 0.9
nx.draw_networkx_edges(self.G, [Link], [(node1, node2)],
edge_color=edge_color, width=4,
alpha=edge_alpha, ax=[Link])
# Draw possible paths
path_colors = ['purple', 'brown', 'pink']
for i, path in enumerate(possible_paths[:3]):
if len(path) > 1:
path_edges = [(path[j], path[j+1]) for j in range(len(path)-1)]
nx.draw_networkx_edges(self.G, [Link], path_edges,
edge_color=path_colors[i], width=3,
alpha=0.5, style='dashed', ax=[Link])
# Draw path taken so far
if len(path_so_far) > 1:
path_edges = [(path_so_far[j], path_so_far[j+1]) for j in range(len(path_so_far)-1)]
nx.draw_networkx_edges(self.G, [Link], path_edges,
edge_color='blue', width=6,
alpha=1.0, ax=[Link])
# Draw nodes
node_colors = []
node_sizes = []
for node in [Link]():
if node == start:
node_colors.append('lightblue')
node_sizes.append(1200)
elif node == end:
node_colors.append('lightcoral')
node_sizes.append(1200)
elif node == current_node:
node_colors.append('gold')
node_sizes.append(1400)
elif node == next_node and step_type == 'analyzing':
node_colors.append('yellow')
node_sizes.append(1000)
elif node in visited:
node_colors.append('lightgreen')
node_sizes.append(800)
else:
node_colors.append('lightgray')
node_sizes.append(600)
nx.draw_networkx_nodes(self.G, [Link], node_color=node_colors,
node_size=node_sizes, alpha=0.8, ax=[Link])
# Draw labels
nx.draw_networkx_labels(self.G, [Link], font_size=10,
font_weight='bold', ax=[Link])
# Add animated agent
if current_node in [Link]:
x, y = [Link][current_node]
# Pulsing effect
pulse_factor = 1 + 0.3 * [Link](frame * 0.5)
circle = Circle((x, y), 0.05 * pulse_factor, color='red',
alpha=0.8, zorder=10)
[Link].add_patch(circle)
# Add title
if step_type == 'start':
title = f"Step {step_data['step_number']}: Starting at Node {current_node}"
elif step_type == 'analyzing':
title = f"Step {step_data['step_number']}: Analyzing from Node {current_node}"
elif step_type == 'moving':
title = f"Step {step_data['step_number']}: Moving to Node {current_node}"
else:
title = f"Step {step_data['step_number']}: Destination Reached!"
[Link].set_title(title, fontsize=16, fontweight='bold', pad=20)
# Add detailed information box
info_text = f"Current Position: Node {current_node}\n"
info_text += f"Destination: Node {end}\n"
info_text += f"Traffic State: {traffic_state}\n"
info_text += f"Visited Nodes: {sorted(list(visited))}\n"
info_text += f"Path: {' -> '.join(map(str, path_so_far))}\n"
info_text += f"Possible Paths: {len(possible_paths)}\n"
if next_node:
info_text += f"Next Node: {next_node}\n"
info_text += f"\nAnalysis:\n{analysis}"
[Link](0.02, 0.98, info_text, transform=[Link],
verticalalignment='top', fontsize=9,
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.9))
# Add legend
legend_elements = [
[Link](color='lightblue', label='Start Node'),
[Link](color='lightcoral', label='End Node'),
[Link](color='gold', label='Current Position'),
[Link](color='yellow', label='Next Target'),
[Link](color='lightgreen', label='Visited'),
[Link](color='blue', label='Path Taken'),
[Link](color='purple', label='Possible Paths'),
[Link](color='red', label='High Traffic'),
[Link](color='orange', label='Medium Traffic'),
[Link](color='green', label='Low Traffic')
]
[Link](handles=legend_elements, loc='upper right',
bbox_to_anchor=(1, 1), fontsize=9)
[Link].set_aspect('equal')
[Link]('off')
# Add algorithm info
algo_text = "Algorithms Used:\n• Bayes Theorem\n• HMM Traffic Prediction\n• Monte
Carlo Simulation"
[Link](0.02, 0.02, algo_text, transform=[Link],
verticalalignment='bottom', fontsize=9,
bbox=dict(boxstyle='round', facecolor='lightblue', alpha=0.8))
# Create animation
frames = len(self.animation_data)
interval = 3000 # 3 seconds per frame
print(f"Creating animation with {frames} frames...")
anim = [Link]([Link], animate_step, frames=frames,
interval=interval, repeat=True, blit=False)
if save_animation:
if video_format.lower() == 'mp4':
print("Saving animation as MP4 video...")
# For MP4, you need ffmpeg installed
try:
writer = FFMpegWriter(fps=1, metadata=dict(artist='RouteOptimizer'), bitrate=1800)
[Link]('enhanced_pathfinding_animation.mp4', writer=writer)
print("Animation saved as 'enhanced_pathfinding_animation.mp4'")
except Exception as e:
print(f"FFmpeg not available, falling back to GIF: {e}")
[Link]('enhanced_pathfinding_animation.gif', writer='pillow', fps=0.33)
print("Animation saved as 'enhanced_pathfinding_animation.gif'")
elif video_format.lower() == 'avi':
print("Saving animation as AVI video...")
try:
writer = FFMpegWriter(fps=1, codec='libx264')
[Link]('enhanced_pathfinding_animation.avi', writer=writer)
print("Animation saved as 'enhanced_pathfinding_animation.avi'")
except Exception as e:
print(f"FFmpeg not available, falling back to GIF: {e}")
[Link]('enhanced_pathfinding_animation.gif', writer='pillow', fps=0.33)
print("Animation saved as 'enhanced_pathfinding_animation.gif'")
else:
print("Saving animation as GIF...")
[Link]('enhanced_pathfinding_animation.gif', writer='pillow', fps=0.33)
print("Animation saved as 'enhanced_pathfinding_animation.gif'")
plt.tight_layout()
[Link]()
return anim
def create_sample_graph(self):
"""Create a sample graph for testing"""
edges = [
(1, 2, 5.0), (1, 3, 3.0), (1, 4, 7.0),
(2, 3, 2.0), (2, 5, 4.0), (2, 6, 6.0),
(3, 4, 4.0), (3, 6, 5.0), (3, 7, 8.0),
(4, 7, 3.0), (4, 8, 6.0),
(5, 6, 3.0), (5, 8, 9.0),
(6, 7, 4.0), (6, 8, 5.0),
(7, 8, 2.0)
]
for start, end, distance in edges:
self.add_edge(start, end, distance)
print(f"Sample graph created with {len(edges)} edges and {len(set([e[0] for e in edges] + [e[1]
for e in edges]))} nodes")
def calculate_path_metrics(self, path: List[int], day: str, time: int,
vehicle_count: int, road_cond: str, weather: str,
special_event: str) -> Dict[str, float]:
"""Calculate various metrics for a given path"""
if len(path) < 2:
return {'total_distance': 0, 'total_time': 0, 'traffic_score': 0, 'avg_traffic_score': 0,
'max_traffic_score': 0, 'num_high_traffic_edges': 0}
total_distance = 0
total_time = 0
traffic_scores = []
for i in range(len(path) - 1):
start_node = path[i]
end_node = path[i + 1]
# Get base distance
base_distance = [Link]((start_node, end_node), 10)
total_distance += base_distance
# Get traffic probabilities
traffic_probs = self.calculate_bayes_probabilities_pytorch(
day, time, vehicle_count, road_cond, weather,
special_event, start_node, end_node
)
# Calculate traffic score (higher = worse traffic)
traffic_score = (traffic_probs['low'] * 1.0 +
traffic_probs['medium'] * 2.0 +
traffic_probs['high'] * 3.0)
traffic_scores.append(traffic_score)
# Estimate time based on traffic
if traffic_score < 1.5:
speed_factor = 1.0 # Normal speed
elif traffic_score < 2.5:
speed_factor = 0.7 # Slower
else:
speed_factor = 0.4 # Much slower
# Assume base speed of 50 km/h
edge_time = (base_distance / 50) * (1 / speed_factor)
total_time += edge_time
return {
'total_distance': total_distance,
'total_time': total_time,
'avg_traffic_score': [Link](traffic_scores) if traffic_scores else 0,
'max_traffic_score': max(traffic_scores) if traffic_scores else 0,
'num_high_traffic_edges': sum(1 for score in traffic_scores if score > 2.5)
}
def run_animated_optimization(self, start: int, end: int, day: str = "Monday",
time: int = 8, vehicle_count: int = 200,
road_cond: str = "Good", weather: str = "No Rain",
special_event: str = "No", save_animation: bool = True,
video_format: str = 'mp4'):
"""Run the animated optimization process"""
print("=" * 80)
print("ENHANCED ANIMATED PYTORCH ROUTE OPTIMIZER")
🚀📅
print("=" * 80)
print(f" Journey: Node {start} → Node {end}")
🕐🚗
print(f" Day: {day}")
print(f" Time: {time}:00")
print(f" 🛣️🌤️
print(f" Vehicle Count: {vehicle_count}")
Road Condition: {road_cond}")
print(f"
🎪 Weather: {weather}")
print(f" Special Event: {special_event}")
🧠
print("=" * 80)
print(" Algorithms Used:")
print(" • Bayes Theorem for traffic state prediction")
print(" • Hidden Markov Model for traffic evolution")
print(" • Monte Carlo simulation for path evaluation")
print(" • Real-time step-by-step pathfinding")
print("=" * 80)
# Reset traffic state
self.current_traffic_state = None
🎬
# Run the animated pathfinding
print("\n Starting animated pathfinding...")
animation_obj = self.animate_pathfinding(
start, end, day, time, vehicle_count,
road_cond, weather, special_event, save_animation, video_format
)
# Calculate and display final path metrics
if self.animation_data:
final_step = self.animation_data[-1]
final_path = final_step['path_so_far']
📊
print("\n" + "=" * 80)
print(" FINAL PATH ANALYSIS")
print("=" * 80)
metrics = self.calculate_path_metrics(
final_path, day, time, vehicle_count,
road_cond, weather, special_event
)
print(f"🛣️📏 Final Path: {' → '.join(map(str, final_path))}")
print(f"⏱️🚦
print(f" Total Distance: {metrics['total_distance']:.2f} km")
Estimated Time: {metrics['total_time']:.2f} hours")
🔴📈
print(f" Average Traffic Score: {metrics['avg_traffic_score']:.2f}/3.0")
print(f" High Traffic Edges: {metrics['num_high_traffic_edges']}")
print(f" Path Quality: {'Excellent' if metrics['avg_traffic_score'] < 1.5 else 'Good' if
metrics['avg_traffic_score'] < 2.5 else 'Poor'}")
return animation_obj
def main():
🚀
"""Main function to demonstrate the Enhanced Animated PyTorch Route Optimizer"""
print(" ENHANCED ANIMATED PYTORCH ROUTE OPTIMIZER")
print("=" * 80)
# Initialize optimizer
optimizer = AnimatedPyTorchRouteOptimizer()
📊
# Create sample graph
print(" Creating sample graph...")
optimizer.create_sample_graph()
🔄
# Generate sample data
print(" Generating sample traffic data...")
data_file = '/content/traffic_new.csv'
📥
# Load the data
print(" Loading and processing data...")
optimizer.load_dataset_from_csv(data_file)
🎬
print("\n" + "="*80)
print(" DEMO 1: MORNING RUSH HOUR PATHFINDING")
print("="*80)
# Run animated optimization - Morning rush hour
animation1 = optimizer.run_animated_optimization(
start=1, end=7, day="Monday", time=8, vehicle_count=350,
road_cond="Good", weather="No Rain", special_event="No",
save_animation=True, video_format='mp4'
)
🎬
print("\n" + "="*80)
# print(" DEMO 2: EVENING WITH BAD WEATHER")
# print("="*80)
# # Run second demo with different conditions
# # animation2 = optimizer.run_animated_optimization(
# # start=1, end=7, day="Friday", time=18, vehicle_count=400,
# # road_cond="Poor", weather="Heavy Rain", special_event="Yes",
# # save_animation=False # Don't save to avoid overwriting
##)
# Compare multiple runs
🔍
print("\n" + "="*80)
print(" ALGORITHM CONSISTENCY ANALYSIS")
print("="*80)
optimizer.compare_multiple_runs(start=1, end=7, num_runs=5)
✅
print("\n" + "="*80)
print(" DEMONSTRATION COMPLETE")
🎯
print("="*80)
print(" Key Features Demonstrated:")
print(" ✓ Step-by-step pathfinding with real-time decision making")
print(" ✓ Bayes theorem for traffic state prediction")
print(" ✓ HMM for traffic evolution modeling")
print(" ✓ Monte Carlo simulation for path evaluation")
print(" ✓ Dynamic visualization with traffic analysis")
🎬📁
print(" ✓ Comprehensive path metrics and comparisons")
print("\n Animation saved as 'enhanced_pathfinding_animation.mp4'")
print(" Live animation displayed in matplotlib window")
if __name__ == "__main__":
main()