raplan.classes

Dataclasses to use and configure maintenance planning and scheduling with.

Component dataclass

Component(
    name: str
    | None = field(default=None, skip_if_default=True),
    age: int | float = 0.0,
    lifetime: int | float = 1.0,
    distribution: Distributions = field(
        default_factory=Weibull
    ),
    maintenance: list[Maintenance] = field(
        default_factory=list, skip_if=_is_empty, repr=False
    ),
    uuid: UUID = field(default_factory=uuid4),
)

Component with a failure distribution.

Parameters:

Name	Type	Description	Default
name	str | None	Name of this component.	field(default=None, skip_if_default=True)
age	int | float	Starting age offset (usually in years).	0.0
distribution	Distributions	Failure distribution to use.	field(default_factory=Weibull)
maintenance	list[Maintenance]	List of maintenance tasks that should be applied over this component's lifespan.	field(default_factory=list, skip_if=_is_empty, repr=False)
uuid	UUID	Automatically generated unique identifier for this component.	field(default_factory=uuid4)

cfp

cfp(x: int | float = 1.0) -> float

Cumulative failure probability density function incorporating maintenance.

Source code in src/raplan/classes.py

def cfp(self, x: int | float = 1.0) -> float:
    """Cumulative failure probability density function incorporating maintenance."""
    return self.distribution.cdf(self.get_age_at(x))

get_age_at

get_age_at(x: int | float = 1.0) -> float

Effective age at a point in time given the currently set schedule.

Source code in src/raplan/classes.py

def get_age_at(self, x: int | float = 1.0) -> float:
    """Effective age at a point in time given the currently set schedule."""
    age = float(self.age)
    last_time = 0.0
    for m in self.get_ordered_maintenance():
        if m.time > x:
            # Maintenance is yet to happen.
            break
        # Apply rejuvenation with the then actual age.
        age = (age + m.time - last_time) * (1.0 - m.task.rejuvenation)
        last_time = m.time
    # Add remaining time since last maintenance.
    age += x - last_time
    return age

get_ordered_maintenance

get_ordered_maintenance() -> list[Maintenance]

Maintenance tasks sorted in time.

Source code in src/raplan/classes.py

def get_ordered_maintenance(self) -> list[Maintenance]:
    """Maintenance tasks sorted in time."""
    return sorted(self.maintenance, key=lambda m: m.time)

schedule_maintenance

schedule_maintenance(maintenance: Maintenance)

Schedule maintenance for a single or all system's component or all components.

Parameters:

Name	Type	Description	Default
maintenance	Maintenance	Maintenance to schedule.	required

Source code in src/raplan/classes.py

def schedule_maintenance(self, maintenance: Maintenance):
    """Schedule maintenance for a single or all system's component or all
    components.

    Arguments:
        maintenance: Maintenance to schedule.
    """
    self.maintenance.append(maintenance)

CyclicStrategy dataclass

CyclicStrategy(
    name: str
    | None = field(default=None, skip_if_default=True),
    tasks: list[Task] = field(
        default_factory=list, skip_if=_is_empty
    ),
    percentages: list[float] = field(
        default_factory=list, skip_if=_is_empty
    ),
    uuid: UUID = field(default_factory=uuid4),
)

Maintenance strategy to renovate or replace a component at certain percentages of a cycle.

Parameters:

Name	Type	Description	Default
name	str | None	Name for this cyclic strategy.	field(default=None, skip_if_default=True)
tasks	list[Task]	List of tasks that should be applied at the corresponding entry in percentages.	field(default_factory=list, skip_if=_is_empty)
percentages	list[float]	List of percentages [0.0, 1.0] at which to apply tasks.	field(default_factory=list, skip_if=_is_empty)
uuid	UUID	Automatically generated unique identifier for this cyclic strategy.	field(default_factory=uuid4)

apply_to_component

apply_to_component(
    component: Component,
    cycle_length: int | float,
    horizon: Horizon,
    repeat: bool = True,
    include_history: bool = True,
    integers: bool = False,
    overwrite: bool = True,
) -> None

Apply this strategy to a component.

Parameters:

Name	Type	Description	Default
component	Component	Component for which to schedule maintenance.	required
cycle_length	int | float	Cycle length.	required
horizon	Horizon	Planning horizon to consider.	required
repeat	bool	Whether the cycle should be repeated until the end of the horizon.	True
include_history	bool	Whether to include historical maintenance entries for components that have a pre-defined age.	True
overwrite	bool	Whether to fully overwrite a component's maintenance planning with this new one or extend it.	True
integers	bool	Whether to force all times to be integers.	False

