from __future__ import annotations
import itertools
import logging
import plotly.graph_objs
from conflowgen.analyses.container_flow_adjustment_by_vehicle_type_analysis import \
ContainerFlowAdjustmentByVehicleTypeAnalysis
from conflowgen.reporting import AbstractReportWithPlotly
[docs]
class ContainerFlowAdjustmentByVehicleTypeAnalysisReport(AbstractReportWithPlotly):
"""
This analysis report takes the data structure as generated by
:class:`.ContainerFlowAdjustmentByVehicleTypeAnalysis` 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 ContainerFlowAdjustmentByVehicleTypeAnalysisReport \
<notebooks/analyses.ipynb#Container-Flow-Adjustment-By-Vehicle-Type-Analysis-Report>`_.
"""
report_description = """
Analyze how many containers needed to change their initial container type.
When containers are generated, in order to obey the maximum dwell time, the vehicle type that is used for
onward transportation might change. The initial outbound vehicle type is the vehicle type that is drawn
randomly for a container at the time of generation. The adjusted vehicle type is the vehicle type that is drawn
in case no vehicle of the initial outbound vehicle type is left within the maximum dwell time.
"""
logger = logging.getLogger("conflowgen")
def __init__(self):
super().__init__()
self.analysis = ContainerFlowAdjustmentByVehicleTypeAnalysis()
[docs]
def get_report_as_text(
self, **kwargs
) -> str:
"""
Keyword Args:
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.
"""
initial_to_adjusted_outbound_flow = self._get_analysis(kwargs)
initial_to_adjusted_outbound_flow_in_containers = initial_to_adjusted_outbound_flow.containers
initial_to_adjusted_outbound_flow_in_teu = initial_to_adjusted_outbound_flow.teu
# create string representation
report = "\n"
report += "initial vehicle type "
report += "adjusted vehicle type "
report += "transported capacity (in TEU) "
report += "transported capacity (in boxes)"
report += "\n"
for vehicle_type_initial, vehicle_type_adjusted in itertools.product(
self.order_of_vehicle_types_in_report, repeat=2):
vehicle_type_from_as_text = str(vehicle_type_initial).replace("_", " ")
vehicle_type_to_as_text = str(vehicle_type_adjusted).replace("_", " ")
transported_capacity_in_teu = initial_to_adjusted_outbound_flow_in_teu[vehicle_type_initial][
vehicle_type_adjusted]
transported_capacity_in_containers = initial_to_adjusted_outbound_flow_in_containers[vehicle_type_initial][
vehicle_type_adjusted]
report += f"{vehicle_type_from_as_text:<21} "
report += f"{vehicle_type_to_as_text:<23} "
report += f"{transported_capacity_in_teu:>28.1f}"
report += f"{transported_capacity_in_containers:>32}"
report += "\n"
report += "(rounding errors might exist)\n"
return report
[docs]
def get_report_as_graph(self, **kwargs) -> plotly.graph_objs.Figure:
"""
The container flow is represented by a Sankey diagram.
Keyword Args:
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 plotly figure of the Sankey diagram.
"""
initial_to_adjusted_outbound_flow = self._get_analysis(kwargs)
unit = "TEU"
initial_to_adjusted_outbound_flow_in_teu = initial_to_adjusted_outbound_flow.teu
vehicle_types = [
str(vehicle_type).replace("_", " ")
for vehicle_type in initial_to_adjusted_outbound_flow_in_teu.keys()
]
source_ids = list(range(len(vehicle_types)))
target_ids = list(range(len(vehicle_types), 2 * len(vehicle_types)))
value_ids = list(itertools.product(source_ids, target_ids))
source_ids_with_duplication = [source_id for (source_id, _) in value_ids]
target_ids_with_duplication = [target_id for (_, target_id) in value_ids]
value = [
initial_to_adjusted_outbound_flow_in_teu[vehicle_type_initial][vehicle_type_adjusted]
for vehicle_type_initial in initial_to_adjusted_outbound_flow_in_teu.keys()
for vehicle_type_adjusted in initial_to_adjusted_outbound_flow_in_teu[vehicle_type_initial].keys()
]
if sum(value) == 0:
self.logger.warning("No data available for plotting")
initial_labels = [
str(vehicle_type_initial).replace("_", " ").capitalize() + " initial:<br>" + str(
round(sum(initial_to_adjusted_outbound_flow_in_teu[vehicle_type_initial].values()), 2)
) + " " + unit
for vehicle_type_initial in initial_to_adjusted_outbound_flow_in_teu.keys()
]
to_adjusted_flow = [0 for _ in range(len(initial_to_adjusted_outbound_flow_in_teu.keys()))]
for vehicle_type_initial, capacity in initial_to_adjusted_outbound_flow_in_teu.items():
for i, vehicle_type_adjusted in enumerate(initial_to_adjusted_outbound_flow_in_teu[vehicle_type_initial]):
to_adjusted_flow[i] += capacity[vehicle_type_adjusted]
adjusted_labels = [
str(vehicle_type_adjusted).replace("_", " ").capitalize() + " adjusted:<br>" + str(
round(to_adjusted_flow[i], 2)) + " " + unit
for i, vehicle_type_adjusted in enumerate(initial_to_adjusted_outbound_flow_in_teu.keys())
]
fig = plotly.graph_objs.Figure(
data=[
plotly.graph_objs.Sankey(
arrangement='perpendicular',
node={
'pad': 15,
'thickness': 20,
'line': {
'color': "black",
'width': 0.5
},
'label': initial_labels + adjusted_labels,
'color': "dimgray",
},
link={
'source': source_ids_with_duplication,
'target': target_ids_with_duplication,
'value': value
}
)
]
)
fig.update_layout(
title_text=f"Container flow from initial vehicle type A to adjusted vehicle type B in {unit} as for some "
"containers<br>"
"the initially intended vehicle type was not available due to constraints "
"(schedules, dwell times, etc.).",
font_size=10,
width=900,
height=700
)
return fig
def _get_analysis(self, kwargs):
start_date = kwargs.pop("start_date", None)
end_date = kwargs.pop("end_date", None)
assert len(kwargs) == 0, f"Keyword(s) {kwargs.keys()} have not been processed"
initial_to_adjusted_outbound_flow = self.analysis.get_initial_to_adjusted_outbound_flow(
start_date=start_date,
end_date=end_date
)
return initial_to_adjusted_outbound_flow