Algorithm modification for transshipment operations

conflowgen/__init__.py

@@ -83,7 +83,7 @@
 from conflowgen.domain_models.data_types.storage_requirement import StorageRequirement
 
 # List of functions
-from conflowgen.logging.logging import setup_logger
+from conflowgen.log.log import setup_logger
 from conflowgen.analyses import run_all_analyses
 from conflowgen.previews import run_all_previews
 

conflowgen/analyses/container_dwell_time_analysis_report.py

@@ -33,7 +33,7 @@ def __init__(self):
 
     def get_report_as_text(self, **kwargs) -> str:
         """
-        The report as a text is represented as a table suitable for logging.
+        The report as a text is represented as a table suitable for log.
         It uses a human-readable formatting style.
         For the exact interpretation of the parameter, check
         :meth:`.ContainerDwellTimeAnalysis.get_container_dwell_times`.

conflowgen/analyses/container_flow_adjustment_by_vehicle_type_analysis_summary_report.py

@@ -35,7 +35,7 @@ def get_report_as_text(
             self, **kwargs
     ) -> str:
         """
-        The report as a text is represented as a table suitable for logging.
+        The report as a text is represented as a table suitable for log.
         It uses a human-readable formatting style.
 
         Keyword Args:

conflowgen/analyses/container_flow_vehicle_type_adjustment_per_vehicle_analysis_report.py

@@ -49,7 +49,7 @@ def __init__(self):
 
     def get_report_as_text(self, **kwargs) -> str:
         """
-        The report as a text is represented as a table suitable for logging. It uses a human-readable formatting style.
+        The report as a text is represented as a table suitable for log. It uses a human-readable formatting style.
 
         Keyword Args:
             initial_vehicle_type (:obj:`typing.Any`): Either ``"all"``, a single vehicle of type

conflowgen/analyses/inbound_to_outbound_vehicle_capacity_utilization_analysis_report.py

@@ -52,7 +52,7 @@ def __init__(self):
 
     def get_report_as_text(self, **kwargs) -> str:
         """
-        The report as a text is represented as a table suitable for logging. It uses a human-readable formatting style.
+        The report as a text is represented as a table suitable for log. It uses a human-readable formatting style.
 
         Keyword Args:
             vehicle_type (:py:obj:`Any`): Either ``"scheduled vehicles"``, a single vehicle of type

conflowgen/analyses/modal_split_analysis_report.py

@@ -30,7 +30,7 @@ def get_report_as_text(
             self, **kwargs
     ) -> str:
         """
-        The report as a text is represented as a table suitable for logging. It uses a human-readable formatting style.
+        The report as a text is represented as a table suitable for log. It uses a human-readable formatting style.
 
         Keyword Args:
             start_date (datetime.datetime):

conflowgen/analyses/yard_capacity_analysis_report.py

@@ -40,7 +40,7 @@ def __init__(self):
 
     def get_report_as_text(self, **kwargs) -> str:
         """
-        The report as a text is represented as a table suitable for logging. It uses a human-readable formatting style.
+        The report as a text is represented as a table suitable for log. It uses a human-readable formatting style.
 
         Keyword Args:
             storage_requirement: Either a single storage requirement of type :class:`.StorageRequirement` or a whole

conflowgen/api/container_flow_generation_manager.py

@@ -27,7 +27,9 @@ def set_properties(
             start_date: datetime.date,
             end_date: datetime.date,
             name: typing.Optional[str] = None,
-            transportation_buffer: typing.Optional[float] = None
+            transportation_buffer: typing.Optional[float] = None,
+            ramp_up_period: typing.Optional[datetime.timedelta] = None,
+            ramp_down_period: typing.Optional[datetime.timedelta] = None,
     ) -> None:
         """
         Args:
@@ -38,7 +40,13 @@ def set_properties(
             name: The name of the generated synthetic container flow which helps to distinguish different scenarios.
             transportation_buffer: Determines how many percent more of the inbound journey capacity is used at most to
                 transport containers on the outbound journey.
+            ramp_up_period: The period at the beginning during which yard occupancy gradually increases.
+                During the ramp-up period, the share of transshipment containers on outbound journey is reduced.
+            ramp_down_period: The period at the end during which operations gradually fade out.
+                During the ramp-down period, inbound container volumes are scaled down, reducing the number of new
+                import, export, and transshipment containers in the yard during this period..
         """
+
         properties = self.container_flow_generation_properties_repository.get_container_flow_generation_properties()
 
         if name is not None:
