diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f968d4..57430c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,19 @@ All notable changes to `pddlpy` are documented here. This project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Added +- **Temporal planner** (#84): `get("temporal")` *solves* durative-action + domains. It enforces the full time-tagged contract — `at start`, `over all` + (invariant across the duration), and `at end` — applying start effects at the + start point and end effects at start+duration, under **sequential** semantics + (actions never overlap; instantaneous actions participate as zero-duration + steps). Returns a `TemporalPlan` whose `ScheduledAction` steps carry start + times and durations and whose `makespan` is the plan cost. Registered with + the `:durative-actions` capability. Required concurrency (mandatory overlap) + remains out of scope. + ## [1.1.2] - 2026-07-12 Infrastructure release — no library changes. diff --git a/README.md b/README.md index 219c289..36ab0e5 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,9 @@ The orginal grammar file was authored by Zeyn Saigol from University of Birmingh ### NOTICE ### -Durative actions are now recovered into the object model (see _Planning API_). -The reference planners, however, are non-temporal and do not solve durative -domains — that is documented as out of scope for this round. +Durative actions are recovered into the object model and now *solved*: the +`temporal` planner (#84) schedules durative-action domains sequentially. The +reference STRIPS planners remain non-temporal; see _Planning API_. ### What this project is not? ### @@ -177,9 +177,27 @@ Action costs (`:action-costs`, `(increase (total-cost) ...)`, `(:metric minimize Durative actions (`:durative-actions`) are recovered into a `DurativeAction` type with time-tagged conditions (`at start` / `over all` / `at end`) and effects (`at start` / `at end`) plus a duration. `DomainProblem.durative_operators()` -and `ground_durative_operator(name)` expose them. The reference planners are -non-temporal, so they do not solve durative domains — this is documented as out -of scope for now. +and `ground_durative_operator(name)` expose them. + +The `temporal` planner (#84) *solves* these domains. It enforces the full +time-tagged contract — `at start` at the start point, `over all` as an +invariant across the duration, and `at end` at start+duration — under +**sequential** semantics (actions never overlap; instantaneous actions +participate as zero-duration steps). It returns a `TemporalPlan` whose `steps` +carry per-action start times and durations and whose `makespan` is the plan +cost. Required concurrency (mandatory overlap, e.g. match/cellar) is out of +scope. + +```python +>>> import pddlpy +>>> from pddlpy.planning import get +>>> dp = pddlpy.DomainProblem('weld-domain.pddl', 'weld-problem.pddl') +>>> plan = get('temporal').solve(dp) +>>> plan.makespan +6.0 +>>> [(s.start, s.action.operator_name, s.duration) for s in plan.steps] +[(0.0, 'steady-weld', 6.0)] +``` ```python >>> import pddlpy @@ -295,15 +313,16 @@ There are wonderful material at the the University of Edinburgh: ### Future development ### -* A temporal planner that solves durative-action domains. +* Concurrent temporal planning (required action overlap, e.g. match/cellar). * Heuristic improvements for the reference planners. * ADL (conditional effects, quantifiers) and a full and/or/not precondition tree. Done recently: case-insensitive keywords, `:requirements` capture/enforcement, a planner interface with BFS/A*/GBFS/UCS reference planners, numeric fluents (`:functions`, numeric preconditions/effects), action costs (`total-cost` + -cost-aware search), durative-action recovery into the object model, and a -measured test suite with full coverage of the object model and planning layer. +cost-aware search), durative-action recovery into the object model, a +sequential `temporal` planner that schedules durative-action domains (#84), and +a measured test suite with full coverage of the object model and planning layer. ### Advanced ### diff --git a/docs/object-model.md b/docs/object-model.md index e968878..1b88281 100644 --- a/docs/object-model.md +++ b/docs/object-model.md @@ -66,8 +66,8 @@ evaluates a numeric precondition; `NumericEffect.apply(valuation)` returns the ### `DurativeAction` (#23) A durative action with time-tagged `condition_pos`/`condition_neg` keyed by `'start'`/`'over'`/`'end'`, `effect_pos`/`effect_neg` keyed by -`'start'`/`'end'`, and a `duration`. Recovered into the model but **not** -solved by the reference planners (they are non-temporal). +`'start'`/`'end'`, and a `duration`. Recovered into the model; the reference +planners are non-temporal, but the `temporal` planner (below) schedules them. ## Planning layer (`pddlpy.planning`) @@ -123,6 +123,42 @@ plan = get("astar").solve(dp) print(plan.cost, plan.action_names()) ``` +### Temporal planner (#84) + +The `temporal` planner *solves* durative-action domains, enforcing the full +time-tagged contract that `DurativeState` (at-start only) cannot. It declares +the `:durative-actions` capability and returns a `TemporalPlan`. + +- **`apply_durative(state, action)`** — applies one grounded `DurativeAction` + under **sequential** semantics: it checks `at start` in `state`, applies the + start effects, then checks the `over all` invariant and the `at end` + conditions against that mid-state (with no concurrency it persists for the + whole duration), and finally applies the end effects at `start + duration`. + Returns the successor `State`, or `None` if any condition fails. +- **`TemporalPlanner`** (`get("temporal")`) — uniform-cost search over + accumulated duration; instantaneous actions participate as zero-duration + steps, so mixed domains work. Returns a makespan-minimal sequential schedule. +- **`TemporalPlan`** — a `Plan` whose `steps` are `ScheduledAction`s + (`action`, `start`, `duration`, `end`); its `makespan` doubles as the plan + `cost`. + +```python +from pddlpy import DomainProblem +from pddlpy.planning import get + +dp = DomainProblem("durative-domain.pddl", "durative-problem.pddl") +plan = get("temporal").solve(dp) +if plan is not None: + print("makespan", plan.makespan) + for step in plan.steps: + print(step.start, step.action.operator_name, step.duration) +``` + +Because semantics are sequential (v1), actions never overlap: `over all` and +`at end` are constrained only by the action's own start effects. **Required +concurrency** (actions that must overlap, e.g. match/cellar) is out of scope — +a follow-up to this planner. + ## Known limitations - **Union types:** the `:types` hierarchy is supported (#22) — a parameter @@ -130,5 +166,6 @@ print(plan.cost, plan.action_names()) `(either ...)` union types are not handled. - **Disjunctive/ADL preconditions:** the `or` connective is preserved but not evaluated; the reference planners reject such domains via capability checks. -- **Durative actions:** validated and checked for `at start` applicability - (see above), but not solved — the reference planners are non-temporal. +- **Durative actions:** solved sequentially by the `temporal` planner (#84); + the reference STRIPS planners remain non-temporal. **Required concurrency** + (mandatory action overlap) and PDDL+ processes/events are out of scope. diff --git a/pddlpy/planning/__init__.py b/pddlpy/planning/__init__.py index e13adc3..25d346b 100644 --- a/pddlpy/planning/__init__.py +++ b/pddlpy/planning/__init__.py @@ -28,6 +28,14 @@ UniformCostPlanner, ) from .state import Plan, State, atom_tuple +from .temporal import ( + TEMPORAL_CAPABILITIES, + ScheduledAction, + TemporalPlan, + TemporalPlanner, + TemporalTask, + apply_durative, +) __all__ = [ "State", @@ -53,4 +61,10 @@ "action_cost", "plan_cost", "TOTAL_COST", + "TemporalPlanner", + "TemporalPlan", + "TemporalTask", + "ScheduledAction", + "apply_durative", + "TEMPORAL_CAPABILITIES", ] diff --git a/pddlpy/planning/state.py b/pddlpy/planning/state.py index 29825d7..5bd5549 100644 --- a/pddlpy/planning/state.py +++ b/pddlpy/planning/state.py @@ -7,10 +7,25 @@ """ from __future__ import annotations -from typing import TYPE_CHECKING, Any, Dict, Iterable, Iterator, List, Mapping, Optional, Tuple +from typing import ( + TYPE_CHECKING, + Any, + Dict, + Iterable, + Iterator, + List, + Mapping, + Optional, + Tuple, + Union, +) if TYPE_CHECKING: - from pddlpy.pddl import DomainProblem, Operator + from pddlpy.pddl import DomainProblem, DurativeAction, Operator + + #: Any grounded action a plan step can carry — instantaneous or durative + #: (#84); both expose ``operator_name`` and ``variable_list``. + PlanAction = Union[Operator, DurativeAction] #: A ground atom, e.g. ``("on", "a", "b")``. GroundAtom = Tuple[str, ...] @@ -133,22 +148,22 @@ def __repr__(self) -> str: class Plan: """An ordered sequence of grounded actions that reaches a goal. - ``actions`` is a tuple of grounded ``Operator`` instances. ``cost`` is - the plan cost (defaults to the number of actions; action-cost domains - override this in a later phase). + ``actions`` is a tuple of grounded ``Operator`` (or, for temporal plans, + ``DurativeAction``) instances. ``cost`` is the plan cost (defaults to the + number of actions; action-cost domains override this in a later phase). """ __slots__ = ("actions", "cost") def __init__( self, - actions: Iterable["Operator"] = (), + actions: Iterable["PlanAction"] = (), cost: Optional[float] = None, ) -> None: - self.actions: Tuple["Operator", ...] = tuple(actions) + self.actions: Tuple["PlanAction", ...] = tuple(actions) self.cost: float = len(self.actions) if cost is None else cost - def __iter__(self) -> Iterator["Operator"]: + def __iter__(self) -> Iterator["PlanAction"]: return iter(self.actions) def __len__(self) -> int: diff --git a/pddlpy/planning/temporal.py b/pddlpy/planning/temporal.py new file mode 100644 index 0000000..f51e0b3 --- /dev/null +++ b/pddlpy/planning/temporal.py @@ -0,0 +1,227 @@ +"""Sequential temporal planner over durative actions (#84). + +Solves durative-action domains by compressing each grounded +``DurativeAction`` into an atomic transition. Under **sequential semantics** +actions never overlap, so the state is constant while an action executes: + +1. the ``at start`` conditions must hold in the current state; +2. start effects are applied at the start time point; +3. the ``over all`` invariant and the ``at end`` conditions are checked + against the mid-state (after the start effects) — the state that persists + through the open interval and at the end point; +4. end effects are applied at ``start + duration``. + +An action whose own start effects violate its ``over all`` / ``at end`` +conditions is therefore never executable — the constraint an +``at start``-only check (:class:`~pddlpy.planning.DurativeState`) cannot +express. Instantaneous actions participate as zero-duration steps, so mixed +domains work. + +``TemporalPlanner`` (registered as ``"temporal"``) runs uniform-cost search +over accumulated duration and returns a :class:`TemporalPlan` carrying +per-step start times, durations and a ``makespan`` (= ``cost``), minimal for +sequential schedules. *Required concurrency* (actions that must overlap, e.g. +matchcellar) is out of scope — a follow-up to this planner. +""" +from __future__ import annotations + +import heapq +import itertools +from typing import ( + TYPE_CHECKING, + Any, + Dict, + FrozenSet, + Iterable, + Iterator, + List, + Optional, + Tuple, +) + +from .base import Planner, validate_requirements +from .durative import validate_durative_actions +from .registry import register +from .search import STRIPS_CAPABILITIES +from .state import Plan, State, atom_tuple + +if TYPE_CHECKING: + from pddlpy.pddl import DomainProblem, DurativeAction, Operator + + from .state import PlanAction + +#: Parent-pointer map used to reconstruct a schedule. +_CameFrom = Dict[State, Tuple[State, "PlanAction", float]] + +#: Requirements the temporal planner supports: the STRIPS set plus +#: :durative-actions. +TEMPORAL_CAPABILITIES: FrozenSet[str] = STRIPS_CAPABILITIES | {":durative-actions"} + + +class ScheduledAction: + """One step of a temporal plan: a grounded action, its start time and + its duration (``end == start + duration``; 0 for instantaneous steps).""" + + __slots__ = ("action", "start", "duration") + + def __init__(self, action: "PlanAction", start: float, duration: float) -> None: + self.action = action + self.start = start + self.duration = duration + + @property + def end(self) -> float: + return self.start + self.duration + + def __repr__(self) -> str: + return "%g: %s(%s) [%g]" % ( + self.start, + self.action.operator_name, + ", ".join(str(v) for v in self.action.variable_list.values()), + self.duration, + ) + + +class TemporalPlan(Plan): + """A :class:`Plan` whose steps carry start times and durations (#84). + + ``steps`` is a tuple of :class:`ScheduledAction`; ``actions`` / + ``action_names()`` keep the base ``Plan`` contract. ``makespan`` is the + end time of the schedule and doubles as the plan ``cost``. + """ + + __slots__ = ("steps", "makespan") + + def __init__(self, steps: Iterable[ScheduledAction] = ()) -> None: + self.steps: Tuple[ScheduledAction, ...] = tuple(steps) + self.makespan: float = max((s.end for s in self.steps), default=0.0) + super().__init__((s.action for s in self.steps), cost=self.makespan) + + def __repr__(self) -> str: + return "TemporalPlan[makespan=%g]: %s" % ( + self.makespan, + ", ".join(repr(s) for s in self.steps), + ) + + +def _holds(state: State, pos: Iterable[Any], neg: Iterable[Any]) -> bool: + """Conjunctive check: every atom in ``pos`` holds, none in ``neg`` does.""" + return {atom_tuple(a) for a in pos} <= state.atoms and { + atom_tuple(a) for a in neg + }.isdisjoint(state.atoms) + + +def _apply_effects(state: State, add: Iterable[Any], delete: Iterable[Any]) -> State: + """Add-after-delete effect application, preserving the numeric valuation.""" + return State( + (state.atoms - {atom_tuple(a) for a in delete}) + | {atom_tuple(a) for a in add}, + state.fluents, + ) + + +def apply_durative(state: State, action: "DurativeAction") -> Optional[State]: + """Apply a grounded durative action under sequential semantics. + + Returns the successor ``State``, or ``None`` when the action cannot run: + the ``at start`` conditions fail in ``state``, or the ``over all`` + invariant / ``at end`` conditions fail in the mid-state (after the start + effects — with no concurrent activity, that state holds for the whole + duration and at the end point). + """ + if not _holds(state, action.condition_pos["start"], action.condition_neg["start"]): + return None + mid = _apply_effects(state, action.effect_pos["start"], action.effect_neg["start"]) + if not _holds(mid, action.condition_pos["over"], action.condition_neg["over"]): + return None + if not _holds(mid, action.condition_pos["end"], action.condition_neg["end"]): + return None + return _apply_effects(mid, action.effect_pos["end"], action.effect_neg["end"]) + + +class TemporalTask: + """The temporal counterpart of ``GroundedTask``: initial state, goals and + every grounded action — durative, plus instantaneous ones as zero-duration + steps (so mixed domains work).""" + + def __init__(self, domainproblem: "DomainProblem") -> None: + self.domainproblem = domainproblem + self.initial: State = State.from_problem(domainproblem) + self.goals = domainproblem.goals() + self.durative: List["DurativeAction"] = [ + g + for name in domainproblem.durative_operators() + for g in domainproblem.ground_durative_operator(name) + ] + self.instantaneous: List["Operator"] = [ + g + for name in domainproblem.operators() + for g in domainproblem.ground_operator(name) + ] + + def successors(self, state: State) -> Iterator[Tuple["PlanAction", float, State]]: + """Yield ``(action, duration, successor)`` for every executable action.""" + for op in self.instantaneous: + if state.applicable(op): + yield op, 0.0, state.apply(op) + for da in self.durative: + succ = apply_durative(state, da) + if succ is not None: + yield da, float(da.duration or 0.0), succ + + def is_goal(self, state: State) -> bool: + """True if ``state`` satisfies the goal.""" + return state.satisfies(self.goals) + + +def _schedule(came_from: _CameFrom, state: State) -> TemporalPlan: + """Walk parent pointers back to the start, then lay the actions end to + end — sequential semantics: each starts when the previous one finishes.""" + pairs: List[Tuple["PlanAction", float]] = [] + while state in came_from: + prev, action, duration = came_from[state] + pairs.append((action, duration)) + state = prev + pairs.reverse() + steps: List[ScheduledAction] = [] + clock = 0.0 + for action, duration in pairs: + steps.append(ScheduledAction(action, clock, duration)) + clock += duration + return TemporalPlan(steps) + + +class TemporalPlanner(Planner): + """Uniform-cost search over accumulated duration (Dijkstra): returns a + makespan-minimal *sequential* schedule as a :class:`TemporalPlan`, or + ``None``. Durative actions are validated (#23) before search so malformed + durations fail fast.""" + + capabilities = TEMPORAL_CAPABILITIES + + def solve(self, domainproblem: "DomainProblem") -> Optional[TemporalPlan]: + validate_requirements(domainproblem) + self.check_capabilities(domainproblem) + validate_durative_actions(domainproblem) + task = TemporalTask(domainproblem) + start = task.initial + counter = itertools.count() # tie-breaker; keeps States out of compares + frontier: List[Tuple[float, int, State]] = [(0.0, next(counter), start)] + best_g: Dict[State, float] = {start: 0.0} + came_from: _CameFrom = {} + while frontier: + g, _, state = heapq.heappop(frontier) + if task.is_goal(state): + return _schedule(came_from, state) + if g > best_g.get(state, g): + continue # pragma: no cover - stale heap entry (lazy deletion) + for action, duration, succ in task.successors(state): + ng = g + duration + if ng < best_g.get(succ, ng + 1.0): + best_g[succ] = ng + came_from[succ] = (state, action, duration) + heapq.heappush(frontier, (ng, next(counter), succ)) + return None + + +register("temporal", TemporalPlanner) diff --git a/tests/corpus/temporal-charge-domain.pddl b/tests/corpus/temporal-charge-domain.pddl new file mode 100644 index 0000000..9c73bf5 --- /dev/null +++ b/tests/corpus/temporal-charge-domain.pddl @@ -0,0 +1,22 @@ +;; Mixed instantaneous + durative domain (#84): the instantaneous `plug` +;; enables the durative `charge-battery`. Instantaneous actions participate in +;; the schedule as zero-duration steps. +(define (domain charge) + (:requirements :strips :typing :durative-actions) + (:types device) + (:predicates + (plugged ?d - device) + (unplugged ?d - device) + (charged ?d - device)) + + (:action plug + :parameters (?d - device) + :precondition (unplugged ?d) + :effect (and (not (unplugged ?d)) (plugged ?d))) + + (:durative-action charge-battery + :parameters (?d - device) + :duration (= ?duration 4) + :condition (and (at start (plugged ?d)) + (over all (plugged ?d))) + :effect (at end (charged ?d)))) diff --git a/tests/corpus/temporal-charge-problem.pddl b/tests/corpus/temporal-charge-problem.pddl new file mode 100644 index 0000000..fbffd99 --- /dev/null +++ b/tests/corpus/temporal-charge-problem.pddl @@ -0,0 +1,6 @@ +;; Solvable: plug the phone (t=0), then charge it (t=0..4). Makespan 4. +(define (problem charge-phone) + (:domain charge) + (:objects phone - device) + (:init (unplugged phone)) + (:goal (charged phone))) diff --git a/tests/corpus/temporal-weld-domain.pddl b/tests/corpus/temporal-weld-domain.pddl new file mode 100644 index 0000000..c7d074e --- /dev/null +++ b/tests/corpus/temporal-weld-domain.pddl @@ -0,0 +1,38 @@ +;; Temporal demonstrator (#84): 'over all' genuinely constrains the plan. +;; +;; Both welds are applicable *at start* (they only need a held part), so an +;; at-start-only checker would happily emit the cheaper `quick-weld`. But +;; quick-weld lets go of the part *at start* while its own 'over all' invariant +;; demands the part stay held for the whole duration — so it is never actually +;; executable. Only `steady-weld`, which keeps holding the part until the end, +;; yields a valid schedule (makespan 6). +(define (domain weld) + (:requirements :strips :typing :durative-actions) + (:types part) + (:predicates + (holding ?p - part) + (welding ?p - part) + (welded ?p - part) + (torch-lit)) + + ;; Drops the clamp at start -> violates its own 'over all (holding)'. + (:durative-action quick-weld + :parameters (?p - part) + :duration (= ?duration 2) + :condition (and (at start (holding ?p)) + (over all (holding ?p)) + (at end (torch-lit))) + :effect (and (at start (not (holding ?p))) + (at end (welded ?p)))) + + ;; Keeps holding the part across the interval; releases only at end. + (:durative-action steady-weld + :parameters (?p - part) + :duration (= ?duration 6) + :condition (and (at start (holding ?p)) + (over all (holding ?p)) + (at end (torch-lit))) + :effect (and (at start (welding ?p)) + (at end (not (welding ?p))) + (at end (welded ?p)) + (at end (not (holding ?p)))))) diff --git a/tests/corpus/temporal-weld-notorch-problem.pddl b/tests/corpus/temporal-weld-notorch-problem.pddl new file mode 100644 index 0000000..8d353d5 --- /dev/null +++ b/tests/corpus/temporal-weld-notorch-problem.pddl @@ -0,0 +1,7 @@ +;; Unsolvable: no lit torch, so steady-weld's 'at end (torch-lit)' condition +;; fails and quick-weld self-violates its 'over all' -- no valid schedule. +(define (problem weld-notorch) + (:domain weld) + (:objects p1 - part) + (:init (holding p1)) + (:goal (welded p1))) diff --git a/tests/corpus/temporal-weld-problem.pddl b/tests/corpus/temporal-weld-problem.pddl new file mode 100644 index 0000000..e056937 --- /dev/null +++ b/tests/corpus/temporal-weld-problem.pddl @@ -0,0 +1,6 @@ +;; Solvable: torch is lit, so steady-weld's 'at end (torch-lit)' holds. +(define (problem weld-p1) + (:domain weld) + (:objects p1 - part) + (:init (holding p1) (torch-lit)) + (:goal (welded p1))) diff --git a/tests/test_temporal.py b/tests/test_temporal.py new file mode 100644 index 0000000..23b7398 --- /dev/null +++ b/tests/test_temporal.py @@ -0,0 +1,145 @@ +"""#84 temporal planner: schedules durative-action domains under sequential +semantics (start/end events; 'at start', 'over all', 'at end'). + +The planner enforces the full time-tagged contract that +:class:`~pddlpy.planning.DurativeState` (at-start only) cannot, applies start +effects at the start point and end effects at start+duration, and returns a +:class:`~pddlpy.planning.TemporalPlan` carrying per-step start times, durations +and a makespan. +""" +import os + +from pddlpy import DomainProblem +from pddlpy.planning import ( + DurativeState, + ScheduledAction, + TemporalPlan, + TemporalPlanner, + apply_durative, + get, +) +from pddlpy.planning.state import State + +CORPUS = os.path.join(os.path.dirname(__file__), "corpus") + + +def _dp(domain, problem): + return DomainProblem(os.path.join(CORPUS, domain), os.path.join(CORPUS, problem)) + + +def _go(): + return _dp("durative-action-domain.pddl", "durative-action-problem.pddl") + + +def _weld(problem="temporal-weld-problem.pddl"): + return _dp("temporal-weld-domain.pddl", problem) + + +def _charge(): + return _dp("temporal-charge-domain.pddl", "temporal-charge-problem.pddl") + + +# -- registry / plumbing -------------------------------------------------- +def test_registered_as_temporal(): + assert isinstance(get("temporal"), TemporalPlanner) + + +def test_declares_durative_actions_capability(): + assert ":durative-actions" in TemporalPlanner.capabilities + + +# -- simple durative schedule (go) --------------------------------------- +def test_go_schedule(): + plan = get("temporal").solve(_go()) + assert plan is not None + assert plan.makespan == 5.0 + assert plan.action_names() == [("go", {"?from": "a", "?to": "b"})] + (step,) = plan.steps + assert (step.start, step.duration, step.end) == (0.0, 5.0, 5.0) + assert plan.cost == plan.makespan + + +# -- 'over all' genuinely constrains the plan ----------------------------- +def test_over_all_forces_steady_weld(): + dp = _weld() + plan = get("temporal").solve(dp) + assert plan is not None + # steady-weld (makespan 6) is picked over the shorter quick-weld, which an + # at-start-only checker would (wrongly) accept. + assert plan.action_names() == [("steady-weld", {"?p": "p1"})] + assert plan.makespan == 6.0 + + +def test_quick_weld_naive_vs_temporal(): + dp = _weld() + quick = next(g for g in dp.ground_durative_operator("quick-weld")) + # An at-start-only view accepts quick-weld ... + assert DurativeState.from_problem(dp).applicable(quick) is True + # ... but the temporal planner rejects it: its start effect drops the part, + # violating its own 'over all (holding)' invariant. + assert apply_durative(State.from_problem(dp), quick) is None + + +def test_at_end_condition_unsatisfiable_is_unsolvable(): + # No lit torch: steady-weld's 'at end (torch-lit)' fails, quick-weld + # self-violates 'over all' -> no schedule. + assert get("temporal").solve(_weld("temporal-weld-notorch-problem.pddl")) is None + + +# -- mixed instantaneous + durative -------------------------------------- +def test_mixed_instantaneous_and_durative(): + plan = get("temporal").solve(_charge()) + assert plan is not None + assert plan.makespan == 4.0 + assert plan.action_names() == [ + ("plug", {"?d": "phone"}), + ("charge-battery", {"?d": "phone"}), + ] + plug, charge = plan.steps + assert (plug.start, plug.duration) == (0.0, 0.0) # zero-duration step + assert (charge.start, charge.duration, charge.end) == (0.0, 4.0, 4.0) + + +# -- effect timing: start effects at start, end effects at start+duration -- +def test_start_and_end_effect_timing(): + dp = _go() + state = State.from_problem(dp) + go_ab = next( + g + for g in dp.ground_durative_operator("go") + if g.variable_list == {"?from": "a", "?to": "b"} + ) + succ = apply_durative(state, go_ab) + assert succ is not None + # start effect removed (at a); end effects added (at b), (visited b). + assert ("at", "a") not in succ.atoms + assert {("at", "b"), ("visited", "b")} <= succ.atoms + + +def test_at_start_condition_gates_application(): + dp = _go() + state = State.from_problem(dp) # (at a), (road a b) + go_ba = next( + g + for g in dp.ground_durative_operator("go") + if g.variable_list == {"?from": "b", "?to": "a"} + ) + # 'at start (at b)' does not hold in the initial state. + assert apply_durative(state, go_ba) is None + + +# -- value types ---------------------------------------------------------- +def test_scheduled_action_repr_and_end(): + plan = get("temporal").solve(_go()) + (step,) = plan.steps + assert isinstance(step, ScheduledAction) + assert step.end == step.start + step.duration + assert "go(a, b)" in repr(step) + + +def test_temporal_plan_repr(): + plan = get("temporal").solve(_go()) + assert isinstance(plan, TemporalPlan) + text = repr(plan) + assert "makespan=5" in text + assert "go(a, b)" in text