core/berth_core/check.py

"""Feasibility and cost of a plan. Pure Python: the server grades with this and needs no solver."""

from __future__ import annotations

from dataclasses import dataclass, field

from .model import MOVE_PENALTY, Plan, Task


@dataclass
class ShipResult:
    ship: int
    berth_hour: int | None
    section: int | None
    departure: int | None
    delay_h: int
    moved: bool
    cost: int
    problems: list[str] = field(default_factory=list)
    cranes: int | None = None


@dataclass
class PlanResult:
    feasible: bool
    ships: list[ShipResult]
    cost: int | None           # None when the plan is infeasible
    delay_cost: int | None
    moves: int | None

    @property
    def clean_fraction(self) -> float:
        return sum(1 for s in self.ships if not s.problems) / max(1, len(self.ships))

    @property
    def violations(self) -> list[dict]:
        return [{"ship": s.ship, "problem": p} for s in self.ships for p in s.problems]


def _span(a0: int, a1: int) -> str:
    return f"{a0}" if a0 == a1 else f"{a0}-{a1}"


def _hours(hs: list[int]) -> str:
    """[3, 4, 5, 9] -> '3-5, 9'"""
    out, run = [], []
    for h in sorted(set(hs)):
        if run and h == run[-1] + 1:
            run.append(h)
        else:
            if run:
                out.append(_span(run[0], run[-1]))
            run = [h]
    if run:
        out.append(_span(run[0], run[-1]))
    return ", ".join(out[:4]) + (" …" if len(out) > 4 else "")


def evaluate(task: Task, plan: Plan) -> PlanResult:
    """Check every ship of `plan` against the task's rules and price it.

    Rules: berth at or after arrival; stay within the quay; never share a section-hour with another ship, a ship
    already alongside, or a closed section. On crane-rule tasks also: cranes within the ship's min-max, cranes in use
    at every hour within the pool (less any outage), and at most max_moves_per_hour berthings + departures per hour.
    Cost: sum over ships of sections x hours late against the planned departure, plus MOVE_PENALTY for each
    scheduled ship placed on different sections than planned."""
    lo, hi = task.first_section, task.last_section
    out: list[ShipResult] = []
    rects: dict[int, tuple[int, int, int, int]] = {}
    for s in task.ships:
        if s.id not in plan:
            out.append(ShipResult(s.id, None, None, None, 0, False, 0, ["missing from the plan"]))
            continue
        entry = plan[s.id]
        h, sec = entry[0], entry[1]
        c = entry[2] if task.cranes and len(entry) > 2 else (s.std_cranes if task.cranes else None)
        bad_cranes = task.cranes and not (s.min_cranes <= c <= s.max_cranes)
        work = s.handling_for(s.std_cranes if bad_cranes else c)   # never divide by a zero or negative count
        dep = task.departure(s, h + work) if task.cranes else h + work
        stay = dep - h
        r = ShipResult(s.id, h, sec, dep, max(0, dep - s.due),
                       s.planned_section is not None and sec != s.planned_section, 0, cranes=c)
        if task.cranes:
            for a, b in task.no_move_windows(s):
                if a <= h < b:
                    r.problems.append(f"berths at hour {h}, inside the no-movement window {a}-{b}")
        if bad_cranes:
            r.problems.append(f"gets {c} cranes but can be worked by {s.min_cranes}-{s.max_cranes}")
        if h < s.arrival:
            r.problems.append(f"berths at hour {h} but cannot arrive before hour {s.arrival}")
        last = sec + s.sections - 1
        if sec < lo or last > hi:
            r.problems.append(f"needs sections {_span(sec, last)} but the quay has sections {lo}-{hi}")
        for b in task.blocks:
            if sec <= b.last and b.first <= last and h < b.end and b.start < h + stay:
                what = f"{b.label} (alongside)" if b.kind == "alongside" else "closed sections"
                r.problems.append(f"overlaps {what} {_span(b.first, b.last)} during hours {b.start}-{b.end}")
        rects[s.id] = (h, h + stay, sec, last, h + work)
        out.append(r)
    ids = sorted(rects)
    for i, a in enumerate(ids):
        t0, t1, m0, m1 = rects[a][:4]
        for b in ids[i + 1:]:
            u0, u1, n0, n1 = rects[b][:4]
            if t0 < u1 and u0 < t1 and m0 <= n1 and n0 <= m1:
                hrs, secs = f"{max(t0, u0)}-{min(t1, u1)}", _span(max(m0, n0), min(m1, n1))
                out[a].problems.append(f"overlaps ship {b} ({task.ships[b].name}) in sections {secs} during hours {hrs}")
                out[b].problems.append(f"overlaps ship {a} ({task.ships[a].name}) in sections {secs} during hours {hrs}")
    if task.cranes:
        _crane_and_move_limits(task, out, rects)
    feasible = not any(r.problems for r in out)
    for r in out:
        if r.berth_hour is not None:
            r.cost = ship_delay_cost(task.ships[r.ship], r.berth_hour, r.delay_h) + (MOVE_PENALTY if r.moved else 0)
    if not feasible:
        return PlanResult(False, out, None, None, None)
    delay = sum(ship_delay_cost(task.ships[r.ship], r.berth_hour, r.delay_h) for r in out)
    moves = sum(r.moved for r in out)
    return PlanResult(True, out, delay + MOVE_PENALTY * moves, delay, moves)


