Distinguish betwee flow_direction and journey_direction, fix tests. R…

…an the demo scripts without any modification for ramp-up and ramp-down. Things still look fine.

conflowgen/api/container_flow_generation_manager.py

@@ -54,8 +54,15 @@ def set_properties(
         properties.start_date = start_date
         properties.end_date = end_date
 
-        properties.ramp_up_period = ramp_up_period.total_seconds() / 86400  # in days as float
-        properties.ramp_down_period = ramp_down_period.total_seconds() / 86400 # in days as float
+        if ramp_up_period:
+            properties.ramp_up_period = ramp_up_period.total_seconds() / 86400  # in days as float
+        else:
+            properties.ramp_up_period = 0
+
+        if ramp_down_period:
+            properties.ramp_down_period = ramp_down_period.total_seconds() / 86400  # in days as float
+        else:
+            properties.ramp_down_period = 0
 
         if transportation_buffer is not None:
             properties.transportation_buffer = transportation_buffer

conflowgen/application/models/container_flow_generation_properties.py

@@ -27,7 +27,7 @@ class ContainerFlowGenerationProperties(BaseModel):
         help_text="The last day of the generated container flow"
     )
 
-    ramp_up_period= FloatField(
+    ramp_up_period = FloatField(
         default=0,
         help_text="Number of days for the ramp-up period"
     )

conflowgen/application/reports/container_flow_statistics_report.py

@@ -4,6 +4,7 @@
 import statistics
 from typing import List, Type, Dict
 
+from conflowgen.descriptive_datatypes import FlowDirection
 from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
 from conflowgen.domain_models.repositories.large_scheduled_vehicle_repository import LargeScheduledVehicleRepository
 from conflowgen.domain_models.vehicle import AbstractLargeScheduledVehicle, LargeScheduledVehicle
@@ -44,7 +45,8 @@ def _generate_free_capacity_statistics(self, vehicles_of_types):
                     vehicle
                 )
                 free_capacity_outbound = self.large_scheduled_vehicle_repository.get_free_capacity_for_outbound_journey(
-                    vehicle
+                    vehicle,
+                    FlowDirection.undefined
                 )
                 assert free_capacity_inbound <= large_scheduled_vehicle.capacity_in_teu, \
                     f"A vehicle can only load at maximum its capacity, but for vehicle {vehicle} the free capacity " \

conflowgen/descriptive_datatypes/__init__.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import enum
 import typing
 
 from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
@@ -127,3 +128,18 @@ class ContainersTransportedByTruck(typing.NamedTuple):
 
     #: The number of containers moved on the outbound journey
     outbound: float
+
+
+class FlowDirection(enum.Enum):
+    """
+    Represents the flow direction based on the terminology of
+    *Handbook of Terminal Planning*, edited by Jürgen W. Böse (https://link.springer.com/book/10.1007/978-3-030-39990-0)
+    """
+
+    import_flow = "import"
+
+    export_flow = "export"
+
+    transshipment_flow = "transshipment"
+
+    undefined = "undefined"

conflowgen/domain_models/container.py

@@ -14,6 +14,7 @@
 from .vehicle import LargeScheduledVehicle
 from .vehicle import Truck
 from .data_types.storage_requirement import StorageRequirement
+from ..descriptive_datatypes import FlowDirection
 from ..domain_models.data_types.mode_of_transport import ModeOfTransport
 
 
@@ -117,15 +118,17 @@ def occupied_teu(self) -> float:
         return CONTAINER_LENGTH_TO_OCCUPIED_TEU[self.length]
 
     @property
-    def flow_direction(self) -> str:
+    def flow_direction(self) -> FlowDirection:
         if (self.delivered_by in [ModeOfTransport.truck, ModeOfTransport.train, ModeOfTransport.barge]
                 and self.picked_up_by in [ModeOfTransport.feeder, ModeOfTransport.deep_sea_vessel]):
-            return "export"
+            return FlowDirection.export_flow
         if (self.picked_up_by in [ModeOfTransport.truck, ModeOfTransport.train, ModeOfTransport.barge]
                 and self.delivered_by in [ModeOfTransport.feeder, ModeOfTransport.deep_sea_vessel]):
-            return "import"
-        else:
-            return "transshipment"
+            return FlowDirection.import_flow
+        if (self.picked_up_by in [ModeOfTransport.feeder, ModeOfTransport.deep_sea_vessel]
+                and self.delivered_by in [ModeOfTransport.feeder, ModeOfTransport.deep_sea_vessel]):
+            return FlowDirection.transshipment_flow
+        return FlowDirection.undefined
 
     def get_arrival_time(self) -> datetime.datetime:
 

