from __future__ import annotations
import datetime
import typing
import matplotlib.axes
import pandas as pd
from matplotlib import pyplot as plt
from matplotlib.ticker import FuncFormatter
from conflowgen.analyses.container_flow_vehicle_type_adjustment_per_vehicle_analysis import \
ContainerFlowVehicleTypeAdjustmentPerVehicleAnalysis
from conflowgen.analyses.inbound_to_outbound_vehicle_capacity_utilization_analysis import VehicleIdentifier
from conflowgen.reporting import AbstractReportWithMatplotlib
from conflowgen.reporting.no_data_plot import no_data_graph
[docs]
class ContainerFlowVehicleTypeAdjustmentPerVehicleAnalysisReport(AbstractReportWithMatplotlib):
"""
This analysis report takes the data structure as generated by
:class:`.ContainerFlowVehicleTypeAdjustmentPerVehicleAnalysis` 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 ContainerFlowVehicleTypeAdjustmentPerVehicleAnalysisReport \
<notebooks/analyses.ipynb#Container-Flow-Vehicle-Type-Adjustment-Per-Vehicle-Analysis-Report>`_.
"""
report_description = """
Analyze how many of the containers loaded by a vehicle on its outbound journey have been originally destined for
another vehicle type.
Generally, it expected that such an adjustment should only occur occasionally in the beginning.
Only towards the end, difficulties to find the desired vehicle type are expected, e.g., if a feeder delivers a
container to be transshipped to a deep sea vessel but the last deep sea vessel has already departed, then the
container must be placed on a vehicle of another type.
If no vehicle is available, a corresponding truck is created to pick up the container from the terminal.
A large fraction of re-assigned containers before the end are considered noteworthy but depending on the input data
it might be acceptable.
"""
plot_title = "Re-assignment of containers to a new vehicle type"
def __init__(self):
super().__init__()
self._df = None
self.analysis = ContainerFlowVehicleTypeAdjustmentPerVehicleAnalysis(
transportation_buffer=self.transportation_buffer
)
[docs]
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.
Keyword Args:
initial_vehicle_type (:obj:`typing.Any`): Either ``"all"``, a single vehicle of type
:class:`.ModeOfTransport` or a whole collection of vehicle types, e.g., passed as a :class:`list` or
:class:`set`.
Only the vehicles that correspond to the provided vehicle type(s) are considered in the analysis.
adjusted_vehicle_type (:obj:`typing.Any`): Either ``"all"``, a single vehicle of type
:class:`.ModeOfTransport` or a whole collection of vehicle types, e.g., passed as a :class:`list` or
:class:`set`.
Only the vehicles that correspond to the provided vehicle type(s) are considered in the analysis.
Defaults to {:class:.`ModeOfTransport.deep_sea_vessel`, :class:.`ModeOfTransport.feeder`,
:class:.`ModeOfTransport.barge`, :class:.`ModeOfTransport.train`}
start_date (datetime.datetime):
Only include containers that arrive after the given start time.
end_date (datetime.datetime):
Only include containers that depart before the given end time.
Returns:
The report in text format spanning over several lines.
"""
(
initial_vehicle_type, adjusted_vehicle_type, start_date, end_date, fraction_per_vehicle
) = self._get_analysis(kwargs)
assert len(kwargs) == 0, f"Keyword(s) {list(kwargs.keys())} have not been processed."
report = "\n"
report += self._get_filter_condition(
adjusted_vehicle_type=adjusted_vehicle_type,
end_date=end_date,
initial_vehicle_type=initial_vehicle_type,
start_date=start_date
)
report += "vehicle identifier "
report += "fraction of adjusted containers (in containers)"
report += "\n"
for vehicle_identifier, fraction in fraction_per_vehicle.items():
vehicle_name = self._vehicle_identifier_to_text(vehicle_identifier)
report += f"{vehicle_name:<50} " # align this with cls.maximum_length_for_readable_name!
report += f"{fraction:<5.1%}".rstrip()
report += "\n"
if len(fraction_per_vehicle) == 0:
report += "--no vehicles exist--\n"
else:
report += "(rounding errors might exist)\n"
return report
def _get_filter_condition(
self,
adjusted_vehicle_type: typing.Any,
initial_vehicle_type: typing.Any,
start_date: datetime.datetime | None,
end_date: datetime.datetime | None,
) -> str:
initial_vehicle_type = self._get_vehicle_representation(initial_vehicle_type)
adjusted_vehicle_type = self._get_vehicle_representation(adjusted_vehicle_type)
start_date = self._get_datetime_representation(start_date)
end_date = self._get_datetime_representation(end_date)
filter_condition_text = ""
filter_condition_text += "initial vehicle type = " + initial_vehicle_type + "\n"
filter_condition_text += "adjusted vehicle type = " + adjusted_vehicle_type + "\n"
filter_condition_text += "start date = " + start_date + "\n"
filter_condition_text += "end date = " + end_date + "\n"
return filter_condition_text
def _get_analysis(
self, kwargs: dict
) -> typing.Tuple[typing.Any, typing.Any, datetime.datetime, datetime.datetime,
typing.Dict[VehicleIdentifier, float]]:
initial_vehicle_type = kwargs.pop("initial_vehicle_type", "scheduled vehicles")
adjusted_vehicle_type = kwargs.pop("adjusted_vehicle_type", "scheduled vehicles")
start_date = kwargs.pop("start_date", None)
end_date = kwargs.pop("end_date", None)
fraction_per_vehicle = self.analysis.get_vehicle_type_adjustments_per_vehicle(
initial_vehicle_type=initial_vehicle_type,
adjusted_vehicle_type=adjusted_vehicle_type,
start_date=start_date,
end_date=end_date
)
return initial_vehicle_type, adjusted_vehicle_type, start_date, end_date, fraction_per_vehicle
[docs]
def get_report_as_graph(self, **kwargs) -> matplotlib.axes.Axes:
"""
The report as a graph is represented as a scatter plot using pandas.
Keyword Args:
initial_vehicle_type (:obj:`typing.Any`): Either ``"all"``, a single vehicle of type
:class:`.ModeOfTransport` or a whole collection of vehicle types, e.g., passed as a :class:`list` or
:class:`set`.
Only the vehicles that correspond to the provided vehicle type(s) are considered in the analysis.
adjusted_vehicle_type (:obj:`typing.Any`): Either ``"all"``, a single vehicle of type
:class:`.ModeOfTransport` or a whole collection of vehicle types, e.g., passed as a :class:`list` or
:class:`set`.
Only the vehicles that correspond to the provided vehicle type(s) are considered in the analysis.
start_date (datetime.datetime):
Only include containers that arrive after the given start time.
end_date (datetime.datetime):
Only include containers that depart before the given end time.
Returns:
The matplotlib figure
"""
(
initial_vehicle_type, adjusted_vehicle_type, start_date, end_date, fraction_per_vehicle
) = self._get_analysis(kwargs)
assert len(kwargs) == 0, f"Keyword(s) {list(kwargs.keys())} have not been processed."
if len(fraction_per_vehicle) == 0:
fig, ax = no_data_graph()
ax.set_title(self.plot_title)
return ax
self._df = self._convert_analysis_result_to_df(fraction_per_vehicle)
filter_conditions = self._get_filter_condition(
adjusted_vehicle_type=adjusted_vehicle_type,
end_date=end_date,
initial_vehicle_type=initial_vehicle_type,
start_date=start_date
)
ax = self._df.plot.scatter(x="arrival time", y="percentage of adjusted containers")
ax.yaxis.set_major_formatter(FuncFormatter('{0:.0%}'.format)) # pylint: disable=consider-using-f-string
ax.set_title(self.plot_title + "\n" + filter_conditions)
plt.setp(ax.get_xticklabels(), rotation=30, horizontalalignment='right')
ax.grid(color='lightgray', linestyle=':', linewidth=.5)
return ax
def _convert_analysis_result_to_df(
self,
analysis_result: typing.Dict[VehicleIdentifier, float]
) -> pd.DataFrame:
rows = []
for vehicle_identifier, fraction in analysis_result.items():
vehicle_name = self._vehicle_identifier_to_text(vehicle_identifier)
rows.append({
"vehicle name": vehicle_name,
"vehicle type": vehicle_identifier.mode_of_transport,
"arrival time": vehicle_identifier.vehicle_arrival_time,
"percentage of adjusted containers": fraction
})
df = pd.DataFrame(rows)
return df