Optymalizacja cięcia belek

main.py

import time
import json
import enum
import csv
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
from itertools import chain
from pathlib import Path

from ortools.sat.python import cp_model

from qtpy import QtWidgets as qtw
from qtpy import QtGui as qtg
from qtpy import QtCore as qtc
from qtpy.QtCore import Qt


DECIMALS = 3
LENGTH_DEFAULT = 1
LENGTH_MIN = 0.01
LENGTH_MAX = 100

QUANTITY_DEFAULT = 1
QUANTITY_MIN = 1
QUANTITY_MAX = 100

SAVE_PATH = Path.home() / ".pglomski/optymalizacja_ciecia/optymalizacja.json"
SAVE_USEFUL_WASTE = "useful_waste"


class SolutionCollector(cp_model.CpSolverSolutionCallback):
    def __init__(self, model: cp_model.CpModel, objective_vars: dict):
        cp_model.CpSolverSolutionCallback.__init__(self)
        self._model = model
        self._objective_vars = objective_vars
        self.solutions = []

    def on_solution_callback(self):
        objective_value = self.ObjectiveValue()

        solution = {}
        for idx, var in self._objective_vars.items():
            solution[idx] = self.Value(var)
        self.solutions.append((objective_value, solution))


@dataclass
class QuantityDesc:
    ordered: int
    extra: int

    @property
    def combined(self) -> int:
        return self.ordered + self.extra


def sat(
    elements: dict[float, QuantityDesc],
    bins: list[float],
    cut_width_mm: int,
    max_time: float,
) -> list[list[tuple[float, int]]]:
    elements: dict[int, QuantityDesc] = {
        round(k * (10**DECIMALS)): v for k, v in elements.items() if v.combined > 0
    }
    bins = [round(length * (10**DECIMALS)) for length in bins]

    bin_indices = range(len(bins))

    model = cp_model.CpModel()

    # Variables
    # x[b] = 0 or 1      --->   1 if `bins[b]` is used
    # x[length, b] = n   --->   `length` is packed `n` times in `bins[b]`
    x = {}
    for length, quantity in elements.items():
        for b in bin_indices:
            x[b] = model.NewIntVar(0, 1, f"x_{b}")
            # Preprocessing to avoid creating variables for impossible item-bin combinations
            if length <= bins[b]:
                x[length, b] = model.NewIntVar(0, quantity.combined, f"x_{length}_{b}")

    # Constraints
    # Ensure we do not pack more of each item than we have
    for length, quantity in elements.items():
        count_from_all_beams = sum(x[length, b] for b in bin_indices if (length, b) in x)
        if quantity.extra == 0:
            model.Add(count_from_all_beams == quantity.ordered)
        elif quantity.ordered != 0:
            model.Add(count_from_all_beams >= quantity.ordered)
            model.Add(count_from_all_beams <= quantity.combined)

    # The amount packed in each bin cannot exceed its capacity
    waste_list = []
    for b in bin_indices:
        number_of_items = sum(x[length, b] for length in elements.keys() if (length, b) in x)
        packed_length = sum(x[length, b] * length for length in elements if (length, b) in x)
        waste = x[b] * bins[b] - packed_length
        model.Add(waste >= cut_width_mm * (number_of_items - 1))

        waste_list.append(waste)

    # Objective
    # minimize the "useless waste"
    model.Minimize(sum(waste_list))

    solver = cp_model.CpSolver()
    solver.parameters.max_time_in_seconds = max_time
    collector = SolutionCollector(model, objective_vars=x)
    status = solver.Solve(model, collector)

    solution = []
    if status in [cp_model.OPTIMAL, cp_model.FEASIBLE]:
        for b in bin_indices:
            solution.append(
                [
                    (length * (10**-DECIMALS), count)
                    for length in elements.keys()
                    if (length, b) in x and (count := solver.Value(x[length, b])) > 0
                ]
            )
    return solution


def validate_float(text: str, default: float, min_val: float, max_val: float) -> float:
    try:
        return round(min(max(float(text), min_val), max_val), DECIMALS)
    except ValueError:
        return default


def validate_int(text: str, default: int, min_val: int, max_val: int) -> int:
    try:
        return min(max(round(float(text)), min_val), max_val)
    except ValueError:
        return default


