PySCIPOpt/tests/test_recipe_realtime_trace_jsonl.py at 8edf0cc2f3d632119f9d819113533d1703bf1880 · scipopt/PySCIPOpt · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import json

import pytest
from helpers.utils import bin_packing_model

from pyscipopt import SCIP_EVENTTYPE, Eventhdlr
from pyscipopt.recipes.realtime_trace_jsonl import optimizeNogilTrace, optimizeTrace


@pytest.fixture(
    params=[optimizeTrace, optimizeNogilTrace], ids=["optimize", "optimize_nogil"]
)
def optimize(request):
    return request.param


def test_realtime_trace_in_memory(optimize):
    sizes = list(range(1, 41)) * 3
    model = bin_packing_model(sizes=sizes, capacity=50)
    model.setParam("limits/time", 5)

    model.data = {"test": True}

    optimize(model, path=None)

    assert "test" in model.data
    assert "trace" in model.data

    required_fields = {"time", "primalbound", "dualbound", "gap", "nodes", "nsol"}

    types = [r["type"] for r in model.data["trace"]]
    assert ("bestsol_found" in types) or ("dualbound_improved" in types)

    for record in model.data["trace"]:
        if record["type"] != "run_end":
            assert required_fields <= set(record.keys())

    primalbounds = [r["primalbound"] for r in model.data["trace"] if "primalbound" in r]
    for i in range(1, len(primalbounds)):
        assert primalbounds[i] <= primalbounds[i - 1]

    dualbounds = [r["dualbound"] for r in model.data["trace"] if "dualbound" in r]
    for i in range(1, len(dualbounds)):
        assert dualbounds[i] >= dualbounds[i - 1]

    assert "run_end" in types


def test_realtime_trace_file_output(optimize, tmp_path):
    sizes = list(range(1, 41)) * 3
    model = bin_packing_model(sizes=sizes, capacity=50)
    model.setParam("limits/time", 5)

    path = tmp_path / "trace.jsonl"

    optimize(model, path=str(path))

    assert path.exists()

    records = [json.loads(line) for line in path.read_text().splitlines()]
    assert len(records) > 0

    types = [r["type"] for r in records]
    assert "run_end" in types


class _InterruptOnBest(Eventhdlr):
    def eventinit(self):
        self.model.catchEvent(SCIP_EVENTTYPE.BESTSOLFOUND, self)

    def eventexec(self, event):
        self.model.interruptSolve()


def test_optimize_with_trace_records_run_end_on_interrupt(optimize):
    sizes = list(range(1, 41)) * 3
    model = bin_packing_model(sizes=sizes, capacity=50)
    model.setParam("limits/time", 5)

    model.includeEventhdlr(_InterruptOnBest(), "stopper", "Interrupt on bestsol")

    optimize(model, path=None)

    types = [r["type"] for r in model.data["trace"]]
    assert "bestsol_found" in types
    assert "run_end" in types