0% found this document useful (0 votes)
5 views18 pages

PyTorch Route Optimization Animation

The document outlines a PyTorch-based route optimization system that utilizes a graph structure to model transportation routes and traffic conditions. It incorporates Monte Carlo simulations, Hidden Markov Models (HMM), and Bayesian probability calculations to evaluate and predict traffic states based on various factors such as time, weather, and vehicle counts. Additionally, it includes functionality for visualizing the pathfinding process through animations using Matplotlib and NetworkX.

Uploaded by

cs23b2017
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
5 views18 pages

PyTorch Route Optimization Animation

The document outlines a PyTorch-based route optimization system that utilizes a graph structure to model transportation routes and traffic conditions. It incorporates Monte Carlo simulations, Hidden Markov Models (HMM), and Bayesian probability calculations to evaluate and predict traffic states based on various factors such as time, weather, and vehicle counts. Additionally, it includes functionality for visualizing the pathfinding process through animations using Matplotlib and NetworkX.

Uploaded by

cs23b2017
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

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()

You might also like