Dev trip planner

co2calculator/constants.py

@@ -53,6 +53,18 @@ class CarBusFuel(str, enum.Enum):
     HYDROGEN = "hydrogen"
 
 
+@enum.unique
+class BusFuel(str, enum.Enum):
+    """Enum for bus fuel types"""
+
+    ELECTRIC = "electric"
+    DIESEL = "diesel"
+    AVERAGE = "average"
+    HYDROGEN = "hydrogen"
+    CNG = "cng"
+
+
+
 @enum.unique
 class Size(str, enum.Enum):
     """Enum for car sizes"""

co2calculator/distances.py

@@ -11,16 +11,18 @@
 import numpy as np
 import openrouteservice
 import pandas as pd
+import pydantic
 from dotenv import load_dotenv
 from openrouteservice.directions import directions
 from openrouteservice.geocode import pelias_search, pelias_structured
 from pydantic import BaseModel, ValidationError, Extra, confloat
 from thefuzz import fuzz
 from thefuzz import process
+from iso3166 import countries
 
 from ._types import Kilometer
+from .enums import TransportationMode
 from .constants import (
-    TransportationMode,
     CountryCode2,
     CountryCode3,
     CountryName,
@@ -39,7 +41,7 @@
 script_path = str(Path(__file__).parent)
 
 
-class StructuredLocation(BaseModel, extra=Extra.forbid):
+class StructuredLocation(BaseModel, extra=Extra.ignore):
     address: Optional[str]
     locality: str
     country: Union[CountryCode2, CountryCode3, CountryName]
@@ -50,9 +52,19 @@ class StructuredLocation(BaseModel, extra=Extra.forbid):
     neighbourhood: Optional[str]
 
 
-class TrainStation(BaseModel):
+class TrainStation(BaseModel, extra=Extra.ignore):
     station_name: str
-    country: CountryCode2
+    country_code: CountryCode2 = None
+    country: CountryName = None
+
+    @pydantic.root_validator(pre=False)
+    def get_country_code(cls, values):
+        if values["country_code"] is None:
+            try:
+                values["country_code"] = countries.get(values["country"]).alpha2
+            except KeyError as e:
+                raise ValueError(f"Invalid country: {values['country']}.")
+        return values
 
 
 class Airport(BaseModel):
@@ -305,7 +317,7 @@ def geocoding_train_stations(loc_dict):
     # remove stations with no coordinates
     stations_df.dropna(subset=["latitude", "longitude"], inplace=True)
     countries_eu = stations_df["country"].unique()
-    country_code = station.country
+    country_code = station.country_code
     if country_code not in countries_eu:
         warnings.warn(
             "The provided country is not within Europe. "
@@ -493,34 +505,34 @@ def create_distance_request(
 
     try:
         if transportation_mode in [
-            TransportationMode.CAR,
-            TransportationMode.MOTORBIKE,
-            TransportationMode.BUS,
-            TransportationMode.FERRY,
+            TransportationMode.Car,
+            TransportationMode.Motorbike,
+            TransportationMode.Bus,
+            TransportationMode.Ferry,
         ]:
             return DistanceRequest(
                 transportation_mode=transportation_mode,
                 start=StructuredLocation(**start),
                 destination=StructuredLocation(**destination),
             )
 
-        if transportation_mode in [TransportationMode.TRAIN]:
+        if transportation_mode in [TransportationMode.Train]:
             return DistanceRequest(
                 transportation_mode=transportation_mode,
                 start=TrainStation(**start),
                 destination=TrainStation(**destination),
             )
 
-        if transportation_mode in [TransportationMode.PLANE]:
+        if transportation_mode in [TransportationMode.Plane]:
             return DistanceRequest(
                 transportation_mode=transportation_mode,
                 start=Airport(iata_code=start),
                 destination=Airport(iata_code=destination),
             )
 
     except ValidationError as e:
-        raise InvalidSpatialInput(e)
-
+        #raise InvalidSpatialInput(e)
+        raise InvalidSpatialInput(f"unknown transportation_mode: '{transportation_mode}'")
     raise InvalidSpatialInput(f"unknown transportation_mode: '{transportation_mode}'")
 
 
@@ -536,25 +548,25 @@ def get_distance(request: DistanceRequest) -> Kilometer:
     """
 
     detour_map = {
-        TransportationMode.CAR: False,
-        TransportationMode.MOTORBIKE: False,
-        TransportationMode.BUS: True,
-        TransportationMode.TRAIN: True,
-        TransportationMode.PLANE: True,
-        TransportationMode.FERRY: False,
+        TransportationMode.Car: False,
+        TransportationMode.Motorbike: False,
+        TransportationMode.Bus: True,
+        TransportationMode.Train: True,
+        TransportationMode.Plane: True,
+        TransportationMode.Ferry: True,
     }
 
     if request.transportation_mode in [
-        TransportationMode.CAR,
-        TransportationMode.MOTORBIKE,
+        TransportationMode.Car,
+        TransportationMode.Motorbike,
     ]:
         coords = []
         for loc in [request.start, request.destination]:
             _, _, loc_coords, _ = geocoding_structured(loc.dict())
             coords.append(loc_coords)
         return get_route(coords, "driving-car")
 
-    if request.transportation_mode == TransportationMode.BUS:
+    if request.transportation_mode == TransportationMode.Bus:
         # Same as car (StructuredLocation)
         # TODO: Validate with BaseModel
         # TODO: Question: Why are we not calculating the bus trip like `driving-car` routes?
@@ -570,7 +582,7 @@ def get_distance(request: DistanceRequest) -> Kilometer:
             )
         return _apply_detour(distance, request.transportation_mode)
 
-    if request.transportation_mode == TransportationMode.TRAIN:
+    if request.transportation_mode == TransportationMode.Train:
 
         distance = 0
         coords = []
@@ -591,17 +603,22 @@ def get_distance(request: DistanceRequest) -> Kilometer:
             )
         return _apply_detour(distance, request.transportation_mode)
 
-    if request.transportation_mode == TransportationMode.PLANE:
+    if request.transportation_mode == TransportationMode.Plane:
         # Stops are IATA code of airports
         # TODO: Validate stops with BaseModel
 
         _, geom_start, _ = geocoding_airport(request.start.iata_code)
         _, geom_dest, _ = geocoding_airport(request.destination.iata_code)
+        #_, _, geom_start, _ = geocoding_structured(request.start.dict())
+        #_, _, geom_dest, _ = geocoding_structured(request.destination.dict())
 
         distance = haversine(geom_start[1], geom_start[0], geom_dest[1], geom_dest[0])
         return _apply_detour(distance, request.transportation_mode)
 
-    if request.transportation_mode == TransportationMode.FERRY:
+    if request.transportation_mode == TransportationMode.Ferry:
+        # todo: Do we have a way of checking if there even exists a ferry connection between the given cities (or if the
+        #  cities even have a port?
+
         _, _, geom_start, _ = geocoding_structured(request.start.dict())
         _, _, geom_dest, _ = geocoding_structured(request.destination.dict())
 

co2calculator/enums.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""Enums holding parameter choices derived from data set"""
+
+__author__ = "Christina Ludwig, GIScience Research Group, Heidelberg University"
+__email__ = "christina.ludwig@uni-heidelberg.de"
+
+from enum import Enum
+import pandas as pd
+from pathlib import Path
+import numpy as np
+
+script_path = str(Path(__file__).parent)
+
+
+class EnumCreator:
+
+    def __init__(self):
+        self.df = pd.read_csv(f"{script_path}/../data/emission_factors.csv")
+
+    def create_enum(self, mode, column, name):
+        df = self.df[self.df["subcategory"] == mode]
+        if df[column].dtype == np.float64:
+            data = df[column].map(lambda x: "c_{:.0f}".format(x)).astype("str")
+            return Enum(name, {x: float(x[2:]) for x in data}, type=str)
+        else:
+            data = df[column].fillna("nan").unique()
+            return Enum(name, {x.capitalize().replace("-", "_"): x for x in data}, type=str)
+
+    def create_enum_transport(self, column, name):
+        data = self.df[column].fillna("nan").astype("str").unique()
+        return Enum(name, {x.capitalize(): x for x in data}, type=str)
+
+
+enum_creator = EnumCreator()
+TransportationMode = enum_creator.create_enum_transport("subcategory", "TransportationMode")
+
+# Fuel types ------------------------------------------------------
+
+TrainFuelType = enum_creator.create_enum(TransportationMode.Train,
+                            "fuel_type",
+                            "TrainFuelType")
+CarFuelType = enum_creator.create_enum(TransportationMode.Car,
+                            "fuel_type",
+                          "CarFuelType")
+BusFuelType = enum_creator.create_enum(TransportationMode.Bus,
+                            "fuel_type",
+                          "BusFuelType")
+TramFuelType = enum_creator.create_enum(TransportationMode.Tram,
+                            "fuel_type", "TramFuelType")
+PlaneFuelType = enum_creator.create_enum(TransportationMode.Plane,
+                            "fuel_type",
+                            "PlaneFuelType")
+FerryFuelType = enum_creator.create_enum(TransportationMode.Ferry,
+                            "fuel_type",
+                            "FerryFuelType")
+BicycleFuelType = enum_creator.create_enum(TransportationMode.Bicycle,
+                            "fuel_type",
+                              "BicycleFuelType")
+PedelecFuelType = enum_creator.create_enum(TransportationMode.Pedelec,
+                            "fuel_type",
+                              "PedelecFuelType")
+
+# Size
+TrainSize = enum_creator.create_enum(TransportationMode.Train,
+                            "size",
+                            "TrainSize")
+CarSize = enum_creator.create_enum(TransportationMode.Car,
+                            "size",
+                          "CarSize")
+BusSize = enum_creator.create_enum(TransportationMode.Bus,
+                            "size",
+                          "BusSize")
+MotorbikeSize = enum_creator.create_enum(TransportationMode.Motorbike,
+                            "size",
+                          "MotorbikeSize")
+TramSize = enum_creator.create_enum(TransportationMode.Tram,
+                            "size",
+                            "TramSize")
+
+# Vehicle range
+TrainRange = enum_creator.create_enum(TransportationMode.Train,
+                            "range",
+                            "TrainRange")
+BusRange = enum_creator.create_enum(TransportationMode.Bus,
+                            "range",
+                          "BusRange")
+
+PlaneRange = enum_creator.create_enum(TransportationMode.Plane,
+                            "range",
+                          "PlaneRange")
+
+# Seating class
+PlaneSeatingClass = enum_creator.create_enum(TransportationMode.Plane,
+                            "seating",
+                            "PlaneSeatingClass")
+FerrySeatingClass = enum_creator.create_enum(TransportationMode.Ferry,
+                            "seating",
+                          "FerrySeatingClass")
+
+# Occupancy
+TrainOccupancy = enum_creator.create_enum(TransportationMode.Train,
+                            "occupancy",
+                            "TrainOccupancy")
+BusOccupancy = enum_creator.create_enum( TransportationMode.Bus,
+                            "occupancy",
+                            "BusOccupancy")

co2calculator/exceptions.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""__description__"""
+
+__author__ = "Christina Ludwig, GIScience Research Group, Heidelberg University"
+__email__ = "christina.ludwig@uni-heidelberg.de"
+
+
+class ConversionFactorNotFound(Exception):
+
+    def __init__(self, message):
+        self.message = message
+