Introduce 07-independentset

07_independentset/models/binary_linear/dat_reader.py

@@ -0,0 +1,56 @@
+def read_dimacs_gph(path: str):
+    """
+    Read an undirected graph in DIMACS format (.gph).
+    The function converts vertices to 0-based indexing and removes self-loops
+    or duplicate edges. Edge endpoints are normalized such that the smaller
+    index comes first.
+
+    Args:
+        path (str): Path to the `.gph` DIMACS graph file.
+
+    Returns:
+        tuple[int, list[list[int]]]:
+            - N: The number of vertices.
+            - E: A list of edges, where each edge is represented as a
+              2-element list `[u, v]` with 0-based vertex indices.
+    """
+    N = None
+    E = []
+    with open(path, "r") as f:
+        for raw in f:
+            line = raw.strip()
+            if not line or line.startswith("c"):
+                # Skip empty and comment lines
+                continue
+            if line.startswith("p"):
+                # Example: "p edge 17 39"
+                parts = line.split()
+                if len(parts) < 4 or parts[1] != "edge":
+                    raise ValueError(f"Invalid p-line: {line}")
+                N = int(parts[2])
+                # M = int(parts[3])  # Optional: use this for edge count validation
+            elif line.startswith("e"):
+                # Example: "e 7 17"
+                _, u, v = line.split()
+                u0 = int(u) - 1  # Convert to 0-based
+                v0 = int(v) - 1
+                if u0 == v0:
+                    # Ignore self-loops if present
+                    continue
+                # Normalize to (smaller, larger) to avoid duplicates
+                if v0 < u0:
+                    u0, v0 = v0, u0
+                E.append([u0, v0])
+
+    if N is None:
+        raise ValueError("Missing 'p edge N M' header.")
+
+    # Deduplicate edges just in case
+    E = sorted(set(tuple(e) for e in E))
+    E = [list(e) for e in E]
+
+    # Basic validation: ensure all endpoints are in range
+    if any(u < 0 or v < 0 or u > N - 1 or v > N - 1 for u, v in E):
+        raise ValueError("Edge endpoint out of range for declared N.")
+
+    return N, E

07_independentset/models/binary_linear/model.py

@@ -0,0 +1,41 @@
+import jijmodeling as jm
+
+
+def build_mis_problem() -> jm.Problem:
+    """Create Maximum Independent Set (MIS) optimization model.
+
+    Formulates the maximum independent set problem using binary decision
+    variables, where the objective is to maximize the number of selected
+    vertices subject to adjacency constraints.
+
+    Returns:
+        jm.Problem: JijModeling problem instance with all constraints and
+        variables defined for the maximum independent set problem.
+    """
+    # Placeholders
+    N = jm.Placeholder("N", description="number of nodes")
+    E = jm.Placeholder("E", ndim=2, description="edge list as pairs (u,v), 0-based")
+
+    # Decision variable: x[i] ∈ {0,1}
+    x = jm.BinaryVar("x", shape=(N,), description="1 if vertex i is selected")
+
+    # Objective: maximize sum_v x[v]
+    v = jm.Element("v", belong_to=(0, N))
+    obj = jm.sum(v, x[v])
+
+    # Problem
+    probem = jm.Problem("maximum_independent_set", sense=jm.ProblemSense.MAXIMIZE)
+    probem += obj
+
+    # Constraints: for all (u,v) in E, x[u] + x[v] <= 1
+    e = jm.Element("e", belong_to=E)
+    probem += jm.Constraint("no_adjacent", x[e[0]] + x[e[1]] <= 1, forall=e)
+
+    problem = jm.Problem("maximum_independent_set", sense=jm.ProblemSense.MAXIMIZE)
+    problem += obj
+
+    # Constraints: for all (u,v) in E, x[u] + x[v] <= 1
+    e = jm.Element("e", belong_to=E)
+    problem += jm.Constraint("no_adjacent", x[e[0]] + x[e[1]] <= 1, forall=e)
+
+    return problem

07_independentset/models/binary_linear/ommx_create.py

