File size: 4,042 Bytes
069dbdc
758eef3
735d438
 
758eef3
 
 
 
 
069dbdc
758eef3
 
 
 
735d438
e929ca4
758eef3
b645561
 
758eef3
 
b645561
 
758eef3
b645561
758eef3
 
 
 
 
b0e0875
 
 
758eef3
b0e0875
758eef3
735d438
e7da8f8
b645561
758eef3
 
 
 
 
b645561
758eef3
b645561
758eef3
b645561
758eef3
e929ca4
b645561
 
 
e929ca4
b645561
 
 
758eef3
 
 
 
 
 
 
 
 
 
 
b0e0875
069dbdc
758eef3
 
b0e0875
758eef3
b0e0875
 
 
758eef3
 
b0e0875
 
 
 
 
 
 
758eef3
 
 
b0e0875
758eef3
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
import math
from geopy.geocoders import Nominatim

class GreenRouteOptimizer:
    """
    Calculates and compares green routes for shipments, providing data on emissions,
    travel time, and costs, formatted for the GreenPath Streamlit app.
    """
    def __init__(self, user_agent="green_path_app"):
        self.geolocator = Nominatim(user_agent=user_agent)
        # Constants used for calculations
        self.emission_factors = {"road": 0.21, "rail": 0.041, "ship": 0.014}  # kg CO₂/tonne-km
        self.avg_speeds_kmh = {"road": 80, "rail": 100, "ship": 30}          # km/h
        self.carbon_tax_rate_usd = 50.0                                     # $ per tonne of CO₂

    def geocode(self, place_name):
        """Geocodes a place name, returning (lat, lon) or None on failure."""
        try:
            location = self.geolocator.geocode(place_name, timeout=10)
            if location:
                return (location.latitude, location.longitude)
        except Exception:
            return None
        return None

    def calculate_route_metrics(self, distance_km, mode, weight_tonnes):
        """Calculates all metrics for a single route."""
        emissions_kg = distance_km * self.emission_factors[mode] * weight_tonnes
        travel_time_hours = distance_km / self.avg_speeds_kmh[mode]
        carbon_tax_usd = (emissions_kg / 1000) * self.carbon_tax_rate_usd
        
        return {
            "transport_mode": mode,
            "co2_emissions_kg": emissions_kg,
            "estimated_travel_time_hours": travel_time_hours,
            "carbon_tax_cost_usd": carbon_tax_usd,
        }

    def recommend_green_routes(self, start_place, end_place, weight_tonnes):
        """
        The main method that finds and processes route recommendations.
        Returns a dictionary compatible with the Streamlit frontend.
        """
        # --- 1. Input Validation ---
        if not start_place or not start_place.strip():
            return {"error": "Origin location cannot be empty."}
        if not end_place or not end_place.strip():
            return {"error": "Destination location cannot be empty."}

        # --- 2. Geocoding ---
        start_coords = self.geocode(start_place)
        if start_coords is None:
            return {"error": f"Could not find location: '{start_place}'"}

        end_coords = self.geocode(end_place)
        if end_coords is None:
            return {"error": f"Could not find location: '{end_place}'"}

        # --- 3. Calculate Distance (Haversine Formula) ---
        R = 6371  # Radius of Earth in km
        lat1, lon1 = start_coords
        lat2, lon2 = end_coords
        dlat = math.radians(lat2 - lat1)
        dlon = math.radians(lon2 - lon1)
        a = math.sin(dlat / 2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2)**2
        c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
        distance_km = R * c

        # --- 4. Calculate Metrics for All Modes ---
        all_routes = []
        for mode in self.emission_factors:
            route_metrics = self.calculate_route_metrics(distance_km, mode, weight_tonnes)
            all_routes.append(route_metrics)

        # --- 5. Post-Processing for Comparison Metrics ---
        if not all_routes:
            return {"error": "Could not calculate any routes."}

        # Find the worst emission value to calculate savings against it
        worst_emission = max(route['co2_emissions_kg'] for route in all_routes)

        for route in all_routes:
            if worst_emission > 0:
                reduction = (worst_emission - route['co2_emissions_kg']) / worst_emission * 100
                route['emission_reduction_percent'] = reduction
            else:
                route['emission_reduction_percent'] = 0
        
        # Sort routes by emissions (best first)
        all_routes.sort(key=lambda x: x['co2_emissions_kg'])

        # --- 6. Return Data in the Exact Format the App Expects ---
        return {"recommendations": all_routes}