Add truck gate throughput preview (#182)

Co-authored-by: Luc Edes <[email protected]>

.github/workflows/docs.yaml

@@ -29,9 +29,12 @@ jobs:
         sudo apt-get update && sudo apt-get upgrade && sudo apt-get install pandoc
 
     - uses: actions/checkout@v3
-      with:
-        lfs: 'true'
-    - run: git lfs pull
+    - run: |
+        curl -LJO "https://media.tuhh.de/mls/software/conflowgen/docs/data/prepared_dbs/demo_poc.sqlite"
+        curl -LJO "https://media.tuhh.de/mls/software/conflowgen/docs/data/prepared_dbs/demo_deham_cta.sqlite"
+        mkdir -p docs/notebooks/data/prepared_dbs
+        mv demo_poc.sqlite docs/notebooks/data/prepared_dbs/
+        mv demo_deham_cta.sqlite docs/notebooks/data/prepared_dbs/
 
     - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v4

.github/workflows/unittests.yaml

@@ -20,11 +20,10 @@ jobs:
       uses: fkirc/skip-duplicate-actions@v5
 
     - uses: actions/checkout@v2
-      with:
-        lfs: 'true'
     - run: |
-        git lfs fetch -p -I '**/notebooks/data/prepared_dbs/demo_poc.sqlite'
-        git lfs checkout
+        curl -LJO "https://media.tuhh.de/mls/software/conflowgen/docs/data/prepared_dbs/demo_poc.sqlite"
+        mkdir -p docs/notebooks/data/prepared_dbs
+        mv demo_poc.sqlite docs/notebooks/data/prepared_dbs/
 
     - name: Set up Python 3.10
       uses: actions/setup-python@v4

conflowgen/__init__.py

@@ -25,6 +25,8 @@
     VehicleCapacityUtilizationOnOutboundJourneyPreviewReport
 from conflowgen.previews.modal_split_preview import ModalSplitPreview
 from conflowgen.previews.modal_split_preview_report import ModalSplitPreviewReport
+from conflowgen.previews.truck_gate_throughput_preview import TruckGateThroughputPreview
+from conflowgen.previews.truck_gate_throughput_preview_report import TruckGateThroughputPreviewReport
 
 # Analyses and their reports
 from conflowgen.analyses.inbound_and_outbound_vehicle_capacity_analysis import \

conflowgen/domain_models/distribution_repositories/container_length_distribution_repository.py

@@ -53,3 +53,15 @@ def set_distribution(cls, container_lengths: Dict[ContainerLength, float]):
                 container_length=container_length,
                 fraction=fraction
             ).save()
+
+    @classmethod
+    def get_teu_factor(cls) -> float:
+        """
+        Calculates and returns the TEU factor based on the container length distribution.
+        """
+        # Loop through container lengths and calculate weighted average of all container lengths
+        container_length_weighted_average = 0.0
+        container_length_distribution = cls.get_distribution()
+        for container_length, fraction in container_length_distribution.items():
+            container_length_weighted_average += ContainerLength.get_factor(container_length) * fraction
+        return container_length_weighted_average

conflowgen/previews/__init__.py

@@ -3,6 +3,7 @@
 from .inbound_and_outbound_vehicle_capacity_preview_report import InboundAndOutboundVehicleCapacityPreviewReport
 from .container_flow_by_vehicle_type_preview_report import ContainerFlowByVehicleTypePreviewReport
 from .modal_split_preview_report import ModalSplitPreviewReport
+from .truck_gate_throughput_preview_report import TruckGateThroughputPreviewReport
 from .vehicle_capacity_exceeded_preview_report import VehicleCapacityUtilizationOnOutboundJourneyPreviewReport
 from ..reporting import AbstractReport
 from ..reporting.auto_reporter import AutoReporter
@@ -14,6 +15,7 @@
     VehicleCapacityUtilizationOnOutboundJourneyPreviewReport,
     ContainerFlowByVehicleTypePreviewReport,
     ModalSplitPreviewReport,
+    TruckGateThroughputPreviewReport
 ]
 
 

conflowgen/previews/truck_gate_throughput_preview.py