Source code in src/raplan/classes.py

def apply_to_component(
    self,
    component: Component,
    cycle_length: int | float,
    horizon: Horizon,
    repeat: bool = True,
    include_history: bool = True,
    integers: bool = False,
    overwrite: bool = True,
) -> None:
    """Apply this strategy to a component.

    Arguments:
        component: Component for which to schedule maintenance.
        cycle_length: Cycle length.
        horizon: Planning horizon to consider.
        repeat: Whether the cycle should be repeated until the end of the horizon.
        include_history: Whether to include historical maintenance entries for
            components that have a pre-defined age.
        overwrite: Whether to fully overwrite a component's maintenance planning
            with this new one or extend it.
        integers: Whether to force all times to be integers.
    """
    maintenance = self.get_maintenance(
        component.age,
        cycle_length,
        horizon,
        repeat=repeat,
        include_history=include_history,
        integers=integers,
        prefix=component.name,
    )
    if overwrite:
        component.maintenance = maintenance
    else:
        component.maintenance.extend(maintenance)

get_maintenance

get_maintenance(
    age: int | float,
    cycle_length: int | float,
    horizon: Horizon,
    prefix: str | None = None,
    repeat: bool = True,
    include_history: bool = True,
    integers: bool = False,
) -> list[Maintenance]

Get maintenance list for this strategy.

Parameters:

Name	Type	Description	Default
age	int | float	Starting age of a virtual component.	required
cycle_length	int | float	Cycle length.	required
horizon	Horizon	Planning horizon to consider.	required
prefix	str | None	Maintenance naming prefix.	None
repeat	bool	Whether the cycle should be repeated until the end of the horizon.	True
include_history	bool	Whether to include historical maintenance entries for components that have a pre-defined age.	True
integers	bool	Whether to force all times to be integers.	False

Returns:

Type	Description
list[Maintenance]	Maintenance list.

Source code in src/raplan/classes.py

def get_maintenance(
    self,
    age: int | float,
    cycle_length: int | float,
    horizon: Horizon,
    prefix: str | None = None,
    repeat: bool = True,
    include_history: bool = True,
    integers: bool = False,
) -> list[Maintenance]:
    """Get maintenance list for this strategy.

    Arguments:
        age: Starting age of a virtual component.
        cycle_length: Cycle length.
        horizon: Planning horizon to consider.
        prefix: Maintenance naming prefix.
        repeat: Whether the cycle should be repeated until the end of the horizon.
        include_history: Whether to include historical maintenance entries for
            components that have a pre-defined age.
        integers: Whether to force all times to be integers.

    Returns:
        Maintenance list.
    """

    start: int | float = -age
    end = horizon.end - horizon.start if horizon.end else cycle_length

    offsets = [p * cycle_length for p in self.percentages]
    tasks = sorted(zip(offsets, self.tasks), key=lambda x: x[0])
    n_tasks = len(tasks)

    maintenance = []
    cycles_offset: int | float = 0
    while start + cycles_offset < end:
        for index, (offset, task) in enumerate(tasks):
            time = offset + cycles_offset + start
            if integers:
                time = round(time)

            # No further planning beyond this point.
            if time > end:
                break

            if not include_history and time < 0:
                continue

            _prefix = f"{prefix} | " if prefix else ""
            _name = f"{self.name} | " if self.name else ""

            maintenance.append(
                Maintenance(
                    f"{_prefix}{_name}{index + 1}/{n_tasks} | {task.name} @ {time}",
                    task=task,
                    time=time,
                )
            )

        # No further planning beyond this point.
        if not repeat or time > end:
            break

        cycles_offset = cycles_offset + cycle_length

    return maintenance

Horizon dataclass

Horizon(
    start: int | float = 0.0,
    end: int
    | float
    | None = field(default=None, skip_if_default=True),
    uuid: UUID = field(default_factory=uuid4),
)

Maintenance planning and scheduling horizon.

Parameters:

Name	Type	Description	Default
start	int | float	Start of the planning horizon.	0.0
end	int | float | None	End of the planning horizon. Optional, as it is otherwise derived from the final task in the schedule.	field(default=None, skip_if_default=True)
uuid	UUID	Automatically generated unique identifier for this Horizon.	field(default_factory=uuid4)

