distro/src/plan.py

import graphlib
from dataclasses import dataclass
from pathlib import Path

from src.layout import Layout
from src.recipe import Recipe, RecipeSet

BUILD_PHASES = ("prepare", "configure", "build")
PHASE_STAMPS = {
    "prepare": "prepared",
    "configure": "configured",
    "build": "built",
}


def _key(r: Recipe) -> str:
    return r.key  # "host:<name>" or "<name>"


def transitive_host_deps(rs: RecipeSet, r: Recipe) -> list[str]:
    """Returns transitive host_deps for `r`, in topological order (deepest first)."""
    seen: dict[str, Recipe] = {}
    order: list[str] = []

    def visit(name: str) -> None:
        if name in seen:
            return
        h = rs.host.get(name)
        if h is None:
            raise KeyError(f"unknown host dep: {name}")
        seen[name] = h
        for hh in h.host_deps:
            visit(hh)
        order.append(name)

    for d in r.host_deps:
        visit(d)
    return order


def is_built(layout: Layout, r: Recipe) -> bool:
    if r.kind == "host":
        return layout.host_pkg_marker(r.name, r.version, r.revision).is_file()
    for out in r.outputs:
        if not layout.apk_path(out, r.version, r.revision).is_file():
            return False
    return True


def stamp_dir(layout: Layout, r: Recipe) -> Path:
    wd = layout.build_workdir(r.name if r.kind == "target" else f"host-{r.name}")
    return wd / ".orchid-stamps"


def stamp_token(r: Recipe) -> str:
    return f"{r.version}-r{r.revision}\n"


def stamp_valid(layout: Layout, r: Recipe, phase: str) -> bool:
    name = PHASE_STAMPS.get(phase)
    if name is None:
        return False
    p = stamp_dir(layout, r) / name
    try:
        return p.read_text() == stamp_token(r)
    except FileNotFoundError:
        return False


def planned_stages(
    layout: Layout, r: Recipe, *, forced: bool = False
) -> tuple[str, ...]:
    stages: list[str] = []
    for phase in BUILD_PHASES:
        if phase in r.phases and (forced or not stamp_valid(layout, r, phase)):
            stages.append(phase)
    if "install" in r.phases:
        stages.append("install")
    stages.append("package" if r.kind == "target" else "finalize")
    return tuple(stages)


@dataclass
class Plan:
    order: list[str]  # ordered list of recipe keys
    recipes: dict[str, Recipe]  # all referenced recipes
    forced: set[str]  # keys to rebuild even if built
    stages: dict[str, tuple[str, ...]]  # stages each planned recipe will run

    def __iter__(self):
        return iter(self.order)


def _collect_targets(rs: RecipeSet, requested: list[str] | None) -> list[Recipe]:
    if not requested:
        return [r for r in rs.target.values() if r.enabled] + [
            r for r in rs.host.values() if r.enabled
        ]
    out: list[Recipe] = []
    for spec in requested:
        r = rs.get(spec)
        if not r.enabled:
            raise ValueError(f"{spec}: disabled by build_if = False")
        out.append(r)
    return out


def build_plan(
    rs: RecipeSet, layout: Layout, requested: list[str] | None, *, rebuild: bool = False
) -> Plan:
    seen: dict[str, Recipe] = {}
    ts: graphlib.TopologicalSorter[str] = graphlib.TopologicalSorter()
    forced: set[str] = set()

    def add(r: Recipe) -> None:
        k = _key(r)
        if k in seen:
            return
        seen[k] = r
        deps: list[str] = []
        for h in r.host_deps:
            hr = rs.host.get(h)
            if hr is None:
                raise KeyError(f"{r.name}: unknown host dep {h!r}")
            if not hr.enabled:
                raise ValueError(f"{r.name}: host dep {h!r} disabled by build_if")
            add(hr)
            deps.append(_key(hr))
        if r.kind == "target":
            for d in (*r.deps, *r.run_deps):
                tr = rs.target.get(d)
                if tr is None:
                    raise KeyError(f"{r.name}: unknown dep {d!r}")
                if not tr.enabled:
                    raise ValueError(f"{r.name}: dep {d!r} disabled by build_if")
                add(tr)
                deps.append(_key(tr))
        else:
            # host recipes may declare target `deps` that need to land in /sysroot
            for d in r.deps:
                tr = rs.target.get(d)
                if tr is None:
                    raise KeyError(f"host:{r.name}: unknown target dep {d!r}")
                if not tr.enabled:
                    raise ValueError(f"host:{r.name}: dep {d!r} disabled by build_if")
                add(tr)
                deps.append(_key(tr))
        ts.add(k, *deps)

    requested_recipes = _collect_targets(rs, requested)
    for r in requested_recipes:
        add(r)
        if rebuild:
            forced.add(_key(r))

    order = list(ts.static_order())

    stages: dict[str, tuple[str, ...]] = {}
    final_order: list[str] = []
    for k in order:
        r = seen[k]
        if k in forced:
            stages[k] = planned_stages(layout, r, forced=True)
            final_order.append(k)
            continue
        if is_built(layout, r):
            continue
        stages[k] = planned_stages(layout, r)
        final_order.append(k)

    return Plan(order=final_order, recipes=seen, forced=forced, stages=stages)