class TextWidget(qtw.QWidget):
    def __init__(self, text: str, *, font_size_offset: float = 0, bold: bool = False):
        super().__init__()

        font = self.font()
        font.setBold(bold)
        font.setPointSizeF(font.pointSizeF() + font_size_offset)

        self._label = qtw.QLabel(text)
        self._label.setFont(font)
        self._label.setWordWrap(True)

        layout = qtw.QHBoxLayout(self)
        layout.addWidget(self._label, 0, Qt.AlignmentFlag.AlignCenter)

    @property
    def text(self) -> str:
        return self._label.text()

    def set_text(self, text: str) -> None:
        self._label.setText(text)

    def resizeEvent(self, event: qtg.QResizeEvent) -> None:
        return super().resizeEvent(event)


class ItemsEntry(qtw.QFrame):
    class Column(enum.IntEnum):
        LENGTH = 0
        QUANTITY = enum.auto()
        REMOVE = enum.auto()
        COLUMNS_NUM = enum.auto()

    COLUMN_NAME = {
        Column.LENGTH: "Długość",
        Column.QUANTITY: "Ilość",
        Column.REMOVE: "Usuń",
    }

    def __init__(self, title: str):
        super().__init__()
        self.setMinimumSize(240, 120)
        self.setFrameShape(qtw.QFrame.Shape.StyledPanel)

        add_button = qtw.QPushButton("+")
        add_button.clicked.connect(self.add_new_item)

        top_layout = qtw.QHBoxLayout()
        top_layout.setAlignment(Qt.AlignmentFlag.AlignTop)
        top_layout.addWidget(TextWidget(title))
        top_layout.addStretch()
        top_layout.addWidget(add_button)

        self._table = qtw.QTableWidget(0, 3)
        # self._table.setSizePolicy(
        #     qtw.QSizePolicy.Policy.Expanding,
        #     qtw.QSizePolicy.Policy.Expanding,
        # )
        self._table.setSizeAdjustPolicy(qtw.QAbstractScrollArea.SizeAdjustPolicy.AdjustIgnored)
        # self._table.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOn)
        self._table.setAlternatingRowColors(True)
        # self._table.setStyleSheet("QTableWidget::item { padding: 3px }")
        self._table.setHorizontalHeaderLabels(ItemsEntry.COLUMN_NAME.values())
        self._table.horizontalHeader().setSectionResizeMode(0, qtw.QHeaderView.ResizeMode.Stretch)
        self._table.horizontalHeader().setSectionResizeMode(1, qtw.QHeaderView.ResizeMode.Stretch)
        self._table.horizontalHeader().setSectionResizeMode(
            2, qtw.QHeaderView.ResizeMode.ResizeToContents
        )

        main_layout = qtw.QVBoxLayout(self)
        main_layout.addLayout(top_layout)
        main_layout.addWidget(self._table)  # , 1, Qt.AlignmentFlag.AlignTop)

        self.setMaximumWidth(self.sizeHint().width())

    def add_new_item(
        self, length: float = LENGTH_DEFAULT, quantity: int = QUANTITY_DEFAULT
    ) -> None:
        row = self._table.rowCount()
        self._table.insertRow(row)

        remove_button = qtw.QPushButton("-")
        remove_button.setFixedSize(x := min(remove_button.sizeHint().toTuple()), x)
        # remove_button.setSizePolicy(qtw.QSizePolicy.Minimum, qtw.QSizePolicy.Minimum)

        remove_button_widget = qtw.QWidget()
        remove_button_widget_layout = qtw.QHBoxLayout(remove_button_widget)
        remove_button_widget_layout.setContentsMargins(0, 0, 0, 0)
        remove_button_widget_layout.addWidget(remove_button, 0, Qt.AlignmentFlag.AlignCenter)
        remove_button.clicked.connect(
            lambda: self._table.removeRow(
                self._table.indexAt(remove_button_widget.geometry().center()).row()
            )
        )

        def validate(item: qtw.QTableWidgetItem):
            if item.column() == 0:
                value = validate_float(item.text(), LENGTH_DEFAULT, LENGTH_MIN, LENGTH_MAX)
                item.setText(f"{value:.{DECIMALS}f}")
            elif item.column() == 1:
                value = validate_int(item.text(), QUANTITY_DEFAULT, QUANTITY_MIN, QUANTITY_MAX)
                item.setText(f"{value}")

        self._table.itemChanged.connect(validate)

        length_item = qtw.QTableWidgetItem(f"{length}")
        length_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)

        quantity_item = qtw.QTableWidgetItem(f"{quantity}")
        quantity_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)

        self._table.setItem(row, ItemsEntry.Column.LENGTH, length_item)
        self._table.setItem(row, ItemsEntry.Column.QUANTITY, quantity_item)
        self._table.setCellWidget(row, ItemsEntry.Column.REMOVE, remove_button_widget)

    def load(self, save_dict: list[dict[int, str]]) -> None:
        for _ in range(self._table.rowCount()):
            self._table.removeRow(0)
        for row_dict in save_dict:
            self.add_new_item(
                row_dict.get(ItemsEntry.Column.LENGTH.name, LENGTH_DEFAULT),
                row_dict.get(ItemsEntry.Column.QUANTITY.name, QUANTITY_DEFAULT),
            )

    def save(self) -> list[dict[int, str]]:
        save = self.collect()
        for row in save:
            with_names = {column.name: value for column, value in row.items()}
            row.clear()
            row.update(with_names)
        return save

    def collect(self) -> list[dict[Column, str]]:
        save = []
        for row in range(self._table.rowCount()):
            length_item = self._table.item(row, ItemsEntry.Column.LENGTH)
            quantity_item = self._table.item(row, ItemsEntry.Column.QUANTITY)
            save.append(
                {
                    ItemsEntry.Column.LENGTH: length_item.text(),
                    ItemsEntry.Column.QUANTITY: quantity_item.text(),
                }
            )
        return save