conflowgen/domain_models/repositories/large_scheduled_vehicle_repository.py

@@ -4,15 +4,16 @@
 import typing
 from typing import Dict, List, Callable, Type
 
+from conflowgen.descriptive_datatypes import FlowDirection
 from conflowgen.domain_models.container import Container
 from conflowgen.domain_models.data_types.container_length import ContainerLength
 from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
 from conflowgen.domain_models.vehicle import LargeScheduledVehicle, AbstractLargeScheduledVehicle
 
 
-class _Direction(enum.Enum):
-    inbound = 0
-    outbound = 1
+class JourneyDirection(enum.Enum):
+    inbound = "inbound"
+    outbound = "outbound"
 
 
 class LargeScheduledVehicleRepository:
@@ -33,9 +34,9 @@ def set_transportation_buffer(self, transportation_buffer: float):
 
     def set_ramp_up_and_down_times(
             self,
-            ramp_up_period_end: typing.Optional[datetime.datetime],
-            ramp_down_period_start: typing.Optional[datetime.datetime]
-    ):
+            ramp_up_period_end: typing.Optional[datetime.datetime] = None,
+            ramp_down_period_start: typing.Optional[datetime.datetime] = None
+    ) -> None:
         self.ramp_up_period_end = ramp_up_period_end
         self.ramp_down_period_start = ramp_down_period_start
 
