raplan.classes ¶

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

def gen_maintenance(self) -> Generator[Maintenance, None, None]:
    """Yield all component maintenance items from a generator (unordered)."""
    yield from self.maintenance

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_moment = 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.
        moment = min(m.end, x)
        current_age = age + moment - last_moment
        rejuvenation = m.get_progress(x) * m.task.rejuvenation
        age = current_age * (1.0 - rejuvenation)
        last_moment = moment
    # Add remaining time since last maintenance.
    age += x - last_moment
    return age

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

def get_unordered_maintenance(self) -> list[Maintenance]:
    """List of maintenance as-is."""
    return self.maintenance

def reschedule_maintenance(self, data: dict[UUID, int | float], reason: str | None) -> None:
    """Reschedule component maintenance based on the input data.

    Arguments:
        data: Dictionary from maintenance UUIDs to assigned times.
    """
    for m in self.maintenance:
        m.reschedule_from(data=data, reason=reason)

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)

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)

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: int | float = horizon.end - horizon.start if horizon.end else cycle_length

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

    maintenance: list[Maintenance] = []
    cycles_offset: int | float = 0
    while start + cycles_offset < end:
        for index, (offset, task) in enumerate(tasks):
            time: float | int = 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: str = f"{prefix} | " if prefix else ""
            _name: str = 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

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)]

def is_satisfied_by(self, x: int | float) -> bool:
    """Whether the given time falls within this horizon."""
    if x < self.start:
        return False
    if self.end is not None and x > self.end:
        return False
    return True

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)

def reschedule_from(self, data: dict[UUID, int | float], reason: str | None) -> None:
    """Reschedule maintenance based on the input data.

    Arguments:
        data: Dictionary from maintenance UUIDs to assigned times.
    """
    new = data.get(self.uuid)
    if new is None or new == self.time:
        return

    self.time = new
    self.rescheduling.append(Rescheduling(time=new, reason=reason))

def apply_to(
    self, target: Project | System | Component | Maintenance, reason: str | None = None
) -> None:
    """Apply this procedure's schedule (maintenance timing) to a target."""
    data: dict[UUID, int | float] = {m.uuid: m.time for m in self.maintenance}
    if isinstance(target, Maintenance):
        target.reschedule_from(data=data, reason=reason or self.name)
    else:
        target.reschedule_maintenance(data=data, reason=reason or self.name)

def reschedule(self, time: int | float, reason: str | None = None):
    """Reschedule this procedure's starting time."""
    delta = time - self.time
    for m in self.maintenance:
        m.time += delta
        m.rescheduling.append(Rescheduling(m.time, reason=reason))

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

def gen_maintenance(self) -> Generator[Maintenance, None, None]:
    """Yield all component maintenance items from a generator (unordered)."""
    for s in self.systems:
        yield from s.gen_maintenance()

def gen_procedures(
    self,
    prefix: str | None = None,
    uuid_fn: Callable[[list[Maintenance]], UUID] = with_uuid4s_to_uuid5,
) -> Generator["Procedure"]:
    """Generate procedures from the current maintenance schedule."""
    for s in self.systems:
        yield from s.gen_procedures(prefix=prefix, uuid_fn=uuid_fn)

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

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

def get_procedures(
    self,
    uuid_fn: Callable[[list[Maintenance]], UUID] = with_uuid4s_to_uuid5,
) -> list["Procedure"]:
    """Get the procedures per system per time."""
    return list(self.gen_procedures(uuid_fn=uuid_fn))

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
        ]
    )

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.")

def get_unordered_maintenance(self) -> list[Maintenance]:
    """Get all maintenance per system per component."""
    return list(self.gen_maintenance())

def reschedule_maintenance(self, data: dict[UUID, int | float], reason: str | None) -> None:
    """Reschedule system maintenance based on the input data.

    Arguments:
        data: Dictionary from maintenance UUIDs to assigned times.
    """
    for s in self.systems:
        s.reschedule_maintenance(data=data, reason=reason)

def schedule_maintenance(
    self,
    maintenance: Maintenance,
    system: str | None = None,
    component: str | None = 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)

@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])

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

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

def gen_maintenance(self) -> Generator[Maintenance, None, None]:
    """Yield all component maintenance items from a generator (unordered)."""
    for c in self.components:
        yield from c.maintenance

def gen_procedures(
    self,
    prefix: str | None = None,
    uuid_fn: Callable[[list[Maintenance]], UUID] = with_uuid4s_to_uuid5,
) -> Generator["Procedure", None, None]:
    """Generate procedures from the current maintenance schedule."""
    system: str = self.name or str(self.uuid)
    _prefix: str = f"{system}" if prefix is None else f"{prefix} | {system}"
    idx = 1
    proc = Procedure(name=f"{_prefix} | {idx}", maintenance=[])

    for m in self.get_ordered_maintenance():
        if not proc.maintenance or m.time < proc.end:
            proc.maintenance.append(m)
        else:
            proc.uuid = uuid_fn(proc.maintenance)
            yield proc
            idx += 1
            proc = Procedure(name=f"{_prefix} | {idx}", maintenance=[m])

    if proc.maintenance:
        proc.uuid = uuid_fn(proc.maintenance)
        yield proc

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.")

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

def get_procedures(
    self,
    prefix: str | None = None,
    uuid_fn: Callable[[list[Maintenance]], UUID] = with_uuid4s_to_uuid5,
) -> list["Procedure"]:
    """Get a list of procedures per system per time."""
    return list(self.gen_procedures(prefix=prefix, uuid_fn=uuid_fn))

def get_unordered_maintenance(self) -> list[Maintenance]:
    """Get all maintenance per component."""
    return list(self.gen_maintenance())

def reschedule_maintenance(self, data: dict[UUID, int | float], reason: str | None) -> None:
    """Reschedule system maintenance based on the input data.

    Arguments:
        data: Dictionary from maintenance UUIDs to assigned times.
    """
    for c in self.components:
        c.reschedule_maintenance(data=data, reason=reason)

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)

def is_satisfied_by(self, x: int | float) -> bool:
    """Whether this window is satisfied by the given time."""
    if self.latest < x or x < self.earliest:
        return False
    return True