class O12N(qtw.QFrame):
    class Column(enum.IntEnum):
        BEAM = 0
        ORDERED = enum.auto()
        WASTE = enum.auto()
        WASTE_USEFUL = enum.auto()
        WASTE_USELESS = enum.auto()
        COLUMNS_NUM = enum.auto()

    COLUMN_NAME = {
        Column.BEAM: "Belka",
        Column.ORDERED: "Zamówione",
        Column.WASTE: "Odpad",
        Column.WASTE_USEFUL: "Odpad użyteczny",
        Column.WASTE_USELESS: "Odpad bezużyteczny",
    }

    def __init__(self):
        super().__init__()
        self.setMinimumSize(120, 120)
        self.setFrameShape(qtw.QFrame.Shape.StyledPanel)

        self._run_button = qtw.QPushButton("Przelicz")
        self._max_time_input = qtw.QLineEdit("4.00")
        self._max_time_input.setMaximumWidth(
            qtg.QFontMetrics(self._max_time_input.font()).horizontalAdvance("3600.00")
        )
        # self._max_time_input.textEdited.connect(
        #     lambda: self._max_time_input.setText(
        #         str(validate_float(self._max_time_input.text(), default=4, min_val=0, max_val=3600))
        #     )
        # )

        self._cut_width = qtw.QLineEdit("7")
        self._cut_width.setMaximumWidth(
            qtg.QFontMetrics(self._cut_width.font()).horizontalAdvance("999")
        )
        # self._cut_width.textEdited.connect(
        #     lambda: self._cut_width.setText(
        #         str(validate_int(self._cut_width.text(), default=7, min_val=0, max_val=99))
        #     )
        # )

        top_layout = qtw.QHBoxLayout()
        top_layout.addWidget(TextWidget("Propozycja cięć"))
        top_layout.addStretch()
        top_layout.addWidget(TextWidget("Rzaz [mm]"))
        top_layout.addWidget(self._cut_width)
        top_layout.addWidget(TextWidget("Czas obliczeń"))
        top_layout.addWidget(self._max_time_input)
        top_layout.addWidget(self._run_button)

        self._table = qtw.QTableWidget(0, O12N.Column.COLUMNS_NUM)
        self._table.setSizeAdjustPolicy(qtw.QAbstractScrollArea.SizeAdjustPolicy.AdjustIgnored)
        # self._table.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOn)
        self._table.setAlternatingRowColors(True)
        # self._table.setStyleSheet("QTableWidget::item { padding: 3px }")
        self._table.setHorizontalHeaderLabels(O12N.COLUMN_NAME.values())
        self._table.verticalHeader().setSectionResizeMode(
            qtw.QHeaderView.ResizeMode.ResizeToContents
        )
        self._table.horizontalHeader().setSectionResizeMode(
            qtw.QHeaderView.ResizeMode.ResizeToContents
        )
        self._table.setMinimumWidth(self._table.horizontalHeader().width())
        self._table.horizontalHeader().setSectionResizeMode(qtw.QHeaderView.ResizeMode.Stretch)

        save_output_button = qtw.QPushButton("Zapisz")
        save_output_button.clicked.connect(self._save_table_values)

        main_layout = qtw.QVBoxLayout(self)
        main_layout.addLayout(top_layout)
        main_layout.addWidget(self._table)
        main_layout.addWidget(save_output_button)

    @property
    def optimize_event(self) -> qtc.SignalInstance:
        return self._run_button.clicked

    @property
    def cut_width_mm(self) -> int:
        return int(self._cut_width.text())

    @property
    def max_time(self) -> float:
        return float(self._max_time_input.text())

    def _save_table_values(self) -> None:
        with open("optymalizacja.csv", "w", encoding="utf-8") as file:
            writer = csv.writer(file, quoting=csv.QUOTE_MINIMAL)
            writer.writerow(
                O12N.COLUMN_NAME[column]
                for column in O12N.Column
                if column != O12N.Column.COLUMNS_NUM
            )

            for row_idx in range(self._table.rowCount()):
                writer.writerow(
                    self._table.cellWidget(row_idx, col_idx).text
                    for col_idx in range(self._table.columnCount())
                )

    def clear(self) -> None:
        for _ in range(self._table.rowCount()):
            self._table.removeRow(0)

    def add_item(
        self,
        beam_length: float,
        order_lengths_quantities: list[tuple[float, int]],
        waste_lengths_quantities: list[tuple[float, int]],
    ):
        if self._table.rowCount():
            row = self._table.rowCount() - 1
        else:
            row = 0
            self._table.insertRow(row)

        self._table.setCellWidget(row, O12N.Column.BEAM, TextWidget(f"{beam_length} m"))
        self._table.setCellWidget(
            row,
            O12N.Column.ORDERED,
            TextWidget(
                "\n".join(
                    f"{quantity} szt. x {length} m" for length, quantity in order_lengths_quantities
                )
            ),
        )
        self._table.setCellWidget(
            row,
            O12N.Column.WASTE_USEFUL,
            TextWidget(
                "\n".join(
                    f"{quantity} szt. x {length} m" for length, quantity in waste_lengths_quantities
                )
            ),
        )

        order_len = sum(length * quantity for length, quantity in order_lengths_quantities)
        waste_len = sum(length * quantity for length, quantity in waste_lengths_quantities)
        useless_waste = beam_length - (order_len + waste_len)
        all_waste = waste_len + useless_waste
        self._table.setCellWidget(
            row, O12N.Column.WASTE_USELESS, TextWidget(f"{useless_waste:.{DECIMALS}f} m")
        )
        self._table.setCellWidget(row, O12N.Column.WASTE, TextWidget(f"{all_waste:.{DECIMALS}f} m"))
        self._table.resizeRowToContents(row)

        # add the sum of all the useless waste

        all_useless_waste_sum = sum(
            float(self._table.cellWidget(i, O12N.Column.WASTE_USELESS).text[:-1])
            for i in range(self._table.rowCount())
        )

        self._table.insertRow(row + 1)
        for column in O12N.Column:
            text = "-"
            if column == O12N.Column.WASTE_USELESS:
                text = f"Suma {all_useless_waste_sum:.{DECIMALS}f} m"
            self._table.setCellWidget(row + 1, column, TextWidget(text))
        self._table.resizeRowToContents(row + 1)


