diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..c8be00f --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,189 @@ +name: release + +# Publish a release to PyPI and cut the GitHub Release to match. +# +# Push a tag and this does the rest: +# +# git tag -a v2.3.0 -m "jupyddl 2.3.0" && git push origin v2.3.0 +# +# Authentication is PyPI **Trusted Publishing** (OIDC), so there is no API +# token in this repository's secrets to leak or rotate. That requires a +# one-time setup on PyPI — see `docs/RELEASING.md`. Until it is done the +# `pypi` job fails and everything before it still succeeds, so a tag never +# leaves you with half a release and no artifacts. +# +# `workflow_dispatch` runs the same pipeline against TestPyPI, which is how to +# validate a change to this file without spending a real version number. PyPI +# releases are effectively permanent: a version can be yanked but never +# replaced, so the dry run is worth the two minutes. + +on: + push: + tags: ["v*"] + workflow_dispatch: + inputs: + target: + description: "Where to publish" + required: true + default: testpypi + type: choice + options: [testpypi, pypi] + +concurrency: + group: release-${{ github.ref }} + cancel-in-progress: false + +permissions: + contents: read + +jobs: + build: + name: build and verify + runs-on: ubuntu-latest + outputs: + version: ${{ steps.version.outputs.version }} + steps: + - uses: actions/checkout@v7 + with: + submodules: false + fetch-depth: 0 + + - uses: actions/setup-python@v7 + with: + python-version: '3.12' + + - name: Read the version the package declares + id: version + run: | + version=$(python -c "import re,pathlib; \ + print(re.search(r'^version = \"([^\"]+)\"', \ + pathlib.Path('pyproject.toml').read_text(), re.M).group(1))") + echo "version=$version" >> "$GITHUB_OUTPUT" + echo "pyproject declares $version" + + # A tag that disagrees with the package version publishes something + # nobody asked for under a name nobody expects, and PyPI will not let + # you take it back. Refuse before anything is built. + - name: Refuse a tag that disagrees with the package version + if: startsWith(github.ref, 'refs/tags/') + run: | + tag="${GITHUB_REF_NAME#v}" + declared="${{ steps.version.outputs.version }}" + if [ "$tag" != "$declared" ]; then + echo "::error::tag ${GITHUB_REF_NAME} does not match pyproject version ${declared}." \ + "Retag, or fix the version, but do not publish a mismatch." + exit 1 + fi + echo "tag ${GITHUB_REF_NAME} matches ${declared}" + + # jupyddl exports __version__ as well, and a package whose metadata and + # runtime disagree is a support ticket waiting to happen. + - name: Refuse a package whose __version__ disagrees with its metadata + run: | + runtime=$(python -c "import re,pathlib; \ + print(re.search(r'^__version__ = \"([^\"]+)\"', \ + pathlib.Path('jupyddl/__init__.py').read_text(), re.M).group(1))") + declared="${{ steps.version.outputs.version }}" + if [ "$runtime" != "$declared" ]; then + echo "::error::jupyddl.__version__ is ${runtime} but pyproject says ${declared}" + exit 1 + fi + echo "__version__ matches ${declared}" + + - name: Build the sdist and the wheel + run: | + python -m pip install --upgrade pip build twine + python -m build + twine check --strict dist/* + + # The same smoke test the `build` workflow runs on every push, repeated + # here because this is the artefact that actually goes out. + - name: Install the wheel clean and plan with it + working-directory: /tmp + run: | + python -m venv /tmp/fresh + /tmp/fresh/bin/pip install "$GITHUB_WORKSPACE"/dist/*.whl + installed=$(/tmp/fresh/bin/python -c "import jupyddl; print(jupyddl.__version__)") + if [ "$installed" != "${{ steps.version.outputs.version }}" ]; then + echo "::error::installed wheel reports $installed" + exit 1 + fi + /tmp/fresh/bin/jupyddl generate gripper -n 3 --seed 1 -o /tmp/smoke + /tmp/fresh/bin/jupyddl solve \ + /tmp/smoke/gripper-03-1/domain.pddl \ + /tmp/smoke/gripper-03-1/problem.pddl \ + -s astar -H lmcut | tee /tmp/plan.txt + grep -q "Valid: True" /tmp/plan.txt + + - name: Extract this version's changelog section + run: | + python - <<'PY' > release-notes.md + import pathlib, re + version = "${{ steps.version.outputs.version }}" + text = pathlib.Path("CHANGELOG.md").read_text(encoding="utf-8") + # Everything between this version's heading and the next one. + match = re.search( + rf"^## \[{re.escape(version)}\][^\n]*\n(.*?)(?=^## \[|\Z)", + text, re.M | re.S, + ) + print(match.group(1).strip() if match else + f"See CHANGELOG.md for what changed in {version}.") + PY + cat release-notes.md + + - uses: actions/upload-artifact@v4 + with: + name: distributions + path: | + dist/ + release-notes.md + + pypi: + name: publish to ${{ github.event.inputs.target || 'pypi' }} + needs: build + runs-on: ubuntu-latest + # The environment is what a PyPI trusted publisher is scoped to, and it is + # also where a required reviewer can be attached if you ever want a human + # to approve each publish. + environment: + name: ${{ github.event.inputs.target || 'pypi' }} + url: https://pypi.org/project/jupyddl/${{ needs.build.outputs.version }} + permissions: + id-token: write # mint the OIDC token PyPI verifies; no secrets involved + steps: + - uses: actions/download-artifact@v4 + with: + name: distributions + + - name: Publish + uses: pypa/gh-action-pypi-publish@release/v1 + with: + repository-url: >- + ${{ github.event.inputs.target == 'testpypi' + && 'https://test.pypi.org/legacy/' || '' }} + # A tag re-run after a partial failure should not fall over on the + # files that already made it. + skip-existing: true + + github-release: + name: cut the GitHub release + needs: [build, pypi] + if: startsWith(github.ref, 'refs/tags/') + runs-on: ubuntu-latest + permissions: + contents: write # create the release and attach the distributions + steps: + - uses: actions/download-artifact@v4 + with: + name: distributions + + - name: Create the release + uses: softprops/action-gh-release@v2 + with: + name: jupyddl ${{ needs.build.outputs.version }} + body_path: release-notes.md + files: dist/* + draft: false + prerelease: ${{ contains(needs.build.outputs.version, 'rc') + || contains(needs.build.outputs.version, 'a') + || contains(needs.build.outputs.version, 'b') }} diff --git a/AGENTS.md b/AGENTS.md index 2d9b9a6..ccc5de7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,6 +42,14 @@ native build step, and the core has zero runtime dependencies. rebuilds `web/dist` in the same commit; reformatting a bundled source and committing without the rebuild would leave main in a state where `pages` refuses to deploy. +- **`release`** — fires on a `v*` tag. Builds, refuses a tag that disagrees + with `pyproject.toml` or a `__version__` that disagrees with either, runs + `twine check --strict`, installs the wheel clean and plans with it, publishes + to PyPI over OIDC (no stored token), then cuts the GitHub Release from the + changelog section for that version. `docs/RELEASING.md` is the runbook, + including the one-time PyPI trusted-publisher setup only a maintainer can do. + **Bump the version in two places** — `pyproject.toml` and + `jupyddl/__init__.py` — and rebuild `web/dist`, which carries it too. - **`pages`** — bundles, refuses to deploy a stale `web/dist`, then deploys. Its job is called `bundle`, not `build`, because `.mergify.yml` keys a merge rule on a check named `build` and that has to mean the packaging workflow. diff --git a/CHANGELOG.md b/CHANGELOG.md index cc01cf7..8364710 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [2.3.0] - 2026-07-31 ### Added - **`jupyddl.learn`: heuristics trained from your own solved plans.** Every @@ -24,7 +24,8 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). distribution shift, bootstrapping for the instances too hard to label, and the cross-entropy method over the weight vector with the planner as a black box. On blocksworld this took a held-out set from 366 expansions and 0.90 - coverage to 131 and 1.00. + coverage to 137 and 1.00 — though see `.docs/rl-for-search.md`: that mean is + dominated by one hard instance, and the durable claim is the coverage one. - **`jupyddl learn`**, and `learned:` accepted anywhere a heuristic name is — `solve`, `benchmark`, the API. `make_heuristic` also passes through an already-built heuristic, so callers holding a trained model need not diff --git a/README.md b/README.md index 7db3282..3853278 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,8 @@ actually *watch*. ✨
+[![PyPI](https://img.shields.io/pypi/v/jupyddl.svg)](https://pypi.org/project/jupyddl/) +[![Python versions](https://img.shields.io/pypi/pyversions/jupyddl.svg)](https://pypi.org/project/jupyddl/) ![tests](https://github.com/APLA-Toolbox/PythonPDDL/workflows/tests/badge.svg?branch=main) ![build](https://github.com/APLA-Toolbox/PythonPDDL/workflows/build/badge.svg?branch=main) [![GitHub license](https://img.shields.io/github/license/Apla-Toolbox/PythonPDDL.svg)](./LICENSE) @@ -70,11 +72,19 @@ is trivial to install, embed, teach with, and build on. ## Install 💾 -Requires Python ≥ 3.9. Using [uv](https://docs.astral.sh/uv/) (recommended): +Requires Python ≥ 3.9, and nothing else: + +```bash +pip install jupyddl # the framework and the CLI +pip install "jupyddl[viz]" # + matplotlib, for the charts and animations +pip install "jupyddl[viz,learn]" # + numpy, which only makes learning faster +``` + +To work on it, from a clone, using [uv](https://docs.astral.sh/uv/): ```bash uv venv -uv pip install -e ".[dev,viz,learn]" # viz = matplotlib charts, learn = numpy (speed only) +uv pip install -e ".[dev,viz,learn]" ``` or with plain pip: diff --git a/docs/RELEASING.md b/docs/RELEASING.md new file mode 100644 index 0000000..77390c5 --- /dev/null +++ b/docs/RELEASING.md @@ -0,0 +1,110 @@ +# Releasing jupyddl + +Releases are cut by pushing a tag. `.github/workflows/release.yml` builds, +verifies, publishes to PyPI and creates the GitHub Release from the changelog. + +```bash +git tag -a v2.3.0 -m "jupyddl 2.3.0" +git push origin v2.3.0 +``` + +## One-time setup on PyPI (a human has to do this) + +The workflow authenticates with **Trusted Publishing** (OIDC), so there is no +API token stored in this repository — nothing to leak, nothing to rotate, and +nothing that keeps working if someone walks off with a laptop. The trade is one +piece of setup that can only be done by a PyPI maintainer of the project. + +On , add a +publisher with **exactly** these values: + +| Field | Value | +|---|---| +| Owner | `APLA-Toolbox` | +| Repository name | `PythonPDDL` | +| Workflow name | `release.yml` | +| Environment name | `pypi` | + +The environment name matters: the `pypi` job declares +`environment: pypi`, and PyPI will reject a token minted from anywhere else. +That scoping is the point — a workflow added later by someone else cannot +publish unless it also runs in that environment. + +Repeat on with environment name `testpypi` if you want +the dry run below to work. + +Until this exists the `pypi` job fails with an OIDC error. Everything before it +still succeeds, so a tag pushed early leaves you with verified artifacts and no +partial publish — re-run the job once the publisher is configured. + +### Optional: require a human to approve each publish + +In **Settings → Environments → pypi**, add yourself as a required reviewer. +GitHub then pauses the `pypi` job until someone approves it. Worth it if you +would rather a mistaken tag not reach PyPI unattended. + +## Dry run + +Validate the whole pipeline without spending a version number: + +**Actions → release → Run workflow → target: `testpypi`** + +A PyPI release is effectively permanent. A version can be *yanked*, which hides +it from resolvers, but it can never be replaced or re-uploaded — so the version +number is spent either way. Two minutes on TestPyPI is cheap next to that. + +## Cutting a release + +1. **Land everything on `main`** and confirm CI is green. +2. **Bump the version in two places** — they are checked against each other and + against the tag, and a mismatch fails the build rather than publishing a + surprise: + - `pyproject.toml` → `version` + - `jupyddl/__init__.py` → `__version__` +3. **Close the changelog section.** Rename `## [Unreleased]` to + `## [X.Y.Z] - YYYY-MM-DD`. The workflow extracts exactly this section as the + GitHub Release body, so what you write here is what people read. +4. **Rebuild the browser bundle** — `web/dist/build.json` carries the version: + ```bash + python tools/build_web.py + ``` +5. Commit, push, merge. +6. **Tag the merge commit** and push the tag. + +## What the workflow refuses to do + +Each of these is a way a release goes wrong quietly, so each one is a hard +failure rather than a warning: + +- **Tag disagrees with `pyproject.toml`.** Publishing `2.3.0` from a tag reading + `v2.4.0` cannot be undone. +- **`jupyddl.__version__` disagrees with the metadata.** A package that reports + a different version at runtime than the one you installed is a support ticket + with no obvious cause. +- **`twine check --strict` finds anything.** Malformed metadata renders as raw + text on the project page and is only fixable with another release. +- **The built wheel does not work.** It is installed into a clean environment, + away from the source tree, and made to generate and solve an instance. An + editable install hides a module missing from the wheel; this does not. + +## Versioning + +[Semantic versioning](https://semver.org). In practice: + +- **patch** — fixes that change no API and no plan output. +- **minor** — new planners, heuristics, generators, CLI commands, or PDDL + requirement support. Everything that worked still works. +- **major** — a removal or a change in behaviour that existing code would + notice. The 1.0.0 rewrite that removed Julia is the model. + +Note that changing what a heuristic *returns* can change which plan comes back +even when the API is untouched. Plans are not part of the compatibility +promise; costs and validity are. + +## History + +PyPI carried `0.4.1` — the original Julia-backed wrapper — for a long time +after the pure-Python rewrite landed here, because there was no release +automation and nobody published by hand. `pip install jupyddl` therefore gave +people a different library than this repository documented. That is what this +workflow exists to prevent. diff --git a/jupyddl/__init__.py b/jupyddl/__init__.py index 6ff50c0..e9c0cb1 100644 --- a/jupyddl/__init__.py +++ b/jupyddl/__init__.py @@ -29,7 +29,7 @@ TraceRecorder, ) -__version__ = "2.2.0" +__version__ = "2.3.0" __all__ = [ "solve", diff --git a/pyproject.toml b/pyproject.toml index 59e1bd6..4547728 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "jupyddl" -version = "2.2.0" +version = "2.3.0" description = "A pure-Python PDDL planning framework: parser, grounding, classical-to-SOTA planners, heuristics and benchmarking." readme = "README.md" requires-python = ">=3.9" @@ -12,8 +12,21 @@ license = { text = "Apache-2.0" } authors = [{ name = "Erwin Lejeune" }] keywords = ["pddl", "planning", "heuristic-search", "a-star", "lm-cut", "automated-planning"] classifiers = [ - "Programming Language :: Python :: 3", + # Beta rather than Stable on purpose: the planning core has been steady for + # a while, but `jupyddl.learn` is new and its API will move. + "Development Status :: 4 - Beta", + "Intended Audience :: Science/Research", + "Intended Audience :: Developers", "License :: OSI Approved :: Apache Software License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: Implementation :: CPython", "Topic :: Scientific/Engineering :: Artificial Intelligence", ] # The core framework is intentionally dependency-free (stdlib only) so it is @@ -36,6 +49,9 @@ jupyddl = "jupyddl.cli:main" [project.urls] Homepage = "https://github.com/APLA-Toolbox/PythonPDDL" Repository = "https://github.com/APLA-Toolbox/PythonPDDL" +Changelog = "https://github.com/APLA-Toolbox/PythonPDDL/blob/main/CHANGELOG.md" +Issues = "https://github.com/APLA-Toolbox/PythonPDDL/issues" +Workbench = "https://apla-toolbox.github.io/PythonPDDL/" [tool.hatch.build.targets.wheel] packages = ["jupyddl"] diff --git a/web/dist/build.json b/web/dist/build.json index 10c6cce..b444326 100644 --- a/web/dist/build.json +++ b/web/dist/build.json @@ -1 +1 @@ -{"version": "2.2.0", "modules": 32} \ No newline at end of file +{"version": "2.3.0", "modules": 32} \ No newline at end of file diff --git a/web/dist/jupyddl-sources.json b/web/dist/jupyddl-sources.json index 9443289..7158e01 100644 --- a/web/dist/jupyddl-sources.json +++ b/web/dist/jupyddl-sources.json @@ -1 +1 @@ -{"jupyddl/__init__.py": "\"\"\"jupyddl: a pure-Python PDDL planning framework.\n\nQuickstart::\n\n from jupyddl import solve, build_task, trace_search, validate_plan\n\n result = solve(\"domain.pddl\", \"problem.pddl\", search=\"astar\", heuristic=\"lmcut\")\n print(result.solved, result.cost, result.plan_names())\n\n # ...and watch the search itself\n task = build_task(\"domain.pddl\", \"problem.pddl\")\n result, trace = trace_search(task, \"astar\", \"lmcut\")\n trace.save(\"run.json\")\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom .api import build_task, solve, solve_task, trace_search, validate_plan\nfrom .grounding import ground, ground_files\nfrom .heuristics import HEURISTICS, make_heuristic\nfrom .parser import PDDLError, UnsupportedFeatureError, parse\nfrom .search import PLANNERS, SearchResult, make_planner\nfrom .task import Operator, Task\nfrom .trace import (\n MultiObserver,\n SearchEvent,\n SearchObserver,\n SearchTrace,\n TraceRecorder,\n)\n\n__version__ = \"2.2.0\"\n\n__all__ = [\n \"solve\",\n \"solve_task\",\n \"build_task\",\n \"trace_search\",\n \"validate_plan\",\n \"ground\",\n \"ground_files\",\n \"make_planner\",\n \"make_heuristic\",\n \"PLANNERS\",\n \"HEURISTICS\",\n \"SearchResult\",\n \"SearchTrace\",\n \"SearchEvent\",\n \"SearchObserver\",\n \"MultiObserver\",\n \"TraceRecorder\",\n \"Task\",\n \"Operator\",\n \"parse\",\n \"PDDLError\",\n \"UnsupportedFeatureError\",\n \"__version__\",\n]\n", "jupyddl/api.py": "\"\"\"High-level convenience API: parse + ground + plan + validate.\"\"\"\n\nfrom __future__ import annotations\n\nfrom .grounding import ground_files\nfrom .heuristics import make_heuristic\nfrom .search import make_planner\nfrom .search.result import SearchResult, make_budget\nfrom .task import Task\nfrom .trace import TraceRecorder\n\n\ndef build_task(domain_path: str, problem_path: str) -> Task:\n \"\"\"Parse and ground a domain/problem pair into a :class:`Task`.\"\"\"\n return ground_files(domain_path, problem_path)\n\n\ndef solve_task(\n task: Task,\n search: str = \"astar\",\n heuristic=None,\n observer=None,\n max_expansions=None,\n time_limit=None,\n **planner_kwargs,\n) -> SearchResult:\n \"\"\"Run ``search`` (optionally with ``heuristic``) on an already-ground task.\n\n Pass ``observer`` (see :mod:`jupyddl.trace`) to record or live-render the\n search as it runs. ``max_expansions`` and ``time_limit`` bound the run; when\n either is hit the planner stops and sets ``result.truncated``, so an empty\n result means \"gave up\", not \"proved unsolvable\".\n \"\"\"\n planner = make_planner(search, **planner_kwargs)\n heur = None\n name = heuristic if heuristic else (\"hff\" if planner.requires_heuristic else None)\n if name is not None:\n heur = make_heuristic(name, task)\n budget = make_budget(max_expansions, time_limit)\n return planner.search(task, heur, observer=observer, budget=budget)\n\n\ndef solve(\n domain_path: str,\n problem_path: str,\n search: str = \"astar\",\n heuristic=\"lmcut\",\n observer=None,\n max_expansions=None,\n time_limit=None,\n **planner_kwargs,\n) -> SearchResult:\n \"\"\"Parse, ground and solve a PDDL instance in one call.\"\"\"\n task = build_task(domain_path, problem_path)\n return solve_task(\n task,\n search=search,\n heuristic=heuristic,\n observer=observer,\n max_expansions=max_expansions,\n time_limit=time_limit,\n **planner_kwargs,\n )\n\n\ndef trace_search(\n task: Task,\n search: str = \"astar\",\n heuristic=None,\n max_events: int = 20000,\n record_generated: bool = False,\n observer=None,\n max_expansions=None,\n time_limit=None,\n **planner_kwargs,\n):\n \"\"\"Solve ``task`` while recording the search.\n\n Returns ``(result, trace)`` where ``trace`` is a\n :class:`~jupyddl.trace.SearchTrace` ready to plot, save or replay. An extra\n ``observer`` (a live dashboard, say) is notified alongside the recorder.\n \"\"\"\n recorder = TraceRecorder(max_events=max_events, record_generated=record_generated)\n if observer is not None:\n from .trace import MultiObserver\n\n sink = MultiObserver(recorder, observer)\n else:\n sink = recorder\n result = solve_task(\n task,\n search=search,\n heuristic=heuristic,\n observer=sink,\n max_expansions=max_expansions,\n time_limit=time_limit,\n **planner_kwargs,\n )\n return result, recorder.trace\n\n\ndef validate_plan(task: Task, plan) -> bool:\n \"\"\"Return ``True`` iff applying ``plan`` from the initial state reaches the goal.\n\n Replays through the task rather than the raw operators, so derived\n predicates are closed and numeric fluents are carried at every step \u2014\n validating against ``task.init`` directly would see neither.\n \"\"\"\n state = task.initial_state()\n for op in plan or ():\n if not op.applicable(state):\n return False\n state = task.apply(op, state)\n return task.goal_reached(state)\n", "jupyddl/benchmark.py": "\"\"\"Comparative benchmarking of planners/heuristics over PDDL instances.\n\nExample::\n\n from jupyddl.benchmark import discover_instances, run_benchmark, to_csv\n rows = run_benchmark(discover_instances(\"pddl-examples\"),\n [(\"astar\", \"lmcut\"), (\"gbfs\", \"hff\")])\n to_csv(rows, \"results.csv\")\n\"\"\"\n\nfrom __future__ import annotations\n\nimport csv\nimport glob\nimport os\nfrom dataclasses import asdict, dataclass\nfrom typing import Optional\n\nfrom .api import build_task, solve_task, validate_plan\nfrom .parser import PDDLError\n\n\n@dataclass\nclass Instance:\n name: str\n domain: str\n problem: str\n\n\n@dataclass\nclass BenchmarkRow:\n instance: str\n planner: str\n heuristic: str\n solved: bool\n valid: bool\n cost: Optional[int]\n plan_length: Optional[int]\n expanded: int\n generated: int\n evaluated: int\n runtime: float\n error: str = \"\"\n truncated: bool = False\n\n\ndef discover_instances(root: str) -> list:\n \"\"\"Find ``/*/`` folders containing both ``domain.pddl`` and ``problem.pddl``.\"\"\"\n instances = []\n for domain in sorted(glob.glob(os.path.join(root, \"*\", \"domain.pddl\"))):\n folder = os.path.dirname(domain)\n problem = os.path.join(folder, \"problem.pddl\")\n if os.path.exists(problem):\n instances.append(Instance(os.path.basename(folder), domain, problem))\n return instances\n\n\ndef run_benchmark(instances, configs, max_expansions=None, time_limit=None) -> list:\n \"\"\"Run each ``(planner, heuristic[, kwargs])`` config on each instance.\n\n ``configs`` entries are ``(planner_name, heuristic_name_or_None)`` or\n ``(planner_name, heuristic_name_or_None, planner_kwargs)``.\n\n ``max_expansions`` and ``time_limit`` bound every individual run, which is\n what keeps one pathological instance from stalling a whole benchmark. A run\n that stops on its budget is recorded with ``truncated=True`` and does not\n count towards coverage -- \"we stopped looking\" is not \"no plan exists\".\n \"\"\"\n rows: list = []\n for inst in instances:\n try:\n task = build_task(inst.domain, inst.problem)\n except (PDDLError, ValueError) as exc:\n for cfg in configs:\n planner, heuristic = cfg[0], (cfg[1] or \"\")\n rows.append(\n BenchmarkRow(\n inst.name,\n planner,\n heuristic,\n False,\n False,\n None,\n None,\n 0,\n 0,\n 0,\n 0.0,\n f\"{type(exc).__name__}: {exc}\",\n )\n )\n continue\n for cfg in configs:\n planner = cfg[0]\n heuristic = cfg[1]\n kwargs = cfg[2] if len(cfg) > 2 else {}\n try:\n result = solve_task(\n task,\n planner,\n heuristic,\n max_expansions=max_expansions,\n time_limit=time_limit,\n **kwargs,\n )\n valid = bool(result.solved and validate_plan(task, result.plan))\n rows.append(\n BenchmarkRow(\n inst.name,\n planner,\n heuristic or \"\",\n result.solved,\n valid,\n result.cost,\n result.plan_length,\n result.stats.expanded,\n result.stats.generated,\n result.stats.evaluated,\n round(result.stats.runtime, 6),\n \"\",\n result.stats.truncated,\n )\n )\n except Exception as exc: # keep the benchmark going on a single failure\n rows.append(\n BenchmarkRow(\n inst.name,\n planner,\n heuristic or \"\",\n False,\n False,\n None,\n None,\n 0,\n 0,\n 0,\n 0.0,\n f\"{type(exc).__name__}: {exc}\",\n )\n )\n return rows\n\n\ndef to_csv(rows, path: str) -> None:\n fieldnames = (\n list(asdict(rows[0]).keys())\n if rows\n else [f.name for f in BenchmarkRow.__dataclass_fields__.values()]\n )\n with open(path, \"w\", newline=\"\", encoding=\"utf-8\") as handle:\n writer = csv.DictWriter(handle, fieldnames=fieldnames)\n writer.writeheader()\n for row in rows:\n writer.writerow(asdict(row))\n\n\ndef summarize(rows) -> dict:\n \"\"\"Aggregate coverage and totals per ``planner/heuristic`` configuration.\"\"\"\n summary: dict = {}\n for row in rows:\n key = f\"{row.planner}/{row.heuristic}\" if row.heuristic else row.planner\n agg = summary.setdefault(\n key, {\"coverage\": 0, \"expanded\": 0, \"runtime\": 0.0, \"instances\": 0}\n )\n agg[\"instances\"] += 1\n agg[\"coverage\"] += int(row.valid)\n agg[\"expanded\"] += row.expanded\n agg[\"runtime\"] += row.runtime\n return summary\n\n\ndef plot_summary(rows, path: str, metric: str = \"expanded\") -> None:\n \"\"\"Bar chart of a metric per configuration (requires the ``viz`` extra).\"\"\"\n import matplotlib\n\n matplotlib.use(\"Agg\")\n import matplotlib.pyplot as plt\n\n summary = summarize(rows)\n labels = list(summary.keys())\n values = [summary[k][metric] for k in labels]\n fig, ax = plt.subplots(figsize=(max(6, len(labels) * 0.9), 4))\n ax.bar(labels, values, color=\"#4C72B0\")\n ax.set_ylabel(f\"total {metric}\")\n ax.set_title(f\"Planner comparison ({metric})\")\n plt.xticks(rotation=45, ha=\"right\")\n plt.tight_layout()\n fig.savefig(path, dpi=120)\n plt.close(fig)\n", "jupyddl/cli.py": "\"\"\"Command-line interface.\n\n``solve``, ``benchmark``, ``animate``, ``demo``, ``requirements``, ``generate``\nand ``learn``. Every long-running command accepts ``--max-expansions`` and\n``--time-limit``; when a search stops on one of those it says so rather than\nreporting the instance unsolvable.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport os\nimport sys\n\nfrom .api import build_task, solve_task, trace_search, validate_plan\nfrom .benchmark import (\n discover_instances,\n plot_summary,\n run_benchmark,\n summarize,\n to_csv,\n)\nfrom .heuristics import HEURISTICS, LOADERS\nfrom .search import INFORMED_PLANNERS, PLANNERS\n\nINFORMED = set(INFORMED_PLANNERS)\n\n\ndef heuristic_spec(value: str) -> str:\n \"\"\"Validate ``-H``: a registry name, ``none``, or ``kind:argument``.\n\n ``choices=`` cannot express this \u2014 a trained heuristic is identified by a\n path that does not exist until someone trains one \u2014 so the check is a type\n function instead. It still rejects typos eagerly, which is the only reason\n ``choices=`` was worth having.\n \"\"\"\n if value in HEURISTICS or value == \"none\":\n return value\n kind, sep, argument = value.partition(\":\")\n if sep and kind in LOADERS:\n if not argument:\n raise argparse.ArgumentTypeError(\n f\"'{kind}:' needs an argument, e.g. {kind}:model.json\"\n )\n return value\n raise argparse.ArgumentTypeError(\n f\"unknown heuristic '{value}'; expected one of \"\n f\"{sorted(HEURISTICS) + ['none']} \"\n f\"or {'/'.join(sorted(LOADERS))}:\"\n )\n\n\ndef _add_solve(sub):\n p = sub.add_parser(\"solve\", help=\"solve a single PDDL instance\")\n p.add_argument(\"domain\")\n p.add_argument(\"problem\")\n p.add_argument(\"-s\", \"--search\", default=\"astar\", choices=sorted(PLANNERS))\n p.add_argument(\n \"-H\",\n \"--heuristic\",\n default=\"lmcut\",\n type=heuristic_spec,\n metavar=\"NAME\",\n help=\"a heuristic name, none, or learned:\",\n )\n p.add_argument(\n \"-w\", \"--weight\", type=float, default=2.0, help=\"weight for weighted A*\"\n )\n p.add_argument(\n \"--live\",\n action=\"store_true\",\n help=\"watch the search live in the terminal (no dependencies)\",\n )\n p.add_argument(\"--trace\", default=None, help=\"write the search trace as JSON\")\n p.add_argument(\n \"--plot\",\n default=None,\n help=\"write a four-panel search-progress chart (PNG; needs the viz extra)\",\n )\n p.add_argument(\n \"--tree\", default=None, help=\"write the radial search-wavefront chart (PNG)\"\n )\n p.add_argument(\n \"--plan-plot\", default=None, help=\"write the plan timeline chart (PNG)\"\n )\n p.add_argument(\"--dark\", action=\"store_true\", help=\"render charts for dark mode\")\n p.add_argument(\n \"--max-expansions\",\n type=int,\n default=None,\n help=\"stop after this many node expansions and report what was found\",\n )\n p.add_argument(\n \"--time-limit\",\n type=float,\n default=None,\n help=\"stop after this many seconds and report what was found\",\n )\n p.add_argument(\"--quiet\", action=\"store_true\", help=\"do not print the plan\")\n p.set_defaults(func=_cmd_solve)\n\n\ndef _add_benchmark(sub):\n p = sub.add_parser(\"benchmark\", help=\"compare planners over a folder of instances\")\n p.add_argument(\"root\", help=\"folder containing /domain.pddl + problem.pddl\")\n p.add_argument(\"--planners\", default=\"bfs,dijkstra,astar,gbfs,wastar,ehc\")\n p.add_argument(\"--heuristic\", default=\"hff\", help=\"heuristic for informed planners\")\n p.add_argument(\"--csv\", default=None, help=\"write per-run results to this CSV\")\n p.add_argument(\"--plot\", default=None, help=\"write a comparison bar chart (PNG)\")\n p.add_argument(\n \"--dashboard\",\n default=None,\n help=\"write the full benchmark dashboard: coverage, effort, time, heatmap\",\n )\n p.add_argument(\"--metric\", default=\"expanded\")\n p.add_argument(\"--dark\", action=\"store_true\", help=\"render charts for dark mode\")\n p.add_argument(\n \"--max-expansions\",\n type=int,\n default=None,\n help=\"stop after this many node expansions and report what was found\",\n )\n p.add_argument(\n \"--time-limit\",\n type=float,\n default=None,\n help=\"stop after this many seconds and report what was found\",\n )\n p.set_defaults(func=_cmd_benchmark)\n\n\ndef _add_animate(sub):\n p = sub.add_parser(\"animate\", help=\"replay a search as an animation (MP4 or GIF)\")\n p.add_argument(\"domain\")\n p.add_argument(\"problem\")\n p.add_argument(\"-o\", \"--output\", default=\"search.mp4\")\n p.add_argument(\"-s\", \"--search\", default=\"astar\", choices=sorted(PLANNERS))\n p.add_argument(\n \"-H\",\n \"--heuristic\",\n default=\"lmcut\",\n type=heuristic_spec,\n metavar=\"NAME\",\n help=\"a heuristic name, none, or learned:\",\n )\n p.add_argument(\"--fps\", type=int, default=30)\n p.add_argument(\"--seconds\", type=float, default=8.0)\n p.add_argument(\"--dark\", action=\"store_true\")\n p.set_defaults(func=_cmd_animate)\n\n\ndef _add_demo(sub):\n p = sub.add_parser(\n \"demo\",\n help=\"run the bundled demo instances and write every chart to a folder\",\n )\n p.add_argument(\"-o\", \"--output\", default=\"gallery\", help=\"output folder\")\n p.add_argument(\"--root\", default=\"demos\", help=\"folder of demo instances\")\n p.add_argument(\n \"--both-modes\",\n action=\"store_true\",\n help=\"render a light and a dark variant of every chart\",\n )\n p.add_argument(\n \"--animate\", action=\"store_true\", help=\"also render the search animations\"\n )\n p.set_defaults(func=_cmd_demo)\n\n\n# --------------------------------------------------------------------------\ndef _observers(args):\n \"\"\"Build the observer for a solve, honouring --live/--trace/--plot.\"\"\"\n from .trace import MultiObserver, TraceRecorder\n\n wants_trace = bool(\n args.trace\n or args.plot\n or getattr(args, \"tree\", None)\n or getattr(args, \"plan_plot\", None)\n )\n recorder = TraceRecorder() if wants_trace else None\n dashboard = None\n if args.live:\n from .live import TerminalDashboard\n\n dashboard = TerminalDashboard()\n if recorder and dashboard:\n return MultiObserver(recorder, dashboard), recorder\n return (dashboard or recorder), recorder\n\n\ndef _cmd_solve(args) -> int:\n task = build_task(args.domain, args.problem)\n heuristic = None if args.heuristic == \"none\" else args.heuristic\n kwargs = {\"weight\": args.weight} if args.search == \"wastar\" else {}\n observer, recorder = _observers(args)\n result = solve_task(\n task,\n args.search,\n heuristic,\n observer=observer,\n max_expansions=args.max_expansions,\n time_limit=args.time_limit,\n **kwargs,\n )\n\n trace = recorder.trace if recorder else None\n if trace is not None:\n if args.trace:\n trace.save(args.trace)\n print(f\"Wrote trace to {args.trace}\")\n _write_charts(args, trace)\n\n if not result.solved:\n if not args.live:\n if result.truncated:\n print(\n \"No plan found within the budget \"\n \"(the instance may still be solvable).\"\n )\n else:\n print(\"No plan found.\")\n _print_stats(result)\n return 1\n valid = validate_plan(task, result.plan)\n visible = task.visible_plan(result.plan)\n if not args.quiet:\n header = f\"Plan ({len(visible)} steps, cost {result.cost}\"\n if task.temporal:\n header += f\", makespan {task.makespan(result.plan):g}\"\n print(header + \"):\")\n for i, op in enumerate(visible):\n print(f\" {i + 1:3d}. {op.base_name}\")\n print(f\"Valid: {valid}\")\n if not args.live:\n _print_stats(result)\n return 0 if valid else 2\n\n\ndef _write_charts(args, trace) -> None:\n targets = [\n (args.plot, \"plot_search_progress\"),\n (getattr(args, \"tree\", None), \"plot_search_tree\"),\n (getattr(args, \"plan_plot\", None), \"plot_plan_timeline\"),\n ]\n if not any(path for path, _ in targets):\n return\n try:\n from . import viz\n except ImportError as exc:\n print(f\"Charts need the viz extra: {exc}\", file=sys.stderr)\n return\n for path, function in targets:\n if not path:\n continue\n getattr(viz, function)(trace, path, dark=args.dark)\n print(f\"Wrote {path}\")\n\n\ndef _cmd_benchmark(args) -> int:\n instances = discover_instances(args.root)\n if not instances:\n print(f\"No instances found under {args.root}\", file=sys.stderr)\n return 1\n configs = []\n for planner in args.planners.split(\",\"):\n planner = planner.strip()\n configs.append((planner, args.heuristic if planner in INFORMED else None))\n\n rows = run_benchmark(\n instances,\n configs,\n max_expansions=args.max_expansions,\n time_limit=args.time_limit,\n )\n _print_summary(summarize(rows))\n if args.csv:\n to_csv(rows, args.csv)\n print(f\"\\nWrote per-run results to {args.csv}\")\n if args.plot:\n plot_summary(rows, args.plot, metric=args.metric)\n print(f\"Wrote plot to {args.plot}\")\n if args.dashboard:\n from .viz import plot_benchmark_dashboard\n\n plot_benchmark_dashboard(rows, args.dashboard, dark=args.dark)\n print(f\"Wrote dashboard to {args.dashboard}\")\n return 0\n\n\ndef _cmd_animate(args) -> int:\n from .viz import animate_search\n\n task = build_task(args.domain, args.problem)\n heuristic = None if args.heuristic == \"none\" else args.heuristic\n _, trace = trace_search(task, args.search, heuristic)\n animate_search(\n trace, args.output, dark=args.dark, fps=args.fps, seconds=args.seconds\n )\n print(f\"Wrote animation to {args.output}\")\n return 0\n\n\ndef _add_requirements(sub):\n p = sub.add_parser(\n \"requirements\",\n help=\"show which PDDL requirement flags are supported, and how\",\n )\n p.add_argument(\n \"--support\",\n default=None,\n help=\"filter by support level: native, compiled, partial or rejected\",\n )\n p.add_argument(\"--json\", action=\"store_true\", help=\"emit machine-readable JSON\")\n p.add_argument(\"--verbose\", action=\"store_true\", help=\"include the full notes\")\n p.set_defaults(func=_cmd_requirements)\n\n\ndef _add_generate(sub):\n from .generator import GENERATORS\n\n p = sub.add_parser(\n \"generate\", help=\"generate PDDL instances reproducibly from a seed\"\n )\n p.add_argument(\"kind\", choices=sorted(GENERATORS))\n p.add_argument(\"-o\", \"--output\", default=None, help=\"write into this folder\")\n p.add_argument(\"-n\", \"--size\", type=int, default=4, help=\"instance size\")\n p.add_argument(\"--seed\", type=int, default=0)\n p.add_argument(\n \"--count\",\n type=int,\n default=1,\n help=\"generate a ladder of instances with increasing size\",\n )\n p.add_argument(\n \"--step\", type=int, default=1, help=\"size increment between ladder rungs\"\n )\n p.set_defaults(func=_cmd_generate)\n\n\ndef _add_learn(sub):\n from .generator import GENERATORS\n\n p = sub.add_parser(\n \"learn\",\n help=\"train a heuristic from solved plans, then reinforce it on search cost\",\n description=(\n \"Generate a ladder of instances, solve the small ones, fit a network \"\n \"to the cost-to-go their plans reveal, and optionally tune it against \"\n \"the number of nodes search actually expands. Writes a model that \"\n \"'-H learned:' accepts everywhere.\"\n ),\n )\n p.add_argument(\"kind\", choices=sorted(GENERATORS))\n p.add_argument(\"-o\", \"--output\", default=None, help=\"write the model here\")\n p.add_argument(\n \"--sizes\",\n default=\"3-6\",\n help=\"training ladder, as 'lo-hi' (default 3-6). Keep these small: they \"\n \"have to be solvable optimally\",\n )\n p.add_argument(\n \"--seeds-per-size\",\n type=int,\n default=2,\n help=\"instances per rung; more seeds usually beats more rungs\",\n )\n p.add_argument(\"--seed\", type=int, default=0)\n p.add_argument(\"--epochs\", type=int, default=60)\n p.add_argument(\n \"--rank-weight\",\n type=float,\n default=0.8,\n help=\"share of the objective spent on ordering rather than magnitude; \"\n \"GBFS only reads the order\",\n )\n p.add_argument(\n \"--dagger\",\n type=int,\n default=0,\n metavar=\"ROUNDS\",\n help=\"retrain on the states the heuristic's own search visits\",\n )\n p.add_argument(\n \"--bootstrap\",\n default=None,\n metavar=\"LO-HI\",\n help=\"grow the corpus with harder instances as they become solvable\",\n )\n p.add_argument(\n \"--cem\",\n type=int,\n default=0,\n metavar=\"ITERATIONS\",\n help=\"optimise expansions directly (needs instances with headroom; \"\n \"see --cem-sizes)\",\n )\n p.add_argument(\n \"--cem-sizes\",\n default=None,\n metavar=\"LO-HI\",\n help=\"instances to tune search cost on; defaults to a rung above the \"\n \"training ladder, because the training ladder has no headroom left\",\n )\n p.add_argument(\n \"--evaluate\",\n default=None,\n metavar=\"LO-HI\",\n help=\"after training, benchmark against hff/goalcount on these sizes\",\n )\n p.set_defaults(func=_cmd_learn)\n\n\ndef _range(spec: str):\n \"\"\"Parse ``'3-6'`` or ``'5'`` into a range.\"\"\"\n text = str(spec).strip()\n if \"-\" in text:\n lo, _, hi = text.partition(\"-\")\n return range(int(lo), int(hi) + 1)\n return range(int(text), int(text) + 1)\n\n\ndef _cmd_learn(args) -> int:\n from .learn import RLConfig, TrainConfig, learn_heuristic\n from .learn.pipeline import (\n evaluate_transfer,\n summarise_transfer,\n tasks_from_generator,\n )\n\n output = args.output or f\"{args.kind}.heur.json\"\n try:\n bundle = learn_heuristic(\n args.kind,\n sizes=_range(args.sizes),\n seeds_per_size=args.seeds_per_size,\n seed=args.seed,\n train_config=TrainConfig(\n epochs=args.epochs, rank_weight=args.rank_weight, seed=args.seed\n ),\n rl_config=RLConfig(seed=args.seed, verbose=True),\n dagger_rounds=args.dagger,\n bootstrap_sizes=_range(args.bootstrap) if args.bootstrap else None,\n cem_iterations=args.cem,\n cem_sizes=_range(args.cem_sizes) if args.cem_sizes else None,\n verbose=True,\n )\n except RuntimeError as exc:\n print(f\"error: {exc}\", file=sys.stderr)\n return 1\n\n bundle.save(output)\n print(f\"\\nWrote {output}\")\n print(\n f\" features {bundle.space.size} over {len(bundle.space.vocabulary)} predicates\"\n )\n print(f\" parameters {bundle.model.num_parameters}\")\n metrics = bundle.metrics\n print(\n f\" held out MAE {metrics.get('mae', 0):.2f}\"\n f\" top-1 {metrics.get('top1', 0):.3f}\"\n f\" over {metrics.get('held_out_instances', 0)} instances\"\n )\n print(f\"\\nUse it:\\n jupyddl solve -s gbfs -H learned:{output}\")\n\n if args.evaluate:\n tasks = tasks_from_generator(\n args.kind, _range(args.evaluate), seed=args.seed + 7777, seeds_per_size=2\n )\n print(f\"\\nBenchmark on {len(tasks)} unseen instances (gbfs):\")\n rows = evaluate_transfer(bundle, tasks)\n summary = summarise_transfer(rows)\n print(\n f\" {'heuristic':<12}{'coverage':>10}{'expanded':>12}\"\n f\"{'seconds':>10}{'cost':>8}\"\n )\n for name in [\"learned\"] + [k for k in summary if k != \"learned\"]:\n agg = summary[name]\n print(\n f\" {name:<12}{agg['coverage']:>10.2f}{agg['mean_expanded']:>12.0f}\"\n f\"{agg['mean_seconds']:>10.3f}{agg['mean_cost']:>8.1f}\"\n )\n return 0\n\n\ndef _cmd_requirements(args) -> int:\n from .requirements import as_rows, summary\n\n rows = as_rows()\n if args.support:\n rows = [row for row in rows if row[\"support\"] == args.support]\n if not rows:\n print(\n f\"No requirements with support level '{args.support}'.\", file=sys.stderr\n )\n return 1\n\n if args.json:\n import json\n\n print(json.dumps({\"requirements\": rows, \"summary\": summary()}, indent=2))\n return 0\n\n counts = summary()\n print(\n f\"jupyddl PDDL support: {counts['native']} native, \"\n f\"{counts['compiled']} compiled, {counts['partial']} partial, \"\n f\"{counts['rejected']} rejected\\n\"\n )\n print(f\"{'requirement':<30}{'PDDL':<7}{'support':<11}summary\")\n print(\"-\" * 100)\n for row in rows:\n print(f\"{row['name']:<30}{row['pddl']:<7}{row['support']:<11}{row['summary']}\")\n if args.verbose and row[\"note\"]:\n for line in _wrap(row[\"note\"], 92):\n print(f\"{'':<48}{line}\")\n if not args.verbose:\n print(\"\\nRun with --verbose for the details of each compilation.\")\n return 0\n\n\ndef _wrap(text: str, width: int) -> list:\n words = text.split()\n lines, current = [], \"\"\n for word in words:\n if len(current) + len(word) + 1 > width:\n lines.append(current)\n current = word\n else:\n current = f\"{current} {word}\".strip()\n if current:\n lines.append(current)\n return lines\n\n\ndef _cmd_generate(args) -> int:\n from .generator import generate, write_instance\n\n sizes = [args.size + i * args.step for i in range(max(1, args.count))]\n if args.output is None:\n if len(sizes) > 1:\n print(\n \"Generating a ladder needs --output; a single instance can go \"\n \"to stdout but several cannot.\",\n file=sys.stderr,\n )\n return 1\n domain, problem = generate(args.kind, size=args.size, seed=args.seed)\n print(\";; ---------- domain.pddl ----------\")\n print(domain)\n print(\";; ---------- problem.pddl ----------\")\n print(problem)\n return 0\n\n for size in sizes:\n folder = write_instance(args.kind, args.output, size=size, seed=args.seed)\n print(f\"Wrote {folder}\")\n return 0\n\n\ndef _cmd_demo(args) -> int:\n \"\"\"Render the whole gallery: per-instance charts plus a benchmark dashboard.\"\"\"\n from .viz import (\n plot_benchmark_dashboard,\n plot_plan_timeline,\n plot_planner_comparison,\n plot_search_progress,\n plot_search_tree,\n )\n\n instances = discover_instances(args.root)\n if not instances:\n print(f\"No instances found under {args.root}\", file=sys.stderr)\n return 1\n os.makedirs(args.output, exist_ok=True)\n modes = [False, True] if args.both_modes else [False]\n\n configs = [(\"astar\", \"lmcut\"), (\"astar\", \"hmax\"), (\"gbfs\", \"hff\"), (\"bfs\", None)]\n for instance in instances:\n print(f\"-- {instance.name}\")\n try:\n task = build_task(instance.domain, instance.problem)\n except Exception as exc:\n print(f\" skipped: {type(exc).__name__}: {exc}\")\n continue\n traces = []\n for planner, heuristic in configs:\n try:\n _, trace = trace_search(task, planner, heuristic)\n traces.append(trace)\n except Exception as exc:\n print(f\" {planner}/{heuristic}: {type(exc).__name__}: {exc}\")\n if not traces:\n continue\n for dark in modes:\n suffix = \"-dark\" if dark else \"\"\n base = os.path.join(args.output, instance.name)\n plot_search_progress(traces[0], f\"{base}-progress{suffix}.png\", dark=dark)\n plot_search_tree(traces[0], f\"{base}-tree{suffix}.png\", dark=dark)\n plot_plan_timeline(traces[0], f\"{base}-plan{suffix}.png\", dark=dark)\n plot_planner_comparison(traces, f\"{base}-compare{suffix}.png\", dark=dark)\n if args.animate:\n from .viz import animate_search\n\n animate_search(\n traces[0], os.path.join(args.output, f\"{instance.name}-search.mp4\")\n )\n\n rows = run_benchmark(\n instances,\n [\n (\"astar\", \"lmcut\"),\n (\"astar\", \"hmax\"),\n (\"gbfs\", \"hff\"),\n (\"ehc\", \"hff\"),\n (\"bfs\", None),\n (\"dijkstra\", None),\n ],\n )\n to_csv(rows, os.path.join(args.output, \"benchmark.csv\"))\n for dark in modes:\n suffix = \"-dark\" if dark else \"\"\n plot_benchmark_dashboard(\n rows, os.path.join(args.output, f\"benchmark{suffix}.png\"), dark=dark\n )\n print(f\"\\nGallery written to {args.output}/\")\n return 0\n\n\ndef _print_stats(result) -> None:\n s = result.stats\n print(\n f\"Stats: expanded={s.expanded} generated={s.generated} \"\n f\"evaluated={s.evaluated} reopened={s.reopened} \"\n f\"deadends={s.deadends} runtime={s.runtime:.4f}s\"\n )\n\n\ndef _print_summary(summary) -> None:\n print(f\"{'config':<20}{'coverage':>12}{'expanded':>12}{'runtime(s)':>12}\")\n print(\"-\" * 56)\n for key, agg in summary.items():\n cov = f\"{agg['coverage']}/{agg['instances']}\"\n print(f\"{key:<20}{cov:>12}{agg['expanded']:>12}{agg['runtime']:>12.3f}\")\n\n\ndef main(argv=None) -> int:\n parser = argparse.ArgumentParser(\n prog=\"jupyddl\",\n description=\"Pure-Python PDDL planning framework.\",\n )\n sub = parser.add_subparsers(dest=\"command\", required=True)\n _add_solve(sub)\n _add_benchmark(sub)\n _add_animate(sub)\n _add_demo(sub)\n _add_requirements(sub)\n _add_generate(sub)\n _add_learn(sub)\n args = parser.parse_args(argv)\n return args.func(args)\n\n\nif __name__ == \"__main__\": # pragma: no cover\n sys.exit(main())\n", "jupyddl/compile.py": "\"\"\"Compile PDDL 3 constructs down to the classical core.\n\nPreferences, trajectory constraints, timed initial literals and object fluents\nare all rewritten here, at the AST level, *before* grounding. The grounder and\nthe search never learn they existed \u2014 which is the point: one representation to\noptimise, and every new front-end feature is a source-to-source transformation\nrather than another special case in the hot loop.\n\nEach compilation is documented in :mod:`jupyddl.requirements`, including what it\ncosts you. The synthetic actions introduced along the way are all named with a\nleading ``__``; :attr:`jupyddl.task.Task.synthetic` collects them so a printed\nplan shows only the actions the domain author wrote.\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom dataclasses import replace\n\nfrom .parser.ast import (\n Action,\n AddEffect,\n And,\n Atom,\n Comparison,\n ConjunctiveEffect,\n DelEffect,\n Domain,\n EqualityConstraint,\n Exists,\n Forall,\n ForallEffect,\n FluentRef,\n IncreaseCostEffect,\n Literal,\n Number,\n NumericEffect,\n Or,\n PDDLError,\n Predicate,\n Preference,\n Problem,\n Truth,\n UnsupportedFeatureError,\n WhenEffect,\n)\n\n# Every fact and action this module invents is prefixed, so they cannot collide\n# with a domain's own names and are easy to filter out of a plan.\nPREFIX = \"__\"\nCLOCK = FluentRef(\"__time\", ())\n\n__all__ = [\"compile_problem\", \"SYNTHETIC_PREFIX\"]\n\nSYNTHETIC_PREFIX = PREFIX\n\n\n# --------------------------------------------------------------------------\n# small AST helpers\n# --------------------------------------------------------------------------\ndef _fact(name: str, *args) -> Atom:\n return Atom(name, tuple(args))\n\n\ndef _holds(name: str, *args) -> Literal:\n return Literal(_fact(name, *args), True)\n\n\ndef _absent(name: str, *args) -> Literal:\n return Literal(_fact(name, *args), False)\n\n\ndef _conjoin(*parts):\n \"\"\"And() over the parts, dropping trivially-true ones.\"\"\"\n kept = [p for p in parts if p is not None and not isinstance(p, Truth)]\n if not kept:\n return Truth(True)\n if len(kept) == 1:\n return kept[0]\n return And(tuple(kept))\n\n\ndef _free():\n \"\"\"An explicit zero cost.\n\n Grounding charges 1 for any action without a cost effect, which is right for\n a domain action and wrong for the bookkeeping this module invents: closing\n the plan or observing that a constraint held must not show up in the metric.\n \"\"\"\n return IncreaseCostEffect(0.0)\n\n\ndef _also(effect, *extra):\n \"\"\"Append effects to an existing effect tree.\"\"\"\n parts = list(effect.parts) if isinstance(effect, ConjunctiveEffect) else [effect]\n parts.extend(extra)\n return ConjunctiveEffect(parts)\n\n\ndef _negate(formula):\n \"\"\"Negation normal form of ``not formula``.\"\"\"\n if isinstance(formula, Truth):\n return Truth(not formula.value)\n if isinstance(formula, Literal):\n return Literal(formula.atom, not formula.positive)\n if isinstance(formula, EqualityConstraint):\n return EqualityConstraint(formula.left, formula.right, not formula.positive)\n if isinstance(formula, Comparison):\n flip = {\"<\": \">=\", \"<=\": \">\", \">\": \"<=\", \">=\": \"<\", \"=\": \"!=\", \"!=\": \"=\"}\n return Comparison(flip[formula.op], formula.left, formula.right)\n if isinstance(formula, And):\n return Or(tuple(_negate(p) for p in formula.parts))\n if isinstance(formula, Or):\n return And(tuple(_negate(p) for p in formula.parts))\n if isinstance(formula, Forall):\n return Exists(formula.params, _negate(formula.body))\n if isinstance(formula, Exists):\n return Forall(formula.params, _negate(formula.body))\n raise PDDLError(f\"cannot negate {formula!r}\")\n\n\ndef _is_domain_action(action: Action) -> bool:\n return not action.name.startswith(PREFIX)\n\n\ndef _declare(domain: Domain, name: str, params=()) -> None:\n if all(p.name != name for p in domain.predicates):\n domain.predicates.append(Predicate(name, list(params)))\n\n\ndef _add_precondition(domain: Domain, formula, only_domain_actions=True) -> None:\n \"\"\"Conjoin ``formula`` onto every action's precondition.\"\"\"\n for index, action in enumerate(domain.actions):\n if only_domain_actions and not _is_domain_action(action):\n continue\n domain.actions[index] = replace(\n action, precondition=_conjoin(action.precondition, formula)\n )\n\n\ndef _add_conditional_effect(domain: Domain, condition, body) -> None:\n \"\"\"Give every domain action a ``when condition body`` effect.\n\n Used for monitors that must not be optional: the planner cannot decline to\n notice that a constraint's trigger became true.\n \"\"\"\n for index, action in enumerate(domain.actions):\n if not _is_domain_action(action):\n continue\n domain.actions[index] = replace(\n action, effect=_also(action.effect, WhenEffect(condition, body))\n )\n\n\n# --------------------------------------------------------------------------\n# object fluents\n# --------------------------------------------------------------------------\ndef compile_object_fluents(domain: Domain, problem: Problem):\n \"\"\"Turn ``(location ?p) - place`` into a predicate plus a uniqueness rule.\n\n ``(= (location ?p) ?x)`` becomes ``(__fn-location ?p ?x)``, and\n ``(assign (location ?p) ?y)`` clears the old value before setting the new\n one, so the predicate stays single-valued. Object fluents used as *nested\n terms* \u2014 ``(at ?t (location ?p))`` \u2014 are refused: flattening those needs a\n fresh existential per occurrence, and the equality form above expresses the\n same thing without the guesswork.\n \"\"\"\n if not domain.object_fluents:\n return domain, problem\n\n fluents = {fluent.name: fluent for fluent in domain.object_fluents}\n\n def predicate_name(name: str) -> str:\n return f\"{PREFIX}fn-{name}\"\n\n for fluent in domain.object_fluents:\n params = list(fluent.params) + [(\"?__value\", fluent.result_type)]\n _declare(domain, predicate_name(fluent.name), params)\n\n def rewrite_condition(formula):\n if isinstance(formula, Comparison) and formula.op in (\"=\", \"!=\"):\n left, right = formula.left, formula.right\n for a, b in ((left, right), (right, left)):\n if isinstance(a, FluentRef) and a.name in fluents:\n if isinstance(b, Number):\n raise UnsupportedFeatureError(\n f\"'{a.name}' returns an object, so it cannot be \"\n \"compared with a number\"\n )\n value = b.name if isinstance(b, FluentRef) else str(b)\n if isinstance(b, FluentRef) and b.name in fluents:\n raise UnsupportedFeatureError(\n \"comparing two object fluents directly is not \"\n \"supported; introduce a variable for one of them\"\n )\n atom = _fact(predicate_name(a.name), *a.args, value)\n return Literal(atom, formula.op == \"=\")\n return formula\n if isinstance(formula, And):\n return And(tuple(rewrite_condition(p) for p in formula.parts))\n if isinstance(formula, Or):\n return Or(tuple(rewrite_condition(p) for p in formula.parts))\n if isinstance(formula, Forall):\n return Forall(formula.params, rewrite_condition(formula.body))\n if isinstance(formula, Exists):\n return Exists(formula.params, rewrite_condition(formula.body))\n return formula\n\n def rewrite_effect(effect):\n if isinstance(effect, ConjunctiveEffect):\n return ConjunctiveEffect([rewrite_effect(p) for p in effect.parts])\n if isinstance(effect, ForallEffect):\n return ForallEffect(effect.params, rewrite_effect(effect.body))\n if isinstance(effect, WhenEffect):\n return WhenEffect(\n rewrite_condition(effect.condition), rewrite_effect(effect.body)\n )\n if isinstance(effect, NumericEffect) and effect.target.name in fluents:\n if effect.op != \"assign\":\n raise UnsupportedFeatureError(\n f\"'{effect.op}' is arithmetic, but '{effect.target.name}' \"\n \"returns an object; only 'assign' is meaningful\"\n )\n fluent = fluents[effect.target.name]\n name = predicate_name(fluent.name)\n value = effect.value\n new = value.name if isinstance(value, FluentRef) else str(value)\n # Clear whatever the function used to return, then set the new\n # value: that is what keeps it a function rather than a relation.\n clear = ForallEffect(\n [(\"?__old\", fluent.result_type)],\n WhenEffect(\n _holds(name, *effect.target.args, \"?__old\"),\n DelEffect(_fact(name, *effect.target.args, \"?__old\")),\n ),\n )\n return ConjunctiveEffect(\n [clear, AddEffect(_fact(name, *effect.target.args, new))]\n )\n return effect\n\n for index, action in enumerate(domain.actions):\n domain.actions[index] = replace(\n action,\n precondition=rewrite_condition(action.precondition),\n effect=rewrite_effect(action.effect),\n )\n for index, rule in enumerate(domain.derived):\n domain.derived[index] = replace(rule, body=rewrite_condition(rule.body))\n\n problem.goal = rewrite_condition(problem.goal)\n for index, preference in enumerate(problem.preferences):\n problem.preferences[index] = Preference(\n preference.name, rewrite_condition(preference.body)\n )\n\n # `(= (location p1) depot)` in :init becomes a plain fact.\n for fluent_ref, value in problem.init_objects.items():\n if fluent_ref.name not in fluents:\n raise PDDLError(\n f\"'{fluent_ref.name}' is assigned an object in :init but is not \"\n \"declared as an object fluent in :functions\"\n )\n problem.init.append(\n _fact(predicate_name(fluent_ref.name), *fluent_ref.args, value)\n )\n problem.init_objects = {}\n return domain, problem\n\n\n# --------------------------------------------------------------------------\n# timed initial literals\n# --------------------------------------------------------------------------\ndef compile_timed_initials(domain: Domain, problem: Problem):\n \"\"\"Give the model a clock, and make each timed literal fire off it.\n\n Elapsed time becomes the numeric fluent ``(__time)``, advanced by each\n durative action's duration. Every timed literal gets a zero-cost\n ``__fire-til-k`` action guarded by ``(>= (__time) t)``, and a\n ``__wait-til-k`` action that lets the planner advance the clock to ``t``\n when it wants the literal to happen.\n\n The literal must not be *skipped*: every domain action therefore carries\n ``(or (< (__time) t) (__til-k))`` for each timed literal, so once the clock\n passes ``t`` nothing else may happen until the literal has fired.\n\n Literals must also fire **in time order**, which is a separate constraint:\n ``(at 0 (open))`` and ``(at 3 (not (open)))`` describe a shop that opens then\n shuts, but firing them the other way round would leave it open forever. Each\n firing therefore requires every earlier literal to have fired already.\n\n Because actions do not overlap, a literal scheduled strictly inside an\n action's duration fires immediately after it rather than during it.\n \"\"\"\n if not problem.timed_initials:\n return domain, problem\n\n problem.init_numeric = dict(problem.init_numeric)\n problem.init_numeric.setdefault(CLOCK, 0.0)\n\n # The clock only moves if something moves it.\n for index, action in enumerate(domain.actions):\n if action.duration is None or not _is_domain_action(action):\n continue\n domain.actions[index] = replace(\n action,\n effect=_also(\n action.effect, NumericEffect(\"increase\", CLOCK, action.duration)\n ),\n )\n\n guards = []\n earlier: list = []\n for k, timed in enumerate(sorted(problem.timed_initials, key=lambda t: t.time)):\n marker = f\"{PREFIX}til-{k}\"\n _declare(domain, marker)\n due = Comparison(\">=\", CLOCK, Number(timed.time))\n not_due = Comparison(\"<\", CLOCK, Number(timed.time))\n # Everything scheduled before this must already have happened.\n in_order = _conjoin(*[_holds(name) for name in earlier])\n\n body = (\n AddEffect(timed.literal.atom)\n if timed.literal.positive\n else DelEffect(timed.literal.atom)\n )\n domain.actions.append(\n Action(\n f\"{PREFIX}fire-til-{k}\",\n [],\n _conjoin(due, _absent(marker), in_order),\n ConjunctiveEffect([body, AddEffect(_fact(marker)), _free()]),\n )\n )\n domain.actions.append(\n Action(\n f\"{PREFIX}wait-til-{k}\",\n [],\n # Waiting past an event that has not happened yet would skip it.\n _conjoin(not_due, in_order),\n ConjunctiveEffect(\n [NumericEffect(\"assign\", CLOCK, Number(timed.time)), _free()]\n ),\n )\n )\n guards.append(Or((not_due, _holds(marker))))\n earlier.append(marker)\n\n # A due literal blocks everything else until it has fired.\n for guard in guards:\n _add_precondition(domain, guard)\n problem.timed_initials = []\n return domain, problem\n\n\n# --------------------------------------------------------------------------\n# trajectory constraints\n# --------------------------------------------------------------------------\ndef compile_constraints(domain: Domain, problem: Problem):\n \"\"\"Compile ``(:constraints ...)`` into preconditions, monitors and goals.\n\n * ``always phi`` \u2014 conjoined onto every action's precondition and onto the\n goal. Every state on a plan's trajectory is either the initial state, a\n state an action is taken from, or the final state, so those three checks\n cover all of them.\n * ``at-end phi`` \u2014 conjoined onto the goal.\n * ``sometime phi`` \u2014 a zero-cost ``__observe`` action, applicable exactly\n when ``phi`` holds, sets a monitor fact the goal then requires.\n * ``sometime-before phi psi`` \u2014 the same monitor for ``psi``, plus\n ``always (phi implies monitor)``.\n * ``sometime-after phi psi`` \u2014 a *forced* monitor: every action records an\n outstanding obligation when ``phi`` holds without ``psi``, and discharges\n it when ``psi`` holds. The goal requires nothing outstanding.\n * ``at-most-once phi`` \u2014 forced monitors for \"phi has held\" and \"phi has\n since stopped\", plus ``always not (phi and stopped)``.\n\n Forced monitors ride on conditional effects, which the planner cannot\n decline; optional ones ride on actions, which it applies when convenient.\n The difference matters: a constraint the planner could satisfy by *not\n looking* would not be a constraint.\n \"\"\"\n constraints = list(domain.constraints) + list(problem.constraints)\n if not constraints:\n return domain, problem\n\n for constraint in constraints:\n if isinstance(constraint, Preference):\n raise UnsupportedFeatureError(\n f\"the soft constraint '{constraint.name}' is not supported: \"\n \"preferences are supported over goals, not over trajectory \"\n \"constraints\"\n )\n\n goal_parts = [problem.goal]\n invariants = []\n\n for index, constraint in enumerate(constraints):\n kind = constraint.kind\n if kind == \"always\":\n invariants.append(constraint.args[0])\n elif kind == \"at-end\":\n goal_parts.append(constraint.args[0])\n elif kind == \"sometime\":\n marker = _observation_monitor(domain, index, constraint.args[0])\n goal_parts.append(_holds(marker))\n elif kind == \"sometime-before\":\n trigger, earlier = constraint.args\n marker = _observation_monitor(domain, index, earlier)\n # \"phi implies the monitor\" as an invariant: by the time phi holds,\n # psi must already have been observed.\n invariants.append(Or((_negate(trigger), _holds(marker))))\n elif kind == \"sometime-after\":\n trigger, follower = constraint.args\n marker = f\"{PREFIX}pending-{index}\"\n _declare(domain, marker)\n _add_conditional_effect(\n domain,\n _conjoin(trigger, _negate(follower)),\n AddEffect(_fact(marker)),\n )\n _add_conditional_effect(domain, follower, DelEffect(_fact(marker)))\n goal_parts.append(_absent(marker))\n # ...and the final state must not leave a fresh obligation either.\n goal_parts.append(Or((_negate(trigger), follower)))\n elif kind == \"at-most-once\":\n phi = constraint.args[0]\n seen = f\"{PREFIX}amo-seen-{index}\"\n closed = f\"{PREFIX}amo-closed-{index}\"\n _declare(domain, seen)\n _declare(domain, closed)\n _add_conditional_effect(domain, phi, AddEffect(_fact(seen)))\n _add_conditional_effect(\n domain,\n _conjoin(_negate(phi), _holds(seen)),\n AddEffect(_fact(closed)),\n )\n # Holding again after an interval has closed is the second interval.\n invariants.append(Or((_negate(phi), _absent(closed))))\n else: # pragma: no cover - the parser rejects anything else\n raise UnsupportedFeatureError(f\"unsupported constraint '{kind}'\")\n\n for invariant in invariants:\n # Every action, not just the domain's own. The coverage argument above\n # holds only if each state on the trajectory is either taken from by an\n # action that checks the invariant or is the final state \u2014 and the\n # actions timed initial literals compile to *change facts*. Exempting\n # them let a plan step through a state the invariant forbade: a literal\n # that clears `(safe)` at t=3 and one that restores it at t=4 fire\n # back-to-back with nothing checking the state in between.\n _add_precondition(domain, invariant, only_domain_actions=False)\n goal_parts.append(invariant)\n\n problem.goal = _conjoin(*goal_parts)\n domain.constraints = []\n problem.constraints = []\n return domain, problem\n\n\ndef _observation_monitor(domain: Domain, index: int, formula) -> str:\n \"\"\"A fact the planner can set, for free, whenever ``formula`` holds.\"\"\"\n marker = f\"{PREFIX}seen-{index}\"\n _declare(domain, marker)\n domain.actions.append(\n Action(\n f\"{PREFIX}observe-{index}\",\n [],\n _conjoin(formula, _absent(marker)),\n ConjunctiveEffect([AddEffect(_fact(marker)), _free()]),\n )\n )\n return marker\n\n\n# --------------------------------------------------------------------------\n# preferences\n# --------------------------------------------------------------------------\ndef compile_preferences(domain: Domain, problem: Problem):\n \"\"\"Turn each soft goal into a priced choice: satisfy it, or pay for it.\n\n A closing phase makes this sound. ``__close`` ends the plan \u2014 every domain\n action requires that it has *not* happened \u2014 after which each preference is\n resolved by one of two zero-parameter actions: a free one that requires the\n preference to hold, or one costing the metric's ``(is-violated p)`` weight\n that does not. Cost-optimal search then picks whichever is cheaper, which is\n exactly what minimising the metric means.\n\n Freezing the state first is the point: without it the planner could satisfy\n a preference halfway through and then break it, and still be paid for it.\n \"\"\"\n if not problem.preferences:\n return domain, problem\n\n closed = f\"{PREFIX}closed\"\n _declare(domain, closed)\n _add_precondition(domain, _absent(closed))\n\n domain.actions.append(\n Action(\n f\"{PREFIX}close\",\n [],\n _absent(closed),\n ConjunctiveEffect([AddEffect(_fact(closed)), _free()]),\n )\n )\n\n goal_parts = [problem.goal, _holds(closed)]\n for index, preference in enumerate(problem.preferences):\n done = f\"{PREFIX}pref-{index}\"\n _declare(domain, done)\n weight = float(problem.violation_weights.get(preference.name, 1.0))\n if weight < 0:\n raise PDDLError(\n f\"preference '{preference.name}' has a negative violation \"\n \"weight, which would reward breaking it\"\n )\n domain.actions.append(\n Action(\n f\"{PREFIX}satisfy-{preference.name}\",\n [],\n _conjoin(_holds(closed), _absent(done), preference.body),\n ConjunctiveEffect([AddEffect(_fact(done)), _free()]),\n )\n )\n domain.actions.append(\n Action(\n f\"{PREFIX}violate-{preference.name}\",\n [],\n _conjoin(_holds(closed), _absent(done)),\n ConjunctiveEffect([AddEffect(_fact(done))]), # priced below\n )\n )\n # The penalty rides on the action cost, so plain cost-optimal search\n # optimises the metric without knowing what a preference is.\n domain.actions[-1] = replace(\n domain.actions[-1],\n effect=_also(domain.actions[-1].effect, _increase_cost(weight)),\n )\n goal_parts.append(_holds(done))\n\n problem.goal = _conjoin(*goal_parts)\n problem.preferences = []\n return domain, problem\n\n\ndef _increase_cost(amount: float):\n return IncreaseCostEffect(amount)\n\n\n# --------------------------------------------------------------------------\n# entry point\n# --------------------------------------------------------------------------\ndef compile_problem(domain: Domain, problem: Problem):\n \"\"\"Apply every PDDL 3 compilation, in the order they depend on each other.\n\n Object fluents go first because they rewrite terms everywhere else reads;\n preferences go last because their closing phase must sit outside the\n machinery the other compilations add.\n \"\"\"\n domain, problem = compile_object_fluents(domain, problem)\n domain, problem = compile_timed_initials(domain, problem)\n domain, problem = compile_constraints(domain, problem)\n domain, problem = compile_preferences(domain, problem)\n return domain, problem\n", "jupyddl/generator.py": "\"\"\"Generate PDDL domains and problems, reproducibly.\n\nBenchmarks live or die on their instance set, and hand-writing a scaling ladder\nis tedious. Each generator here takes a size and a ``seed`` and emits PDDL text,\nso a whole difficulty curve is one comprehension \u2014 and re-running the same seed\ngives byte-identical files, which is what makes a published experiment\nreproducible.\n\n::\n\n from jupyddl.generator import generate, GENERATORS\n\n domain, problem = generate(\"blocksworld\", size=10, seed=7)\n ladder = [generate(\"gripper\", size=n, seed=1) for n in range(2, 12)]\n\nEvery generator is exercised by the test suite, which grounds and solves what it\nproduces \u2014 a generator that emits unsolvable or unparseable PDDL is a bug.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport random\nfrom typing import Optional\n\n__all__ = [\n \"GENERATORS\",\n \"generate\",\n \"describe_generators\",\n \"generate_blocksworld\",\n \"generate_gripper\",\n \"generate_logistics\",\n \"generate_rovers\",\n \"generate_random_strips\",\n \"generate_numeric_transport\",\n \"generate_temporal_workshop\",\n]\n\n\ndef _objects(prefix: str, count: int) -> list:\n return [f\"{prefix}{i + 1}\" for i in range(count)]\n\n\n# --------------------------------------------------------------------------\n# blocksworld\n# --------------------------------------------------------------------------\nBLOCKSWORLD_DOMAIN = \"\"\"\\\n;; Blocksworld -- generated by jupyddl.generator\n(define (domain blocksworld)\n (:requirements :strips :typing)\n (:types block - object)\n (:predicates\n (on ?x - block ?y - block)\n (ontable ?x - block)\n (clear ?x - block)\n (handempty)\n (holding ?x - block))\n\n (:action pick-up\n :parameters (?x - block)\n :precondition (and (clear ?x) (ontable ?x) (handempty))\n :effect (and (not (ontable ?x)) (not (clear ?x))\n (not (handempty)) (holding ?x)))\n\n (:action put-down\n :parameters (?x - block)\n :precondition (holding ?x)\n :effect (and (not (holding ?x)) (clear ?x) (handempty) (ontable ?x)))\n\n (:action stack\n :parameters (?x - block ?y - block)\n :precondition (and (holding ?x) (clear ?y))\n :effect (and (not (holding ?x)) (not (clear ?y))\n (clear ?x) (handempty) (on ?x ?y)))\n\n (:action unstack\n :parameters (?x - block ?y - block)\n :precondition (and (on ?x ?y) (clear ?x) (handempty))\n :effect (and (holding ?x) (clear ?y) (not (clear ?x))\n (not (handempty)) (not (on ?x ?y)))))\n\"\"\"\n\n\ndef _random_towers(blocks, rng) -> list:\n \"\"\"Partition ``blocks`` into a random set of towers (bottom-first).\"\"\"\n shuffled = list(blocks)\n rng.shuffle(shuffled)\n towers: list = []\n index = 0\n while index < len(shuffled):\n height = rng.randint(1, max(1, len(shuffled) - index))\n towers.append(shuffled[index : index + height])\n index += height\n return towers\n\n\ndef _tower_facts(towers) -> list:\n facts = []\n for tower in towers:\n facts.append(f\"(ontable {tower[0]})\")\n for lower, upper in zip(tower, tower[1:]):\n facts.append(f\"(on {upper} {lower})\")\n facts.append(f\"(clear {tower[-1]})\")\n return facts\n\n\ndef generate_blocksworld(size: int = 6, seed: int = 0):\n \"\"\"``size`` blocks in random towers, to be rearranged into other towers.\"\"\"\n rng = random.Random(seed)\n blocks = _objects(\"b\", max(2, size))\n start = _random_towers(blocks, rng)\n goal_towers = _random_towers(blocks, rng)\n while goal_towers == start and len(blocks) > 2:\n goal_towers = _random_towers(blocks, rng)\n\n goal_facts = []\n for tower in goal_towers:\n goal_facts.append(f\"(ontable {tower[0]})\")\n for lower, upper in zip(tower, tower[1:]):\n goal_facts.append(f\"(on {upper} {lower})\")\n\n problem = f\"\"\"\\\n;; {len(blocks)} blocks, seed {seed} -- generated by jupyddl.generator\n(define (problem blocksworld-{len(blocks)}-{seed})\n (:domain blocksworld)\n (:objects {' '.join(blocks)} - block)\n (:init\n (handempty)\n {chr(10).join(' ' + f for f in _tower_facts(start)).strip()})\n (:goal (and\n {chr(10).join(' ' + f for f in goal_facts).strip()})))\n\"\"\"\n return BLOCKSWORLD_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# gripper\n# --------------------------------------------------------------------------\nGRIPPER_DOMAIN = \"\"\"\\\n;; Gripper -- generated by jupyddl.generator\n(define (domain gripper)\n (:requirements :strips :typing)\n (:types room ball gripper - object)\n (:predicates\n (at-robby ?r - room)\n (at ?b - ball ?r - room)\n (free ?g - gripper)\n (carry ?b - ball ?g - gripper))\n\n (:action move\n :parameters (?from - room ?to - room)\n :precondition (at-robby ?from)\n :effect (and (at-robby ?to) (not (at-robby ?from))))\n\n (:action pick\n :parameters (?b - ball ?r - room ?g - gripper)\n :precondition (and (at ?b ?r) (at-robby ?r) (free ?g))\n :effect (and (carry ?b ?g) (not (at ?b ?r)) (not (free ?g))))\n\n (:action drop\n :parameters (?b - ball ?r - room ?g - gripper)\n :precondition (and (carry ?b ?g) (at-robby ?r))\n :effect (and (at ?b ?r) (free ?g) (not (carry ?b ?g)))))\n\"\"\"\n\n\ndef generate_gripper(size: int = 4, seed: int = 0, grippers: int = 2):\n \"\"\"``size`` balls to move from room A to room B. Plan length grows linearly.\"\"\"\n balls = _objects(\"ball\", max(1, size))\n hands = _objects(\"gripper\", max(1, grippers))\n init = [\"(at-robby rooma)\"]\n init += [f\"(free {g})\" for g in hands]\n init += [f\"(at {b} rooma)\" for b in balls]\n\n problem = f\"\"\"\\\n;; {len(balls)} balls, {len(hands)} grippers, seed {seed}\n(define (problem gripper-{len(balls)}-{seed})\n (:domain gripper)\n (:objects\n rooma roomb - room\n {' '.join(balls)} - ball\n {' '.join(hands)} - gripper)\n (:init\n {chr(10).join(' ' + f for f in init).strip()})\n (:goal (and\n {chr(10).join(f' (at {b} roomb)' for b in balls).strip()})))\n\"\"\"\n return GRIPPER_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# logistics\n# --------------------------------------------------------------------------\nLOGISTICS_DOMAIN = \"\"\"\\\n;; Logistics with action costs -- generated by jupyddl.generator\n(define (domain logistics)\n (:requirements :strips :typing :action-costs)\n (:types\n truck airplane - vehicle\n package vehicle - thing\n airport location - place\n city - object)\n (:predicates\n (at ?t - thing ?p - place)\n (in ?p - package ?v - vehicle)\n (in-city ?p - place ?c - city))\n (:functions (total-cost))\n\n (:action load\n :parameters (?p - package ?v - vehicle ?l - place)\n :precondition (and (at ?p ?l) (at ?v ?l))\n :effect (and (not (at ?p ?l)) (in ?p ?v) (increase (total-cost) 1)))\n\n (:action unload\n :parameters (?p - package ?v - vehicle ?l - place)\n :precondition (and (in ?p ?v) (at ?v ?l))\n :effect (and (not (in ?p ?v)) (at ?p ?l) (increase (total-cost) 1)))\n\n (:action drive\n :parameters (?t - truck ?from - place ?to - place ?c - city)\n :precondition (and (at ?t ?from) (in-city ?from ?c) (in-city ?to ?c))\n :effect (and (not (at ?t ?from)) (at ?t ?to) (increase (total-cost) 2)))\n\n (:action fly\n :parameters (?a - airplane ?from - airport ?to - airport)\n :precondition (at ?a ?from)\n :effect (and (not (at ?a ?from)) (at ?a ?to) (increase (total-cost) 6))))\n\"\"\"\n\n\ndef generate_logistics(size: int = 3, seed: int = 0, cities: int = 2):\n \"\"\"``size`` packages across ``cities`` cities, each with an airport and depot.\"\"\"\n rng = random.Random(seed)\n cities = max(2, cities)\n packages = _objects(\"pkg\", max(1, size))\n city_names = _objects(\"city\", cities)\n airports = [f\"apt{i + 1}\" for i in range(cities)]\n depots = [f\"depot{i + 1}\" for i in range(cities)]\n trucks = [f\"truck{i + 1}\" for i in range(cities)]\n\n init = [\"(= (total-cost) 0)\"]\n for i in range(cities):\n init.append(f\"(in-city {airports[i]} {city_names[i]})\")\n init.append(f\"(in-city {depots[i]} {city_names[i]})\")\n init.append(f\"(at {trucks[i]} {depots[i]})\")\n init.append(f\"(at plane1 {airports[0]})\")\n\n goals = []\n for package in packages:\n source = rng.randrange(cities)\n target = rng.randrange(cities)\n while target == source and cities > 1:\n target = rng.randrange(cities)\n init.append(f\"(at {package} {rng.choice([airports[source], depots[source]])})\")\n goals.append(f\"(at {package} {rng.choice([airports[target], depots[target]])})\")\n\n problem = f\"\"\"\\\n;; {len(packages)} packages, {cities} cities, seed {seed}\n(define (problem logistics-{len(packages)}-{seed})\n (:domain logistics)\n (:objects\n {' '.join(city_names)} - city\n {' '.join(trucks)} - truck\n plane1 - airplane\n {' '.join(packages)} - package\n {' '.join(airports)} - airport\n {' '.join(depots)} - location)\n (:init\n {chr(10).join(' ' + f for f in init).strip()})\n (:goal (and\n {chr(10).join(' ' + g for g in goals).strip()}))\n (:metric minimize (total-cost)))\n\"\"\"\n return LOGISTICS_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# rovers (ADL: disjunction + quantification)\n# --------------------------------------------------------------------------\nROVERS_DOMAIN = \"\"\"\\\n;; Rovers -- exercises disjunctive and quantified preconditions.\n(define (domain rovers)\n (:requirements :strips :typing :adl)\n (:types rover waypoint objective - object)\n (:predicates\n (at ?r - rover ?w - waypoint)\n (can-traverse ?a ?b - waypoint)\n (visible ?o - objective ?w - waypoint)\n (imaged ?o - objective)\n (sampled ?w - waypoint)\n (analysed ?w - waypoint)\n (reported ?o - objective))\n\n (:action navigate\n :parameters (?r - rover ?from ?to - waypoint)\n :precondition (and (at ?r ?from) (can-traverse ?from ?to))\n :effect (and (not (at ?r ?from)) (at ?r ?to)))\n\n (:action sample\n :parameters (?r - rover ?w - waypoint)\n :precondition (and (at ?r ?w) (not (sampled ?w)))\n :effect (sampled ?w))\n\n (:action analyse\n :parameters (?w - waypoint)\n :precondition (sampled ?w)\n :effect (analysed ?w))\n\n ;; An objective can be imaged from any waypoint that sees it.\n (:action image\n :parameters (?r - rover ?o - objective)\n :precondition (exists (?w - waypoint) (and (at ?r ?w) (visible ?o ?w)))\n :effect (imaged ?o))\n\n ;; Reporting accepts either a picture or a full sample analysis.\n (:action report\n :parameters (?o - objective)\n :precondition (or (imaged ?o)\n (exists (?w - waypoint)\n (and (visible ?o ?w) (analysed ?w))))\n :effect (reported ?o)))\n\"\"\"\n\n\ndef generate_rovers(size: int = 3, seed: int = 0, waypoints: int = 5):\n \"\"\"``size`` objectives over a connected waypoint graph. Uses `or` and `exists`.\"\"\"\n rng = random.Random(seed)\n waypoints = max(2, waypoints)\n points = _objects(\"wp\", waypoints)\n objectives = _objects(\"obj\", max(1, size))\n\n init = [\"(at rover1 wp1)\"]\n # A spanning path guarantees the graph is connected, so every instance is\n # solvable; the extra edges just give the planner choices.\n for a, b in zip(points, points[1:]):\n init.append(f\"(can-traverse {a} {b})\")\n init.append(f\"(can-traverse {b} {a})\")\n for _ in range(waypoints):\n a, b = rng.sample(points, 2)\n init.append(f\"(can-traverse {a} {b})\")\n\n for objective in objectives:\n for point in rng.sample(points, rng.randint(1, min(2, len(points)))):\n init.append(f\"(visible {objective} {point})\")\n\n problem = f\"\"\"\\\n;; {len(objectives)} objectives, {waypoints} waypoints, seed {seed}\n(define (problem rovers-{len(objectives)}-{seed})\n (:domain rovers)\n (:objects\n rover1 - rover\n {' '.join(points)} - waypoint\n {' '.join(objectives)} - objective)\n (:init\n {chr(10).join(' ' + f for f in sorted(set(init))).strip()})\n (:goal (and\n {chr(10).join(f' (reported {o})' for o in objectives).strip()})))\n\"\"\"\n return ROVERS_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# numeric transport\n# --------------------------------------------------------------------------\nNUMERIC_DOMAIN = \"\"\"\\\n;; Transport with fuel -- exercises numeric fluents.\n(define (domain numeric-transport)\n (:requirements :strips :typing :numeric-fluents)\n (:types truck location package - object)\n (:predicates\n (at ?t - truck ?l - location)\n (road ?a ?b - location)\n (carrying ?p - package ?t - truck)\n (package-at ?p - package ?l - location))\n (:functions\n (fuel ?t - truck)\n (distance ?a ?b - location))\n\n (:action drive\n :parameters (?t - truck ?from ?to - location)\n :precondition (and (at ?t ?from) (road ?from ?to)\n (>= (fuel ?t) (distance ?from ?to)))\n :effect (and (not (at ?t ?from)) (at ?t ?to)\n (decrease (fuel ?t) (distance ?from ?to))))\n\n (:action refuel\n :parameters (?t - truck)\n :precondition (< (fuel ?t) 40)\n :effect (assign (fuel ?t) 60))\n\n (:action load\n :parameters (?p - package ?t - truck ?l - location)\n :precondition (and (at ?t ?l) (package-at ?p ?l))\n :effect (and (not (package-at ?p ?l)) (carrying ?p ?t)))\n\n (:action unload\n :parameters (?p - package ?t - truck ?l - location)\n :precondition (and (at ?t ?l) (carrying ?p ?t))\n :effect (and (not (carrying ?p ?t)) (package-at ?p ?l))))\n\"\"\"\n\n\ndef generate_numeric_transport(size: int = 2, seed: int = 0, locations: int = 4):\n \"\"\"``size`` packages on a fuel-limited road network.\"\"\"\n rng = random.Random(seed)\n locations = max(2, locations)\n places = _objects(\"loc\", locations)\n packages = _objects(\"pkg\", max(1, size))\n\n init = [\"(at truck1 loc1)\", \"(= (fuel truck1) 50)\"]\n for a, b in zip(places, places[1:]):\n distance = rng.choice([10, 15, 20])\n init.append(f\"(road {a} {b})\")\n init.append(f\"(road {b} {a})\")\n init.append(f\"(= (distance {a} {b}) {distance})\")\n init.append(f\"(= (distance {b} {a}) {distance})\")\n\n goals = []\n for package in packages:\n init.append(f\"(package-at {package} {places[0]})\")\n goals.append(f\"(package-at {package} {places[-1]})\")\n\n problem = f\"\"\"\\\n;; {len(packages)} packages, {locations} locations, seed {seed}\n(define (problem numeric-transport-{len(packages)}-{seed})\n (:domain numeric-transport)\n (:objects\n truck1 - truck\n {' '.join(places)} - location\n {' '.join(packages)} - package)\n (:init\n {chr(10).join(' ' + f for f in init).strip()})\n (:goal (and\n {chr(10).join(' ' + g for g in goals).strip()})))\n\"\"\"\n return NUMERIC_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# temporal workshop\n# --------------------------------------------------------------------------\nTEMPORAL_DOMAIN = \"\"\"\\\n;; Workshop -- exercises durative actions (sequential compilation).\n(define (domain workshop)\n (:requirements :strips :typing :durative-actions)\n (:types part - object)\n (:predicates\n (cut ?p - part)\n (drilled ?p - part)\n (painted ?p - part)\n (finished ?p - part))\n\n (:durative-action cut\n :parameters (?p - part)\n :duration (= ?duration 2)\n :condition (and (at start (not (cut ?p))))\n :effect (and (at end (cut ?p))))\n\n (:durative-action drill\n :parameters (?p - part)\n :duration (= ?duration 3)\n :condition (and (over all (cut ?p)))\n :effect (and (at end (drilled ?p))))\n\n (:durative-action paint\n :parameters (?p - part)\n :duration (= ?duration 5)\n :condition (and (over all (drilled ?p)))\n :effect (and (at end (painted ?p))))\n\n (:durative-action inspect\n :parameters (?p - part)\n :duration (= ?duration 1)\n :condition (and (over all (painted ?p)))\n :effect (and (at end (finished ?p)))))\n\"\"\"\n\n\ndef generate_temporal_workshop(size: int = 2, seed: int = 0):\n \"\"\"``size`` parts through a cut/drill/paint/inspect pipeline, with durations.\"\"\"\n parts = _objects(\"part\", max(1, size))\n problem = f\"\"\"\\\n;; {len(parts)} parts, seed {seed}\n(define (problem workshop-{len(parts)}-{seed})\n (:domain workshop)\n (:objects {' '.join(parts)} - part)\n (:init )\n (:goal (and\n {chr(10).join(f' (finished {p})' for p in parts).strip()})))\n\"\"\"\n return TEMPORAL_DOMAIN, problem\n\n\n# --------------------------------------------------------------------------\n# random STRIPS\n# --------------------------------------------------------------------------\ndef generate_random_strips(\n size: int = 8,\n seed: int = 0,\n actions: int = 10,\n goal_size: int = 3,\n):\n \"\"\"A random STRIPS instance built backwards from a guaranteed solution.\n\n Purely random operators almost always give an unsolvable problem, which\n makes for a useless benchmark. Instead this plants a random *chain*: each\n step in the chain has an action that turns the previous state into the next,\n so a plan of known length exists. The remaining actions are noise the\n planner has to search past.\n \"\"\"\n rng = random.Random(seed)\n size = max(4, size)\n facts = [f\"p{i + 1}\" for i in range(size)]\n chain_length = max(2, min(actions // 2, size))\n\n init_facts = set(rng.sample(facts, max(1, size // 3)))\n state = set(init_facts)\n operators = []\n\n # --- the planted solution ---------------------------------------------\n for step in range(chain_length):\n candidates = [f for f in facts if f not in state]\n if not candidates:\n break\n added = rng.choice(candidates)\n precondition = sorted(rng.sample(sorted(state), min(2, len(state))))\n deleted = []\n if len(state) > 1 and rng.random() < 0.4:\n deleted = [\n (\n rng.choice([f for f in sorted(state) if f not in precondition])\n if len(state) > len(precondition)\n else None\n )\n ]\n deleted = [d for d in deleted if d]\n operators.append((f\"solve{step + 1}\", precondition, [added], deleted))\n state.add(added)\n state.difference_update(deleted)\n\n goal_facts = sorted(rng.sample(sorted(state), min(goal_size, len(state))))\n\n # --- distractors -------------------------------------------------------\n for index in range(max(0, actions - len(operators))):\n precondition = sorted(rng.sample(facts, rng.randint(1, 2)))\n added = sorted(rng.sample(facts, rng.randint(1, 2)))\n deleted = [f for f in rng.sample(facts, 1) if f not in added]\n operators.append((f\"noise{index + 1}\", precondition, added, deleted))\n\n def action_text(name, pre, add, delete):\n conditions = \" \".join(f\"({p})\" for p in pre) or \"\"\n effects = \" \".join(f\"({a})\" for a in add)\n effects += \" \" + \" \".join(f\"(not ({d}))\" for d in delete)\n return (\n f\" (:action {name}\\n\"\n f\" :precondition (and {conditions})\\n\"\n f\" :effect (and {effects.strip()}))\"\n )\n\n domain = \"\\n\".join(\n [\n \";; Random STRIPS -- generated by jupyddl.generator\",\n f\"(define (domain random-strips-{size}-{seed})\",\n \" (:requirements :strips)\",\n \" (:predicates \" + \" \".join(f\"({f})\" for f in facts) + \")\",\n \"\",\n \"\\n\\n\".join(action_text(*op) for op in operators),\n \")\",\n ]\n )\n problem = f\"\"\"\\\n;; {size} facts, {len(operators)} actions, seed {seed}\n(define (problem random-strips-{size}-{seed})\n (:domain random-strips-{size}-{seed})\n (:init {' '.join(f'({f})' for f in sorted(init_facts))})\n (:goal (and {' '.join(f'({f})' for f in goal_facts)})))\n\"\"\"\n return domain, problem\n\n\n# --------------------------------------------------------------------------\n# registry\n# --------------------------------------------------------------------------\nGENERATORS = {\n \"blocksworld\": generate_blocksworld,\n \"gripper\": generate_gripper,\n \"logistics\": generate_logistics,\n \"rovers\": generate_rovers,\n \"numeric-transport\": generate_numeric_transport,\n \"workshop\": generate_temporal_workshop,\n \"random-strips\": generate_random_strips,\n}\n\n# What each generator is meant to stress, for the CLI and the playground.\nGENERATOR_NOTES = {\n \"blocksworld\": \"Classic block stacking. Deep plans, heavy plateaus.\",\n \"gripper\": \"Balls between two rooms. High branching factor.\",\n \"logistics\": \"Trucks and a plane with action costs.\",\n \"rovers\": \"ADL: disjunctive and existential preconditions.\",\n \"numeric-transport\": \"Numeric fluents: fuel consumption and refuelling.\",\n \"workshop\": \"Durative actions; plans report a makespan.\",\n \"random-strips\": \"Random operators around a planted solution chain.\",\n}\n\n\ndef generate(kind: str, size: int = 4, seed: int = 0, **kwargs):\n \"\"\"Generate ``(domain_text, problem_text)`` for a named generator.\"\"\"\n try:\n factory = GENERATORS[kind]\n except KeyError:\n raise ValueError(\n f\"Unknown generator '{kind}'. Available: {sorted(GENERATORS)}\"\n ) from None\n return factory(size=size, seed=seed, **kwargs)\n\n\ndef describe_generators() -> list:\n \"\"\"Serialisable metadata for the CLI and the web workbench.\"\"\"\n return [\n {\"name\": name, \"summary\": GENERATOR_NOTES.get(name, \"\")}\n for name in sorted(GENERATORS)\n ]\n\n\ndef write_instance(\n kind: str,\n folder: str,\n size: int = 4,\n seed: int = 0,\n name: Optional[str] = None,\n **kwargs,\n) -> str:\n \"\"\"Generate an instance and write ``//{domain,problem}.pddl``.\"\"\"\n import os\n\n domain, problem = generate(kind, size=size, seed=seed, **kwargs)\n target = os.path.join(folder, name or f\"{kind}-{size:02d}-{seed}\")\n os.makedirs(target, exist_ok=True)\n with open(os.path.join(target, \"domain.pddl\"), \"w\", encoding=\"utf-8\") as handle:\n handle.write(domain)\n with open(os.path.join(target, \"problem.pddl\"), \"w\", encoding=\"utf-8\") as handle:\n handle.write(problem)\n return target\n", "jupyddl/grounding.py": "\"\"\"Grounding: turn a parsed :class:`Domain` + :class:`Problem` into a\ngrounded :class:`~jupyddl.task.Task`.\n\nPipeline:\n\n1. Build the ``type -> objects`` table (with type-hierarchy closure).\n2. Instantiate every action over all type-consistent parameter tuples. The\n precondition formula is expanded (quantifiers over the object pool) and\n distributed into DNF; **each disjunct becomes its own grounded operator**, so\n the search only ever sees conjunctive preconditions.\n3. Detect static predicates (never added, deleted, or derived) and use the\n initial state to prune infeasible action instances and simplify conditions.\n4. Compile negative preconditions/goals into *positive normal form* by\n introducing complement facts ``(not ...)`` and maintaining them on every\n operator that touches the underlying atom.\n5. Ground derived-predicate rules into :class:`~jupyddl.task.Axiom` objects.\n6. Collect numeric fluents, compile their expressions into closures over the\n state's value vector, and encode everything as integer ids.\n\nA disjunctive goal is compiled to a single artificial goal fact achieved by one\nzero-cost operator per disjunct; those operators are recorded in\n``Task.synthetic`` so they can be hidden when a plan is printed.\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom collections import defaultdict\nfrom dataclasses import dataclass, field\nfrom itertools import product\n\nfrom .parser.ast import (\n AddEffect,\n And,\n Arithmetic,\n Atom,\n Comparison,\n ConjunctiveEffect,\n DelEffect,\n Domain,\n EqualityConstraint,\n Exists,\n Forall,\n ForallEffect,\n FluentRef,\n IncreaseCostEffect,\n Literal,\n Number,\n NumericEffect,\n Or,\n PDDLError,\n Problem,\n Truth,\n UnsupportedFeatureError,\n WhenEffect,\n)\nfrom .compile import CLOCK as CLOCK_FLUENT\nfrom .compile import SYNTHETIC_PREFIX, compile_problem\nfrom .parser.parser import parse_domain_file, parse_problem_file\nfrom .task import Axiom, CondEffect, Operator, Task\n\n# Distributing a deeply disjunctive precondition into DNF can blow up\n# combinatorially. Stop with a clear message rather than exhausting memory.\nMAX_DISJUNCTS = 20000\n\nGOAL_FACT = Atom(\"__goal-reached__\", ())\n\n\ndef _ground_atom(atom: Atom, subst: dict) -> Atom:\n return Atom(atom.predicate, tuple(subst.get(a, a) for a in atom.args))\n\n\ndef _ground_fluent(ref: FluentRef, subst: dict) -> FluentRef:\n return FluentRef(ref.name, tuple(subst.get(a, a) for a in ref.args))\n\n\ndef _ground_expression(expr, subst: dict):\n if isinstance(expr, Number):\n return expr\n if isinstance(expr, FluentRef):\n return _ground_fluent(expr, subst)\n if isinstance(expr, Arithmetic):\n return Arithmetic(\n expr.op,\n _ground_expression(expr.left, subst),\n _ground_expression(expr.right, subst),\n )\n raise PDDLError(f\"unexpected numeric expression node: {expr!r}\")\n\n\ndef _build_type_objects(domain: Domain, problem: Problem):\n \"\"\"Map every type to the set of objects that inhabit it (hierarchy-aware).\"\"\"\n parent = dict(domain.types)\n\n def ancestors(typ: str):\n chain = [typ]\n seen = {typ}\n cur = typ\n while cur in parent and parent[cur] not in (None, \"object\", cur):\n cur = parent[cur]\n if cur in seen:\n break\n seen.add(cur)\n chain.append(cur)\n return chain\n\n type_objects: dict = defaultdict(set)\n declared: set = set()\n for name, typ in list(domain.constants) + list(problem.objects):\n declared.add(name)\n for anc in ancestors(typ):\n type_objects[anc].add(name)\n type_objects[\"object\"].add(name)\n\n # Robustness: some (toy) problems omit the :objects section and only mention\n # constants in :init/:goal. Treat any such undeclared constant as an object\n # of the root type so untyped domains still ground.\n for name in _harvest_constants(problem):\n if name not in declared:\n type_objects[\"object\"].add(name)\n\n for typ in set(list(parent) + list(parent.values())):\n type_objects.setdefault(typ, set())\n return {typ: tuple(sorted(objs)) for typ, objs in type_objects.items()}\n\n\ndef _harvest_constants(problem: Problem) -> set:\n found: set = set()\n for atom in problem.init:\n found.update(arg for arg in atom.args if not arg.startswith(\"?\"))\n for ref in problem.init_numeric:\n found.update(arg for arg in ref.args if not arg.startswith(\"?\"))\n\n def walk(formula):\n if isinstance(formula, Literal):\n found.update(a for a in formula.atom.args if not a.startswith(\"?\"))\n elif isinstance(formula, (And, Or)):\n for part in formula.parts:\n walk(part)\n elif isinstance(formula, (Exists, Forall)):\n walk(formula.body)\n\n walk(problem.goal)\n return found\n\n\n# --------------------------------------------------------------------------\n# condition formulas -> DNF\n# --------------------------------------------------------------------------\n@dataclass\nclass _Disjunct:\n \"\"\"One ground conjunction: positive atoms, negative atoms, comparisons.\"\"\"\n\n pos: set = field(default_factory=set)\n neg: set = field(default_factory=set)\n comparisons: list = field(default_factory=list)\n\n def merged(self, other: \"_Disjunct\") -> \"_Disjunct\":\n return _Disjunct(\n self.pos | other.pos,\n self.neg | other.neg,\n self.comparisons + other.comparisons,\n )\n\n @property\n def contradictory(self) -> bool:\n return bool(self.pos & self.neg)\n\n\nTRUE_DNF = [_Disjunct()]\nFALSE_DNF: list = []\n\n\ndef _dnf(formula, subst: dict, type_objects: dict) -> list:\n \"\"\"Expand quantifiers and distribute ``formula`` into a list of disjuncts.\n\n An empty list means *unsatisfiable*; a list holding one empty disjunct means\n *trivially true*.\n \"\"\"\n if formula is None:\n return TRUE_DNF\n\n if isinstance(formula, Truth):\n return TRUE_DNF if formula.value else FALSE_DNF\n\n if isinstance(formula, Literal):\n atom = _ground_atom(formula.atom, subst)\n if formula.positive:\n return [_Disjunct(pos={atom})]\n return [_Disjunct(neg={atom})]\n\n if isinstance(formula, EqualityConstraint):\n left = subst.get(formula.left, formula.left)\n right = subst.get(formula.right, formula.right)\n holds = (left == right) == formula.positive\n return TRUE_DNF if holds else FALSE_DNF\n\n if isinstance(formula, Comparison):\n grounded = Comparison(\n formula.op,\n _ground_expression(formula.left, subst),\n _ground_expression(formula.right, subst),\n )\n return [_Disjunct(comparisons=[grounded])]\n\n if isinstance(formula, And):\n result = TRUE_DNF\n for part in formula.parts:\n result = _cross(result, _dnf(part, subst, type_objects))\n if not result:\n return FALSE_DNF\n return result\n\n if isinstance(formula, Or):\n out: list = []\n for part in formula.parts:\n out.extend(_dnf(part, subst, type_objects))\n if len(out) > MAX_DISJUNCTS:\n raise UnsupportedFeatureError(\n \"disjunctive condition expands past \"\n f\"{MAX_DISJUNCTS} cases; simplify the domain or split the action\"\n )\n return out\n\n if isinstance(formula, Forall):\n result = TRUE_DNF\n for sub in _quantifier_substitutions(formula.params, subst, type_objects):\n result = _cross(result, _dnf(formula.body, sub, type_objects))\n if not result:\n return FALSE_DNF\n return result\n\n if isinstance(formula, Exists):\n out = []\n for sub in _quantifier_substitutions(formula.params, subst, type_objects):\n out.extend(_dnf(formula.body, sub, type_objects))\n if len(out) > MAX_DISJUNCTS:\n raise UnsupportedFeatureError(\n \"existential condition expands past \"\n f\"{MAX_DISJUNCTS} cases; the object pool is too large\"\n )\n return out\n\n raise PDDLError(f\"unexpected condition node: {formula!r}\")\n\n\ndef _quantifier_substitutions(params, subst: dict, type_objects: dict):\n pools = [type_objects.get(typ, ()) for (_, typ) in params]\n for combo in product(*pools):\n extended = dict(subst)\n for (var, _), obj in zip(params, combo):\n extended[var] = obj\n yield extended\n\n\ndef _cross(left: list, right: list) -> list:\n \"\"\"Distribute a conjunction of two DNFs, dropping contradictory disjuncts.\"\"\"\n if not left or not right:\n return FALSE_DNF\n out = []\n for a in left:\n for b in right:\n merged = a.merged(b)\n if not merged.contradictory:\n out.append(merged)\n if len(out) > MAX_DISJUNCTS:\n raise UnsupportedFeatureError(\n f\"conjunction of disjunctions expands past {MAX_DISJUNCTS} cases\"\n )\n return out\n\n\n# --------------------------------------------------------------------------\n# effects\n# --------------------------------------------------------------------------\n@dataclass\nclass _RawOp:\n name: str\n pre_pos: set\n pre_neg: set\n comparisons: list\n add: set\n delete: set\n cond: list # (cpos, cneg, cadd, cdel)\n numeric: list # (op, FluentRef, expression)\n cost: float\n duration: float\n synthetic: bool = False\n\n\ndef _collect_effect(eff, subst, type_objects, cpos, cneg, acc):\n if isinstance(eff, ConjunctiveEffect):\n for part in eff.parts:\n _collect_effect(part, subst, type_objects, cpos, cneg, acc)\n elif isinstance(eff, AddEffect):\n atom = _ground_atom(eff.atom, subst)\n if cpos or cneg:\n acc[\"cond\"].append((frozenset(cpos), frozenset(cneg), {atom}, set()))\n else:\n acc[\"add\"].add(atom)\n elif isinstance(eff, DelEffect):\n atom = _ground_atom(eff.atom, subst)\n if cpos or cneg:\n acc[\"cond\"].append((frozenset(cpos), frozenset(cneg), set(), {atom}))\n else:\n acc[\"delete\"].add(atom)\n elif isinstance(eff, IncreaseCostEffect):\n acc[\"cost\"] += eff.amount\n acc[\"has_cost\"] = True\n elif isinstance(eff, NumericEffect):\n if cpos or cneg:\n raise UnsupportedFeatureError(\n \"numeric effects inside a 'when' are not supported\"\n )\n acc[\"numeric\"].append(\n (\n eff.op,\n _ground_fluent(eff.target, subst),\n _ground_expression(eff.value, subst),\n )\n )\n elif isinstance(eff, ForallEffect):\n for sub in _quantifier_substitutions(eff.params, subst, type_objects):\n _collect_effect(eff.body, sub, type_objects, cpos, cneg, acc)\n elif isinstance(eff, WhenEffect):\n # A disjunctive effect condition splits into one conditional effect per\n # disjunct, which is exactly equivalent.\n for disjunct in _dnf(eff.condition, subst, type_objects):\n if disjunct.comparisons:\n raise UnsupportedFeatureError(\n \"numeric comparisons inside a 'when' condition are not supported\"\n )\n _collect_effect(\n eff.body,\n subst,\n type_objects,\n cpos | disjunct.pos,\n cneg | disjunct.neg,\n acc,\n )\n else:\n raise PDDLError(f\"unexpected effect node: {eff!r}\")\n\n\ndef _effect_predicates(domain: Domain) -> set:\n preds: set = set()\n\n def walk(eff):\n if isinstance(eff, ConjunctiveEffect):\n for part in eff.parts:\n walk(part)\n elif isinstance(eff, (AddEffect, DelEffect)):\n preds.add(eff.atom.predicate)\n elif isinstance(eff, (ForallEffect, WhenEffect)):\n walk(eff.body)\n\n for action in domain.actions:\n walk(action.effect)\n return preds\n\n\ndef _ground_raw_operators(domain, problem, type_objects) -> list:\n raw = []\n for action in domain.actions:\n pools = [type_objects.get(typ, ()) for (_, typ) in action.parameters]\n for combo in product(*pools):\n subst = {var: obj for (var, _), obj in zip(action.parameters, combo)}\n disjuncts = _dnf(action.precondition, subst, type_objects)\n if not disjuncts:\n continue # precondition is unsatisfiable for this instance\n\n acc = {\n \"add\": set(),\n \"delete\": set(),\n \"cond\": [],\n \"numeric\": [],\n \"cost\": 0.0,\n \"has_cost\": False,\n }\n _collect_effect(action.effect, subst, type_objects, set(), set(), acc)\n\n args = \",\".join(combo)\n base = f\"{action.name}({args})\" if combo else action.name\n duration = 0.0\n if action.duration is not None:\n duration = _constant_value(action.duration, subst, action.name)\n if acc[\"has_cost\"]:\n cost = acc[\"cost\"]\n elif duration:\n # A temporal action with no explicit cost: optimise makespan.\n cost = duration\n else:\n cost = 1\n\n for index, disjunct in enumerate(disjuncts):\n # Only tag the name when the split is real, so classical domains\n # keep the operator names their users expect.\n name = base if len(disjuncts) == 1 else f\"{base}#{index + 1}\"\n raw.append(\n _RawOp(\n name=name,\n pre_pos=set(disjunct.pos),\n pre_neg=set(disjunct.neg),\n comparisons=list(disjunct.comparisons),\n add=set(acc[\"add\"]),\n delete=set(acc[\"delete\"]),\n cond=list(acc[\"cond\"]),\n numeric=list(acc[\"numeric\"]),\n cost=cost,\n duration=duration,\n )\n )\n return raw\n\n\ndef _constant_value(expr, subst, action_name):\n \"\"\"Evaluate a duration expression that must be constant.\"\"\"\n grounded = _ground_expression(expr, subst)\n if isinstance(grounded, Number):\n return float(grounded.value)\n raise UnsupportedFeatureError(\n f\"the duration of '{action_name}' must be a constant; \"\n \"durations that read numeric fluents are not supported\"\n )\n\n\n# --------------------------------------------------------------------------\n# numeric compilation\n# --------------------------------------------------------------------------\ndef _collect_fluents(expr, out: set) -> None:\n if isinstance(expr, FluentRef):\n out.add(expr)\n elif isinstance(expr, Arithmetic):\n _collect_fluents(expr.left, out)\n _collect_fluents(expr.right, out)\n\n\ndef _compile_expression(expr, index_of: dict):\n \"\"\"Compile a ground numeric expression into a ``values -> float`` closure.\"\"\"\n if isinstance(expr, Number):\n constant = float(expr.value)\n return lambda values: constant\n if isinstance(expr, FluentRef):\n index = index_of[expr]\n return lambda values: values[index]\n if isinstance(expr, Arithmetic):\n left = _compile_expression(expr.left, index_of)\n right = _compile_expression(expr.right, index_of)\n if expr.op == \"+\":\n return lambda values: left(values) + right(values)\n if expr.op == \"-\":\n return lambda values: left(values) - right(values)\n if expr.op == \"*\":\n return lambda values: left(values) * right(values)\n if expr.op == \"/\":\n\n def divide(values):\n denominator = right(values)\n if denominator == 0:\n # Undefined rather than crashing mid-search: an infinite\n # value makes every comparison against it fail.\n return float(\"inf\")\n return left(values) / denominator\n\n return divide\n raise PDDLError(f\"unknown arithmetic operator '{expr.op}'\")\n raise PDDLError(f\"unexpected numeric expression: {expr!r}\")\n\n\ndef _compile_comparison(comparison: Comparison, index_of: dict):\n left = _compile_expression(comparison.left, index_of)\n right = _compile_expression(comparison.right, index_of)\n op = comparison.op\n if op == \"<\":\n return lambda values: left(values) < right(values)\n if op == \"<=\":\n return lambda values: left(values) <= right(values)\n if op == \">\":\n return lambda values: left(values) > right(values)\n if op == \">=\":\n return lambda values: left(values) >= right(values)\n if op == \"=\":\n return lambda values: left(values) == right(values)\n if op == \"!=\":\n return lambda values: left(values) != right(values)\n raise PDDLError(f\"unknown comparison operator '{op}'\")\n\n\ndef _compile_numeric_effect(op: str, index: int, value, index_of: dict):\n compute = _compile_expression(value, index_of)\n if op == \"assign\":\n return (index, compute)\n if op == \"increase\":\n return (index, lambda values: values[index] + compute(values))\n if op == \"decrease\":\n return (index, lambda values: values[index] - compute(values))\n if op == \"scale-up\":\n return (index, lambda values: values[index] * compute(values))\n if op == \"scale-down\":\n\n def scale_down(values):\n divisor = compute(values)\n return float(\"inf\") if divisor == 0 else values[index] / divisor\n\n return (index, scale_down)\n raise PDDLError(f\"unknown numeric assignment '{op}'\")\n\n\n# --------------------------------------------------------------------------\n# encoding\n# --------------------------------------------------------------------------\n@dataclass\nclass _Encoder:\n fact_ids: dict = field(default_factory=dict)\n comp_ids: dict = field(default_factory=dict)\n names: list = field(default_factory=list)\n\n def fact(self, atom: Atom) -> int:\n if atom not in self.fact_ids:\n self.fact_ids[atom] = len(self.names)\n self.names.append(str(atom))\n return self.fact_ids[atom]\n\n def comp(self, atom: Atom) -> int:\n if atom not in self.comp_ids:\n self.comp_ids[atom] = len(self.names)\n self.names.append(f\"(not {atom})\")\n return self.comp_ids[atom]\n\n\ndef _ground_axioms(domain: Domain, type_objects: dict) -> list:\n \"\"\"Ground every derived-predicate rule into (head atom, disjunct) pairs.\"\"\"\n grounded = []\n for rule in domain.derived:\n for subst in _quantifier_substitutions(rule.params, {}, type_objects):\n head = _ground_atom(rule.head, subst)\n for disjunct in _dnf(rule.body, subst, type_objects):\n if disjunct.comparisons:\n raise UnsupportedFeatureError(\n \"numeric comparisons in a derived predicate are not supported\"\n )\n grounded.append((head, disjunct))\n return grounded\n\n\ndef ground(domain: Domain, problem: Problem) -> Task:\n \"\"\"Ground ``domain`` + ``problem`` into a :class:`Task`.\"\"\"\n # PDDL 3 constructs (preferences, trajectory constraints, timed literals,\n # object fluents) are rewritten into the classical core first, so nothing\n # below this line has to know they exist.\n requirements = tuple(domain.requirements)\n domain, problem = compile_problem(domain, problem)\n type_objects = _build_type_objects(domain, problem)\n init_atoms = set(problem.init)\n\n derived_preds = {rule.head.predicate for rule in domain.derived}\n all_preds = {p.name for p in domain.predicates} | derived_preds\n # A derived predicate never appears in an effect, but it is emphatically not\n # static -- the axioms compute it.\n static_preds = all_preds - _effect_predicates(domain) - derived_preds\n\n def is_static(atom: Atom) -> bool:\n return atom.predicate in static_preds\n\n raw_ops = _ground_raw_operators(domain, problem, type_objects)\n axiom_rules = _ground_axioms(domain, type_objects)\n\n enc = _Encoder()\n tracked_neg: set = set()\n\n def resolve_literals(pos_atoms, neg_atoms):\n \"\"\"Drop static literals that hold, fail on ones that do not.\"\"\"\n pos_fluent, neg_fluent = set(), set()\n for atom in pos_atoms:\n if is_static(atom):\n if atom not in init_atoms:\n return None\n else:\n pos_fluent.add(atom)\n for atom in neg_atoms:\n if is_static(atom):\n if atom in init_atoms:\n return None\n else:\n neg_fluent.add(atom)\n tracked_neg.add(atom)\n return pos_fluent, neg_fluent\n\n # --- resolve static literals, drop infeasible operators ------------------\n resolved = []\n for op in raw_ops:\n pre = resolve_literals(op.pre_pos, op.pre_neg)\n if pre is None:\n continue\n pre_pos_fluent, pre_neg_fluent = pre\n\n add = set(op.add)\n delete = set(op.delete)\n cond = []\n for cpos, cneg, cadd, cdel in op.cond:\n trigger = resolve_literals(cpos, cneg)\n if trigger is None:\n continue # this conditional effect can never fire\n cpos_f, cneg_f = trigger\n if not cpos_f and not cneg_f:\n add |= cadd\n delete |= cdel\n else:\n cond.append((cpos_f, cneg_f, cadd, cdel))\n resolved.append((op, pre_pos_fluent, pre_neg_fluent, add, delete, cond))\n\n # --- goal ----------------------------------------------------------------\n goal_disjuncts = _dnf(problem.goal, {}, type_objects)\n if not goal_disjuncts:\n raise ValueError(\"Goal condition is self-contradictory\")\n\n resolved_goals = []\n unsolvable = True\n for disjunct in goal_disjuncts:\n parts = resolve_literals(disjunct.pos, disjunct.neg)\n if parts is None:\n continue # this way of satisfying the goal is statically impossible\n unsolvable = False\n resolved_goals.append((parts[0], parts[1], disjunct.comparisons))\n if not resolved_goals:\n resolved_goals = [(set(), set(), [])]\n\n # --- axioms --------------------------------------------------------------\n resolved_axioms = []\n for head, disjunct in axiom_rules:\n parts = resolve_literals(disjunct.pos, disjunct.neg)\n if parts is None:\n continue\n resolved_axioms.append((head, parts[0], parts[1]))\n\n # --- numeric fluents -----------------------------------------------------\n fluents: set = set(problem.init_numeric)\n for op, *_ in resolved:\n for comparison in op.comparisons:\n _collect_fluents(comparison.left, fluents)\n _collect_fluents(comparison.right, fluents)\n for _, target, value in op.numeric:\n fluents.add(target)\n _collect_fluents(value, fluents)\n for _, _, comparisons in resolved_goals:\n for comparison in comparisons:\n _collect_fluents(comparison.left, fluents)\n _collect_fluents(comparison.right, fluents)\n # `total-cost` is bookkeeping handled by operator costs, not a state variable.\n fluents = {ref for ref in fluents if ref.name != \"total-cost\"}\n\n ordered_fluents = sorted(fluents, key=str)\n index_of = {ref: i for i, ref in enumerate(ordered_fluents)}\n numeric_names = tuple(str(ref) for ref in ordered_fluents)\n init_values = tuple(\n float(problem.init_numeric.get(ref, 0.0)) for ref in ordered_fluents\n )\n\n # --- encode facts --------------------------------------------------------\n init_ids = {enc.fact(a) for a in init_atoms if not is_static(a)}\n for atom in tracked_neg:\n cid = enc.comp(atom)\n if atom not in init_atoms:\n init_ids.add(cid)\n\n def encode_add_del(add_atoms, del_atoms):\n add_ids = {enc.fact(a) for a in add_atoms}\n del_ids = {enc.fact(a) for a in del_atoms}\n for a in add_atoms:\n if a in tracked_neg:\n del_ids.add(enc.comp(a))\n for a in del_atoms:\n if a in tracked_neg:\n add_ids.add(enc.comp(a))\n return frozenset(add_ids), frozenset(del_ids)\n\n operators = []\n for op, pp, pn, add, delete, cond in resolved:\n precond = {enc.fact(a) for a in pp} | {enc.comp(a) for a in pn}\n add_ids, del_ids = encode_add_del(add, delete)\n cond_effects = []\n for cpos_f, cneg_f, cadd, cdel in cond:\n cond_ids = {enc.fact(a) for a in cpos_f} | {enc.comp(a) for a in cneg_f}\n cadd_ids, cdel_ids = encode_add_del(cadd, cdel)\n if cadd_ids or cdel_ids:\n cond_effects.append(CondEffect(frozenset(cond_ids), cadd_ids, cdel_ids))\n numeric_pre = tuple(_compile_comparison(c, index_of) for c in op.comparisons)\n numeric_eff = tuple(\n _compile_numeric_effect(kind, index_of[target], value, index_of)\n for kind, target, value in op.numeric\n )\n operators.append(\n Operator(\n op.name,\n frozenset(precond),\n add_ids,\n del_ids,\n tuple(cond_effects),\n op.cost,\n numeric_pre,\n numeric_eff,\n op.duration,\n )\n )\n\n # --- goal encoding, compiling a disjunction into a synthetic fact --------\n synthetic: set = set()\n goal_numeric: tuple = ()\n if len(resolved_goals) == 1:\n goal_pos, goal_neg, comparisons = resolved_goals[0]\n goals = {enc.fact(a) for a in goal_pos} | {enc.comp(a) for a in goal_neg}\n goal_numeric = tuple(_compile_comparison(c, index_of) for c in comparisons)\n else:\n goal_fact = enc.fact(GOAL_FACT)\n goals = {goal_fact}\n for index, (goal_pos, goal_neg, comparisons) in enumerate(resolved_goals):\n name = f\"__reach-goal__#{index + 1}\"\n synthetic.add(name)\n precond = {enc.fact(a) for a in goal_pos} | {enc.comp(a) for a in goal_neg}\n operators.append(\n Operator(\n name,\n frozenset(precond),\n frozenset({goal_fact}),\n frozenset(),\n (),\n 0,\n tuple(_compile_comparison(c, index_of) for c in comparisons),\n (),\n 0.0,\n )\n )\n\n if unsolvable:\n sentinel = len(enc.names)\n enc.names.append(\"(unsolvable)\")\n goals.add(sentinel) # never produced by any operator\n\n axioms = tuple(\n Axiom(\n enc.fact(head),\n frozenset(\n {enc.fact(a) for a in body_pos} | {enc.comp(a) for a in body_neg}\n ),\n )\n for head, body_pos, body_neg in resolved_axioms\n )\n\n # Anything a compilation introduced is bookkeeping, not something the\n # domain author wrote, so keep it out of printed plans.\n synthetic |= {\n op.name for op in operators if op.base_name.startswith(SYNTHETIC_PREFIX)\n }\n temporal = any(op.duration for op in operators)\n clock_index = index_of.get(CLOCK_FLUENT)\n metric = None\n if problem.metric is not None:\n direction, expression = problem.metric\n metric = f\"{direction} {expression}\"\n\n return Task(\n name=problem.name or domain.name,\n facts=tuple(enc.names),\n init=frozenset(init_ids),\n goals=frozenset(goals),\n operators=tuple(operators),\n metric_cost=problem.metric_minimize_cost,\n axioms=axioms,\n numeric_names=numeric_names,\n init_values=init_values,\n goal_numeric=goal_numeric,\n temporal=temporal,\n clock_index=clock_index,\n metric=metric,\n requirements=requirements,\n synthetic=frozenset(synthetic),\n )\n\n\ndef ground_files(domain_path: str, problem_path: str) -> Task:\n \"\"\"Convenience: parse both files and ground them into a :class:`Task`.\"\"\"\n return ground(parse_domain_file(domain_path), parse_problem_file(problem_path))\n", "jupyddl/heuristics/__init__.py": "\"\"\"Heuristics and a name-based registry.\"\"\"\n\nfrom __future__ import annotations\n\nfrom functools import partial\n\nfrom .base import Heuristic\nfrom .critical_path import CriticalPathHeuristic\nfrom .delete_relaxation import FFHeuristic, HAddHeuristic, HMaxHeuristic\nfrom .lmcut import LMCutHeuristic\nfrom .simple import BlindHeuristic, GoalCountHeuristic\n\n# name -> callable(task) -> Heuristic\nHEURISTICS = {\n \"blind\": BlindHeuristic,\n \"goalcount\": GoalCountHeuristic,\n \"hmax\": HMaxHeuristic,\n \"hadd\": HAddHeuristic,\n \"hff\": FFHeuristic,\n \"lmcut\": LMCutHeuristic,\n \"h1\": partial(CriticalPathHeuristic, m=1),\n \"h2\": partial(CriticalPathHeuristic, m=2),\n \"hm\": CriticalPathHeuristic,\n}\n\n\n# A heuristic that needs more than a task to build takes an argument after a\n# colon: ``learned:blocksworld.heur.json``. Keeping this a *string* spec rather\n# than a Python object is what makes a trained heuristic usable from the CLI,\n# the benchmark harness and the web UI without any of them knowing it exists.\nLOADERS = {}\n\n\ndef _load_learned(path: str, task) -> Heuristic:\n \"\"\"Resolve ``learned:``, importing the learning stack only if asked.\"\"\"\n from ..learn.heuristic import HeuristicBundle\n\n return HeuristicBundle.load(path).bind(task)\n\n\nLOADERS[\"learned\"] = _load_learned\n\n\ndef make_heuristic(name, task) -> Heuristic:\n \"\"\"Instantiate a heuristic by name (see :data:`HEURISTICS`).\n\n Accepts a registry name (``\"lmcut\"``), a parameterised spec\n (``\"learned:model.json\"``), or an already-built heuristic, which is passed\n through so callers holding a trained model need not round-trip it to disk.\n \"\"\"\n if callable(name) and not isinstance(name, str):\n return name\n if \":\" in name:\n kind, _, argument = name.partition(\":\")\n loader = LOADERS.get(kind)\n if loader is None:\n raise ValueError(\n f\"Unknown parameterised heuristic '{kind}'. \"\n f\"Available: {sorted(LOADERS)}\"\n )\n return loader(argument, task)\n try:\n factory = HEURISTICS[name]\n except KeyError:\n raise ValueError(\n f\"Unknown heuristic '{name}'. Available: {sorted(HEURISTICS)}\"\n + (f\" (or {sorted(LOADERS)} with an argument)\" if LOADERS else \"\")\n ) from None\n return factory(task)\n\n\n__all__ = [\n \"Heuristic\",\n \"BlindHeuristic\",\n \"GoalCountHeuristic\",\n \"HMaxHeuristic\",\n \"HAddHeuristic\",\n \"FFHeuristic\",\n \"LMCutHeuristic\",\n \"CriticalPathHeuristic\",\n \"HEURISTICS\",\n \"LOADERS\",\n \"make_heuristic\",\n]\n", "jupyddl/heuristics/base.py": "\"\"\"Heuristic base class.\"\"\"\n\nfrom __future__ import annotations\n\n\nclass Heuristic:\n \"\"\"A heuristic bound to a task; call it on a state to get an estimate.\n\n Returning ``math.inf`` signals that the state is a (relaxed) dead end.\n \"\"\"\n\n name: str = \"heuristic\"\n admissible: bool = False\n\n def __init__(self, task):\n self.task = task\n\n def __call__(self, state: frozenset) -> float: # pragma: no cover\n raise NotImplementedError\n", "jupyddl/heuristics/critical_path.py": "\"\"\"Critical-path heuristics h^m (Haslum & Geffner, 2000).\n\n``h^m`` estimates the cost of the most expensive size-``m`` subset of atoms.\n``h^1`` equals ``h_max``; higher ``m`` is more informative but costs more.\nAll ``h^m`` are admissible.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport math\nfrom itertools import combinations\n\nfrom .base import Heuristic\n\n\nclass CriticalPathHeuristic(Heuristic):\n name = \"hm\"\n admissible = True\n\n def __init__(self, task, m: int = 2):\n super().__init__(task)\n self.m = m\n self.goal = task.goals\n self.ops = list(task.relaxed_operators()) # (pre, add, cost)\n\n def __call__(self, state) -> float:\n return self._hm(state)\n\n def _hm(self, state) -> float:\n m = self.m\n inf = math.inf\n atoms = set(state) | set(self.goal)\n for pre, add, _ in self.ops:\n atoms |= pre\n atoms |= add\n atoms = sorted(atoms)\n\n # Table of costs for every atom-set of size 1..m.\n table: dict = {}\n sets: list = []\n for size in range(1, m + 1):\n for combo in combinations(atoms, size):\n fs = frozenset(combo)\n table[fs] = 0.0 if fs <= state else inf\n sets.append(fs)\n\n def cost_of(subset: frozenset) -> float:\n # Cost of an arbitrary atom-set: table lookup, or (if larger than m)\n # the max over its size-m subsets.\n if not subset:\n return 0.0\n if len(subset) <= m:\n return table.get(subset, inf)\n best = 0.0\n for combo in combinations(sorted(subset), m):\n val = table.get(frozenset(combo), inf)\n if val > best:\n best = val\n return best\n\n # Value iteration to the fixpoint.\n changed = True\n while changed:\n changed = False\n for target in sets:\n if table[target] == 0.0:\n continue\n best = table[target]\n for pre, add, cost in self.ops:\n if not (add & target):\n continue\n regressed = (target - add) | pre\n val = cost + cost_of(regressed)\n if val < best:\n best = val\n if best < table[target]:\n table[target] = best\n changed = True\n\n return cost_of(frozenset(self.goal))\n", "jupyddl/heuristics/delete_relaxation.py": "\"\"\"Delete-relaxation heuristics: h_max, h_add and the FF heuristic.\"\"\"\n\nfrom __future__ import annotations\n\nimport math\nfrom collections import deque\n\nfrom .base import Heuristic\nfrom .relaxation import RelaxedTask, goal_value, propagate_costs\n\n\nclass HMaxHeuristic(Heuristic):\n \"\"\"h_max: the most expensive relaxed goal fact. Admissible.\"\"\"\n\n name = \"hmax\"\n admissible = True\n\n def __init__(self, task):\n super().__init__(task)\n self.rt = RelaxedTask(task)\n\n def __call__(self, state) -> float:\n cost, _ = propagate_costs(self.rt, state, additive=False)\n return goal_value(cost, self.rt.goal, additive=False)\n\n\nclass HAddHeuristic(Heuristic):\n \"\"\"h_add: sum of relaxed goal-fact costs. Informative, not admissible.\"\"\"\n\n name = \"hadd\"\n\n def __init__(self, task):\n super().__init__(task)\n self.rt = RelaxedTask(task)\n\n def __call__(self, state) -> float:\n cost, _ = propagate_costs(self.rt, state, additive=True)\n return goal_value(cost, self.rt.goal, additive=True)\n\n\nclass FFHeuristic(Heuristic):\n \"\"\"FF heuristic: cost of a relaxed plan extracted from the h_add graph.\"\"\"\n\n name = \"hff\"\n\n def __init__(self, task):\n super().__init__(task)\n self.rt = RelaxedTask(task)\n\n def __call__(self, state) -> float:\n cost, supporter = propagate_costs(self.rt, state, additive=True)\n if any(math.isinf(cost[g]) for g in self.rt.goal):\n return math.inf\n relaxed_plan: set = set()\n seen: set = set()\n queue = deque(self.rt.goal)\n while queue:\n fact = queue.popleft()\n if fact in state or fact in seen:\n continue\n seen.add(fact)\n op_idx = supporter[fact]\n if op_idx < 0:\n return math.inf\n if op_idx not in relaxed_plan:\n relaxed_plan.add(op_idx)\n for pre_fact in self.rt.ops[op_idx].pre:\n if pre_fact not in state:\n queue.append(pre_fact)\n return float(sum(self.rt.ops[i].cost for i in relaxed_plan))\n", "jupyddl/heuristics/lmcut.py": "\"\"\"The LM-cut heuristic (Helmert & Domshlak, 2009).\n\nLM-cut repeatedly computes h_max, finds a *landmark cut* of operators separating\nthe initial state from the goal in the justification graph, adds the cut's\ncheapest operator cost to the estimate, and discounts the cut operators. It is\none of the strongest admissible heuristics for optimal planning.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport heapq\nimport math\nfrom dataclasses import dataclass\n\nfrom .base import Heuristic\n\n\n@dataclass\nclass _AugOp:\n idx: int\n pre: frozenset\n add: frozenset\n base_cost: int\n\n\nclass LMCutHeuristic(Heuristic):\n name = \"lmcut\"\n admissible = True\n\n def __init__(self, task):\n super().__init__(task)\n nf = task.num_facts\n self.INIT = nf # artificial \"always true\" fact\n self.GOAL = nf + 1 # artificial goal fact\n self.num = nf + 2\n\n self.ops: list[_AugOp] = []\n for pre, add, cost in task.relaxed_operators():\n # Every operator must have at least one precondition for the\n # justification graph; use the artificial INIT fact if needed.\n pre = pre if pre else frozenset({self.INIT})\n self.ops.append(_AugOp(len(self.ops), pre, add, cost))\n # Artificial goal operator (cost 0) collecting the real goal.\n self.goal_op = len(self.ops)\n self.ops.append(_AugOp(self.goal_op, task.goals, frozenset({self.GOAL}), 0))\n\n self.consumers: list[list[int]] = [[] for _ in range(self.num)]\n for op in self.ops:\n for f in op.pre:\n self.consumers[f].append(op.idx)\n\n def __call__(self, state) -> float:\n base = set(state)\n base.add(self.INIT)\n source = frozenset(base)\n costs = [op.base_cost for op in self.ops]\n total = 0.0\n\n while True:\n hmax, pcf = self._hmax(source, costs)\n if hmax[self.GOAL] == 0:\n return total\n if math.isinf(hmax[self.GOAL]):\n return math.inf\n\n goal_zone = self._goal_zone(costs, pcf)\n cut = self._cut(source, goal_zone, pcf, hmax)\n if not cut: # safety; should not happen when hmax(goal) > 0\n return math.inf\n min_cost = min(costs[oi] for oi in cut)\n total += min_cost\n for oi in cut:\n costs[oi] -= min_cost\n\n def _hmax(self, source, costs):\n inf = math.inf\n cost = [inf] * self.num\n pcf = [-1] * len(self.ops)\n counter = [len(op.pre) for op in self.ops]\n pq: list = []\n for f in source:\n cost[f] = 0\n heapq.heappush(pq, (0, f))\n\n def relax(op: _AugOp):\n supporter = max(op.pre, key=lambda p: cost[p])\n value = cost[supporter] + costs[op.idx]\n pcf[op.idx] = supporter\n if math.isinf(cost[supporter]):\n return\n for f in op.add:\n if value < cost[f]:\n cost[f] = value\n heapq.heappush(pq, (value, f))\n\n for op in self.ops:\n if counter[op.idx] == 0:\n relax(op)\n while pq:\n c, f = heapq.heappop(pq)\n if c > cost[f]:\n continue\n for op_idx in self.consumers[f]:\n counter[op_idx] -= 1\n if counter[op_idx] == 0:\n relax(self.ops[op_idx])\n return cost, pcf\n\n def _goal_zone(self, costs, pcf):\n \"\"\"Facts that reach the goal through zero-cost justification edges.\"\"\"\n zone = {self.GOAL}\n changed = True\n while changed:\n changed = False\n for op in self.ops:\n supporter = pcf[op.idx]\n if (\n costs[op.idx] == 0\n and supporter >= 0\n and supporter not in zone\n and (op.add & zone)\n ):\n zone.add(supporter)\n changed = True\n return zone\n\n def _cut(self, source, goal_zone, pcf, hmax):\n \"\"\"Operators crossing from the init-reachable region into the goal zone.\"\"\"\n # Forward-reachable facts from the initial state that stay out of the\n # goal zone (the \"before\" region).\n before = {f for f in source if f not in goal_zone}\n changed = True\n while changed:\n changed = False\n for op in self.ops:\n supporter = pcf[op.idx]\n if supporter in before and not math.isinf(hmax[supporter]):\n for f in op.add:\n if f not in goal_zone and f not in before:\n before.add(f)\n changed = True\n cut = [\n op.idx for op in self.ops if pcf[op.idx] in before and (op.add & goal_zone)\n ]\n return cut\n", "jupyddl/heuristics/relaxation.py": "\"\"\"Delete-relaxation machinery shared by h_max, h_add, h_FF and LM-cut.\"\"\"\n\nfrom __future__ import annotations\n\nimport heapq\nimport math\nfrom dataclasses import dataclass\n\n\n@dataclass\nclass RelaxedOp:\n idx: int\n pre: frozenset\n add: frozenset\n cost: int\n\n\nclass RelaxedTask:\n \"\"\"Delete-relaxed view of a task: unary-ish operators + goal.\n\n Conditional effects are already expanded into separate relaxed operators by\n :meth:`jupyddl.task.Task.relaxed_operators`.\n \"\"\"\n\n def __init__(self, task):\n self.num_facts = task.num_facts\n self.goal = task.goals\n self.ops = [\n RelaxedOp(i, pre, add, cost)\n for i, (pre, add, cost) in enumerate(task.relaxed_operators())\n ]\n # Index: fact -> operators that have it as a precondition.\n self.consumers: list[list[int]] = [[] for _ in range(self.num_facts)]\n self.no_pre: list[int] = []\n for op in self.ops:\n if op.pre:\n for f in op.pre:\n self.consumers[f].append(op.idx)\n else:\n self.no_pre.append(op.idx)\n\n\ndef propagate_costs(rt: RelaxedTask, state, additive: bool):\n \"\"\"Generalised Dijkstra computing h_max (``additive=False``) or h_add costs.\n\n Returns ``(cost, supporter)`` where ``cost[f]`` is the estimated cost to\n achieve fact ``f`` and ``supporter[f]`` is the operator index that achieved\n it (used for FF's relaxed-plan extraction).\n \"\"\"\n inf = math.inf\n cost = [inf] * rt.num_facts\n supporter = [-1] * rt.num_facts\n counter = [len(op.pre) for op in rt.ops]\n pq: list = []\n\n for f in state:\n if cost[f] != 0:\n cost[f] = 0\n heapq.heappush(pq, (0, f))\n\n def op_value(op: RelaxedOp) -> float:\n if not op.pre:\n return op.cost\n pre_costs = [cost[p] for p in op.pre]\n agg = sum(pre_costs) if additive else max(pre_costs)\n return op.cost + agg\n\n def apply_op(op: RelaxedOp):\n value = op_value(op)\n if math.isinf(value):\n return\n for f in op.add:\n if value < cost[f]:\n cost[f] = value\n supporter[f] = op.idx\n heapq.heappush(pq, (value, f))\n\n for idx in rt.no_pre:\n apply_op(rt.ops[idx])\n\n while pq:\n c, f = heapq.heappop(pq)\n if c > cost[f]:\n continue\n for op_idx in rt.consumers[f]:\n counter[op_idx] -= 1\n if counter[op_idx] == 0:\n apply_op(rt.ops[op_idx])\n return cost, supporter\n\n\ndef goal_value(cost, goal, additive: bool) -> float:\n if not goal:\n return 0.0\n values = [cost[g] for g in goal]\n if any(math.isinf(v) for v in values):\n return math.inf\n return float(sum(values) if additive else max(values))\n", "jupyddl/heuristics/simple.py": "\"\"\"Cheap non-relaxation heuristics.\"\"\"\n\nfrom __future__ import annotations\n\nfrom ..task import facts_of\nfrom .base import Heuristic\n\n\nclass BlindHeuristic(Heuristic):\n \"\"\"0 in a goal state, otherwise the cheapest operator cost. Admissible.\"\"\"\n\n name = \"blind\"\n admissible = True\n\n def __init__(self, task):\n super().__init__(task)\n costs = [op.cost for op in task.operators if op.cost > 0]\n self.min_cost = min(costs) if costs else 1\n\n def __call__(self, state) -> float:\n return 0.0 if self.task.goal_reached(state) else float(self.min_cost)\n\n\nclass GoalCountHeuristic(Heuristic):\n \"\"\"Number of unsatisfied goal facts. Fast, informative, not admissible.\"\"\"\n\n name = \"goalcount\"\n\n def __call__(self, state) -> float:\n return float(len(self.task.goals - facts_of(state)))\n", "jupyddl/live.py": "\"\"\"A live search dashboard that runs in any terminal, with no dependencies.\n\n:class:`TerminalDashboard` is a :class:`~jupyddl.trace.SearchObserver` that\nrepaints a small block of the terminal while the planner works \u2014 sparklines for\nthe heuristic and the ``f`` frontier, a frontier gauge, live counters and a node\nrate. It is pure standard library (ANSI escapes and Unicode block characters),\nso watching a search never costs you a dependency::\n\n from jupyddl import build_task, solve_task\n from jupyddl.live import TerminalDashboard\n\n task = build_task(\"domain.pddl\", \"problem.pddl\")\n solve_task(task, \"astar\", \"lmcut\", observer=TerminalDashboard())\n\nOn a non-interactive stream it degrades to a periodic one-line progress report,\nso it is also safe to use in CI logs and notebooks.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport math\nimport shutil\nimport sys\nimport time\n\nfrom .trace import SearchObserver\n\nSPARKS = \"\u2581\u2582\u2583\u2584\u2585\u2586\u2587\u2588\"\nGAUGE_FULL = \"\u2588\"\nGAUGE_EMPTY = \"\u2591\"\n\n# 256-colour ANSI approximations of the jupyddl palette slots.\n_ANSI = {\n \"blue\": \"\\x1b[38;5;33m\",\n \"orange\": \"\\x1b[38;5;208m\",\n \"aqua\": \"\\x1b[38;5;36m\",\n \"yellow\": \"\\x1b[38;5;178m\",\n \"green\": \"\\x1b[38;5;34m\",\n \"muted\": \"\\x1b[38;5;245m\",\n \"bold\": \"\\x1b[1m\",\n \"reset\": \"\\x1b[0m\",\n}\n\n__all__ = [\"TerminalDashboard\", \"sparkline\"]\n\n\ndef sparkline(values, width: int = 40) -> str:\n \"\"\"Render ``values`` as a Unicode sparkline of at most ``width`` cells.\"\"\"\n values = [v for v in values if v is not None and not math.isinf(v)]\n if not values:\n return \"\"\n if len(values) > width: # keep the shape, drop the resolution\n bucket = len(values) / width\n values = [values[min(len(values) - 1, int(i * bucket))] for i in range(width)]\n low, high = min(values), max(values)\n if high == low:\n return SPARKS[3] * len(values)\n span = high - low\n return \"\".join(SPARKS[min(7, int((v - low) / span * 7.999))] for v in values)\n\n\ndef _gauge(fraction: float, width: int = 16) -> str:\n fraction = min(1.0, max(0.0, fraction))\n filled = int(round(fraction * width))\n return GAUGE_FULL * filled + GAUGE_EMPTY * (width - filled)\n\n\ndef _fmt(value) -> str:\n if value is None:\n return \"\u2013\"\n value = int(value)\n if value >= 1_000_000:\n return f\"{value / 1_000_000:.1f}M\"\n if value >= 10_000:\n return f\"{value / 1000:.1f}k\"\n return f\"{value:,}\"\n\n\nclass TerminalDashboard(SearchObserver):\n \"\"\"Repaint a live view of the search in the terminal.\n\n ``interval`` throttles repaints (seconds); ``history`` bounds the sparkline\n buffers so a multi-million-node search still costs constant memory.\n \"\"\"\n\n def __init__(\n self,\n stream=None,\n interval: float = 0.08,\n history: int = 400,\n color: bool = True,\n ):\n self.stream = stream if stream is not None else sys.stderr\n self.interval = interval\n self.history = history\n self.interactive = bool(getattr(self.stream, \"isatty\", lambda: False)())\n self.color = color and self.interactive\n\n self._h: list = []\n self._f: list = []\n self._open: list = []\n self._peak_open = 1\n self._best_h = math.inf\n self._start = time.perf_counter()\n self._last_paint = 0.0\n self._lines = 0\n self._title = \"jupyddl\"\n self._latest = None\n self._bounds = 0\n\n # ------------------------------------------------------------- painting\n def _c(self, key: str, text: str) -> str:\n if not self.color:\n return text\n return f\"{_ANSI[key]}{text}{_ANSI['reset']}\"\n\n def _width(self) -> int:\n try:\n return max(48, min(shutil.get_terminal_size().columns - 2, 100))\n except Exception: # pragma: no cover - very unusual terminals\n return 72\n\n def _clear(self) -> None:\n if self._lines and self.interactive:\n self.stream.write(f\"\\x1b[{self._lines}A\\x1b[0J\")\n\n def _render(self, final: bool = False) -> None:\n width = self._width()\n spark_width = max(16, width - 30)\n event = self._latest\n expanded = event.expanded if event else 0\n generated = event.generated if event else 0\n evaluated = event.evaluated if event else 0\n elapsed = max(1e-9, time.perf_counter() - self._start)\n rate = expanded / elapsed\n\n current_h = self._h[-1] if self._h else None\n current_f = self._f[-1] if self._f else None\n open_now = self._open[-1] if self._open else 0\n best_h = None if math.isinf(self._best_h) else self._best_h\n\n lines = [\n self._c(\"bold\", self._title),\n self._c(\"muted\", \"\u2500\" * width),\n \" h \"\n + self._c(\"aqua\", sparkline(self._h, spark_width))\n + self._c(\"muted\", f\" now {_num(current_h)} best {_num(best_h)}\"),\n \" f \"\n + self._c(\"blue\", sparkline(self._f, spark_width))\n + self._c(\"muted\", f\" now {_num(current_f)}\"),\n \" frontier \"\n + self._c(\"orange\", _gauge(open_now / max(1, self._peak_open), 18))\n + self._c(\"muted\", f\" {_fmt(open_now)} (peak {_fmt(self._peak_open)})\"),\n \" \"\n + self._c(\"bold\", _fmt(expanded))\n + self._c(\"muted\", \" expanded \")\n + self._c(\"bold\", _fmt(generated))\n + self._c(\"muted\", \" generated \")\n + self._c(\"bold\", _fmt(evaluated))\n + self._c(\"muted\", \" evaluated\"),\n self._c(\"muted\", f\" {_fmt(rate)} nodes/s \u00b7 {elapsed:.2f}s\")\n + (self._c(\"muted\", f\" \u00b7 {self._bounds} bounds\") if self._bounds else \"\"),\n ]\n self._clear()\n self.stream.write(\"\\n\".join(lines) + \"\\n\")\n self.stream.flush()\n self._lines = len(lines)\n\n def _maybe_paint(self) -> None:\n now = time.perf_counter()\n if now - self._last_paint < self.interval:\n return\n self._last_paint = now\n if self.interactive:\n self._render()\n else:\n # Non-tty: a single appended line, much less often.\n if now - self._start > 1 and int(now) % 2 == 0:\n event = self._latest\n if event is not None:\n self.stream.write(\n f\" ... {_fmt(event.expanded)} expanded, \"\n f\"{_fmt(event.generated)} generated, \"\n f\"{now - self._start:.1f}s\\n\"\n )\n self.stream.flush()\n\n # -------------------------------------------------------- observer hooks\n def on_start(self, task, planner: str, heuristic: str = \"\") -> None:\n self._start = time.perf_counter()\n label = f\"{planner}/{heuristic}\" if heuristic else planner\n name = getattr(task, \"name\", \"\") or \"task\"\n facts = getattr(task, \"num_facts\", 0)\n operators = len(getattr(task, \"operators\", ()))\n self._title = (\n f\"jupyddl \u00b7 {label} \u00b7 {name} \"\n f\"({facts} facts, {operators} ground actions)\"\n )\n if self.interactive:\n self._render()\n\n def on_expand(\n self,\n state,\n g: float = 0.0,\n h: float = 0.0,\n f: float = 0.0,\n depth: int = 0,\n open_size: int = 0,\n stats=None,\n parent=None,\n action: str = \"\",\n ) -> None:\n self._latest = stats\n if h is not None and not math.isinf(h):\n self._h.append(h)\n self._best_h = min(self._best_h, h)\n self._f.append(f if f else g + (h or 0))\n self._open.append(open_size)\n self._peak_open = max(self._peak_open, open_size)\n for buffer in (self._h, self._f, self._open):\n if len(buffer) > self.history:\n del buffer[: len(buffer) - self.history]\n self._maybe_paint()\n\n def on_bound(self, threshold: float, iteration: int, stats=None) -> None:\n self._bounds += 1\n self._latest = stats or self._latest\n\n def on_finish(self, result) -> None:\n self._latest = getattr(result, \"stats\", None) or self._latest\n if self.interactive:\n self._render(final=True)\n elapsed = time.perf_counter() - self._start\n if result is not None and getattr(result, \"solved\", False):\n verdict = self._c(\n \"green\",\n f\" \u2714 solved \u00b7 cost {result.cost} \u00b7 \"\n f\"{result.plan_length} actions \u00b7 {elapsed:.2f}s\",\n )\n else:\n verdict = self._c(\"orange\", f\" \u2718 no plan found \u00b7 {elapsed:.2f}s\")\n self.stream.write(verdict + \"\\n\")\n self.stream.flush()\n self._lines = 0\n\n\ndef _num(value) -> str:\n if value is None:\n return \"\u2013\"\n if isinstance(value, float):\n if math.isinf(value):\n return \"\u221e\"\n if value.is_integer():\n return str(int(value))\n return f\"{value:.1f}\"\n return str(value)\n", "jupyddl/parser/__init__.py": "\"\"\"PDDL parsing: tokenizer, AST and recursive-descent parser.\"\"\"\n\nfrom .ast import (\n Action,\n AddEffect,\n And,\n Arithmetic,\n Atom,\n Comparison,\n Conjunct,\n ConjunctiveEffect,\n DelEffect,\n DerivedPredicate,\n Domain,\n EqualityConstraint,\n Exists,\n Forall,\n ForallEffect,\n FluentRef,\n Function,\n IncreaseCostEffect,\n Literal,\n Number,\n NumericEffect,\n Or,\n PDDLError,\n Predicate,\n Problem,\n Truth,\n UnsupportedFeatureError,\n WhenEffect,\n)\nfrom .parser import (\n parse,\n parse_condition,\n parse_domain,\n parse_domain_file,\n parse_effect,\n parse_problem,\n parse_problem_file,\n)\nfrom .tokenizer import tokenize\n\n__all__ = [\n \"Action\",\n \"AddEffect\",\n \"And\",\n \"Arithmetic\",\n \"Atom\",\n \"Comparison\",\n \"Conjunct\",\n \"ConjunctiveEffect\",\n \"DelEffect\",\n \"DerivedPredicate\",\n \"Domain\",\n \"EqualityConstraint\",\n \"Exists\",\n \"Forall\",\n \"ForallEffect\",\n \"FluentRef\",\n \"Function\",\n \"IncreaseCostEffect\",\n \"Literal\",\n \"Number\",\n \"NumericEffect\",\n \"Or\",\n \"PDDLError\",\n \"Predicate\",\n \"Problem\",\n \"Truth\",\n \"UnsupportedFeatureError\",\n \"WhenEffect\",\n \"parse\",\n \"parse_condition\",\n \"parse_domain\",\n \"parse_effect\",\n \"parse_problem\",\n \"parse_domain_file\",\n \"parse_problem_file\",\n \"tokenize\",\n]\n", "jupyddl/parser/ast.py": "\"\"\"Structured AST for PDDL.\n\nConditions are stored as a **formula tree in negation normal form**: the parser\npushes every ``not`` down to the atoms and rewrites ``imply``, so the grounder\nonly ever sees negation applied to a literal. Quantifiers stay in the tree\nbecause expanding them needs the object pool, which only exists at grounding\ntime; the grounder then distributes the formula into DNF and emits one operator\nper disjunct.\n\nNumeric fluents and durative actions have their own small node types. See\n:mod:`jupyddl.requirements` for exactly which PDDL requirement flags are\nsupported and how.\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom dataclasses import dataclass, field\n\n\nclass PDDLError(Exception):\n \"\"\"Base class for all parsing / modelling errors.\"\"\"\n\n\nclass UnsupportedFeatureError(PDDLError):\n \"\"\"Raised when a PDDL construct outside the supported subset is used.\"\"\"\n\n\n@dataclass(frozen=True)\nclass Atom:\n \"\"\"A (possibly lifted) predicate application, e.g. ``(on ?x ?y)``.\n\n ``args`` holds terms as raw strings: variables keep their leading ``?``\n while constants/objects are stored verbatim.\n \"\"\"\n\n predicate: str\n args: tuple = ()\n\n def __str__(self) -> str:\n if not self.args:\n return f\"({self.predicate})\"\n return f\"({self.predicate} {' '.join(self.args)})\"\n\n\n@dataclass(frozen=True)\nclass Literal:\n \"\"\"A positive or negative atom used in preconditions and goals.\"\"\"\n\n atom: Atom\n positive: bool = True\n\n\n@dataclass(frozen=True)\nclass EqualityConstraint:\n \"\"\"An ``(= a b)`` (or its negation) constraint over terms.\"\"\"\n\n left: str\n right: str\n positive: bool = True\n\n\n# --- numeric expressions -----------------------------------------------------\n\n\n@dataclass(frozen=True)\nclass Number:\n \"\"\"A numeric literal.\"\"\"\n\n value: float\n\n\n@dataclass(frozen=True)\nclass FluentRef:\n \"\"\"A reference to a numeric fluent, e.g. ``(fuel ?truck)``.\"\"\"\n\n name: str\n args: tuple = ()\n\n def __str__(self) -> str:\n if not self.args:\n return f\"({self.name})\"\n return f\"({self.name} {' '.join(self.args)})\"\n\n\n@dataclass(frozen=True)\nclass Arithmetic:\n \"\"\"A binary arithmetic expression (``+``, ``-``, ``*``, ``/``).\n\n Unary minus is parsed as ``(- 0 x)``.\n \"\"\"\n\n op: str\n left: object\n right: object\n\n\n@dataclass(frozen=True)\nclass Comparison:\n \"\"\"A numeric comparison used in preconditions and goals.\"\"\"\n\n op: str # one of < <= = >= >\n left: object\n right: object\n\n\n# --- condition formulas (negation normal form) -------------------------------\n\n\n@dataclass(frozen=True)\nclass Truth:\n \"\"\"The constant ``true`` \u2014 what an empty ``(and)`` parses to.\"\"\"\n\n value: bool = True\n\n\n@dataclass(frozen=True)\nclass And:\n parts: tuple = ()\n\n\n@dataclass(frozen=True)\nclass Or:\n parts: tuple = ()\n\n\n@dataclass(frozen=True)\nclass Exists:\n params: tuple = () # ((var, type), ...)\n body: object = None\n\n\n@dataclass(frozen=True)\nclass Forall:\n params: tuple = () # ((var, type), ...)\n body: object = None\n\n\n@dataclass\nclass Conjunct:\n \"\"\"One ground DNF disjunct: a conjunction of literals and comparisons.\n\n Produced by the grounder, not by the parser. ``equalities`` are resolved\n during instantiation and never survive into a grounded operator.\n \"\"\"\n\n literals: list = field(default_factory=list)\n equalities: list = field(default_factory=list)\n comparisons: list = field(default_factory=list)\n\n\n# --- effects -----------------------------------------------------------------\n\n\n@dataclass\nclass AddEffect:\n atom: Atom\n\n\n@dataclass\nclass DelEffect:\n atom: Atom\n\n\n@dataclass\nclass IncreaseCostEffect:\n \"\"\"``(increase (total-cost) k)`` \u2014 the classical action-cost shorthand.\"\"\"\n\n amount: float\n\n\n@dataclass\nclass NumericEffect:\n \"\"\"An assignment to a numeric fluent.\n\n ``op`` is one of ``assign``, ``increase``, ``decrease``, ``scale-up`` or\n ``scale-down``.\n \"\"\"\n\n op: str\n target: FluentRef\n value: object # an arithmetic expression\n\n\n@dataclass\nclass ConjunctiveEffect:\n parts: list = field(default_factory=list)\n\n\n@dataclass\nclass ForallEffect:\n params: list # [(variable, type), ...]\n body: object\n\n\n@dataclass\nclass WhenEffect:\n condition: object # a condition formula\n body: object\n\n\n# --- domain / problem --------------------------------------------------------\n\n\n@dataclass\nclass Predicate:\n name: str\n params: list # [(variable, type), ...]\n\n\n@dataclass\nclass Function:\n \"\"\"A declared numeric function (``:functions``).\"\"\"\n\n name: str\n params: list = field(default_factory=list)\n\n\n@dataclass\nclass Action:\n name: str\n parameters: list # [(variable, type), ...]\n precondition: object # a condition formula\n effect: object # one of the *Effect nodes above\n # Set when the action came from a (:durative-action ...) block; the value is\n # its duration, and the compilation is documented in jupyddl.requirements.\n duration: object = None\n\n\n@dataclass\nclass DerivedPredicate:\n \"\"\"A ``(:derived (head ?x) body)`` axiom.\"\"\"\n\n head: Atom\n params: list # [(variable, type), ...]\n body: object # a condition formula\n\n\n@dataclass\nclass Domain:\n name: str\n requirements: list\n types: dict # child type -> parent type (\"object\" if none)\n constants: list # [(name, type), ...]\n predicates: list\n actions: list\n functions: list = field(default_factory=list)\n derived: list = field(default_factory=list)\n object_fluents: list = field(default_factory=list)\n constraints: list = field(default_factory=list)\n\n @property\n def has_durative_actions(self) -> bool:\n return any(action.duration is not None for action in self.actions)\n\n\n@dataclass\nclass Problem:\n name: str\n domain_name: str\n objects: list # [(name, type), ...]\n init: list # [Atom, ...]\n goal: object # a condition formula\n metric_minimize_cost: bool = False\n init_numeric: dict = field(default_factory=dict) # FluentRef -> float\n metric: object = None # (direction, expression) or None\n preferences: list = field(default_factory=list) # [Preference, ...]\n constraints: list = field(default_factory=list) # [Constraint | Preference]\n timed_initials: list = field(default_factory=list) # [TimedInitial, ...]\n init_objects: dict = field(default_factory=dict) # object-fluent assignments\n violation_weights: dict = field(default_factory=dict) # preference -> weight\n\n\n# --- PDDL 3: preferences and trajectory constraints --------------------------\n\n\n@dataclass(frozen=True)\nclass Preference:\n \"\"\"A named soft goal or soft constraint.\n\n ``body`` is either a condition formula (a goal preference) or a\n :class:`Constraint` (a soft trajectory constraint). Violating it is legal;\n the ``:metric`` says what it costs.\n \"\"\"\n\n name: str\n body: object\n\n\n@dataclass(frozen=True)\nclass Constraint:\n \"\"\"One state-trajectory constraint from a ``(:constraints ...)`` block.\n\n ``kind`` is the PDDL modal operator (``always``, ``sometime``, ``at-end``,\n ``at-most-once``, ``sometime-before``, ``sometime-after``) and ``args`` holds\n its operand formulas, already in negation normal form.\n \"\"\"\n\n kind: str\n args: tuple = ()\n\n\n@dataclass(frozen=True)\nclass TimedInitial:\n \"\"\"A ``(at