* Implemented "PlannedTripCriterion" object to decide what selection …

…criterion to use simply picking a class. #24

jinete/__init__.py

@@ -11,7 +11,7 @@
     Route,
     Planning,
     Result,
-    DirectionObjective,
+    OptimizationDirection,
     Objective,
     DialARideObjective,
     TaxiSharingObjective,

jinete/algorithms/metaheuristics/grasp.py

@@ -46,6 +46,6 @@ def _optimize(self) -> Planning:
         for i in range(self.episodes):
             seed = self.random.randint(0, maxsize)
             current = self.build_algorithm(seed=seed).optimize()
-            best = self.objective.best_result(best, current)
+            best = self.objective.best(best, current)
         logger.info('Optimized!')
         return best.planning

jinete/algorithms/utils/crossers/abc.py

@@ -14,6 +14,7 @@
 )
 from ....models import (
     Route,
+    ShortestTimePlannedTripCriterion,
 )
 
 if TYPE_CHECKING:
@@ -41,6 +42,8 @@ def __init__(self, fleet: Fleet, job: Job):
         self.routes = set(Route(vehicle) for vehicle in self.vehicles)
         self._pending_trips = set(self.trips)
 
+        self.criterion = ShortestTimePlannedTripCriterion()
+
     @property
     def vehicles(self) -> Set[Vehicle]:
         return self.fleet.vehicles
@@ -49,10 +52,6 @@ def vehicles(self) -> Set[Vehicle]:
     def trips(self) -> Set[Trip]:
         return self.job.trips
 
-    @property
-    def objective(self) -> Objective:
-        return self.job.objective
-
     def __iter__(self):
         return self
 

jinete/algorithms/utils/crossers/best_stateless.py

@@ -21,7 +21,5 @@
 class BestStatelessCrosser(StatelessCrosser):
 
     def get_planned_trip(self) -> Optional[PlannedTrip]:
-        best = None
-        for current in self.iterator:
-            best = self.objective.best_planned_trip(best, current)
+        best = self.criterion.best(*self.iterator)
         return best

jinete/algorithms/utils/crossers/ordered.py

@@ -52,7 +52,7 @@ def create_sub_ranking(self, route: Route) -> OrderedDict[UUID, PlannedTrip]:
             if planned_trip is None:
                 continue
             raw_sub_ranking.append(planned_trip)
-        raw_sub_ranking.sort(key=lambda pt: self.objective.planned_trip_scoring_function(pt))
+        self.criterion.sorted(raw_sub_ranking, inplace=True)
         return OrderedDict((item.trip_uuid, item) for item in raw_sub_ranking)
 
     def update_ranking(self, planned_trip: PlannedTrip) -> None:
@@ -74,5 +74,5 @@ def get_planned_trip(self) -> Optional[PlannedTrip]:
             if len(sub_ranking) == 0:
                 continue
             current = next(iter(sub_ranking.values()))
-            best = self.objective.best_planned_trip(best, current)
+            best = self.criterion.best(best, current)
         return best

jinete/algorithms/utils/crossers/randomized.py

@@ -35,14 +35,14 @@ def get_planned_trip(self) -> Optional[PlannedTrip]:
                 continue
             for current in sub_ranking.values():
                 if len(candidates) != 0:
-                    best = self.objective.best_planned_trip(candidates[-1], current)
+                    best = self.criterion.best(candidates[-1], current)
                     if not best != candidates[-1]:
                         break
 
                 if self.randomized_size < len(candidates):
                     candidates.pop()
                 candidates.append(current)
-            candidates.sort(key=lambda pt: self.objective.planned_trip_scoring_function(pt))
+            self.criterion.sorted(candidates, inplace=True)
         if len(candidates) == 0:
             return None
         best = self.random.choice(candidates)

jinete/models/__init__.py

@@ -16,6 +16,13 @@
 from .planned_trips import (
     PlannedTrip,
 )
+from .criterions import (
+    PlannedTripCriterion,
+    ShortestTimePlannedTripCriterion,
+)
+from .constants import (
+    OptimizationDirection,
+)
 from .jobs import (
     Job
 )
