raplan.plot ¶

def gantt_project_procedures(
    project: Project,
    horizon: Horizon,
    x_axis_title: str = "Time [year]",
    y: str = "System",
    y_axis_title: str = "System",
    zero_margin: bool = True,
    duration_scale: float = 1.0,
    **args,
) -> go.Figure:
    """Create a Gantt chart for a given project's systems.

    Arguments:
        project: RaPlan project.
        horizon: Time horizon for which to generate the figure.
        x_axis_title: X-axis label.
        y: Column name of the data dictionary that is built in this function. (see below)
        y_axis_title: Y-axis label.
        zero_margin: Whether to set all margins to zero.
        duration_scale: Conversion factor from task durations to scheduled time. The latter is in
            years, usually.

    The data dictionary contains the following columns:

    - Start
    - End
    - System
    - Tasks
    - Cost
    - Duration
    """
    data = dict(Start=[], End=[], System=[], Tasks=[], Cost=[], Duration=[])

    start, end = _derive_horizon_start_end(project, horizon=horizon)

    for s in project.systems:
        for p in s.gen_procedures():
            data["Start"].append(
                float_years_to_datetime(start + p.time, precision="day"),
            )
            data["End"].append(
                float_years_to_datetime(
                    start + p.time + p.duration * duration_scale, precision="day"
                ),
            )
            data["System"].append(s.name)
            data["Tasks"].append([m.task.name for m in p.maintenance])
            data["Cost"].append(p.cost)
            data["Duration"].append(p.duration)

    fig = px.timeline(
        data,
        x_start="Start",
        x_end="End",
        y=y,
        color="Cost",
        hover_data=["Duration", "Tasks"],
        **args,
    )

    fig.update_yaxes(title=y_axis_title)
    fig.update_xaxes(
        title=x_axis_title,
        range=(float_years_to_datetime(start), float_years_to_datetime(end)),
    )
    if zero_margin:
        fig.layout.update(margin=dict(r=0, t=0, l=0, b=0))

    return fig

def gantt_system(
    system: System,
    horizon: Horizon | None = None,
    x_axis_title: str = "Time [year]",
    y: str = "Component",
    y_axis_title: str = "Component",
    zero_margin: bool = True,
    **args,
) -> go.Figure:
    """Create a Gantt chart for a given system's components.

    Arguments:
        system: RaPlan `System`.
        horizon: Time horizon for which to generate the figure.
        x_axis_title: X-axis label.
        y: Column name of the data dictionary that is built in this function. (see below)
        y_axis_title: Y-axis label.
        zero_margin: Whether to set all margins to zero.

    The data dictionary contains the following columns:

    - Start
    - End
    - Component
    - Task
    - Cost
    - Duration
    """
    data = dict(Start=[], End=[], Component=[], Task=[], Cost=[], Duration=[])
    start, end = _derive_horizon_start_end(system, horizon=horizon)

    for c in system.components:
        for m in c.gen_maintenance():
            data["Start"].append(
                float_years_to_datetime(start + m.time, precision="day"),
            )
            data["End"].append(
                float_years_to_datetime(start + m.end, precision="day"),
            )
            data["Component"].append(c.name)
            data["Task"].append(m.task.name)
            data["Cost"].append(m.task.cost)
            data["Duration"].append(m.task.duration)

    fig = px.timeline(
        data,
        x_start="Start",
        x_end="End",
        y=y,
        color="Task",
        hover_data=["Cost", "Duration"],
        **args,
    )

    fig.update_yaxes(title=y_axis_title)
    fig.update_xaxes(
        title=x_axis_title,
        range=(float_years_to_datetime(start), float_years_to_datetime(end)),
    )
    if zero_margin:
        fig.layout.update(margin=dict(r=0, t=0, l=0, b=0))

    return fig

