raplan.milp ¶

def get_duration_segments(self, delta_x: float) -> int:
    """Calculate the number of segments for a given duration."""
    return duration_segments(delta_x=delta_x, time_segments=self.time_segments)

def get_movement_objective(self, id: UUID, delta_segments: int) -> int | float:
    """Get the movement objective value for a maintenance ID that moves this amount
    with respect to its Window's initial value."""
    dynamic: int | float | None = self.dynamic_movement_objective(
        id, self.per_segment * delta_segments
    )
    if dynamic is None:
        return self.movement_objective * self.per_segment * abs(delta_segments)
    return dynamic

def get_time_segment(self, x: float) -> int:
    """Calculate the time segment for a given time."""
    return time_segment(x, time_segments=self.time_segments)

def get_window_range(self, window: Window) -> range:
    """Calculate the time segment range for a window."""
    return window_segments(window, time_segments=self.time_segments)

@staticmethod
def build(
    builder: LpBuilder,
    items: dict[UUID, LpMoveableItem],
    candidates: dict[int, list[tuple[UUID, highs_var]]],
) -> "LpCombinations":
    """Build a set of synchronization constraints that prohibit overlap without a common
    starting segment.

    Arguments:
        builder: MILP model builder.
        items: Mapping of unique maintenance identifiers to their MILP synchronization
            descriptions.
        candidates: Mapping of segment indices to lists of pairs of maintenance identifiers and
            the scheduling booleans that may lead to occupation of the given segment.
    """

    disjoint_constraints: list[tuple[int, UUID, UUID, highs_cons]] = []
    pairs: list[tuple[int, UUID, UUID, highs_var]] = []
    pair_constraints: list[tuple[int, UUID, UUID, highs_cons]] = []
    exclusion_constraints: list[tuple[int, UUID, UUID, highs_cons]] = []

    observed: set[frozenset[UUID]] = set()
    for entries in candidates.values():
        for combi in combinations((e[0] for e in entries), r=2):
            pair: frozenset[UUID] = frozenset(combi)
            if len(pair) < 2 or pair in observed:
                continue
            observed.add(pair)

            a, b = (items[member] for member in pair)

            if builder.pair_exclusions(pair):
                exclusion_constraints.extend(a.exclude(other=b))
                exclusion_constraints.extend(b.exclude(other=a))
                continue  # because forcing sync or objective combinations become redundant.

            if builder.force_sync:
                disjoint_constraints.extend(a.disable_tail_start(other=b))
                disjoint_constraints.extend(b.disable_tail_start(other=a))

            obj: float | None = builder.pair_objective(pair)
            if obj:
                for segment in shared_range(a.window, b.window):
                    var = builder.lp.addBinary(obj=obj * builder.per_segment)
                    var_a = a.segment_vars[segment - a.window.start]
                    var_b = b.segment_vars[segment - b.window.start]
                    pairs.append(
                        (
                            segment,
                            a.uuid,
                            b.uuid,
                            var,
                        )
                    )
                    pair_constraints.append(
                        (
                            segment,
                            a.uuid,
                            b.uuid,
                            builder.lp.addConstr(2 * var <= var_a + var_b),
                        )
                    )

    return LpCombinations(
        builder=builder,
        pairs=pairs,
        pair_constraints=pair_constraints,
        disjoint_constraints=disjoint_constraints,
        exclusion_constraints=exclusion_constraints,
    )

def get_value(self, var: highs_var | highs_linear_expression) -> int | bool | float:
    """Get a highs variable or expression value."""
    value = self.lp.variableValue(var)
    return value

def run(self) -> HighsModelStatus:
    """Run the solver."""
    self.lp.solve()
    return self.model_status

def solve(self) -> HighsModelStatus:
    """Run the solver if we're not optimal."""

    if self.model_status != HighsModelStatus.kOptimal:
        return self.run()

    return self.model_status