@@ -33,7 +40,7 @@
     Result,
 )
 from .objectives import (
-    DirectionObjective,
+    OptimizationDirection,
     Objective,
     DialARideObjective,
     TaxiSharingObjective,

jinete/models/constants.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from enum import (
+    Enum,
+    unique,
+)
+
+if TYPE_CHECKING:
+    from typing import (
+        Callable,
+        Iterable,
+        TypeVar,
+    )
+
+    T = TypeVar('T')
+
+
+@unique
+class OptimizationDirection(Enum):
+    MAXIMIZATION = max
+    MINIMIZATION = min
+
+    @property
+    def fn(self) -> Callable[[Iterable[T]], T]:
+        return self.value
+
+    def __str__(self):
+        return f'{self.name.capitalize()}'

jinete/models/criterions.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+from .constants import (
+    OptimizationDirection,
+)
+
+if TYPE_CHECKING:
+    from typing import (
+        Iterable,
+        Optional,
+        List,
+    )
+    from .planned_trips import (
+        PlannedTrip,
+    )
+
+
+class PlannedTripCriterion(ABC):
+    def __init__(self, name: str, direction: OptimizationDirection):
+        self.name = name
+        self.direction = direction
+
+    @abstractmethod
+    def scoring(self, planned_trip: PlannedTrip) -> float:
+        pass
+
+    def best(self, *args: PlannedTrip) -> Optional[PlannedTrip]:
+        return self.direction.fn(
+            (arg for arg in args if arg is not None),
+            key=self.scoring,
+            default=None,
+        )
+
+    def sorted(self, arr: List[PlannedTrip], inplace: bool = False) -> List[PlannedTrip]:
+        if inplace:
+            arr.sort(key=self.scoring)
+        else:
+            arr = sorted(arr, key=self.scoring)
+        return arr
+
+
+class ShortestTimePlannedTripCriterion(PlannedTripCriterion):
+
+    def __init__(self):
+        super().__init__(
+            direction=OptimizationDirection.MAXIMIZATION,
+            name='Shortest-Time',
+        )
+
+    def scoring(self, planned_trip: PlannedTrip) -> float:
+        return planned_trip.collection_time - planned_trip.route.last_time

jinete/models/objectives.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 from abc import ABC, abstractmethod
-from enum import Enum
 
 from typing import TYPE_CHECKING
+from .constants import (
+    OptimizationDirection,
+)
 
 if TYPE_CHECKING:
     from typing import (
-        Callable,
-        Sequence,
-        Any,
+        Iterable,
     )
     from .routes import (
         Route,
@@ -24,81 +24,60 @@
     )
 
 
-class DirectionObjective(Enum):
-    MAXIMIZE = 1
-    MINIMIZE = -1
-
-
 class Objective(ABC):
-    direction: DirectionObjective
+    direction: OptimizationDirection
 
-    def __init__(self, name: str, direction: DirectionObjective):
+    def __init__(self, name: str, direction: OptimizationDirection):
         self.name = name
         self.direction = direction
 
-    @property
-    def _direction_function(self) -> Callable[[Sequence[Any]], float]:
-        if self.direction == DirectionObjective.MAXIMIZE:
-            return max
-        else:
-            return min
+    def best(self, *args: Result) -> Result:
+        return self.direction.fn(
+            (arg for arg in args if arg is not None),
+            key=lambda pt: self.scoring(pt)
+        )
 
-    def result_optimization_function(self, result: Result) -> float:
-        return self.planning_optimization_function(result.planning)
+    def scoring(self, result: Result) -> float:
+        return self._planning_optimization_function(result.planning)
 
-    def planning_optimization_function(self, planning: Planning) -> float:
+    def _planning_optimization_function(self, planning: Planning) -> float:
         scoring = 0.0
         for route in planning:
-            scoring += self.route_optimization_function(route)
+            scoring += self._route_optimization_function(route)
         return scoring
 
-    def route_optimization_function(self, route: Route) -> float:
+    def _route_optimization_function(self, route: Route) -> float:
         scoring = 0.0
         for planned_trip in route:
-            scoring += self.planned_trip_optimization_function(planned_trip)
+            scoring += self._planned_trip_optimization_function(planned_trip)
         return scoring
 
     @abstractmethod
-    def planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
+    def _planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
         pass
 
-    def planned_trip_scoring_function(self, planned_trip: PlannedTrip) -> float:
-        return planned_trip.collection_time - planned_trip.route.last_time
-
-    def best_planned_trip(self, *args) -> float:
-        return self._direction_function(
-            (arg for arg in args if arg is not None),
-            key=lambda pt: self.planned_trip_scoring_function(pt),
-        )
-
-    def best_result(self, *args) -> float:
-        return self._direction_function(
-            (arg for arg in args if arg is not None),
-            key=lambda pt: self.result_optimization_function(pt)
-        )
-
 
 class DialARideObjective(Objective):
 
     def __init__(self):
         super().__init__(
             name='Dial-a-Ride',
-            direction=DirectionObjective.MINIMIZE,
+            direction=OptimizationDirection.MINIMIZATION,
         )
 
-    def planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
-        return planned_trip.cost
+    def _planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
+        return planned_trip.distance
 
 
 class TaxiSharingObjective(Objective):
 
     def __init__(self):
         super().__init__(
             name='Taxi-Sharing',
-            direction=DirectionObjective.MAXIMIZE,
+            direction=OptimizationDirection.MAXIMIZATION,
         )
 
-    def planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
+    def _planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
         if planned_trip.capacity == 0:
             return 0.0
         return planned_trip.duration
@@ -108,10 +87,10 @@ class HashCodeObjective(Objective):
     def __init__(self):
         super().__init__(
             name='HashCode-2018',
-            direction=DirectionObjective.MAXIMIZE,
+            direction=OptimizationDirection.MAXIMIZATION,
         )
 
-    def planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
+    def _planned_trip_optimization_function(self, planned_trip: PlannedTrip) -> float:
         if planned_trip.capacity == 0:
             return 0.0
         trip = planned_trip.trip

jinete/models/planned_trips.py

@@ -56,10 +56,6 @@ def distance(self) -> float:
     def duration(self) -> float:
         return self.delivery_time - self.collection_time
 
-    @property
-    def cost(self) -> float:
-        return self.duration
-
     @property
     def capacity(self):
         return self.trip.capacity

jinete/models/plannings.py

@@ -38,10 +38,3 @@ def __iter__(self):
     @property
     def loaded_routes(self):
         return set(route for route in self.routes if route.loaded)
-
-    @property
-    def cost(self) -> float:
-        cost = 0.0
-        for route in self.routes:
-            cost += route.cost
-        return cost

jinete/models/results.py

@@ -10,6 +10,7 @@
         Trip,
         Route,
         Objective,
+        OptimizationDirection,
     )
 
 
@@ -42,9 +43,9 @@ def objective(self) -> Objective:
         return self.job.objective
 
     @property
-    def cost(self) -> float:
-        return self.planning.cost
+    def scoring(self) -> float:
+        return self.objective.scoring(self)
 
     @property
-    def scoring(self) -> float:
-        return self.objective.result_optimization_function(self)
+    def direction(self) -> OptimizationDirection:
+        return self.objective.direction

jinete/models/routes.py

@@ -71,13 +71,6 @@ def feasible(self) -> bool:
                 return False
         return True
 
-    @property
-    def cost(self) -> float:
-        cost = 0.0
-        for planned_trip in self.planned_trips:
-            cost += planned_trip.cost
-        return cost
-
     @property
     def loaded(self):
         return len(self.planned_trips) > 0

jinete/storers/formatters/columnar.py

@@ -55,6 +55,6 @@ def format(self) -> str:
             '\n'.join(f'{self.tab_character}{row}' for row in rows),
             f'Computation time: "{self.result.computation_time:0.4f}" seconds',
             f'Coverage Rate: "{self.result.coverage_rate}"',
-            f'Cost: "{self.result.cost}"',
             f'Scoring: "{self.result.scoring}"',
+            f'Direction: "{self.result.direction}"',
         ))