@@ -0,0 +1,110 @@
+import os
+import glob
+import jijmodeling as jm
+from ommx.artifact import ArtifactBuilder
+from model import build_mis_problem
+from dat_reader import read_dimacs_gph
+from sol_reader import parse_sol_file
+
+
+def _pick_solution_file(sol_dir: str, base: str) -> str | None:
+    """Pick first existing solution file for a basename."""
+    candidates = [
+        os.path.join(sol_dir, f"{base}.opt.sol"),
+        os.path.join(sol_dir, f"{base}.best.sol"),
+        os.path.join(sol_dir, f"{base}.bst.sol"),
+        os.path.join(sol_dir, f"{base}.sol"),
+    ]
+    for p in candidates:
+        if os.path.exists(p):
+            return p
+    return None
+
+
+def batch_process(
+    inst_dir: str = "../../instances",
+    sol_root: str = "../../solutions",
+    output_directory: str = "./ommx_output",
+):
+    """
+    Process instances from a QBench JSON file and corresponding solution files,
+    convert them into OMMX artifacts, and save them to the output directory.
+
+    Parameters:
+        inst_dir (str): Path to the gph..
+        sol_root (str): Path to the root solutions directory.
+        output_directory (str): Path to save the generated .ommx files.
+    """
+    os.makedirs(output_directory, exist_ok=True)
+
+    problem = build_mis_problem()
+
+    gph_paths = sorted(glob.glob(os.path.join(inst_dir, "*.gph")))
+    if not gph_paths:
+        print(f"[MIS] No .gph files found under: {inst_dir}")
+        return
+
+    processed_count = 0
+    error_count = 0
+
+    for gph_path in gph_paths:
+        base = os.path.splitext(os.path.basename(gph_path))[0]
+        try:
+            print(f"[{base}] Reading: {gph_path}")
+            N, E = read_dimacs_gph(gph_path)
+            instance_data = {"N": N, "E": E}
+
+            interpreter = jm.Interpreter(instance_data)
+            ommx_instance = interpreter.eval_problem(problem)
+
+            sol_path = _pick_solution_file(sol_root, base)
+            solution = None
+            if sol_path:
+                try:
+                    print(f"  → Evaluating solution: {sol_path}")
+                    obj_from_file, solution_dict = parse_sol_file(sol_path, N)
+                    solution = ommx_instance.evaluate(solution_dict)
+                    if (
+                        solution.feasible
+                        and abs(solution.objective - obj_from_file["Energy"]) < 1e-6
+                    ):
+                        print(
+                            f"    objective={solution.objective}, feasible={solution.feasible}"
+                        )
+                    else:
+                        print(
+                            "    ! Objective mismatch or infeasible; will save instance only."
+                        )
+                        solution = None
+                except Exception as sol_err:
+                    print(f"    ! Solution evaluation failed: {sol_err}")
+                    solution = None
+
+            out_path = os.path.join(output_directory, f"{base}.ommx")
+            if os.path.exists(out_path):
+                os.remove(out_path)
+
+            builder = ArtifactBuilder.new_archive_unnamed(out_path)
+            builder.add_instance(ommx_instance)
+            if solution is not None:
+                builder.add_solution(solution)
+            builder.build()
+
+            print(f"  ✓ Created: {out_path}")
+            print("-" * 50)
+            processed_count += 1
+
+        except Exception as e:
+            print(f"[{base}] Error: {e}")
+            print("-" * 50)
+            error_count += 1
+
+    print(f"Batch complete — processed: {processed_count}, errors: {error_count}")
+
+
+if __name__ == "__main__":
+    batch_process(
+        inst_dir="../../instances",
+        sol_root="../../solutions",
+        output_directory="./ommx_output",
+    )

07_independentset/models/binary_linear/sol_reader.py

@@ -0,0 +1,99 @@
+import re
+
+
+def parse_sol_file(file_path: str, n: int) -> tuple[dict[str, int], dict[int, int]]:
+    """Parse MIS solution file with automatic format detection.
+
+    Supports two common formats found in `.sol` files:
+
+    **A) Key–Value format (e.g., `opt.sol`, `sol`)**
+        Lines may include:
+            - `# Objective value = <int>`
+            - `x#<idx> <int>` (typically 1-based indexing)
+
+    **B) Index-list format (e.g., `bst.sol`, sometimes `opt.sol`)**
+        Each line contains a single integer representing a selected vertex.
+        Values are interpreted as 0-based or 1-based automatically.
+
+    Args:
+        file_path (str): Path to the solution file.
+        n (int): Number of vertices in the MIS instance.
+
+    Returns:
+        tuple:
+            - dict[str, int]: `{"Energy": <int or None>}`.
+              The objective value if present; otherwise, the count of selected vertices.
+            - dict[int, int]: Mapping of vertex indices (0..n−1) to 0.0/1.0 (or int in KV format).
+    """
+    # read full file
+    with open(file_path, "r") as f:
+        raw_lines = [ln.strip() for ln in f]
+
+    lines = [ln for ln in raw_lines if ln]
+
+    # detect KV format
+    kv_pattern = re.compile(r"^x#(\d+)\s+([-+]?\d*\.?\d+)$", re.IGNORECASE)
+    has_kv = any(kv_pattern.match(ln) for ln in lines)
+
+    # check objective value
+    obj_pattern = re.compile(
+        r"#\s*Objective\s*value\s*=\s*([-+]?\d*\.?\d+)", re.IGNORECASE
+    )
+    obj_value = None
+    for ln in lines:
+        m = obj_pattern.match(ln)
+        if m:
+            obj_value = int(m.group(1))
+            break
+
+    if has_kv:
+        # === KV format ===
+        x_vars = {}
+        for ln in lines:
+            m = kv_pattern.match(ln)
+            if m:
+                idx = int(m.group(1))  # assume 1-based: x#1, x#2, ...
+                val = int(m.group(2))
+                x_vars[idx] = val
+
+        solution_dict = {i - 1: x_vars.get(i, 0.0) for i in range(1, n + 1)}
+        return {"Energy": obj_value}, solution_dict
+
+    # === Index-list format ===
+    selected = []
+    only_ints = True
+    for ln in lines:
+        try:
+            val = int(ln)
+            selected.append(val)
+        except ValueError:
+            only_ints = False
+            break
+
+    if only_ints and selected:
+        all_in_1_based = all(1 <= v <= n for v in selected)
+        all_in_0_based = all(0 <= v <= n - 1 for v in selected)
+
+        if not (all_in_0_based or all_in_1_based):
+            raise ValueError(
+                f"Index list not within [0, {n-1}] or [1, {n}] in {file_path}: {selected[:10]}..."
+            )
+
+        use_one_based = all_in_1_based
+        solution_dict = {i: 0.0 for i in range(n)}
+        if use_one_based:
+            for v in selected:
+                solution_dict[v - 1] = 1.0
+        else:
+            for v in selected:
+                solution_dict[v] = 1.0
+
+        if obj_value is None:
+            obj_value = int(sum(solution_dict.values()))
+
+        return {"Energy": obj_value}, solution_dict
+
+    raise ValueError(
+        f"Unrecognized solution format in {file_path}. "
+        "Expect either 'x#i val' lines or pure index-per-line."
+    )