@staticmethod
def build(
    builder: LpBuilder,
    item: Maintenance | Procedure,
) -> "LpMoveableItem":
    """Build MILP description for a `Maintenance` instance.

    Adds binary toggles for each segment the maintenance could be planned in,
    as well as a constraint that only one can be picked.

    Arguments:
        builder: MILP model builder.
        item: Maintenance instance.

    Returns:
        Linear programming representation of the maintenance instance.
    """

    if builder.only_positive:
        if item.time < 0:
            time_window: Window = Window.new_immovable(item.time)
        else:
            time_window: Window = item.window or builder.window_fallback_fn(item.time)
            if time_window.earliest < 0:
                time_window.max_rush = time_window.initial
    else:
        time_window: Window = item.window or builder.window_fallback_fn(item.time)

    initial: int = builder.get_time_segment(x=time_window.initial)
    window: range = builder.get_window_range(window=time_window)
    if isinstance(item, Procedure):
        segment_vars: list[highs_var] = [
            builder.lp.addBinary(
                obj=sum(
                    builder.get_movement_objective(m.uuid, segment - initial)
                    for m in item.maintenance
                )
            )
            for segment in window
        ]
    else:
        segment_vars: list[highs_var] = [
            builder.lp.addBinary(
                obj=builder.get_movement_objective(item.uuid, segment - initial)
            )
            for segment in window
        ]

    expr = sum(segment_vars) == 1
    assert isinstance(expr, highs_linear_expression), (
        "Expected an expression, are you sure there are segments in this window?"
    )
    picking_constraint = builder.lp.addConstr(expr)

    return LpMoveableItem(
        builder=builder,
        uuid=item.uuid,
        initial=initial,
        window=window,
        duration=builder.get_duration_segments(delta_x=builder.duration_scale * item.duration),
        segment_vars=segment_vars,
        picking_constraint=picking_constraint,
    )

def disable_tail_start(
    self, other: "LpMoveableItem"
) -> Generator[tuple[int, UUID, UUID, highs_cons], None, None]:
    """Disable starting any maintenance in the tail segments of this maintenance.

    Yields:
        Tuple of segment index, own UUID, other UUID and a disabling constraint.
    """

    for segment, toggle in zip(self.window, self.segment_vars):
        # The disabled range is one past A up until it's duration.
        # The index in B's variable list is offset by it's window start.
        tail = slice(
            max(0, segment + 1 - other.window.start),
            segment + self.duration - other.window.start,
        )
        disabled: list[highs_var] = other.segment_vars[tail]

        # Disabled segments could be empty.
        if disabled:
            yield (
                segment,
                self.uuid,
                other.uuid,
                self.lp.addConstr(sum(disabled) <= (1 - toggle)),  # type: ignore
            )

def exclude(
    self, other: "LpMoveableItem"
) -> Generator[tuple[int, UUID, UUID, highs_cons], None, None]:
    """Disallows the other maintenance item to start during this item.

    Yields:
        Tuple of a starting segment, this item's ID, disabled item's ID and constraint.
    """
    for segment, toggle in zip(self.window, self.segment_vars):
        to_disable = slice(
            max(0, segment - other.window.start),
            segment + self.duration - other.window.start,
        )
        disabled: list[highs_var] = other.segment_vars[to_disable]

        # Disabled segments could be empty.
        if disabled:
            yield (
                segment,
                self.uuid,
                other.uuid,
                self.lp.addConstr(sum(disabled) <= (1 - toggle)),  # type: ignore
            )

def get_value(self, var: highs_var | highs_linear_expression) -> int | bool | float:
    """Get a highs variable or expression value."""
    value = self.lp.variableValue(var)
    return value

def is_picked_valid(self) -> bool:
    """Check whether the picking constraint has been satisfied."""
    value = self.get_value(self.picking_constraint.expr())
    return bool(value)

def picked_segment(self) -> int:
    """The currently picked segment for this maintenance instance."""
    try:
        picked: int | None = next(
            segment
            for segment, var in zip(self.window, self.segment_vars)
            if self.get_value(var=var)
        )
        return self.initial if picked is None else picked
    except StopIteration:
        return self.initial

