raplan.analysis

Tools for maintenance planning and scheduling analysis.

SCHEDULE_FN module-attribute

SCHEDULE_FN = dict(
    SYNC_START_EARLIEST=_sync_start_earliest,
    SYNC_START_LATEST=_sync_start_latest,
    SYNC_START_MEAN=_sync_start_mean,
    SYNC_END_EARLIEST=_sync_end_earliest,
    SYNC_END_LATEST=_sync_end_latest,
    SYNC_END_MEAN=_sync_end_mean,
)

Maintenance task scheduling functions.

ClassToKind

Class to Kind conversion.

KindToClass

Kind to Class conversion.

get_maintenance_graph

get_maintenance_graph(
    project: Project, threshold: int | float
) -> Graph

Get a graph of maintenance tasks. Edges are generated with an adjacency value of of the threshold minus the time difference between tasks.

Parameters:

Name	Type	Description	Default
project	Project	Scheduling project to generate a graph for.	required
threshold	int | float	Maximum time difference. Tasks that are within this threshold will be assigned an adjacency edge.	required

Returns:

Type	Description
Graph	Graph of maintenance tasks.

Source code in src/raplan/analysis.py

def get_maintenance_graph(project: Project, threshold: int | float) -> Graph:
    """Get a graph of maintenance tasks. Edges are generated with an adjacency value of
    of the threshold minus the time difference between tasks.

    Arguments:
        project: Scheduling project to generate a graph for.
        threshold: Maximum time difference. Tasks that are within this threshold will be
            assigned an adjacency edge.

    Returns:
        Graph of maintenance tasks.
    """
    graph = Graph(
        labels=[project.name] if project.name else [],
        kind=ClassToKind.PROJECT,
        annotations=dict(data=asdict(project)),
    )

    maintenance_nodes: list[Node] = []
    for system in project.systems:
        if system.uuid in graph.node_uuid_dict:
            s = graph[system.uuid]
        else:
            s = _to_node(system)
            graph.add_node(s)

        for component in system.components:
            if component.uuid in graph.node_uuid_dict:
                c = graph[component.uuid]
            else:
                c = _to_node(component)
                graph.add_node(c)

            graph.add_edge(Edge(source=c, target=s, kind="belongs_to"))
            graph.add_edge(Edge(source=s, target=c, kind="consists_of"))

            for maintenance in component.maintenance:
                if maintenance.uuid in graph.node_uuid_dict:
                    m = graph[maintenance.uuid]
                else:
                    m = _to_node(maintenance)
                    graph.add_node(m)

                # Add edges to both component and system.
                graph.add_edge(Edge(source=m, target=c, kind="maintenance"))
                graph.add_edge(Edge(source=c, target=m, kind="maintenance"))
                graph.add_edge(Edge(source=m, target=s, kind="maintenance"))
                graph.add_edge(Edge(source=s, target=m, kind="maintenance"))

                for mx in maintenance_nodes:
                    if m is mx:
                        continue

                    if graph[mx.name, m.name]:
                        continue

                    delta_t = abs(m.weights["time"] - mx.weights["time"])
                    if not (delta_t < threshold):
                        continue

                    weight = threshold - delta_t

                    graph.add_edge(
                        Edge(
                            source=m,
                            target=mx,
                            kind="adjacency",
                            weights=dict(adjacency=weight),
                        )
                    )
                    graph.add_edge(
                        Edge(
                            source=mx,
                            target=m,
                            kind="adjacency",
                            weights=dict(adjacency=weight),
                        )
                    )

                maintenance_nodes.append(m)

    return graph

process_clustered_maintenance

process_clustered_maintenance(
    graph: Graph,
    schedule_fn: Callable[
        [list[Node]], list[int | float]
    ] = SCHEDULE_FN["SYNC_START_EARLIEST"],
)

Check for maintenance nodes with parents and synchronize their times.

Parameters:

Name	Type	Description	Default
graph	Graph	Maintenance graph.	required
schedule_fn	Callable[[list[Node]], list[int | float]]	A method that takes a list of (maintenance) nodes and returns a new list with modified starting times to incorporate. It should not modify the nodes in-place.	SCHEDULE_FN['SYNC_START_EARLIEST']

Source code in src/raplan/analysis.py

def process_clustered_maintenance(
    graph: Graph,
    schedule_fn: Callable[[list[Node]], list[int | float]] = SCHEDULE_FN["SYNC_START_EARLIEST"],
):
    """Check for maintenance nodes with parents and synchronize their times.

    Arguments:
        graph: Maintenance graph.
        schedule_fn: A method that takes a list of (maintenance) nodes and returns a new
            list with modified starting times to incorporate. It should not modify the
            nodes in-place.
    """
    for n in graph.nodes:
        if n.kind != "maintenance" or n.is_leaf:
            continue
        leafs = n.children
        times = schedule_fn(leafs)
        for i, time in enumerate(times):
            leaf = leafs[i]
            leaf.weights["time"] = time
            leaf.annotations.data["time"] = time

project_from_graph

project_from_graph(graph: Graph) -> Project

Recreate a Project from a Graph. Utilizes the 'data' annotation to build base objects, and overrides it with data found in the graph constructs.

Parameters:

Name	Type	Description	Default
graph	Graph	Graph data.	required

Source code in src/raplan/analysis.py

def project_from_graph(graph: Graph) -> Project:
    """Recreate a Project from a Graph. Utilizes the 'data' annotation to build base
    objects, and overrides it with data found in the graph constructs.

    Arguments:
        graph: Graph data.
    """
    project = _from_node(graph)  # It works, even though naming is odd.
    assert isinstance(project, Project)

    systems = []
    for s in filter(lambda n: n.kind == "system" and n.is_leaf, graph.nodes):
        system = _from_node(s)
        assert isinstance(system, System)

        components: list[Component] = []
        for ec in graph.edges_from(s):
            if ec.kind != "consists_of" or ec.target.kind != "component" or not ec.target.is_leaf:
                continue
            c = ec.target

            component = _from_node(c)
            assert isinstance(component, Component)

            maintenance: list[Maintenance] = []
            for em in graph.edges_from(c):
                if (
                    em.kind != "maintenance"
                    or em.target.kind != "maintenance"
                    or not em.target.is_leaf
                ):
                    continue
                m = em.target

                maintenance_item = _from_node(m)
                assert isinstance(maintenance_item, Maintenance)
                maintenance.append(maintenance_item)

            component.maintenance = maintenance
            components.append(component)

        system.components = components
        systems.append(system)

    project.systems = systems
    return project