class TimeProgressDialog(qtw.QDialog):
    update_event = qtc.Signal(float)

    def __init__(
        self,
        parent: qtw.QWidget | None,
        min_size: tuple[int, int],
        title: str,
        flags: Qt.WindowType = Qt.WindowType.Dialog,
    ):
        super().__init__(parent, flags)
        self.setMinimumSize(*min_size)
        self.setWindowTitle(title)
        self.setWindowModality(Qt.WindowModality.WindowModal)

        self._progress_bar = qtw.QProgressBar()
        self._progress_bar.setMinimum(0)
        self._progress_bar.setMaximum(1000)

        layout = qtw.QVBoxLayout(self)
        layout.addWidget(self._progress_bar, 1, Qt.AlignmentFlag.AlignTop)

        def update(progress: float):
            self._progress_bar.setValue(progress * self._progress_bar.maximum())
            if progress >= 1:
                self.close()

        self.update_event.connect(update)

    def open(self, execution_time: float) -> None:
        start_time = time.time()

        def task():
            progress = (time.time() - start_time) / execution_time
            if progress < 1:
                self.update_event.emit(progress)
            else:
                self.close()

        timer = qtc.QTimer()
        timer.setInterval(17)
        timer.timeout.connect(task)
        timer.start()
        self.exec()
        timer.stop()