def get_cfp_figure(
    subject: Component
    | System
    | Project
    | Compound
    | Sequence[Component | System | Project | Compound],
    xs: list[int | float],
    horizon: Horizon | None = None,
    compound: str | None = None,
    thresholds: dict[str, float] = {"5%": 0.05},
    x_axis_title: str = "Time",
    y_axis_title: str = "CFP",
    y_factor: int | float = 1,
) -> go.Figure:
    """Get a figure displaying the CFP or CFPs of the subject(s).

    Arguments:
        subject: Planning object(s) with a CFP method.
        xs: Time values to calculate CFP values at.
        horizon: Optional horizon to apply. It's start is used as the X-axis offset to
            apply and the end is used to then set the upper bound of the X-axis.
        compound: Whether to show a compound CFP line for the subjects given.
        thresholds: Threshold lines to show.
        x_axis_title: Title for the X-axis (time).
        y_axis_title: Title for the Y-axis (CFP).
        y_factor: Factor to scale the Y-axis values by.

    Returns:
        Plotly figure.
    """
    subjects = _parse_subjects(subject)
    x_offset = 0 if horizon is None else horizon.start

    cfps = [
        _get_cfp_trace(
            s,
            xs,
            x_offset=x_offset,
            y_factor=y_factor,
        )
        for s in subjects
    ]

    if compound and len(subjects) > 1:
        cfps.append(
            _get_cfp_trace(
                _compound_subjects(subjects, name=compound),
                xs,
                x_offset=x_offset,
                y_factor=y_factor,
            )
        )
    threshold_lines = [
        go.Scatter(
            x=[xs[0] + x_offset, xs[-1] + x_offset],
            y=2 * [y_factor * v],
            name=k,
            line=dict(dash="dash", color="crimson"),
            mode="lines",
        )
        for k, v in thresholds.items()
    ]
    fig = go.Figure(
        cfps + threshold_lines,
        layout=dict(
            xaxis=dict(title=x_axis_title),
            yaxis=dict(title=y_axis_title),
        ),
    )
    if horizon:
        _update_xaxis_horizon(fig, horizon)
    return fig

def get_cost_figure(
    subject: Component
    | System
    | Project
    | Compound
    | Sequence[Component | System | Project | Compound],
    horizon: Horizon | None = None,
    bar_width: float | str = "unit",
    x_axis_title: str = "Time",
    y_axis_title: str = "Cost",
    y_factor: int | float = 1,
    **args,
) -> go.Figure:
    """Get a cost bar chart figure.

    Arguments:
        subject: Planning object(s) with maintenance tasks with cost attached.
        horizon: Optional horizon to apply. It's start is used as the X-axis offset to
            apply and the end is used to then set the upper bound of the X-axis.
        bar_width: How to determine widths of bars shown. 'auto' for Plotly defaults,
            'unit' for widths of 1, or a float for a decimal width w.r.t. axis values.
        x_axis_title: Title for the X-axis (time).
        y_axis_title: Title for the Y-axis (cost).
        y_factor: Factor to scale the Y-axis values by.

    Returns:
        Cost bar chart figure.
    """
    x_offset = 0 if horizon is None else horizon.start

    fig = go.Figure(
        _get_barchart(
            subject,
            x_offset=x_offset,
            prop="cost",
            bar_width=bar_width,
            y_factor=y_factor,
        ),
        layout=dict(
            xaxis=dict(title=x_axis_title),
            yaxis=dict(title=y_axis_title),
            **_BAR_STYLE,
        ),
        **args,
    )

    if horizon:
        _update_xaxis_horizon(fig, horizon)

    return fig

def get_duration_figure(
    subject: Component
    | System
    | Project
    | Compound
    | Sequence[Component | System | Project | Compound],
    horizon: Horizon | None = None,
    bar_width: float | str = "unit",
    x_axis_title: str = "Time",
    y_axis_title: str = "Duration",
    y_factor: int | float = 1,
) -> go.Figure:
    """Get a duration bar chart figure.

    Arguments:
        subject: Planning object(s) with maintenance tasks with duration attached.
        horizon: Optional horizon to apply. It's start is used as the X-axis offset to
            apply and the end is used to then set the upper bound of the X-axis.
        bar_width: How to determine widths of bars shown. 'auto' for Plotly defaults,
            'unit' for widths of 1, or a float for a decimal width w.r.t. axis values.
        x_axis_title: Title for the X-axis (time).
        y_axis_title: Title for the Y-axis (duration).
        y_factor: Factor to scale the Y-axis values by.

    Returns:
        Duration bar chart figure.
    """
    x_offset = 0 if horizon is None else horizon.start
    fig = go.Figure(
        _get_barchart(
            subject,
            x_offset=x_offset,
            prop="duration",
            bar_width=bar_width,
            y_factor=y_factor,
        ),
        layout=dict(xaxis=dict(title=x_axis_title), yaxis=dict(title=y_axis_title), **_BAR_STYLE),
    )
    if horizon:
        _update_xaxis_horizon(fig, horizon)
    return fig

