from __future__ import annotations
import typing # noqa, pylint: disable=unused-import # lgtm [py/unused-import] # used in the docstring
import matplotlib.axis
import numpy as np
import pandas as pd
from conflowgen.analyses.container_flow_adjustment_by_vehicle_type_analysis_summary import \
ContainerFlowAdjustmentByVehicleTypeAnalysisSummary, ContainerFlowAdjustedToVehicleType
from conflowgen.reporting import AbstractReportWithMatplotlib
from conflowgen.reporting.no_data_plot import no_data_graph
[docs]
class ContainerFlowAdjustmentByVehicleTypeAnalysisSummaryReport(AbstractReportWithMatplotlib):
"""
This analysis report takes the data structure as generated by
:class:`.ContainerFlowAdjustmentByVehicleTypeAnalysisSummary` 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 ContainerFlowAdjustmentByVehicleTypeAnalysisSummaryReport \
<notebooks/analyses.ipynb#Container-Flow-Adjustment-By-Vehicle-Type-Analysis-Summary-Report>`_.
"""
report_description = """
Analyse whether a container needed to change its vehicle type for the outbound journey and if that was the case,
how many times which vehicle type was chosen in order to not exceed the maximum dwell time.
"""
def __init__(self):
super().__init__()
self.analysis_summary = ContainerFlowAdjustmentByVehicleTypeAnalysisSummary()
[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:
start_date (datetime.datetime):
Only include containers that arrive after the given start time. Defaults to ``None``.
end_date (datetime.datetime):
Only include containers that depart before the given end time. Defaults to ``None``.
Returns:
The report in text format (possibly spanning over several lines).
"""
adjusted_to = self._get_analysis(kwargs)
total_capacity = sum(adjusted_to)
total_capacity = total_capacity if total_capacity else np.nan
report = "\n"
report += " Capacity in TEU\n"
report += f"vehicle type unchanged: {adjusted_to.unchanged:>10.1f} " \
f"({adjusted_to.unchanged / total_capacity:>5.2%})\n"
report += f"changed to deep sea vessel: {adjusted_to.deep_sea_vessel:>10.1f} " \
f"({adjusted_to.deep_sea_vessel / total_capacity:>5.2%})\n"
report += f"changed to feeder: {adjusted_to.feeder:>10.1f} " \
f"({adjusted_to.feeder / total_capacity:>5.2%})\n"
report += f"changed to barge: {adjusted_to.barge:>10.1f} " \
f"({adjusted_to.barge / total_capacity:>5.2%})\n"
report += f"changed to train: {adjusted_to.train:>10.1f} " \
f"({adjusted_to.train / total_capacity:>5.2%})\n"
report += f"changed to truck: {adjusted_to.truck:>10.1f} " \
f"({adjusted_to.truck / total_capacity:>5.2%})\n"
report += "(rounding errors might exist)\n"
return report.replace(" nan", "-")
[docs]
def get_report_as_graph(self, **kwargs) -> matplotlib.axis.Axis:
"""
The report as a graph is represented as a pie chart.
Keyword Args:
start_date (datetime.datetime):
Only include containers that arrive after the given start time. Defaults to ``None``.
end_date (datetime.datetime):
Only include containers that depart before the given end time. Defaults to ``None``.
Returns:
The matplotlib axis of the pie chart.
"""
adjusted_to = self._get_analysis(kwargs)
plot_title = "Adjusted vehicle type (summary)"
if sum(adjusted_to) == 0:
fig, ax = no_data_graph()
ax.set_title(plot_title)
else:
data_series = pd.Series({
"unchanged": adjusted_to.unchanged,
"deep sea vessel": adjusted_to.deep_sea_vessel,
"feeder": adjusted_to.feeder,
"barge": adjusted_to.barge,
"train": adjusted_to.train,
"truck": adjusted_to.truck
}, name="Vehicle type adjustment")
ax = data_series.plot.pie(
legend=False,
autopct=lambda p: f'{p:.1f}%' if p > 0 else '',
label="",
title=plot_title
)
return ax
def _get_analysis(self, kwargs: dict) -> ContainerFlowAdjustedToVehicleType:
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"
adjusted_to = self.analysis_summary.get_summary(
start_date=start_date,
end_date=end_date
)
return adjusted_to