From 52d4ec423167440b2f1fa213eea27fcc4dddb69f Mon Sep 17 00:00:00 2001 From: Gonzalo Vidal <35148159+Gonza10V@users.noreply.github.com> Date: Tue, 5 May 2026 20:47:04 -0600 Subject: [PATCH 1/2] Implement compatibility selector and deterministic route scoring --- src/buildcompiler/inventory/__init__.py | 11 +- src/buildcompiler/inventory/compatibility.py | 58 ++++++++ src/buildcompiler/inventory/selector.py | 148 +++++++++++++++++++ tests/unit/inventory/test_compatibility.py | 13 ++ tests/unit/inventory/test_selector.py | 60 ++++++++ 5 files changed, 289 insertions(+), 1 deletion(-) create mode 100644 src/buildcompiler/inventory/compatibility.py create mode 100644 src/buildcompiler/inventory/selector.py create mode 100644 tests/unit/inventory/test_compatibility.py create mode 100644 tests/unit/inventory/test_selector.py diff --git a/src/buildcompiler/inventory/__init__.py b/src/buildcompiler/inventory/__init__.py index 9558475..403b6e6 100644 --- a/src/buildcompiler/inventory/__init__.py +++ b/src/buildcompiler/inventory/__init__.py @@ -1,5 +1,14 @@ """Inventory package exports for deterministic lookup/indexing contracts.""" +from .compatibility import Lvl1Route, Lvl2Route, RouteScore, RouteSelection from .inventory import Inventory +from .selector import CompatibilitySelector -__all__ = ["Inventory"] +__all__ = [ + "CompatibilitySelector", + "Inventory", + "Lvl1Route", + "Lvl2Route", + "RouteScore", + "RouteSelection", +] diff --git a/src/buildcompiler/inventory/compatibility.py b/src/buildcompiler/inventory/compatibility.py new file mode 100644 index 0000000..52ddcb8 --- /dev/null +++ b/src/buildcompiler/inventory/compatibility.py @@ -0,0 +1,58 @@ +"""Deterministic compatibility route models and score ordering.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from buildcompiler.domain import IndexedBackbone, IndexedPlasmid + + +@dataclass(frozen=True) +class RouteScore: + missing_required_products: int = 0 + missing_domestications: int = 0 + missing_lvl1_plasmids: int = 0 + generated_or_planned_materials: int = 0 + lower_material_state_penalty: int = 0 + constraint_violations: int = 0 + total_assemblies: int = 0 + identity_tiebreak: tuple[str, ...] = () + + def sort_key(self) -> tuple[int, int, int, int, int, int, int, tuple[str, ...]]: + """Lower sort key is a better route.""" + return ( + self.constraint_violations, + self.missing_required_products, + self.missing_domestications, + self.missing_lvl1_plasmids, + self.generated_or_planned_materials, + self.lower_material_state_penalty, + self.total_assemblies, + self.identity_tiebreak, + ) + + +@dataclass(frozen=True) +class Lvl1Route: + request_id: str + part_identities: tuple[str, ...] + selected_part_plasmids: tuple[IndexedPlasmid, ...] + missing_part_identities: tuple[str, ...] + backbone: IndexedBackbone | None + score: RouteScore + + +@dataclass(frozen=True) +class Lvl2Route: + request_id: str + region_order: tuple[str, ...] + selected_lvl1_plasmids: tuple[IndexedPlasmid, ...] + missing_region_identities: tuple[str, ...] + backbone: IndexedBackbone | None + score: RouteScore + + +@dataclass(frozen=True) +class RouteSelection: + selected: Lvl1Route | Lvl2Route | None + rejected: tuple[Lvl1Route | Lvl2Route, ...] = () diff --git a/src/buildcompiler/inventory/selector.py b/src/buildcompiler/inventory/selector.py new file mode 100644 index 0000000..70aae00 --- /dev/null +++ b/src/buildcompiler/inventory/selector.py @@ -0,0 +1,148 @@ +"""Deterministic compatibility selector for lvl1/lvl2 route selection.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from itertools import permutations +from typing import Any + +from buildcompiler.api import BuildOptions +from buildcompiler.domain import BuildStage, MaterialState +from buildcompiler.inventory.compatibility import Lvl1Route, Lvl2Route, RouteScore, RouteSelection +from buildcompiler.inventory.inventory import Inventory + + +_STATE_RANK = { + MaterialState.PLANNED: 0, + MaterialState.GENERATED: 1, + MaterialState.ASSEMBLED: 2, + MaterialState.TRANSFORMED: 3, + MaterialState.PLATED: 4, +} + + +class CompatibilitySelector: + def __init__(self, inventory: Inventory, *, options: BuildOptions | None = None) -> None: + self.inventory = inventory + self.options = options or BuildOptions() + + def _is_generated_or_planned(self, plasmid: Any) -> bool: + source = (plasmid.metadata or {}).get("source", "") + if source: + return source in {"generated", "planned"} + return plasmid.state in {MaterialState.PLANNED, MaterialState.GENERATED} + + def _constraint_filter(self, items: list[Any], constraints: Mapping[str, Any]) -> list[Any]: + allowed = set(constraints.get("allowed_identities", [])) + forbidden = set(constraints.get("forbidden_identities", [])) + antibiotic = constraints.get("antibiotic") + out = [] + for item in items: + if allowed and item.identity not in allowed: + continue + if item.identity in forbidden: + continue + if antibiotic and item.metadata.get("antibiotic") != antibiotic: + continue + out.append(item) + return out + + def _best_candidate(self, candidates: list[Any], constraints: Mapping[str, Any]) -> Any | None: + filtered = self._constraint_filter(candidates, constraints) + if not filtered: + return None + + prefer_existing = self.options.selection.prefer_existing_collection_material + prefer_state = self.options.selection.prefer_higher_material_state + + def _key(p: Any) -> tuple[int, int, str]: + generated_penalty = int(prefer_existing and self._is_generated_or_planned(p)) + state_penalty = -_STATE_RANK[p.state] if prefer_state else 0 + return (generated_penalty, state_penalty, p.identity) + + return sorted(filtered, key=_key)[0] + + def select_lvl1_route(self, *, request_id: str, part_identities: Sequence[str], constraints: Mapping[str, Any] | None = None) -> RouteSelection: + active_constraints = constraints or {} + selected = [] + missing = [] + for part_identity in part_identities: + candidates = self.inventory.find_single_part_plasmids(part_identity, antibiotic=active_constraints.get("antibiotic")) + choice = self._best_candidate(candidates, active_constraints) + if choice is None: + missing.append(part_identity) + else: + selected.append(choice) + + backbone = self.inventory.find_backbone( + fusion_sites=tuple(active_constraints["fusion_sites"]) if "fusion_sites" in active_constraints else None, + antibiotic=active_constraints.get("antibiotic"), + stage=BuildStage.ASSEMBLY_LVL1, + ) + score = RouteScore( + missing_required_products=len(missing), + missing_domestications=len(missing), + generated_or_planned_materials=sum(1 for p in selected if self._is_generated_or_planned(p)), + lower_material_state_penalty=sum((_STATE_RANK[MaterialState.PLATED] - _STATE_RANK[p.state]) for p in selected) + if self.options.selection.prefer_higher_material_state + else 0, + identity_tiebreak=tuple(sorted(p.identity for p in selected)) + tuple(missing), + ) + route = Lvl1Route(request_id, tuple(part_identities), tuple(selected), tuple(missing), backbone, score) + return RouteSelection(selected=route, rejected=()) + + def select_lvl2_route(self, *, request_id: str, region_identities: Sequence[str], constraints: Mapping[str, Any] | None = None) -> RouteSelection: + active_constraints = constraints or {} + max_regions = self.options.planning.lvl2_search.max_exhaustive_region_count + allow_large = self.options.planning.lvl2_search.allow_large_order_search + + if "region_order" in active_constraints: + orders = [tuple(active_constraints["region_order"])] + elif len(region_identities) > max_regions and not allow_large: + blocked = Lvl2Route( + request_id=request_id, + region_order=tuple(region_identities), + selected_lvl1_plasmids=(), + missing_region_identities=tuple(region_identities), + backbone=None, + score=RouteScore( + missing_required_products=len(region_identities), + missing_lvl1_plasmids=len(region_identities), + constraint_violations=1, + identity_tiebreak=tuple(region_identities), + ), + ) + return RouteSelection(selected=None, rejected=(blocked,)) + else: + orders = sorted(set(permutations(region_identities))) + + routes = [] + for order in orders: + selected = [] + missing = [] + for region in order: + candidates = self.inventory.find_lvl1_region_plasmids(region) + choice = self._best_candidate(candidates, active_constraints) + if choice is None: + missing.append(region) + else: + selected.append(choice) + score = RouteScore( + missing_required_products=len(missing), + missing_lvl1_plasmids=len(missing), + generated_or_planned_materials=sum(1 for p in selected if self._is_generated_or_planned(p)), + lower_material_state_penalty=sum((_STATE_RANK[MaterialState.PLATED] - _STATE_RANK[p.state]) for p in selected) + if self.options.selection.prefer_higher_material_state + else 0, + total_assemblies=int(bool(missing)), + identity_tiebreak=tuple(p.identity for p in selected) + tuple(missing), + ) + backbone = self.inventory.find_backbone( + fusion_sites=tuple(active_constraints["fusion_sites"]) if "fusion_sites" in active_constraints else None, + antibiotic=active_constraints.get("antibiotic"), + stage=BuildStage.ASSEMBLY_LVL2, + ) + routes.append(Lvl2Route(request_id, tuple(order), tuple(selected), tuple(missing), backbone, score)) + + ranked = sorted(routes, key=lambda r: r.score.sort_key()) + return RouteSelection(selected=ranked[0] if ranked else None, rejected=tuple(ranked[1:4])) diff --git a/tests/unit/inventory/test_compatibility.py b/tests/unit/inventory/test_compatibility.py new file mode 100644 index 0000000..aa97a3d --- /dev/null +++ b/tests/unit/inventory/test_compatibility.py @@ -0,0 +1,13 @@ +from buildcompiler.inventory import RouteScore + + +def test_route_score_prefers_fewer_missing_domestications(): + assert RouteScore(missing_domestications=0).sort_key() < RouteScore(missing_domestications=1).sort_key() + + +def test_route_score_prefers_fewer_missing_lvl1_plasmids(): + assert RouteScore(missing_lvl1_plasmids=0).sort_key() < RouteScore(missing_lvl1_plasmids=1).sort_key() + + +def test_route_score_uses_stable_identity_tiebreak(): + assert RouteScore(identity_tiebreak=("a",)).sort_key() < RouteScore(identity_tiebreak=("b",)).sort_key() diff --git a/tests/unit/inventory/test_selector.py b/tests/unit/inventory/test_selector.py new file mode 100644 index 0000000..b78b7c9 --- /dev/null +++ b/tests/unit/inventory/test_selector.py @@ -0,0 +1,60 @@ +from buildcompiler.api import BuildOptions +from buildcompiler.domain import IndexedPlasmid, MaterialState +from buildcompiler.inventory import CompatibilitySelector, Inventory + + +def _plasmid(identity: str, insert: str, *, state=MaterialState.PLANNED, source="collection") -> IndexedPlasmid: + return IndexedPlasmid( + identity=identity, + state=state, + metadata={"insert_identities": [insert], "source": source, "antibiotic": "Ampicillin"}, + ) + + +def test_lvl1_missing_parts_are_reported_not_raised(): + inv = Inventory(plasmids=[_plasmid("https://e/p1", "https://e/part1")]) + sel = CompatibilitySelector(inv) + route = sel.select_lvl1_route(request_id="r1", part_identities=["https://e/part1", "https://e/part2"]).selected + assert route is not None + assert route.missing_part_identities == ("https://e/part2",) + + +def test_lvl1_prefers_existing_material_in_tie(): + inv = Inventory( + plasmids=[ + _plasmid("https://e/a", "https://e/part", source="generated", state=MaterialState.GENERATED), + _plasmid("https://e/b", "https://e/part", source="collection", state=MaterialState.GENERATED), + ] + ) + sel = CompatibilitySelector(inv) + route = sel.select_lvl1_route(request_id="r1", part_identities=["https://e/part"]).selected + assert route.selected_part_plasmids[0].identity == "https://e/b" + + +def test_lvl1_hard_constraints_override_selection_preference(): + inv = Inventory(plasmids=[_plasmid("https://e/a", "https://e/part", source="generated")]) + opts = BuildOptions() + opts.selection.prefer_existing_collection_material = True + sel = CompatibilitySelector(inv, options=opts) + route = sel.select_lvl1_route( + request_id="r1", + part_identities=["https://e/part"], + constraints={"allowed_identities": ["https://e/a"]}, + ).selected + assert route.selected_part_plasmids[0].identity == "https://e/a" + + +def test_lvl2_large_order_search_not_silent_without_opt_in(): + inv = Inventory() + sel = CompatibilitySelector(inv) + out = sel.select_lvl2_route(request_id="r2", region_identities=["a", "b", "c", "d", "e"]) + assert out.selected is None + assert out.rejected + + +def test_lvl2_rejected_alternatives_capped_at_3(): + inv = Inventory(plasmids=[_plasmid(f"https://e/p{i}", f"https://e/r{i}") for i in range(4)]) + sel = CompatibilitySelector(inv) + out = sel.select_lvl2_route(request_id="r3", region_identities=["https://e/r0", "https://e/r1", "https://e/r2", "https://e/r3"]) + assert out.selected is not None + assert len(out.rejected) == 3 From 2765225d2c7625a6e20085266d21591ed345307d Mon Sep 17 00:00:00 2001 From: Gonzalo Vidal <35148159+Gonza10V@users.noreply.github.com> Date: Tue, 5 May 2026 21:38:27 -0600 Subject: [PATCH 2/2] Fix lvl2 region_order constraint validation --- src/buildcompiler/inventory/selector.py | 19 ++++++++++++++++++- tests/unit/inventory/test_selector.py | 13 +++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/src/buildcompiler/inventory/selector.py b/src/buildcompiler/inventory/selector.py index 70aae00..07ddd31 100644 --- a/src/buildcompiler/inventory/selector.py +++ b/src/buildcompiler/inventory/selector.py @@ -97,7 +97,24 @@ def select_lvl2_route(self, *, request_id: str, region_identities: Sequence[str] allow_large = self.options.planning.lvl2_search.allow_large_order_search if "region_order" in active_constraints: - orders = [tuple(active_constraints["region_order"])] + constrained_order = tuple(active_constraints["region_order"]) + requested_regions = tuple(region_identities) + if sorted(constrained_order) != sorted(requested_regions): + blocked = Lvl2Route( + request_id=request_id, + region_order=constrained_order, + selected_lvl1_plasmids=(), + missing_region_identities=requested_regions, + backbone=None, + score=RouteScore( + missing_required_products=len(requested_regions), + missing_lvl1_plasmids=len(requested_regions), + constraint_violations=1, + identity_tiebreak=requested_regions, + ), + ) + return RouteSelection(selected=None, rejected=(blocked,)) + orders = [constrained_order] elif len(region_identities) > max_regions and not allow_large: blocked = Lvl2Route( request_id=request_id, diff --git a/tests/unit/inventory/test_selector.py b/tests/unit/inventory/test_selector.py index b78b7c9..fe4c03c 100644 --- a/tests/unit/inventory/test_selector.py +++ b/tests/unit/inventory/test_selector.py @@ -58,3 +58,16 @@ def test_lvl2_rejected_alternatives_capped_at_3(): out = sel.select_lvl2_route(request_id="r3", region_identities=["https://e/r0", "https://e/r1", "https://e/r2", "https://e/r3"]) assert out.selected is not None assert len(out.rejected) == 3 + + +def test_lvl2_constrained_order_must_match_requested_regions(): + inv = Inventory(plasmids=[_plasmid("https://e/p0", "https://e/r0"), _plasmid("https://e/p1", "https://e/r1")]) + sel = CompatibilitySelector(inv) + out = sel.select_lvl2_route( + request_id="r4", + region_identities=["https://e/r0", "https://e/r1"], + constraints={"region_order": ["https://e/r0"]}, + ) + assert out.selected is None + assert out.rejected + assert out.rejected[0].missing_region_identities == ("https://e/r0", "https://e/r1")