get_range

get_range(
    steps: int, zero_based: bool = True
) -> list[int | float]

Range between start and end (inclusive) in the given number of steps.

Source code in src/raplan/classes.py

def get_range(self, steps: int, zero_based: bool = True) -> list[int | float]:
    """Range between start and end (inclusive) in the given number of steps."""
    if self.end is None:
        raise ValueError("Can't calculate a range with no horizon end value.")
    step_size = (self.end - self.start) / steps
    start = type(self.start)(0) if zero_based else self.start
    return [start + i * step_size for i in range(steps + 1)]

Maintenance dataclass

Maintenance(
    name: str
    | None = field(default=None, skip_if_default=True),
    task: Task = field(default_factory=Task),
    time: int | float = 1.0,
    uuid: UUID = field(default_factory=uuid4),
)

Maintenance task scheduled at a point in time.

Parameters:

Name	Type	Description	Default
name	str | None	Name of this maintenance task.	field(default=None, skip_if_default=True)
task	Task	Task information.	field(default_factory=Task)
time	int | float	Time at which this maintenance is scheduled.	1.0
uuid	UUID	Automatically generated unique identifier for this maintenance.	field(default_factory=uuid4)

end property

end: int | float

End time of this maintenance.

get_progress

get_progress(x: int | float) -> float

Percentage of the set task that is completed at a given time.

Source code in src/raplan/classes.py

def get_progress(self, x: int | float) -> float:
    """Percentage of the set task that is completed at a given time."""
    if x < self.time:
        return 0.0
    elif x >= self.end or self.task.duration == 0:
        return 1.0
    else:
        return min(1.0, (x - self.time) / self.task.duration)

Procedure dataclass

Procedure(
    name: str
    | None = field(default=None, skip_if_default=True),
    system: str = "system",
    kind: str = "procedure",
    time: int | float = 1.0,
    cost: int | float = 1.0,
    duration: int | float = 1.0,
    uuid: UUID = field(default_factory=uuid4),
)

A specific grouping of tasks to apply to a System.

Note

Mainly used for plotting purposes.

Parameters:

Name	Type	Description	Default
name	str | None	Name for this procedure.	field(default=None, skip_if_default=True)
system	str	System (name) to which the procedure should be applied.	'system'
kind	str	Kind of procedure (category name).	'procedure'
time	int | float	Time at which the procedure is scheduled.	1.0
cost	int | float	Cost of the procedure.	1.0
duration	int | float	Duration of the procedure.	1.0
uuid	UUID	Automatically generated unique identifier for this procedure.	field(default_factory=uuid4)

Project dataclass

Project(
    name: str
    | None = field(default=None, skip_if_default=True),
    horizon: Horizon = field(default_factory=Horizon),
    systems: list[System] = field(
        default_factory=list, skip_if=_is_empty, repr=False
    ),
    uuid: UUID = field(default_factory=uuid4),
)

Maintenance planning and scheduling project.

Parameters:

Name	Type	Description	Default
name	str | None	Optional name to identify this project by.	field(default=None, skip_if_default=True)
horizon	Horizon	Timescale horizon for this project. What timeframe are we looking at?	field(default_factory=Horizon)
systems	list[System]	List of systems to consider a part of this maintenance project.	field(default_factory=list, skip_if=_is_empty, repr=False)
uuid	UUID	Automatically generated unique identifier for this project.	field(default_factory=uuid4)

cfp

cfp(x: int | float = 1.0) -> float

Cumulative failure probability density function as the sum of its systems' respective function incorporating maintenance.

Source code in src/raplan/classes.py

def cfp(self, x: int | float = 1.0) -> float:
    """Cumulative failure probability density function as the sum of its
    systems' respective function incorporating maintenance.
    """
    if len(self.systems):
        return distributions.compound_probability(s.cfp(x) for s in self.systems)
    else:
        return 0.0

get_horizon_end

get_horizon_end() -> float

Get the end of the planning horizon or last maintenance task.

Source code in src/raplan/classes.py

def get_horizon_end(self) -> float:
    """Get the end of the planning horizon or last maintenance task."""
    if self.horizon.end is None:
        end = 0.0
        try:
            end = max(m.time for s in self.systems for c in s.components for m in c.maintenance)
        except ValueError:
            pass  # arg is an empty sequency: end = 0.0
        finally:
            return end
    return self.horizon.end