def get_overview_figure(
    subject: Component
    | System
    | Project
    | Compound
    | Sequence[Component | System | Project | Compound],
    xs: list[int | float],
    horizon: Horizon | None = None,
    compound: str | None = None,
    bar_width: float | str = "unit",
    x_axis_title: str = "Time",
    cfp_axis_title: str = "CFP",
    cfp_factor: int | float = 1,
    cost_axis_title: str = "Cost",
    cost_log: bool = True,
    cost_factor: int | float = 1,
    duration_axis_title: str = "Duration",
    duration_log: bool = True,
    duration_factor: int | float = 1,
) -> go.Figure:
    """Get an overview figure consisting of a CFP plot, as well as cost and duration
    bar charts.

    Arguments:
        subject: Planning subject(s) with CFP and maintenance tasks.
        xs: Values to calculate the CFP at.
        horizon: Optional horizon to apply. It's start is used as the X-axis offset to
            apply and the end is used to then set the upper bound of the X-axis.
        compound: If not `None`, the title for a compound CFP line.
        bar_width: How to determine widths of bars shown. 'auto' for Plotly defaults,
            'unit' for widths of 1, or a float for a decimal width w.r.t. axis values.
        x_axis_title: Title for the X-axis (time).
        cfp_axis_title: Title for the CFP Y-axis.
        cfp_factor: Scaling for the CFP values.
        cost_axis_title: Title for the cost Y-axis.
        cost_log: Whether to display cost on a logarithmic axis.
        cost_factor: Scaling for the cost values.
        duration_axis_title: Title for the duration Y-axis.
        duration_log: Whether to display duration on a logarithmic axis.
        duration_factor: Scaling for the duration values.

    Returns:
        Subplots overview figure.
    """
    fig = make_subplots(3, 1, shared_xaxes=True, x_title=x_axis_title)
    subjects = _parse_subjects(subject)
    x_offset = 0.0 if horizon is None else horizon.start

    for s in subjects:
        fig.add_trace(
            _get_cfp_trace(
                s,
                xs,
                x_offset=x_offset,
                y_factor=cfp_factor,
            ),
            row=1,
            col=1,
        )

    if compound and len(subjects) > 1:
        fig.add_trace(
            _get_cfp_trace(
                _compound_subjects(subjects, name=compound),
                xs,
                x_offset=x_offset,
                y_factor=cfp_factor,
            )
        )

    for b in _get_barchart(
        _compound_subjects(subjects, "Cost"),
        prop="cost",
        x_offset=x_offset,
        bar_width=bar_width,
        y_factor=cost_factor,
    ):
        fig.add_trace(b, row=2, col=1)

    for b in _get_barchart(
        _compound_subjects(subjects, "Duration"),
        prop="duration",
        x_offset=x_offset,
        bar_width=bar_width,
        y_factor=duration_factor,
    ):
        fig.add_trace(b, row=3, col=1)

    if cost_log:
        fig.update_yaxes(type="log", row=2, col=1)

    if duration_log:
        fig.update_yaxes(type="log", row=3, col=1)

    fig.layout.update(
        yaxis1=dict(title=cfp_axis_title),
        yaxis2=dict(title=cost_axis_title),
        yaxis3=dict(title=duration_axis_title),
        **_BAR_STYLE,
    )

    if horizon:
        _update_xaxis_horizon(fig, horizon)

    return fig