@@ -0,0 +1,104 @@
+import math
+import typing
+from abc import ABC
+from builtins import bool
+from datetime import datetime
+from collections import namedtuple
+
+from conflowgen.previews.inbound_and_outbound_vehicle_capacity_preview import \
+    InboundAndOutboundVehicleCapacityPreview
+from conflowgen.api.truck_arrival_distribution_manager import TruckArrivalDistributionManager
+from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
+from conflowgen.domain_models.distribution_repositories.container_length_distribution_repository import \
+    ContainerLengthDistributionRepository
+from conflowgen.domain_models.distribution_validators import validate_distribution_with_one_dependent_variable
+from conflowgen.previews.abstract_preview import AbstractPreview
+
+
+class TruckGateThroughputPreview(AbstractPreview, ABC):
+    """
+    This preview shows the distribution of truck traffic throughout a given week
+
+    The preview returns a data structure that can be used for generating reports (e.g., in text or as a figure). The
+    preview is intended to provide an estimate of the truck gate throughput for the given inputs. It does not
+    consider all factors that may impact the actual truck gate throughput.
+    """
+
+    def __init__(self, start_date: datetime.date, end_date: datetime.date, transportation_buffer: float):
+        super().__init__(start_date, end_date, transportation_buffer)
+        self.inbound_and_outbound_vehicle_capacity_preview = (
+            InboundAndOutboundVehicleCapacityPreview(
+                self.start_date,
+                self.end_date,
+                self.transportation_buffer,
+            )
+        )
+
+    def hypothesize_with_mode_of_transport_distribution(
+            self,
+            mode_of_transport_distribution: typing.Dict[ModeOfTransport, typing.Dict[ModeOfTransport, float]]
+    ):
+        validate_distribution_with_one_dependent_variable(
+            mode_of_transport_distribution, ModeOfTransport, ModeOfTransport, values_are_frequencies=True
+        )
+        self.inbound_and_outbound_vehicle_capacity_preview.hypothesize_with_mode_of_transport_distribution(
+            mode_of_transport_distribution)
+
+    def _get_total_trucks(self) -> typing.Tuple[int, int]:
+        # Calculate the truck capacity for export containers using the inbound container capacities
+        inbound_used_and_maximum_capacity = self.inbound_and_outbound_vehicle_capacity_preview. \
+            get_inbound_capacity_of_vehicles()
+        outbound_used_and_maximum_capacity = self.inbound_and_outbound_vehicle_capacity_preview.\
+            get_outbound_capacity_of_vehicles()
+
+        # Get the total truck capacity in TEU
+        total_inbound_truck_capacity_in_teu = inbound_used_and_maximum_capacity.teu[ModeOfTransport.truck]
+        total_outbound_truck_capacity_in_teu = outbound_used_and_maximum_capacity.used.teu[ModeOfTransport.truck]
+
+        # Calculate the TEU factor using the container length distribution
+        teu_factor = ContainerLengthDistributionRepository.get_teu_factor()
+
+        # Calculate the total number of containers transported by truck
+        total_inbound_containers_transported_by_truck = \
+            int(math.ceil(total_inbound_truck_capacity_in_teu / teu_factor))
+        total_outbound_containers_transported_by_truck = \
+            int(math.ceil(total_outbound_truck_capacity_in_teu / teu_factor))
+
+        total_containers_transported_by_truck_datatype = \
+            namedtuple('total_containers_transported_by_truck_datatype', 'inbound outbound')
+        total_containers_transported_by_truck = \
+            total_containers_transported_by_truck_datatype(total_inbound_containers_transported_by_truck,
+                                                           total_outbound_containers_transported_by_truck)
+
+        return total_containers_transported_by_truck
+
+    def _get_number_of_trucks_per_week(self) -> typing.Tuple[float, float]:
+        # Calculate average number of trucks per week
+        num_weeks = (self.end_date - self.start_date).days / 7
+        total_trucks = self._get_total_trucks()
+        inbound_trucks_per_week = total_trucks.inbound / num_weeks
+        outbound_trucks_per_week = total_trucks.outbound / num_weeks
+
+        total_weekly_trucks_datatype = namedtuple('total_weekly_trucks_datatype', 'inbound outbound')
+        total_weekly_trucks = total_weekly_trucks_datatype(inbound_trucks_per_week, outbound_trucks_per_week)
+
+        return total_weekly_trucks
+
+    def get_weekly_truck_arrivals(self, inbound: bool = True, outbound: bool = True) -> typing.Dict[int, int]:
+
+        assert inbound or outbound, "At least one of inbound or outbound must be True"
+
+        # Get truck arrival distribution
+        truck_arrival_probability_distribution = TruckArrivalDistributionManager().\
+            get_truck_arrival_distribution()
+
+        truck_arrival_integer_distribution = {}
+        weekly_trucks = self._get_number_of_trucks_per_week()
+        for time, probability in truck_arrival_probability_distribution.items():
+            truck_arrival_integer_distribution[time] = 0
+            if inbound:
+                truck_arrival_integer_distribution[time] += int(round(probability * weekly_trucks.inbound))
+            if outbound:
+                truck_arrival_integer_distribution[time] += int(round(probability * weekly_trucks.outbound))
+
+        return truck_arrival_integer_distribution