def run(self) -> HighsModelStatus:
    """Run the solver."""
    self.lp.solve()
    return self.model_status

def solve(self) -> HighsModelStatus:
    """Run the solver if we're not optimal."""

    if self.model_status != HighsModelStatus.kOptimal:
        return self.run()

    return self.model_status

def apply_to(self, project: Project, new: bool = True) -> Project:
    """Apply rescheduled maintenance results to this project.

    Arguments:
        project: Project to reschedule maintenance for.
        new: Whether to create a new deepcopy of this project.

    Returns:
        Project with updated maintenance.
    """
    if new:
        project = deepcopy(project)

    procedures: dict[UUID, Procedure] = {proc.uuid: proc for proc in project.gen_procedures()}

    for lproc in self.procedures:
        proc: Procedure = procedures[lproc.uuid]
        new_time: int | float = self.builder.per_segment * lproc.picked_segment()
        proc.reschedule(new_time, PROC_REASON)

    return project

@staticmethod
def build(builder: LpBuilder, project: Project) -> "LpProcedureProject":
    """Build a MILP description for a procedure planning project.

    Constraints can be added using the instance methods on `LpSyncProject` itself.

    Arguments:
        project: Project to encode as a procedure optimization problem.

    Returns:
        MILP description for a procedure planning project.
    """

    procedures: dict[UUID, LpMoveableItem] = {
        procedure.uuid: LpMoveableItem.build(builder, procedure)
        for procedure in project.gen_procedures()
    }

    # Create a mapping per segment to all variables that would lead to segment occupation.
    candidates: defaultdict[int, list[tuple[UUID, highs_var]]] = defaultdict(list)
    for p in procedures.values():
        # Add segment occupation.
        for segment, toggle in zip(p.window, p.segment_vars):
            for offset in range(p.duration):
                occupation_by_segment: list[tuple[UUID, highs_var]] = candidates[
                    segment + offset
                ]
                occupation_by_segment.append((p.uuid, toggle))

    # Build segment descriptions with variables and constraints regarding the amount of
    # parallel execution. Can be skipped if the objective is 0 regardless.
    if builder.concurrent_objective != 0:
        occupation: list[LpSegmentOccupation] = [
            LpSegmentOccupation.build(builder, segment=segment, candidates=candidates)
            for segment, candidates in candidates.items()
        ]
    else:
        occupation: list[LpSegmentOccupation] = []

    # For the given system maintenance and segment occupation, calculate the potentially
    # overlapping pairs and constrain them such that they either start synchronously or don't
    # overlap at all.
    combinations = LpCombinations.build(
        builder=builder,
        items=procedures,
        candidates=candidates,
    )

    return LpProcedureProject(
        builder=builder,
        uuid=project.uuid,
        procedures=list(procedures.values()),
        occupation=occupation,
        combinations=combinations,
    )

def get_value(self, var: highs_var | highs_linear_expression) -> int | bool | float:
    """Get a highs variable or expression value."""
    value = self.lp.variableValue(var)
    return value

def run(self) -> HighsModelStatus:
    """Run the solver."""
    self.lp.solve()
    return self.model_status

def solve(self) -> HighsModelStatus:
    """Run the solver if we're not optimal."""

    if self.model_status != HighsModelStatus.kOptimal:
        return self.run()

    return self.model_status