@@ -47,6 +55,16 @@ def set_properties(
         properties.start_date = start_date
         properties.end_date = end_date
 
+        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
 
@@ -67,6 +85,8 @@ def get_properties(self) -> typing.Dict[str, typing.Union[str, datetime.date, fl
             'start_date': properties.start_date,
             'end_date': properties.end_date,
             'transportation_buffer': properties.transportation_buffer,
+            'ramp_up_period': properties.ramp_up_period,
+            'ramp_down_period': properties.ramp_down_period,
         }
 
     def container_flow_data_exists(self) -> bool:

conflowgen/application/models/container_flow_generation_properties.py

@@ -27,6 +27,16 @@ class ContainerFlowGenerationProperties(BaseModel):
         help_text="The last day of the generated container flow"
     )
 
+    ramp_up_period = FloatField(
+        default=0,
+        help_text="Number of days for the ramp-up period"
+    )
+
+    ramp_down_period = FloatField(
+        default=0,
+        help_text="Number of days for the ramp-down period"
+    )
+
     generated_at = DateTimeField(
         default=lambda: datetime.datetime.now().replace(microsecond=0),
         help_text="The date the these properties have been created"

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
 
 
@@ -116,6 +117,19 @@ class Container(BaseModel):
     def occupied_teu(self) -> float:
         return CONTAINER_LENGTH_TO_OCCUPIED_TEU[self.length]
 
+    @property
+    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 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 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:
 
         if self.cached_arrival_time is not None:

conflowgen/domain_models/repositories/large_scheduled_vehicle_repository.py

@@ -1,18 +1,29 @@
+import datetime
+import enum
 import logging
+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 JourneyDirection(enum.Enum):
+    inbound = "inbound"
+    outbound = "outbound"
+
+
 class LargeScheduledVehicleRepository:
 
     ignored_capacity = ContainerLength.get_teu_factor(ContainerLength.other)
 
     def __init__(self):
         self.transportation_buffer = None
+        self.ramp_up_period_end = None
+        self.ramp_down_period_start = None
         self.free_capacity_for_outbound_journey_buffer: Dict[Type[AbstractLargeScheduledVehicle], float] = {}
         self.free_capacity_for_inbound_journey_buffer: Dict[Type[AbstractLargeScheduledVehicle], float] = {}
         self.logger = logging.getLogger("conflowgen")
@@ -21,6 +32,14 @@ def set_transportation_buffer(self, transportation_buffer: float):
         assert -1 < transportation_buffer
         self.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
+    ) -> None:
+        self.ramp_up_period_end = ramp_up_period_end
+        self.ramp_down_period_start = ramp_down_period_start
+
     def reset_cache(self):
         self.free_capacity_for_outbound_journey_buffer = {}
         self.free_capacity_for_inbound_journey_buffer = {}
@@ -87,12 +106,17 @@ def get_free_capacity_for_inbound_journey(self, vehicle: Type[AbstractLargeSched
         free_capacity_in_teu = self._get_free_capacity_in_teu(
             vehicle=vehicle,
             maximum_capacity=total_moved_capacity_for_inbound_transportation_in_teu,
-            container_counter=self._get_number_containers_for_inbound_journey
+            container_counter=self._get_number_containers_for_inbound_journey,
+            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]) -> float:
+    def get_free_capacity_for_outbound_journey(
+            self, vehicle: Type[AbstractLargeScheduledVehicle],
+            flow_direction: FlowDirection
+    ) -> float:
         """Get the free capacity for the outbound journey on a vehicle that moves according to a schedule in TEU.
         """
         assert self.transportation_buffer is not None, "First set the value!"
@@ -115,17 +139,20 @@ def get_free_capacity_for_outbound_journey(self, vehicle: Type[AbstractLargeSche
         free_capacity_in_teu = self._get_free_capacity_in_teu(
             vehicle=vehicle,
             maximum_capacity=total_moved_capacity_for_onward_transportation_in_teu,
-            container_counter=self._get_number_containers_for_outbound_journey
+            container_counter=self._get_number_containers_for_outbound_journey,
+            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]
+            container_counter: Callable[[Type[AbstractLargeScheduledVehicle], ContainerLength], int],
+            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)
@@ -138,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, " \
@@ -148,6 +178,29 @@ def _get_free_capacity_in_teu(
                                           f"loaded_40_foot_containers: {loaded_40_foot_containers}, " \
                                           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
+
+        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 (
+                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
 
     @classmethod

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)