conflowgen/previews/truck_gate_throughput_preview_report.py

@@ -0,0 +1,132 @@
+from abc import ABC
+import typing
+import pandas as pd
+
+import matplotlib
+from matplotlib import pyplot as plt
+
+from conflowgen.domain_models.data_types.mode_of_transport import ModeOfTransport
+from conflowgen.previews.truck_gate_throughput_preview import TruckGateThroughputPreview
+from conflowgen.reporting import AbstractReportWithMatplotlib
+
+
+class TruckGateThroughputPreviewReport(AbstractReportWithMatplotlib, ABC):
+    """
+    This preview report takes the data structure as generated by
+    :class:`.TruckGateThroughputPreview`
+    and creates a comprehensible representation for the user, either as text or as a graph.
+    The visual and table are expected to approximately look like in the
+    `example TruckGateThroughputPreviewReport <notebooks/previews.ipynb#Truck-Gate-Throughput-Preview-Report>`_.
+    """
+
+    report_description = """This report previews the average truck gate throughput throughout the week as defined by
+    schedules and input distributions."""
+
+    def __init__(self):
+        super().__init__()
+        self.preview = TruckGateThroughputPreview(
+            start_date=self.start_date,
+            end_date=self.end_date,
+            transportation_buffer=self.transportation_buffer
+        )
+
+    def hypothesize_with_mode_of_transport_distribution(
+            self,
+            mode_of_transport_distribution: typing.Dict[ModeOfTransport, typing.Dict[ModeOfTransport, float]]
+    ):
+        self.preview.hypothesize_with_mode_of_transport_distribution(mode_of_transport_distribution)
+
+    def _get_updated_preview(self) -> TruckGateThroughputPreview:
+        assert self.start_date is not None
+        assert self.end_date is not None
+        assert self.transportation_buffer is not None
+        self.preview.update(
+            start_date=self.start_date,
+            end_date=self.end_date,
+            transportation_buffer=self.transportation_buffer
+        )
+        return self.preview
+
+    def get_report_as_text(self, inbound: bool = True, outbound: bool = True, **kwargs) -> str:
+        truck_distribution = self.preview.get_weekly_truck_arrivals(inbound, outbound)
+        data = [
+            {'minimum': float('inf'), 'maximum': 0, 'average': 0.0, 'sum': 0}
+            for _ in range(8)  # Monday to Sunday plus week total
+        ]
+
+        fewest_trucks_in_a_day = float('inf')
+        fewest_trucks_day = ''
+        most_trucks_in_a_day = 0
+        most_trucks_day = ''
+        average_trucks_in_a_day = 0.0
+
+        count = 0
+        # Find min, max, and average for each day of the week
+        for time in sorted(truck_distribution):
+            day = time // 24
+            if day == 0:
+                count += 1  # Count the number of data points in a single day
+            data[day]['minimum'] = min(data[day]['minimum'], truck_distribution[time])
+            data[day]['maximum'] = max(data[day]['maximum'], truck_distribution[time])
+            data[day]['sum'] += truck_distribution[time]
+
+        # Calculate average
+        for day in range(7):
+            data[day]['average'] = data[day]['sum'] / count
+            data[7]['minimum'] = min(data[7]['minimum'], data[day]['minimum'])
+            data[7]['maximum'] = max(data[7]['maximum'], data[day]['maximum'])
+            data[7]['sum'] += data[day]['sum']
+            if data[day]['sum'] < fewest_trucks_in_a_day:
+                fewest_trucks_in_a_day = data[day]['sum']
+                fewest_trucks_day = self.days_of_the_week[day]
+            if data[day]['sum'] > most_trucks_in_a_day:
+                most_trucks_in_a_day = data[day]['sum']
+                most_trucks_day = self.days_of_the_week[day]
+            most_trucks_in_a_day = max(most_trucks_in_a_day, data[day]['sum'])
+            average_trucks_in_a_day += data[day]['sum']
+
+        data[7]['average'] = data[7]['sum'] / (count * 7)
+        average_trucks_in_a_day /= 7
+
+        # Create a table with pandas for hourly view
+        df = pd.DataFrame(data, index=self.days_of_the_week + ['Total'])
+        df = df.round()
+        df = df.astype(int)
+
+        df = df.rename_axis('Day of the week')
+        df = df.rename(columns={
+            'minimum': 'Minimum (trucks/h)', 'maximum': 'Maximum (trucks/h)', 'average': 'Average (trucks/h)',
+            'sum': 'Sum (trucks/24h)'})
+
+        table_string = "Hourly view:\n" + df.to_string() + "\n"
+        table_string += \
+            "Fewest trucks in a day: " + str(int(fewest_trucks_in_a_day)) + " on " + fewest_trucks_day + "\n"
+        table_string += \
+            "Most trucks in a day: " + str(int(most_trucks_in_a_day)) + " on " + most_trucks_day + "\n"
+        table_string += \
+            "Average trucks per day: " + str(int(average_trucks_in_a_day))
+
+        return table_string
+
+    def get_report_as_graph(self, inbound: bool = True, outbound: bool = True, **kwargs) -> matplotlib.axes.Axes:
+        # Retrieve the truck distribution
+        truck_distribution = self.preview.get_weekly_truck_arrivals(inbound, outbound)
+
+        # Plot the truck arrival distribution
+        hour_in_week, value = zip(*list(sorted(truck_distribution.items())))
+        weekday_in_week = [x / 24 + 1 for x in hour_in_week]
+
+        fig, ax = plt.subplots(figsize=(15, 3))
+        plt.plot(weekday_in_week, value)
+        plt.xlim([1, 7])  # plot from Monday to Sunday
+        ax.xaxis.grid(True, which="minor", color="lightgray")  # every hour
+        ax.xaxis.grid(True, which="major", color="k")  # every day
+        ax.xaxis.set_minor_locator(matplotlib.ticker.MultipleLocator(1 / 24))  # every hour
+
+        plt.title("Expected truck arrival pattern")
+        ax.set_xticks(list(range(1, 8)))  # every day
+        ax.set_xticklabels(self.days_of_the_week)
+        plt.xlabel("Week day")
+        plt.ylabel("Number of trucks")
+
+        return ax