def get_procedures_plot(
    procedures: list[PlottingProcedure],
    x_axis_title: str = "Time",
    subject_axis_title: str = "Subject",
    cost_axis_title: str = "Cost",
    cost_log: bool = True,
    cost_factor: int | float = 1,
    duration_axis_title: str = "Duration",
    duration_log: bool = True,
    duration_factor: int | float = 1,
) -> go.Figure:
    """Get a procedures plot where systems are to be plotted versus standardized
    procedures (groups of tasks) and their duration and cost.

    Arguments:
        procedures: A list of procedures to be executed.
        x_axis_title: Title for the X-axis (time).
        subject_axis_title: Title for the subject Y-axis.
        cost_axis_title: Title for the cost Y-axis.
        cost_log: Whether to display cost on a logarithmic axis.
        cost_factor: Factor to scale cost values by.
        duration_axis_title: Title for the duration Y-axis.
        duration_log: Whether to display duration on a logarithmic axis.
        duration_factor: Factor to scale duration values by.
    """
    fig = make_subplots(
        rows=3,
        cols=1,
        shared_xaxes=True,
        vertical_spacing=0.1,
        row_heights=[0.6, 0.2, 0.2],
    )

    kinds = sorted(set(p.kind for p in procedures))
    kind_to_index = {value: i for i, value in enumerate(kinds)}
    _times = sorted(set(p.time for p in procedures))
    systems = sorted(set(p.system for p in procedures))
    system_to_index = {value: i for i, value in enumerate(systems)}

    procedures = sorted(procedures, key=lambda p: (p.kind, p.system, p.time, -p.cost, -p.duration))

    colors = {kind: COLORS[i % len(COLORS)] for i, kind in enumerate(kinds)}
    hm_colors = [(float(i), colors[kind]) for i, kind in enumerate(kinds)]

    xs = list(range(int(_times[0]), int(_times[-1] + 1)))

    heatmap_z: list[list[int | None]] = [[None for j in xs] for i in systems]
    heatmap_text: list[list[str | None]] = [[None for j in xs] for i in systems]
    for p in procedures:
        row, col = system_to_index[p.system], int(p.time) - xs[0]
        heatmap_z[row][col] = kind_to_index[p.kind]
        heatmap_text[row][col] = p.kind

    for kind in kinds:
        times: list[int | float] = []
        costs: list[int | float] = []
        duras: list[int | float] = []
        objs: list[int] = []

        for p in procedures:
            if p.kind != kind:
                continue
            times.append(p.time)
            costs.append(cost_factor * p.cost)
            duras.append(duration_factor * p.duration)
            objs.append(system_to_index[p.system])

        # Heatmap, with adjusted colormaps to trick plotly's via legend toggles.
        hm_colors = [colors[kind] if kind == _kind else "rgba(0,0,0,0)" for _kind in kinds]

        fig.add_trace(
            go.Heatmap(
                x=xs,
                y=systems,
                z=heatmap_z,
                legendgroup=kind,
                name="procedures",
                colorscale=hm_colors,
                showlegend=False,
                showscale=False,
                hoverongaps=False,
                xgap=2,
                ygap=2,
                hovertemplate="time: %{x}<br>system: %{y}<br>" + "procedure: %{customdata}",
                customdata=heatmap_text,
            )
        )

        # Cost
        fig.add_trace(
            go.Bar(
                x=times,
                y=costs,
                legendgroup=kind,
                name=kind,
                marker_color=colors[kind],
                hovertemplate="time: %{x}<br>cost: %{y}<br>object: %{customdata}",
                customdata=[systems[obj] for obj in objs],
            ),
            2,
            1,
        )

        # Duration
        fig.add_trace(
            go.Bar(
                x=times,
                y=duras,
                legendgroup=kind,
                name=kind,
                showlegend=False,
                marker_color=colors[kind],
            ),
            3,
            1,
        )

    if cost_log:
        fig.update_yaxes(type="log", row=2, col=1)
    if duration_log:
        fig.update_yaxes(type="log", row=3, col=1)

    fig.layout.update(
        xaxis=dict(showticklabels=True),
        xaxis2=dict(showticklabels=True),
        xaxis3=dict(title=x_axis_title),
        yaxis1=dict(title=subject_axis_title),
        yaxis2=dict(title=cost_axis_title),
        yaxis3=dict(title=duration_axis_title),
        barmode="stack",
        bargap=0.0,
        bargroupgap=0.1,
    )
    return fig