get_ordered_maintenance

get_ordered_maintenance() -> list[Maintenance]

Get all maintenance ordered in time.

Source code in src/raplan/classes.py

def get_ordered_maintenance(self) -> list[Maintenance]:
    """Get all maintenance ordered in time."""
    return sorted(
        [m for s in self.systems for c in s.components for m in c.maintenance],
        key=lambda m: m.time,
    )

get_schedule

get_schedule() -> Schedule

Get a fully generated schedule.

Source code in src/raplan/classes.py

def get_schedule(self) -> "Schedule":
    """Get a fully generated schedule."""
    return Schedule(
        items=[
            ScheduleItem(
                name=m.task.name,
                project=self.name,
                system=s.name,
                component=c.name,
                maintenance=m.name,
                task=m.task.name,
                rejuvenation=m.task.rejuvenation,
                duration=m.task.duration,
                cost=m.task.cost,
                time=m.time,
            )
            for s in self.systems
            for c in s.components
            for m in c.maintenance
        ]
    )

get_system

get_system(name: str) -> System

Get a component by name.

Source code in src/raplan/classes.py

def get_system(self, name: str) -> System:
    """Get a component by name."""
    for s in self.systems:
        if s.name == name:
            return s
    raise KeyError(f"System with name '{name}' does not exist in this project.")

schedule_maintenance

schedule_maintenance(
    maintenance: Maintenance,
    system: str | None = None,
    component: str | None = None,
)

Schedule maintenance for a single or all system's component or all components.

Parameters:

Name	Type	Description	Default
maintenance	Maintenance	Maintenance to schedule.	required
system	str | None	System name. If kept None, it will be applied to all.	None
component	str | None	Component name. If kept None, it will be applied to all.	None

Source code in src/raplan/classes.py

def schedule_maintenance(
    self,
    maintenance: Maintenance,
    system: str | None = None,
    component: str | None = None,
):
    """Schedule maintenance for a single or all system's component or all
    components.

    Arguments:
        maintenance: Maintenance to schedule.
        system: System name. If kept `None`, it will be applied to all.
        component: Component name. If kept `None`, it will be applied to all.
    """
    if system is None:
        for s in self.systems:
            s.schedule_maintenance(maintenance, component=component)
    else:
        self.get_system(system).schedule_maintenance(maintenance, component=component)

Schedule dataclass

Schedule(
    items: list[ScheduleItem] = field(
        default_factory=list, skip_if=_is_empty
    ),
    uuid: UUID = field(default_factory=uuid4),
)

A full maintenance schedule.

Parameters:

Name	Type	Description	Default
items	list[ScheduleItem]	Scheduled tasks.	field(default_factory=list, skip_if=_is_empty)
uuid	UUID	Automatically generated unique identifier for this maintenance schedule.	field(default_factory=uuid4)

from_projects classmethod

from_projects(projects: list[Project]) -> Schedule

Create a schedule for multiple projects.

Source code in src/raplan/classes.py

@classmethod
def from_projects(cls, projects: list[Project]) -> "Schedule":
    """Create a schedule for multiple projects."""

    schedules = [p.get_schedule() for p in projects]
    return cls(items=[i for s in schedules for i in s.items])

get_ordered_maintenance

get_ordered_maintenance() -> list[ScheduleItem]

Get all tasks ordered in time.

Source code in src/raplan/classes.py

def get_ordered_maintenance(self) -> list[ScheduleItem]:
    """Get all tasks ordered in time."""
    return sorted(self.items, key=lambda t: t.time)

ScheduleItem dataclass

ScheduleItem(
    name: str
    | None = field(default=None, skip_if_default=True),
    project: str
    | None = field(default=None, skip_if_default=True),
    system: str
    | None = field(default=None, skip_if_default=True),
    component: str
    | None = field(default=None, skip_if_default=True),
    maintenance: str
    | None = field(default=None, skip_if_default=True),
    task: str
    | None = field(default=None, skip_if_default=True),
    rejuvenation: int | float = 1.0,
    duration: int | float = 1.0,
    cost: int | float = 1.0,
    time: int | float = 1.0,
    uuid: UUID = field(default_factory=uuid4),
)

A schedule item with full detail regarding its system, component and maintenance task info.