conflowgen/reporting/__init__.py

@@ -36,6 +36,9 @@ class AbstractReport(abc.ABC):
         ModeOfTransport.truck
     ]
 
+    #: The days of the week
+    days_of_the_week = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
+
     @property
     @abc.abstractmethod
     def report_description(self) -> str:

conflowgen/reporting/auto_reporter.py

@@ -61,6 +61,7 @@ def present_reports(self, reports: typing.Iterable[typing.Type[AbstractReport]])
                     )
                 else:
                     report_as_text = report_instance.get_report_as_text()
+                assert report_as_text, "Report should not be empty"
                 self.output.display_verbatim(report_as_text)
             if self.as_graph:
                 try:

conflowgen/tests/domain_models/distribution_repositories/test_container_length_distribution_repository.py

@@ -108,3 +108,53 @@ def test_set_container_lengths_which_do_not_add_up_to_one(self) -> None:
                     ContainerLength.other: 0
                 }
             )
+
+    def test_get_teu_factor_all_twenty_feet(self):
+        ContainerLengthDistributionRepository.set_distribution({
+            ContainerLength.twenty_feet: 1,
+            ContainerLength.forty_feet: 0,
+            ContainerLength.forty_five_feet: 0,
+            ContainerLength.other: 0
+        })
+        teu_factor = ContainerLengthDistributionRepository.get_teu_factor()
+        self.assertEqual(teu_factor, 1, "TEU factor should be 1 when all containers are 20 feet.")
+
+    def test_get_teu_factor_when_half_of_containers_are_forty_feet(self):
+        ContainerLengthDistributionRepository.set_distribution(
+            {
+                ContainerLength.twenty_feet: 0.5,
+                ContainerLength.forty_feet: 0.5,
+                ContainerLength.forty_five_feet: 0,
+                ContainerLength.other: 0
+            }
+        )
+        teu_factor = ContainerLengthDistributionRepository.get_teu_factor()
+        self.assertEqual(
+            teu_factor, 1.5,
+            "TEU factor should be 1.5 when half of the containers are 20 feet and half are 40 feet.")
+
+    def test_get_teu_factor_all_forty_feet(self) -> None:
+        ContainerLengthDistributionRepository.set_distribution(
+            {
+                ContainerLength.twenty_feet: 0,
+                ContainerLength.forty_feet: 1,
+                ContainerLength.forty_five_feet: 0,
+                ContainerLength.other: 0
+            }
+        )
+        self.assertEqual(
+            ContainerLengthDistributionRepository.get_teu_factor(), 2,
+            "TEU factor should be 2 when all containers are 40 feet.")
+
+    def test_get_teu_factor_all_forty_five_feet(self) -> None:
+        ContainerLengthDistributionRepository.set_distribution(
+            {
+                ContainerLength.twenty_feet: 0,
+                ContainerLength.forty_feet: 0,
+                ContainerLength.forty_five_feet: 1,
+                ContainerLength.other: 0
+            }
+        )
+        self.assertEqual(
+            ContainerLengthDistributionRepository.get_teu_factor(), 2.25,
+            "TEU factor should be 2.25 when all containers are 45 feet.")