def ship_delay_cost(s, berth_hour: int, delay_h: int) -> int:
    """sections x hours late x weight, plus the emergency penalty for berthing after a genuine deadline."""
    cost = s.sections * delay_h * s.weight
    if s.berth_deadline is not None and berth_hour > s.berth_deadline:
        cost += s.deadline_penalty * (berth_hour - s.berth_deadline)
    return cost


def _crane_and_move_limits(task: Task, out: list[ShipResult], rects: dict) -> None:
    """Crane pool and movement limit, hour by hour; a ship involved in an over-limit hour gets one problem line."""
    use: dict[int, int] = {}
    moves: dict[int, list[int]] = {}
    fixed_moves: dict[int, int] = {}
    for b in task.blocks:
        if b.kind == "alongside":
            for t in range(b.start, b.end):
                use[t] = use.get(t, 0) + b.cranes
            fixed_moves[b.end] = fixed_moves.get(b.end, 0) + 1
    for sid, (t0, t1, _, _, tw) in rects.items():
        c = out[sid].cranes or 0
        for t in range(t0, tw):          # cranes work until the ship finishes, not while it waits for the wind
            use[t] = use.get(t, 0) + c
        moves.setdefault(t0, []).append(sid)
        moves.setdefault(t1, []).append(sid)
    over = {t for t, u in use.items() if u > task.crane_pool_at(t)}
    if over:
        for sid, (t0, t1, _, _, tw) in rects.items():
            hs = [t for t in range(t0, tw) if t in over]
            if hs:
                worst = max(hs, key=lambda t: use[t] - task.crane_pool_at(t))
                out[sid].problems.append(
                    f"cranes over the pool at hours {_hours(hs)} (e.g. hour {worst}: {use[worst]} in use, "
                    f"{task.crane_pool_at(worst)} available)")
    cap = task.rules.get("max_moves_per_hour")
    if cap:
        for t, ids in sorted(moves.items()):
            n = len(ids) + fixed_moves.get(t, 0)
            if n > cap:
                for sid in ids:
                    kind = "berths" if rects[sid][0] == t else "leaves"
                    out[sid].problems.append(f"{kind} at hour {t}, when {n} ships move (limit {cap} per hour)")


def plan_cost(task: Task, plan: Plan) -> int:
    res = evaluate(task, plan)
    if not res.feasible:
        raise ValueError("plan is infeasible: " + "; ".join(v["problem"] for v in res.violations[:3]))
    return res.cost


def unavoidable_cost(task: Task) -> int:
    """A lower bound on any plan's cost: each ship on its own, docking at its earliest legal hour (its arrival, or the
    end of a no-movement window it arrives in) with the most cranes it can take. No plan can do better for that ship,
    so the sum is a floor under the optimum; the reward measures mistakes against the cost above this floor."""
    total = 0
    for s in task.ships:
        h = s.arrival
        if task.cranes:
            for a, b in task.no_move_windows(s):
                if a <= h < b:
                    h = b
            dep = task.departure(s, h + s.handling_for(s.max_cranes))
        else:
            dep = h + s.handling
        total += ship_delay_cost(s, h, max(0, dep - s.due))
    return total