class MainWindow(qtw.QMainWindow):
    def __init__(self, save_dict: dict):
        super().__init__()
        self.setWindowTitle("Optymalizacja cięcia belek")

        self.beams_entry = ItemsEntry("Dostępne belki")
        self.order_entry = ItemsEntry("Zamówienie")
        self.waste_entry = ItemsEntry("Przydatne odpady")
        self.optimalization = O12N()
        self.optimalization.optimize_event.connect(self.optimize)

        if save_dict.get(SAVE_USEFUL_WASTE):
            self.waste_entry.load(save_dict.get(SAVE_USEFUL_WASTE))

        central_widget = qtw.QWidget()
        horizontal_layout = qtw.QHBoxLayout(central_widget)
        horizontal_layout.addWidget(self.beams_entry, 1)
        horizontal_layout.addWidget(self.order_entry, 1)
        horizontal_layout.addWidget(self.waste_entry, 1)
        horizontal_layout.addWidget(self.optimalization, 10)
        self.setCentralWidget(central_widget)

        self._msg_box = qtw.QMessageBox()
        self._progress_dialog = TimeProgressDialog(
            self, (320, 160), "Poszukiwanie rozwiązania w toku"
        )

        self.setMinimumSize(self.sizeHint())

    def optimize(self):
        self.optimalization.clear()

        elements = dict[float, QuantityDesc]()
        for row in self.order_entry.collect():
            length = float(row[ItemsEntry.Column.LENGTH])
            quantity = int(row[ItemsEntry.Column.QUANTITY])
            element = elements.setdefault(length, QuantityDesc(0, 0))
            element.ordered += quantity

        for row in self.waste_entry.collect():
            length = float(row[ItemsEntry.Column.LENGTH])
            quantity = int(row[ItemsEntry.Column.QUANTITY])
            element = elements.setdefault(length, QuantityDesc(0, 0))
            element.extra += quantity

        beam_lengths = []
        for row in self.beams_entry.collect():
            length = float(row[ItemsEntry.Column.LENGTH])
            quantity = int(row[ItemsEntry.Column.QUANTITY])
            beam_lengths += [length] * quantity

        max_time = self.optimalization.max_time

        def task():
            solution = sat(
                elements=elements,
                bins=beam_lengths,
                cut_width_mm=self.optimalization.cut_width_mm,
                max_time=max_time,
            )
            self._progress_dialog.update_event.emit(1.0)
            return solution

        with ThreadPoolExecutor() as executor:
            future = executor.submit(task)

            self._progress_dialog.open(max_time)

            try:
                solution = future.result()
                for length, quantity in elements.items():
                    count_sum = sum(
                        count for b in solution for s_length, count in b if s_length == length
                    )
                    assert count_sum == quantity.ordered, "Zbyt mało belek"
            except Exception as exception:
                self._msg_box.warning(
                    self,
                    "Błąd obliczeń",
                    str(exception),
                )
                return

            for beam_idx, beam_items in enumerate(solution):
                ordered_solution = []
                extra_solution = []
                for length, count in beam_items:
                    ordered_quantity = min(count, elements[length].ordered)
                    elements[length].ordered -= ordered_quantity
                    count -= ordered_quantity

                    extra_quantity = min(count, elements[length].extra)
                    elements[length].extra -= extra_quantity

                    if ordered_quantity > 0:
                        ordered_solution.append((length, ordered_quantity))
                    if extra_quantity > 0:
                        extra_solution.append((length, extra_quantity))

                if ordered_solution or extra_solution:
                    self.optimalization.add_item(
                        beam_length=beam_lengths[beam_idx],
                        order_lengths_quantities=ordered_solution,
                        waste_lengths_quantities=extra_solution,
                    )

    def save(self) -> dict:
        return {SAVE_USEFUL_WASTE: self.waste_entry.save()}


def main() -> None:
    Path.mkdir(SAVE_PATH.parent, parents=True, exist_ok=True)
    with open(SAVE_PATH, "r+" if Path.exists(SAVE_PATH) else "a+", encoding="utf-8") as save_file:
        try:
            save_dict = json.load(save_file)
        except json.JSONDecodeError:
            save_dict = {}

        app = qtw.QApplication()
        main_window = MainWindow(save_dict)
        main_window.show()
        app.exec_()

        save_file.seek(0)
        json.dump(main_window.save(), save_file, indent=True)
        save_file.truncate()


if __name__ == "__main__":
    main()

req.txt

qtpy
pyside6
ortools