Parameters:

Name	Type	Description	Default
name	str | None	Name for this action.	field(default=None, skip_if_default=True)
project	str | None	Name of the project this item belongs to if any.	field(default=None, skip_if_default=True)
system	str | None	Name of the system to which this maintenance is applied.	field(default=None, skip_if_default=True)
component	str | None	Name of the component to which this maintenance is applied.	field(default=None, skip_if_default=True)
maintenance	str | None	Name of the maintenance schedule this item belongs to if any.	field(default=None, skip_if_default=True)
rejuvenation	int | float	Rejuvenation factor between [0.0-1.0]. Percentage of age that is regained. Therefore, 1.0 would mean a full replacement.	1.0
duration	int | float	Duration of the maintenance. Usually in years.	1.0
cost	int | float	Cost of the maintenance. Usually expressed in a currency or equivalent.	1.0
time	int | float	Time at which this maintenance is scheduled.	1.0
uuid	UUID	Automatically generated unique identifier for this schedule item.	field(default_factory=uuid4)

System dataclass

System(
    name: str
    | None = field(default=None, skip_if_default=True),
    components: list[Component] = field(
        default_factory=list, skip_if=_is_empty, repr=False
    ),
    uuid: UUID = field(default_factory=uuid4),
)

A system consisting of multiple components.

Parameters:

Name	Type	Description	Default
name	str | None	Name of this system.	field(default=None, skip_if_default=True)
components	list[Component]	Components of this system.	field(default_factory=list, skip_if=_is_empty, repr=False)
uuid	UUID	Automatically generated unique identifier for this system.	field(default_factory=uuid4)

cfp

cfp(x: int | float = 1.0) -> float

Cumulative failure probability density function as the sum of its components' respective function incorporating maintenance.

Source code in src/raplan/classes.py

def cfp(self, x: int | float = 1.0) -> float:
    """Cumulative failure probability density function as the sum of its
    components' respective function incorporating maintenance.
    """
    if len(self.components):
        return distributions.compound_probability(c.cfp(x) for c in self.components)
    else:
        return 0.0

get_component

get_component(name: str) -> Component

Get a component by name.

Source code in src/raplan/classes.py

def get_component(self, name: str) -> Component:
    """Get a component by name."""
    for c in self.components:
        if c.name == name:
            return c
    raise KeyError(f"Component with name '{name}' does not exist in this system.")

get_ordered_maintenance

get_ordered_maintenance() -> list[Maintenance]

Get all maintenance ordered in time.

Source code in src/raplan/classes.py

def get_ordered_maintenance(self) -> list[Maintenance]:
    """Get all maintenance ordered in time."""
    return sorted([m for c in self.components for m in c.maintenance], key=lambda m: m.time)

schedule_maintenance

schedule_maintenance(
    maintenance: Maintenance, component: str | None = None
)

Schedule maintenance for a single or all system's component or all components.

Parameters:

Name	Type	Description	Default
maintenance	Maintenance	Maintenance to schedule.	required
component	str | None	Component name. If kept None, it will be applied to all.	None

Source code in src/raplan/classes.py

def schedule_maintenance(self, maintenance: Maintenance, component: str | None = None):
    """Schedule maintenance for a single or all system's component or all
    components.

    Arguments:
        maintenance: Maintenance to schedule.
        component: Component name. If kept `None`, it will be applied to all.
    """
    if component is None:
        for c in self.components:
            c.schedule_maintenance(maintenance)
    else:
        self.get_component(component).schedule_maintenance(maintenance)

Task dataclass

Task(
    name: str
    | None = field(default=None, skip_if_default=True),
    rejuvenation: int | float = 1.0,
    duration: int | float = 1.0,
    cost: int | float = 1.0,
    uuid: UUID = field(default_factory=uuid4),
)

Maintenance task to apply to a component.

Parameters:

Name	Type	Description	Default
name	str | None	Name for this action.	field(default=None, skip_if_default=True)
rejuvenation	int | float	Rejuvenation factor between [0.0-1.0]. Percentage of age that is regained. Therefore, 1.0 would mean a full replacement.	1.0
duration	int | float	Duration of the maintenance. Usually in years.	1.0
cost	int | float	Cost of the maintenance. Usually expressed in a currency or equivalent.	1.0
uuid	UUID	Automatically generated unique identifier for this task.	field(default_factory=uuid4)