@@ -106,14 +107,15 @@ def get_free_capacity_for_inbound_journey(self, vehicle: Type[AbstractLargeSched
             vehicle=vehicle,
             maximum_capacity=total_moved_capacity_for_inbound_transportation_in_teu,
             container_counter=self._get_number_containers_for_inbound_journey,
-            direction=_Direction.inbound
+            journey_direction=JourneyDirection.inbound,
+            flow_direction=FlowDirection.undefined
         )
         self.free_capacity_for_inbound_journey_buffer[vehicle] = free_capacity_in_teu
         return free_capacity_in_teu
 
     def get_free_capacity_for_outbound_journey(
             self, vehicle: Type[AbstractLargeScheduledVehicle],
-            flow_direction: str
+            flow_direction: FlowDirection
     ) -> float:
         """Get the free capacity for the outbound journey on a vehicle that moves according to a schedule in TEU.
         """
@@ -138,20 +140,19 @@ def get_free_capacity_for_outbound_journey(
             vehicle=vehicle,
             maximum_capacity=total_moved_capacity_for_onward_transportation_in_teu,
             container_counter=self._get_number_containers_for_outbound_journey,
-            direction=_Direction.outbound,
+            journey_direction=JourneyDirection.outbound,
             flow_direction=flow_direction
         )
         self.free_capacity_for_outbound_journey_buffer[vehicle] = free_capacity_in_teu
         return free_capacity_in_teu
 
-    # noinspection PyUnresolvedReferences
-    @staticmethod
     def _get_free_capacity_in_teu(
+            self,
             vehicle: Type[AbstractLargeScheduledVehicle],
             maximum_capacity: int,
             container_counter: Callable[[Type[AbstractLargeScheduledVehicle], ContainerLength], int],
-            direction: _Direction,
-            flow_direction: str
+            journey_direction: JourneyDirection,
+            flow_direction: FlowDirection
     ) -> float:
         loaded_20_foot_containers = container_counter(vehicle, ContainerLength.twenty_feet)
         loaded_40_foot_containers = container_counter(vehicle, ContainerLength.forty_feet)
@@ -164,7 +165,10 @@ def _get_free_capacity_in_teu(
                 - loaded_45_foot_containers * ContainerLength.get_teu_factor(ContainerLength.forty_five_feet)
                 - loaded_other_containers * ContainerLength.get_teu_factor(ContainerLength.other)
         )
+
+        # noinspection PyUnresolvedReferences
         vehicle_name = vehicle.large_scheduled_vehicle.vehicle_name
+
         assert free_capacity_in_teu >= 0, f"vehicle {vehicle} of type {vehicle.get_mode_of_transport()} with the " \
                                           f"name '{vehicle_name}' " \
                                           f"is overloaded, " \
@@ -175,12 +179,26 @@ def _get_free_capacity_in_teu(
                                           f"loaded_45_foot_containers: {loaded_45_foot_containers} and " \
                                           f"loaded_other_containers: {loaded_other_containers}"
 
+        # noinspection PyUnresolvedReferences
         arrival_time: datetime.datetime = vehicle.large_scheduled_vehicle.scheduled_arrival
 
-        #TODO use flow_direction
-        if direction == _Direction.outbound and arrival_time < ramp_up_period_end:
+        if (
+                journey_direction == JourneyDirection.outbound
+                and flow_direction == FlowDirection.transshipment_flow
+                and self.ramp_up_period_end is not None
+                and arrival_time < self.ramp_up_period_end
+        ):
+            # keep transshipment containers in the yard longer during the ramp-up period to fill the yard faster
+            # by offering less transport capacity on the outbound journey of deep sea vessels and feeders
             free_capacity_in_teu = maximum_capacity * 0.1
-        elif direction == _Direction.inbound and arrival_time >= ramp_down_period_start:
+
+        elif (
+                journey_direction == JourneyDirection.inbound
+                and self.ramp_down_period_start is not None
+                and arrival_time >= self.ramp_down_period_start
+        ):
+            # decrease number of inbound containers (any direction) during the ramp-down period
+            # by offering less transport capacity on the inbound journey (all types of vehicles, excluding trucks)
             free_capacity_in_teu = maximum_capacity * 0.1
 
         return free_capacity_in_teu

conflowgen/domain_models/repositories/schedule_repository.py

@@ -1,7 +1,9 @@
 import datetime
+import typing
 from typing import List, Type
 import logging
 
+from conflowgen.descriptive_datatypes import FlowDirection
 from conflowgen.domain_models.container import Container
 from conflowgen.domain_models.data_types.container_length import ContainerLength
 from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
@@ -18,12 +20,23 @@ def __init__(self):
     def set_transportation_buffer(self, transportation_buffer: float):
         self.large_scheduled_vehicle_repository.set_transportation_buffer(transportation_buffer)
 
+    def set_ramp_up_and_down_times(
+            self,
+            ramp_up_period_end: typing.Optional[datetime.datetime] = None,
+            ramp_down_period_start: typing.Optional[datetime.datetime] = None
+    ):
+        self.large_scheduled_vehicle_repository.set_ramp_up_and_down_times(
+            ramp_up_period_end=ramp_up_period_end,
+            ramp_down_period_start=ramp_down_period_start
+        )
+
     def get_departing_vehicles(
             self,
             start: datetime.datetime,
             end: datetime.datetime,
             vehicle_type: ModeOfTransport,
-            required_capacity: ContainerLength
+            required_capacity: ContainerLength,
+            flow_direction: FlowDirection
     ) -> List[Type[AbstractLargeScheduledVehicle]]:
         """Gets the available vehicles for the required capacity of the required type and within the time range.
         """
@@ -46,7 +59,7 @@ def get_departing_vehicles(
         vehicle: Type[AbstractLargeScheduledVehicle]
         for vehicle in vehicles:
             free_capacity_in_teu = self.large_scheduled_vehicle_repository.get_free_capacity_for_outbound_journey(
-                vehicle
+                vehicle, flow_direction
             )
             if free_capacity_in_teu >= required_capacity_in_teu:
                 vehicles_with_sufficient_capacity.append(vehicle)

conflowgen/flow_generator/allocate_space_for_containers_delivered_by_truck_service.py

@@ -3,6 +3,7 @@
 from typing import Dict, Type, List
 
 from conflowgen.application.repositories.random_seed_store_repository import get_initialised_random_object
+from conflowgen.descriptive_datatypes import FlowDirection
 from conflowgen.domain_models.container import Container
 from conflowgen.domain_models.distribution_repositories.mode_of_transport_distribution_repository import \
     ModeOfTransportDistributionRepository
@@ -115,7 +116,7 @@ def allocate(self) -> None:
                     continue  # try again with another vehicle type (refers to while loop)
 
                 free_capacity_of_vehicle = self.large_scheduled_vehicle_repository.\
-                    get_free_capacity_for_outbound_journey(vehicle, flow_direction="export")
+                    get_free_capacity_for_outbound_journey(vehicle, flow_direction=FlowDirection.export_flow)
 
                 if free_capacity_of_vehicle <= self.ignored_capacity:
 
@@ -175,7 +176,8 @@ def _pick_vehicle(
         # Make it more likely that a container ends up on a large vessel than on a smaller one
         vehicle: Type[AbstractLargeScheduledVehicle]
         vehicle_distribution: Dict[Type[AbstractLargeScheduledVehicle], float] = {
-            vehicle: self.large_scheduled_vehicle_repository.get_free_capacity_for_outbound_journey(vehicle)
+            vehicle: self.large_scheduled_vehicle_repository.get_free_capacity_for_outbound_journey(
+                vehicle, FlowDirection.export_flow)
             for vehicle in vehicles_of_type
         }
         all_free_capacities = list(vehicle_distribution.values())

conflowgen/flow_generator/large_scheduled_vehicle_for_onward_transportation_manager.py

@@ -44,12 +44,17 @@ def __init__(self):
     def reload_properties(
             self,
             transportation_buffer: float,
-            ramp_up_period_end: typing.Optional[datetime.datetime],
-            ramp_down_period_start: typing.Optional[datetime.datetime],
+            ramp_up_period_end: typing.Optional[datetime.datetime] = None,
+            ramp_down_period_start: typing.Optional[datetime.datetime] = None,
     ):
         assert -1 < transportation_buffer
         self.schedule_repository.set_transportation_buffer(transportation_buffer)
 
+        self.schedule_repository.set_ramp_up_and_down_times(
+            ramp_up_period_end=ramp_up_period_end,
+            ramp_down_period_start=ramp_down_period_start
+        )
+
         self.logger.debug(f"Using transportation buffer of {transportation_buffer} when choosing the departing "
                           f"vehicles that adhere a schedule.")
 
@@ -129,7 +134,8 @@ def choose_departing_vehicle_for_containers(self) -> None:
                 start=(container_arrival + datetime.timedelta(hours=minimum_dwell_time_in_hours)),
                 end=(container_arrival + datetime.timedelta(hours=maximum_dwell_time_in_hours)),
                 vehicle_type=initial_departing_vehicle_type,
-                required_capacity=container.length
+                required_capacity=container.length,
+                flow_direction=container.flow_direction
             )
 
             if len(available_vehicles) > 0:
@@ -197,7 +203,7 @@ def _draw_vehicle(
             return available_vehicles[0]
 
         vehicles_and_their_respective_free_capacity = {}
-        
+
         for vehicle in available_vehicles:
 
             free_capacity = self.large_scheduled_vehicle_repository.get_free_capacity_for_outbound_journey(
@@ -321,7 +327,8 @@ def _find_alternative_mode_of_transportation(
                 start=(container_arrival + datetime.timedelta(hours=minimum_dwell_time_in_hours)),
                 end=(container_arrival + datetime.timedelta(hours=maximum_dwell_time_in_hours)),
                 vehicle_type=vehicle_type,
-                required_capacity=container.length
+                required_capacity=container.length,
+                flow_direction=container.flow_direction
             )
             if len(available_vehicles) > 0:  # There is a vehicle of a new type available, so it is picked
                 vehicle = self._pick_vehicle_for_container(available_vehicles, container)

conflowgen/tests/domain_models/reposistories/test_large_scheduled_vehicle_repository.py

@@ -1,6 +1,7 @@
 import datetime
 import unittest
 
+from conflowgen.descriptive_datatypes import FlowDirection
 from conflowgen.domain_models.container import Container
 from conflowgen.domain_models.data_types.container_length import ContainerLength
 from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
@@ -58,7 +59,9 @@ def test_free_capacity_for_one_teu(self):
             picked_up_by_large_scheduled_vehicle=self.train_lsv,
         )
 
-        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(self.train)
+        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(
+            self.train, FlowDirection.undefined
+        )
         self.assertEqual(free_capacity_in_teu, 2)
 
     def test_free_capacity_for_one_ffe(self):
@@ -72,7 +75,9 @@ def test_free_capacity_for_one_ffe(self):
             picked_up_by_large_scheduled_vehicle=self.train_lsv,
         )
 
-        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(self.train)
+        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(
+            self.train, FlowDirection.undefined
+        )
         self.assertEqual(free_capacity_in_teu, 1)
 
     def test_free_capacity_for_45_foot_container(self):
@@ -86,7 +91,9 @@ def test_free_capacity_for_45_foot_container(self):
             picked_up_by_large_scheduled_vehicle=self.train_lsv,
         )
 
-        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(self.train)
+        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(
+            self.train, FlowDirection.undefined
+        )
         self.assertEqual(free_capacity_in_teu, 0.75)
 
     def test_free_capacity_for_other_container(self):
@@ -100,5 +107,7 @@ def test_free_capacity_for_other_container(self):
             picked_up_by_large_scheduled_vehicle=self.train_lsv,
         )
 
-        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(self.train)
+        free_capacity_in_teu = self.lsv_repository.get_free_capacity_for_outbound_journey(
+            self.train, FlowDirection.undefined
+        )
         self.assertEqual(free_capacity_in_teu, 0.5)