@staticmethod
def build(
    builder: LpBuilder, segment: int, candidates: list[tuple[UUID, highs_var]]
) -> "LpSegmentOccupation":
    """Build a MILP description for a segment where system maintenance may be
    planned in. This method ties the potential benefits of concurrent task execution during a
    segment to the start time of the candidate tasks.
    """
    upper: int = len(candidates)

    occupation: highs_var | Literal[0] = sum(c[1] for c in candidates)
    concurrent: highs_var = builder.lp.addIntegral(
        ub=len(candidates), obj=builder.per_segment * builder.concurrent_objective
    )
    is_occupied: highs_var = builder.lp.addBinary()

    # Constraint that states that `is_occupied` should turn true when there is occupation.
    occupation_constraint: highs_cons = builder.lp.addConstr(occupation <= is_occupied * upper)  # type: ignore

    # Concurrency equals occupation minus one except when unoccupied.
    concurrent_constraint: highs_cons = builder.lp.addConstr(
        concurrent == occupation - is_occupied
    )

    return LpSegmentOccupation(
        builder=builder,
        segment=segment,
        occupation=occupation,
        concurrent=concurrent,
        is_occupied=is_occupied,
        occupation_constraint=occupation_constraint,
        concurrent_constraint=concurrent_constraint,
    )

def get_value(self, var: highs_var | highs_linear_expression) -> int | bool | float:
    """Get a highs variable or expression value."""
    value = self.lp.variableValue(var)
    return value

def run(self) -> HighsModelStatus:
    """Run the solver."""
    self.lp.solve()
    return self.model_status

def solve(self) -> HighsModelStatus:
    """Run the solver if we're not optimal."""

    if self.model_status != HighsModelStatus.kOptimal:
        return self.run()

    return self.model_status

def apply_to(self, component: Component, new: bool = True) -> Component:
    """Apply rescheduled maintenance results to this component.

    Arguments:
        component: Component to reschedule maintenance for.
        new: Whether to create a new deepcopy of this component.

    Returns:
        Component with updated maintenance.
    """
    if new:
        component = deepcopy(component)
    data: dict[UUID, float] = {k: v for k, v in self.gen_maintenance_times()}
    component.reschedule_maintenance(data=data, reason=SYNC_REASON)
    return component

@staticmethod
def build(builder: LpBuilder, component: Component) -> "LpSyncComponent":
    """Build a linear programming problem for a `Component` instance.

    Arguments:
        component: Component instance.
    """

    return LpSyncComponent(
        builder=builder,
        uuid=component.uuid,
        maintenance=[
            LpMoveableItem.build(builder, maintenance)
            for maintenance in component.maintenance
            if not builder.only_positive or maintenance.end >= 0
        ],
    )

def gen_maintenance_segments(self) -> Generator[tuple[UUID, int], Never, None]:
    """Generate all allocated maintenance segments for this component."""
    for m in self.maintenance:
        yield (m.uuid, m.picked_segment())

def gen_maintenance_times(self) -> Generator[tuple[UUID, int | float], Never, None]:
    """Generate all maintenance times for this component."""
    for m in self.maintenance:
        yield (m.uuid, m.picked_time())

def get_value(self, var: highs_var | highs_linear_expression) -> int | bool | float:
    """Get a highs variable or expression value."""
    value = self.lp.variableValue(var)
    return value

def run(self) -> HighsModelStatus:
    """Run the solver."""
    self.lp.solve()
    return self.model_status

def solve(self) -> HighsModelStatus:
    """Run the solver if we're not optimal."""

    if self.model_status != HighsModelStatus.kOptimal:
        return self.run()

    return self.model_status

def apply_to(self, project: Project, new: bool = True) -> Project:
    """Apply rescheduled maintenance results to this project.

    Arguments:
        project: Project to reschedule maintenance for.
        new: Whether to create a new deepcopy of this project.

    Returns:
        Project with updated maintenance.
    """
    if new:
        project = deepcopy(project)
    data: dict[UUID, float] = {k: v for k, v in self.gen_maintenance_times()}
    project.reschedule_maintenance(data=data, reason=SYNC_REASON)
    return project

@staticmethod
def build(builder: LpBuilder, project: Project) -> "LpSyncProject":
    """Build a linear programming problem for a `Project`.

    Arguments:
        project: `Project` instance.
    """
    return LpSyncProject(
        builder=builder,
        uuid=project.uuid,
        systems=[LpSyncSystem.build(builder, system) for system in project.systems],
    )